有界 MPMC 字节队列内核 / Bounded MPMC byte-queue core More...

#include <mpmc_queue_base.hpp>

Inheritance diagram for LibXR::MPMCQueueBase:

Collaboration diagram for LibXR::MPMCQueueBase:

Data Structures
struct	SequenceCell
	每个逻辑槽对应的序号单元。 Sequence cell for one logical slot. More...

Public Types
using	SequenceType = size_t
	单调递增的逻辑序号类型 / Monotonic logical sequence type.

using	SequenceDiffType
	序号差值判定类型 / Signed type used for sequence-delta checks.

Public Member Functions
	MPMCQueueBase (size_t element_size, size_t capacity)
	构造一个字节队列内核 / Construct one byte-queue core

	~MPMCQueueBase ()
	析构字节队列内核 / Destroy the byte-queue core

ErrorCode	PushBytes (const void *value)
	按字节入队一个 payload / Enqueue one payload by bytes

ErrorCode	PopBytes (void *value=nullptr)
	按字节出队一个 payload；传空指针时只丢弃队头元素 / Dequeue one payload by bytes; pass null to discard the front item only

size_t	MaxSize () const
	获取队列最大容量 / Get the maximum queue capacity

size_t	Size () const
	获取并发快照下的当前元素数 / Get the current approximate element count

size_t	EmptySize () const
	获取剩余空槽数 / Get the current free-slot count

size_t	ElementSize () const
	获取单个 payload 的字节数 / Get the byte size of one payload

Private Member Functions
void *	PayloadPtr (size_t index)
	获取指定槽位 payload 起始地址。 Get the payload base address of one slot.

const void *	PayloadPtr (size_t index) const
	获取指定槽位 payload 起始地址（只读）。 Get the payload base address of one slot (const).

	MPMCQueueBase (const MPMCQueueBase &)
	禁止拷贝构造。 Non-copyable.

MPMCQueueBase &	operator= (const MPMCQueueBase &)
	禁止拷贝赋值。 Non-copy-assignable.

	MPMCQueueBase (MPMCQueueBase &&)
	禁止移动构造。 Non-movable.

MPMCQueueBase &	operator= (MPMCQueueBase &&)
	禁止移动赋值。 Non-move-assignable.

Static Private Member Functions
static size_t	AlignUpChecked (size_t value, size_t align)
	安全地向上对齐字节数。 Safely align one byte count upward.

static size_t	MultiplyChecked (size_t lhs, size_t rhs)
	安全地计算乘积。 Safely multiply two size values.

Private Attributes
const size_t	element_size_
	单个 payload 的字节数。 Byte size of one payload.

const size_t	capacity_
	队列容量。 Queue capacity.

const size_t	payload_stride_
	相邻 payload 槽位之间的步长。 Byte stride between adjacent payload slots.

SequenceCell *	sequences_
	槽序号数组。 Array of per-slot sequence cells.

std::byte *	payloads_
	payload 字节缓冲区。 Byte buffer storing payloads.

std::atomic< SequenceType >	head_
	下一个待出队的逻辑位置。 Next logical dequeue position.

std::atomic< SequenceType >	tail_
	下一个待入队的逻辑位置。 Next logical enqueue position.

Static Private Attributes
static constexpr size_t	PAYLOAD_ALLOC_ALIGN
	payload 缓冲区整体分配对齐 / Allocation alignment used for the whole payload buffer

Detailed Description

有界 MPMC 字节队列内核 / Bounded MPMC byte-queue core

这个内核把并发协议和字节搬运集中在一个非模板实现里，以减少不同 payload 类型各自实例化一整套无锁协议所带来的 flash 膨胀。它只负责搬运固定大小、默认字宽对齐的字节 payload；类型语义由上层薄包装负责。

This core keeps the concurrency protocol and byte-copying logic in one non-template implementation so different payload types do not each instantiate a full copy of the lock-free protocol. It only moves fixed-size, word-aligned byte payloads; type semantics are handled by thin wrappers above it.

Definition at line 28 of file mpmc_queue_base.hpp.

Member Typedef Documentation

◆ SequenceDiffType

using LibXR::MPMCQueueBase::SequenceDiffType

Initial value:

std::make_signed_t<SequenceType>

序号差值判定类型 / Signed type used for sequence-delta checks.

Definition at line 32 of file mpmc_queue_base.hpp.

◆ SequenceType

