LibXR::MPMCQueueBase Class Reference

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

#include <mpmc_queue_base.hpp>

Inheritance diagram for LibXR::MPMCQueueBase:

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

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

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 33 of file mpmc_queue_base.hpp.

SequenceType

using LibXR::MPMCQueueBase::SequenceType

Initial value:

 
      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)
{
  ASSERT(element_size_ > 0);
  ASSERT(capacity_ > 1);
  ASSERT(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

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 96 of file mpmc_queue_base.hpp.

{ 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 91 of file mpmc_queue_base.hpp.

{ return capacity_ - Size(); }

MaxSize()

size_t LibXR::MPMCQueueBase::MaxSize ( ) const

inlinenodiscard

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

Returns

队列容量 / Queue capacity

Definition at line 72 of file mpmc_queue_base.hpp.

{ return capacity_; }

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_;
}

---

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