using LibXR::MPMCQueueBase::SequenceType = size_t

单调递增的逻辑序号类型 / Monotonic logical sequence type.

Definition at line 31 of file mpmc_queue_base.hpp.

Constructor & Destructor Documentation

◆ MPMCQueueBase()

LibXR::MPMCQueueBase::MPMCQueueBase	(	size_t	element_size,
		size_t	capacity )

构造一个字节队列内核 / Construct one byte-queue core

构造字节队列内核 / Construct the byte-queue core

Parameters

element_size	单个 payload 的字节数 / Byte size of one payload
capacity	队列容量 / Queue capacity

Definition at line 15 of file mpmc_queue_base.cpp.

    : element_size_(element_size),
      capacity_(capacity),
      payload_stride_(AlignUpChecked(element_size_, alignof(size_t))),
      sequences_(nullptr),
      payloads_(nullptr),
      head_(0),
      tail_(0)
{
  REQUIRE(element_size_ > 0);
  REQUIRE(capacity_ > 1);
  REQUIRE(capacity_ <= static_cast<size_t>(std::numeric_limits<SequenceDiffType>::max()));
 
  const size_t payload_bytes = MultiplyChecked(payload_stride_, capacity_);
  sequences_ = new (std::align_val_t(alignof(SequenceCell))) SequenceCell[capacity_];
 
  payloads_ = static_cast<std::byte*>(
      ::operator new[](payload_bytes, std::align_val_t(PAYLOAD_ALLOC_ALIGN)));
 
  for (size_t index = 0; index < capacity_; ++index)
  {
    sequences_[index].value.store(static_cast<SequenceType>(index),
                                  std::memory_order_relaxed);
  }
}

◆ ~MPMCQueueBase()

LibXR::MPMCQueueBase::~MPMCQueueBase ( )

析构字节队列内核 / Destroy the byte-queue core

Definition at line 44 of file mpmc_queue_base.cpp.

{
  ::operator delete[](payloads_, std::align_val_t(PAYLOAD_ALLOC_ALIGN));
  delete[] sequences_;
}

Member Function Documentation

◆ AlignUpChecked()

size_t LibXR::MPMCQueueBase::AlignUpChecked	(	size_t	value,
		size_t	align )

staticnodiscardprivate

安全地向上对齐字节数。 Safely align one byte count upward.

向上对齐到指定粒度 / Align one byte count upward to the target granularity

Parameters

value	待对齐字节数 / Byte count to align
align	目标对齐粒度 / Target alignment granularity

Definition at line 171 of file mpmc_queue_base.cpp.

{
  REQUIRE(align > 0);
  REQUIRE(value <= std::numeric_limits<size_t>::max() - (align - 1));
  return ((value + align - 1) / align) * align;
}

◆ ElementSize()

size_t LibXR::MPMCQueueBase::ElementSize ( ) const

inlinenodiscard

获取单个 payload 的字节数 / Get the byte size of one payload

Returns: 单个 payload 的字节数 / Byte size of one payload

Definition at line 94 of file mpmc_queue_base.hpp.

94{ return element_size_; }

◆ EmptySize()

size_t LibXR::MPMCQueueBase::EmptySize ( ) const

inlinenodiscard

获取剩余空槽数 / Get the current free-slot count

Returns: 当前空槽个数 / Current number of free slots

Definition at line 89 of file mpmc_queue_base.hpp.

89{ return capacity_ - Size(); }

LibXR::MPMCQueueBase::Size

size_t Size() const

获取并发快照下的当前元素数 / Get the current approximate element count

Definition mpmc_queue_base.cpp:139

◆ MaxSize()

size_t LibXR::MPMCQueueBase::MaxSize ( ) const

inlinenodiscard

获取队列最大容量 / Get the maximum queue capacity

Returns: 队列容量 / Queue capacity

Definition at line 70 of file mpmc_queue_base.hpp.

70{ return capacity_; }

◆ MultiplyChecked()

size_t LibXR::MPMCQueueBase::MultiplyChecked	(	size_t	lhs,
		size_t	rhs )

staticnodiscardprivate

安全地计算乘积。 Safely multiply two size values.

安全地计算两个字节数的乘积 / Safely multiply two byte counts

Parameters

lhs	左操作数 / Left operand
rhs	右操作数 / Right operand

Returns: 乘积结果 / Product result

Definition at line 184 of file mpmc_queue_base.cpp.

{
  if (lhs == 0 || rhs == 0)
  {
    return 0;
  }
 
  REQUIRE(lhs <= std::numeric_limits<size_t>::max() / rhs);
  return lhs * rhs;
}

◆ PayloadPtr() [1/2]

void * LibXR::MPMCQueueBase::PayloadPtr ( size_t index )

nodiscardprivate

获取指定槽位 payload 起始地址。 Get the payload base address of one slot.

获取指定槽位 payload 起始地址 / Get the payload base address of one slot

Parameters

index 槽位下标 / Slot index

Definition at line 151 of file mpmc_queue_base.cpp.

{
  return payloads_ + index * payload_stride_;
}

◆ PayloadPtr() [2/2]

const void * LibXR::MPMCQueueBase::PayloadPtr ( size_t index ) const

nodiscardprivate

获取指定槽位 payload 起始地址（只读）。 Get the payload base address of one slot (const).

获取指定槽位 payload 起始地址（只读） / Get the payload base address of one slot (const)

Parameters

index 槽位下标 / Slot index

Definition at line 161 of file mpmc_queue_base.cpp.

{
  return payloads_ + index * payload_stride_;
}

◆ PopBytes()

ErrorCode LibXR::MPMCQueueBase::PopBytes ( void * value = nullptr )

按字节出队一个 payload；传空指针时只丢弃队头元素 / Dequeue one payload by bytes; pass null to discard the front item only

按字节出队一个 payload / Dequeue one payload by bytes

Parameters

value 用于接收 payload 的缓冲区；传 nullptr 时仅丢弃 / Buffer that receives the payload; pass nullptr to discard only

Returns: 成功返回 ErrorCode::OK；队列空返回 ErrorCode::EMPTY Returns ErrorCode::OK on success; returns ErrorCode::EMPTY when the queue is empty

Parameters

value 用于接收 payload 的缓冲区；传 nullptr 时仅丢弃队头元素 / Buffer that receives the payload; pass nullptr to discard the front element only

Definition at line 96 of file mpmc_queue_base.cpp.

{
  SequenceType position = head_.load(std::memory_order_relaxed);
 
  while (true)
  {
    SequenceCell& slot = sequences_[position % capacity_];
    const SequenceType sequence = slot.value.load(std::memory_order_acquire);
    const SequenceType expected_ready = position + 1;
    const SequenceDiffType diff =
        static_cast<SequenceDiffType>(sequence - expected_ready);
 
    if (diff == 0)
    {
      if (head_.compare_exchange_weak(position, position + 1, std::memory_order_relaxed,
                                      std::memory_order_relaxed))
      {
        if (value != nullptr)
        {
          LibXR::Memory::FastCopy(value, PayloadPtr(position % capacity_), element_size_);
        }
        slot.value.store(position + static_cast<SequenceType>(capacity_),
                         std::memory_order_release);
        return ErrorCode::OK;
      }
      continue;
    }
 
    if (diff < 0)
    {
      return ErrorCode::EMPTY;
    }
 
    position = head_.load(std::memory_order_relaxed);
  }
}

◆ PushBytes()

ErrorCode LibXR::MPMCQueueBase::PushBytes ( const void * value )

按字节入队一个 payload / Enqueue one payload by bytes

Parameters

value 指向待入队 payload 的指针 / Pointer to the payload to enqueue

Returns: 成功返回 ErrorCode::OK；队列满返回 ErrorCode::FULL；空指针返回 ErrorCode::PTR_NULL Returns ErrorCode::OK on success; ErrorCode::FULL when the queue is full; ErrorCode::PTR_NULL when value is null

Parameters

value 指向待入队 payload 的指针 / Pointer to the payload to enqueue

Definition at line 54 of file mpmc_queue_base.cpp.

{
  if (value == nullptr)
  {
    return ErrorCode::PTR_NULL;
  }
 
  SequenceType position = tail_.load(std::memory_order_relaxed);
 
  while (true)
  {
    SequenceCell& slot = sequences_[position % capacity_];
    const SequenceType sequence = slot.value.load(std::memory_order_acquire);
    const SequenceDiffType diff = static_cast<SequenceDiffType>(sequence - position);
 
    if (diff == 0)
    {
      if (tail_.compare_exchange_weak(position, position + 1, std::memory_order_relaxed,
                                      std::memory_order_relaxed))
      {
        LibXR::Memory::FastCopy(PayloadPtr(position % capacity_), value, element_size_);
        slot.value.store(position + 1, std::memory_order_release);
        return ErrorCode::OK;
      }
      continue;
    }
 
    if (diff < 0)
    {
      return ErrorCode::FULL;
    }
 
    position = tail_.load(std::memory_order_relaxed);
  }
}

◆ Size()

size_t LibXR::MPMCQueueBase::Size ( ) const

nodiscard

获取并发快照下的当前元素数 / Get the current approximate element count

Note

该值是近似快照：

在并发入队/出队时，该值可能已经过期；
这里按 SequenceType 的模差值估算已用槽数，再钳到 [0, MaxSize()]，因此即使极端长寿命下序号回绕后，它仍然只是近似值，而不是精确值。 This value is an approximate snapshot:
it may already be stale while producers/consumers are progressing;
it is computed from modular SequenceType differences and then clamped to [0, MaxSize()], so it remains approximate rather than exact even after very long-lived sequence wraparound.

Returns: 并发快照下的元素数，范围被钳在 [0, MaxSize()] Approximate element count from a concurrent snapshot, clamped to [0, MaxSize()]

Definition at line 139 of file mpmc_queue_base.cpp.

{
  const SequenceType head_snapshot = head_.load(std::memory_order_acquire);
  const SequenceType tail_snapshot = tail_.load(std::memory_order_acquire);
  const SequenceType used = tail_snapshot - head_snapshot;
  return (used <= capacity_) ? used : capacity_;
}

Field Documentation

◆ capacity_

const size_t LibXR::MPMCQueueBase::capacity_

private

队列容量。 Queue capacity.

Definition at line 127 of file mpmc_queue_base.hpp.

◆ element_size_

const size_t LibXR::MPMCQueueBase::element_size_

private

单个 payload 的字节数。 Byte size of one payload.

Definition at line 126 of file mpmc_queue_base.hpp.

◆ head_

std::atomic<SequenceType> LibXR::MPMCQueueBase::head_

private

下一个待出队的逻辑位置。 Next logical dequeue position.

Definition at line 133 of file mpmc_queue_base.hpp.

◆ PAYLOAD_ALLOC_ALIGN

size_t LibXR::MPMCQueueBase::PAYLOAD_ALLOC_ALIGN

staticconstexprprivate

Initial value:

=

std::max(alignof(size_t), alignof(std::max_align_t))

payload 缓冲区整体分配对齐 / Allocation alignment used for the whole payload buffer

Definition at line 114 of file mpmc_queue_base.hpp.

◆ payload_stride_

const size_t LibXR::MPMCQueueBase::payload_stride_

private

相邻 payload 槽位之间的步长。 Byte stride between adjacent payload slots.

Definition at line 128 of file mpmc_queue_base.hpp.

◆ payloads_

std::byte* LibXR::MPMCQueueBase::payloads_

private

payload 字节缓冲区。 Byte buffer storing payloads.

Definition at line 130 of file mpmc_queue_base.hpp.

◆ sequences_

SequenceCell* LibXR::MPMCQueueBase::sequences_

private

槽序号数组。 Array of per-slot sequence cells.

Definition at line 129 of file mpmc_queue_base.hpp.

◆ tail_

std::atomic<SequenceType> LibXR::MPMCQueueBase::tail_

private

下一个待入队的逻辑位置。 Next logical enqueue position.

Definition at line 135 of file mpmc_queue_base.hpp.

The documentation for this class was generated from the following files:

src/structure/queue/mpmc_queue_base.hpp
src/structure/queue/mpmc_queue_base.cpp

Data Structures

Public Types

Public Member Functions

Private Member Functions

Static Private Member Functions

Private Attributes

Static Private Attributes

Detailed Description

Member Typedef Documentation

◆ SequenceDiffType

◆ SequenceType

Constructor & Destructor Documentation

◆ MPMCQueueBase()

◆ ~MPMCQueueBase()

Member Function Documentation

◆ AlignUpChecked()

◆ ElementSize()

◆ EmptySize()

◆ MaxSize()

◆ MultiplyChecked()

◆ PayloadPtr() [1/2]

◆ PayloadPtr() [2/2]

◆ PopBytes()

◆ PushBytes()

◆ Size()

Field Documentation

◆ capacity_

◆ element_size_

◆ head_

◆ PAYLOAD_ALLOC_ALIGN

◆ payload_stride_

◆ payloads_

◆ sequences_

◆ tail_