单生产者单消费者字节队列内核 / Single-producer single-consumer byte-queue core More...

#include <spsc_queue_base.hpp>

Inheritance diagram for LibXR::SPSCQueueBase:

Public Types
using	IndexType = size_t
	环形缓冲区索引类型 / Ring-buffer index type.

Public Member Functions
	SPSCQueueBase (size_t element_size, size_t capacity)
	构造 SPSC 字节队列内核 / Construct the SPSC byte-queue core

	SPSCQueueBase (size_t element_size, size_t element_align, size_t capacity)
	构造 SPSC 字节队列内核并指定元素对齐 / Construct the SPSC byte-queue core with explicit element alignment

	~SPSCQueueBase ()
	析构 SPSC 字节队列内核 / Destroy the SPSC 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

ErrorCode	PeekBytes (void *value)
	按字节查看一个队头 payload 但不出队 / Peek one front payload by bytes without dequeuing it

ErrorCode	PushBatchBytes (const void *data, size_t count)
	按字节批量入队多个 payload / Enqueue multiple payloads by bytes

template<typename Writer >
ErrorCode	PushBytesWithWriter (size_t count, Writer &&writer)
	通过写入器回调批量入队 payload / Enqueue payloads through a writer callback

template<typename Writer >
size_t	ProduceWithWriter (size_t limit, Writer &&writer)
	通过一次双 span 回调生产 payload 前缀 / Produce a payload prefix through one two-span writer callback.

ErrorCode	PopBatchBytes (void *data, size_t count)
	按字节批量出队多个 payload / Dequeue multiple payloads by bytes

template<typename Reader >
ErrorCode	PopBytesWithReader (size_t count, Reader &&reader)
	通过读取器回调批量出队 payload / Dequeue payloads through a reader callback

template<typename Reader >
size_t	ConsumeWithReader (size_t limit, Reader &&reader)
	通过一次双 span 回调消费 payload 前缀 / Consume a payload prefix through one two-span reader callback.

ErrorCode	PeekBatchBytes (void *data, size_t count)
	按字节批量查看多个 payload 但不出队 / Peek multiple payloads by bytes without dequeuing them

void	Reset ()
	从 consumer 侧丢弃当前可见的全部 payload / Discard all currently available payloads from the consumer side

size_t	Size () const
	获取当前已用元素数 / Get the current element count

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

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

Detailed Description

单生产者单消费者字节队列内核 / Single-producer single-consumer byte-queue core

这个内核只负责 SPSC ring 的索引推进、容量判断与原始字节存储，不包含多消费者抢占逻辑，也不暴露任何 CAS 路径。上层 SPSCQueue<Data> 只负责把具体类型 Data 映射到这些字节槽位。

This core contains only the SPSC ring progression, capacity checks, and raw byte storage. It does not include any multi-consumer claiming logic or expose any CAS-based path. The upper SPSCQueue<Data> wrapper is responsible only for mapping the concrete Data type onto these byte slots.

Definition at line 27 of file spsc_queue_base.hpp.

Member Typedef Documentation

◆ IndexType

using LibXR::SPSCQueueBase::IndexType = size_t

环形缓冲区索引类型 / Ring-buffer index type.

Definition at line 30 of file spsc_queue_base.hpp.

Constructor & Destructor Documentation

◆ SPSCQueueBase() [1/2]

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

inline

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

Parameters

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

Definition at line 37 of file spsc_queue_base.hpp.

      : element_size_(element_size),
        capacity_(capacity),
        payload_alloc_align_(alignof(std::max_align_t)),
        payload_stride_(ComputeStride(element_size, alignof(std::byte))),
        payloads_(nullptr),
        head_(0),
        tail_(0)
  {
    InitStorage();
  }

◆ SPSCQueueBase() [2/2]

LibXR::SPSCQueueBase::SPSCQueueBase	(	size_t	element_size,
		size_t	element_align,
		size_t	capacity )

inline

构造 SPSC 字节队列内核并指定元素对齐 / Construct the SPSC byte-queue core with explicit element alignment

Parameters

element_size	单个 payload 的字节数 / Byte size of one payload
element_align	单个 payload 的对齐要求 / Alignment requirement of one payload
capacity	队列容量 / Queue capacity

Definition at line 56 of file spsc_queue_base.hpp.

      : element_size_(element_size),
        capacity_(capacity),
        payload_alloc_align_(alignof(std::max_align_t)),
        payload_stride_(ComputeStride(element_size, element_align)),
        payloads_(nullptr),
        head_(0),
        tail_(0)
  {
    InitStorage();
  }

◆ ~SPSCQueueBase()

LibXR::SPSCQueueBase::~SPSCQueueBase ( )

inline

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

Definition at line 88 of file spsc_queue_base.hpp.

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

Member Function Documentation

◆ ConsumeWithReader()

template<typename Reader >

size_t LibXR::SPSCQueueBase::ConsumeWithReader	(	size_t	limit,
		Reader &&	reader )

inline

通过一次双 span 回调消费 payload 前缀 / Consume a payload prefix through one two-span reader callback.

Template Parameters

Reader 读取器类型 / Reader callback type.

Parameters

limit	最多提供的 payload 个数 / Maximum number of offered payloads.
reader	接收两段数据和元素数，返回已接受的前缀个数 / Receives two data pointers and element counts; returns the accepted prefix count.

Returns: 实际从队列移除的 payload 个数 / Number of payloads removed from the queue.

Precondition: 队列只有一个 consumer；回调期间不得重入同侧操作或 Reset / The queue has one consumer; no same-side operation or Reset during the callback.

Note: 可提供数量为零时不调用回调；返回值不得超过可提供数量；所有指针只在回调期间有效 / The callback is skipped when the offered count is zero and must return no more than the offered count; all pointers are valid only during the callback.

Definition at line 438 of file spsc_queue_base.hpp.

  {
    if (limit == 0U)
    {
      return 0U;
    }
 
    const auto current_head = head_.load(std::memory_order_relaxed);
    const auto current_tail = tail_.load(std::memory_order_acquire);
    const size_t capacity = RingCapacity();
    const size_t available = (current_tail >= current_head)
                                 ? (current_tail - current_head)
                                 : (capacity - current_head + current_tail);
    const size_t offered = std::min(limit, available);
    if (offered == 0U)
    {
      return 0U;
    }
 
    const size_t first_count = std::min(offered, capacity - current_head);
    const size_t second_count = offered - first_count;
    const void* const second = second_count == 0U ? nullptr : PayloadPtr(0U);
    const size_t accepted =
        reader(PayloadPtr(current_head), first_count, second, second_count);
    ASSERT(accepted <= offered);
    if (accepted != 0U)
    {
      head_.store((current_head + accepted) % capacity, std::memory_order_release);
    }
    return accepted;
  }

◆ EmptySize()

size_t LibXR::SPSCQueueBase::EmptySize ( ) const

inline

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

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

Definition at line 549 of file spsc_queue_base.hpp.

549{ return capacity_ - Size(); }

LibXR::SPSCQueueBase::Size

size_t Size() const

获取当前已用元素数 / Get the current element count

Definition spsc_queue_base.hpp:536

◆ MaxSize()

size_t LibXR::SPSCQueueBase::MaxSize ( ) const

inline

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

Returns: 队列容量 / Queue capacity

Definition at line 555 of file spsc_queue_base.hpp.

555{ return capacity_; }

◆ PeekBatchBytes()

ErrorCode LibXR::SPSCQueueBase::PeekBatchBytes	(	void *	data,
		size_t	count )

inline

按字节批量查看多个 payload 但不出队 / Peek multiple payloads by bytes without dequeuing them

Parameters

data	用于接收 payload 的字节缓冲区 / Byte buffer receiving payloads
count	payload 个数 / Number of payloads

Returns: 成功返回 ErrorCode::OK；元素不足返回 ErrorCode::EMPTY； count 不为 0 且 data 为空时返回 ErrorCode::PTR_NULL Returns ErrorCode::OK on success; returns ErrorCode::EMPTY when there are not enough payloads available; returns ErrorCode::PTR_NULL when data is null and count is not 0

Definition at line 481 of file spsc_queue_base.hpp.

  {
    if (count == 0U)
    {
      return ErrorCode::OK;
    }
    if (data == nullptr)
    {
      return ErrorCode::PTR_NULL;
    }
 
    const auto current_head = head_.load(std::memory_order_relaxed);
    const auto current_tail = tail_.load(std::memory_order_acquire);
    const size_t capacity = RingCapacity();
    const size_t available = (current_tail >= current_head)
                                 ? (current_tail - current_head)
                                 : (capacity - current_head + current_tail);
 
    if (available < count)
    {
      return ErrorCode::EMPTY;
    }
 
    auto* dst = static_cast<std::byte*>(data);
    for (size_t index = 0; index < count; ++index)
    {
      LibXR::Memory::FastCopy(dst + index * element_size_,
                              PayloadPtr((current_head + index) % capacity),
                              element_size_);
    }
    return ErrorCode::OK;
  }

◆ PeekBytes()

ErrorCode LibXR::SPSCQueueBase::PeekBytes ( void * value )

inline

按字节查看一个队头 payload 但不出队 / Peek one front payload by bytes without dequeuing it

Parameters

value 用于接收 payload 的缓冲区 / Buffer that receives the payload

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

Definition at line 157 of file spsc_queue_base.hpp.

  {
    if (value == nullptr)
    {
      return ErrorCode::PTR_NULL;
    }
 
    const auto current_head = head_.load(std::memory_order_relaxed);
    if (current_head == tail_.load(std::memory_order_acquire))
    {
      return ErrorCode::EMPTY;
    }
 
    LibXR::Memory::FastCopy(value, PayloadPtr(current_head), element_size_);
    return ErrorCode::OK;
  }

◆ PopBatchBytes()

ErrorCode LibXR::SPSCQueueBase::PopBatchBytes	(	void *	data,
		size_t	count )

inline

按字节批量出队多个 payload / Dequeue multiple payloads by bytes

Parameters

data	用于接收 payload 的字节缓冲区；传 `nullptr` 时仅丢弃 / Byte buffer receiving payloads; pass `nullptr` to discard only
count	payload 个数 / Number of payloads

Returns: 成功返回 ErrorCode::OK；元素不足返回 ErrorCode::EMPTY Returns ErrorCode::OK on success; returns ErrorCode::EMPTY when there are not enough payloads available

Definition at line 331 of file spsc_queue_base.hpp.

  {
    if (count == 0U)
    {
      return ErrorCode::OK;
    }
 
    const auto current_head = head_.load(std::memory_order_relaxed);
    const auto current_tail = tail_.load(std::memory_order_acquire);
    const size_t capacity = RingCapacity();
    const size_t available = (current_tail >= current_head)
                                 ? (current_tail - current_head)
                                 : (capacity - current_head + current_tail);
 
    if (available < count)
    {
      return ErrorCode::EMPTY;
    }
 
    auto* dst = static_cast<std::byte*>(data);
    if (dst != nullptr)
    {
      for (size_t index = 0; index < count; ++index)
      {
        LibXR::Memory::FastCopy(dst + index * element_size_,
                                PayloadPtr((current_head + index) % capacity),
                                element_size_);
      }
    }
 
    head_.store((current_head + count) % capacity, std::memory_order_release);
    return ErrorCode::OK;
  }

◆ PopBytes()

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

inline

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

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

Definition at line 130 of file spsc_queue_base.hpp.

  {
    const auto current_head = head_.load(std::memory_order_relaxed);
 
    if (current_head == tail_.load(std::memory_order_acquire))
    {
      return ErrorCode::EMPTY;
    }
 
    if (value != nullptr)
    {
      LibXR::Memory::FastCopy(value, PayloadPtr(current_head), element_size_);
    }
 
    head_.store(Increment(current_head), std::memory_order_release);
    return ErrorCode::OK;
  }

◆ PopBytesWithReader()

template<typename Reader >

ErrorCode LibXR::SPSCQueueBase::PopBytesWithReader	(	size_t	count,
		Reader &&	reader )

inline

通过读取器回调批量出队 payload / Dequeue payloads through a reader callback

Template Parameters

Reader 读取器类型 / Reader callback type

Parameters

count	payload 个数 / Number of payloads
reader	读取器，签名为 `ErrorCode(const void* buffer, size_t chunk_count)` / Reader with signature `ErrorCode(const void* buffer, size_t chunk_count)`

Returns: 成功返回 ErrorCode::OK；元素不足返回 ErrorCode::EMPTY；否则返回读取器错误码 / Returns ErrorCode::OK on success, ErrorCode::EMPTY when there are not enough payloads, otherwise returns the reader error code

Note: 读取器每次收到一段连续 payload 存储区，字节长度为 chunk_count * element_size / The reader receives one contiguous payload storage chunk each time, with byte length chunk_count * element_size

Definition at line 382 of file spsc_queue_base.hpp.

  {
    if (count == 0U)
    {
      return ErrorCode::OK;
    }
 
    const auto current_head = head_.load(std::memory_order_relaxed);
    const auto current_tail = tail_.load(std::memory_order_acquire);
    const size_t capacity = RingCapacity();
    const size_t available = (current_tail >= current_head)
                                 ? (current_tail - current_head)
                                 : (capacity - current_head + current_tail);
 
    if (available < count)
    {
      return ErrorCode::EMPTY;
    }
 
    const size_t first_chunk = std::min(count, capacity - current_head);
    Reader& reader_ref = reader;
    const ErrorCode first_ec = reader_ref(PayloadPtr(current_head), first_chunk);
    if (first_ec != ErrorCode::OK)
    {
      return first_ec;
    }
 
    if (count > first_chunk)
    {
      const ErrorCode second_ec = reader_ref(PayloadPtr(0), count - first_chunk);
      if (second_ec != ErrorCode::OK)
      {
        return second_ec;
      }
    }
 
    head_.store((current_head + count) % capacity, std::memory_order_release);
    return ErrorCode::OK;
  }

◆ ProduceWithWriter()

template<typename Writer >

size_t LibXR::SPSCQueueBase::ProduceWithWriter	(	size_t	limit,
		Writer &&	writer )

inline

通过一次双 span 回调生产 payload 前缀 / Produce a payload prefix through one two-span writer callback.

Template Parameters

Writer 写入器类型 / Writer callback type.

Parameters

limit	最多提供的 payload 个数 / Maximum number of offered payloads.
writer	接收两段存储和元素数，返回已写入的前缀个数 / Receives two storage pointers and element counts; returns the written prefix count.

Returns: 实际发布到队列的 payload 个数 / Number of payloads published to the queue.

Precondition: 队列只有一个 producer；回调期间不得重入同侧操作或 Reset / The queue has one producer; no same-side operation or Reset during the callback.

Note: 可提供数量为零时不调用回调；第二段为空时指针为 nullptr，所有指针只在回调期间有效 / The callback is skipped when the offered count is zero; an empty second span is nullptr, and all pointers are valid only during the callback.

Definition at line 290 of file spsc_queue_base.hpp.

  {
    if (limit == 0U)
    {
      return 0U;
    }
 
    const auto current_tail = tail_.load(std::memory_order_relaxed);
    const auto current_head = head_.load(std::memory_order_acquire);
    const size_t capacity = RingCapacity();
    const size_t free_space = (current_tail >= current_head)
                                  ? (capacity - (current_tail - current_head) - 1U)
                                  : (current_head - current_tail - 1U);
    const size_t offered = std::min(limit, free_space);
    if (offered == 0U)
    {
      return 0U;
    }
 
    const size_t first_count = std::min(offered, capacity - current_tail);
    const size_t second_count = offered - first_count;
    void* const second = second_count == 0U ? nullptr : PayloadPtr(0U);
    const size_t produced =
        writer(PayloadPtr(current_tail), first_count, second, second_count);
    ASSERT(produced <= offered);
    if (produced != 0U)
    {
      tail_.store((current_tail + produced) % capacity, std::memory_order_release);
    }
    return produced;
  }

◆ PushBatchBytes()

ErrorCode LibXR::SPSCQueueBase::PushBatchBytes	(	const void *	data,
		size_t	count )

inline

按字节批量入队多个 payload / Enqueue multiple payloads by bytes

Parameters

data	指向 payload 数组的字节指针 / Byte pointer to the payload array
count	payload 个数 / Number of payloads

Returns: 成功返回 ErrorCode::OK；队列满返回 ErrorCode::FULL； count 不为 0 且 data 为空时返回 ErrorCode::PTR_NULL Returns ErrorCode::OK on success; returns ErrorCode::FULL when the queue is full; returns ErrorCode::PTR_NULL when data is null and count is not 0

Definition at line 184 of file spsc_queue_base.hpp.

  {
    if (count == 0U)
    {
      return ErrorCode::OK;
    }
    if (data == nullptr)
    {
      return ErrorCode::PTR_NULL;
    }
 
    const auto current_tail = tail_.load(std::memory_order_relaxed);
    const auto current_head = head_.load(std::memory_order_acquire);
    const size_t capacity = RingCapacity();
    const size_t free_space = (current_tail >= current_head)
                                  ? (capacity - (current_tail - current_head) - 1)
                                  : (current_head - current_tail - 1);
 
    if (free_space < count)
    {
      return ErrorCode::FULL;
    }
 
    const auto* src = static_cast<const std::byte*>(data);
    for (size_t index = 0; index < count; ++index)
    {
      LibXR::Memory::FastCopy(PayloadPtr((current_tail + index) % capacity),
                              src + index * element_size_, element_size_);
    }
 
    tail_.store((current_tail + count) % capacity, std::memory_order_release);
    return ErrorCode::OK;
  }

◆ PushBytes()

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

inline

按字节入队一个 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; returns ErrorCode::FULL when the queue is full; returns ErrorCode::PTR_NULL when value is null

Definition at line 101 of file spsc_queue_base.hpp.

  {
    if (value == nullptr)
    {
      return ErrorCode::PTR_NULL;
    }
 
    const auto current_tail = tail_.load(std::memory_order_relaxed);
    const auto next_tail = Increment(current_tail);
 
    if (next_tail == head_.load(std::memory_order_acquire))
    {
      return ErrorCode::FULL;
    }
 
    LibXR::Memory::FastCopy(PayloadPtr(current_tail), value, element_size_);
    tail_.store(next_tail, std::memory_order_release);
    return ErrorCode::OK;
  }

◆ PushBytesWithWriter()

template<typename Writer >

ErrorCode LibXR::SPSCQueueBase::PushBytesWithWriter	(	size_t	count,
		Writer &&	writer )

inline

通过写入器回调批量入队 payload / Enqueue payloads through a writer callback

Template Parameters

Writer 写入器类型 / Writer callback type

Parameters

count	payload 个数 / Number of payloads
writer	写入器，签名为 `ErrorCode(void* buffer, size_t chunk_count)` / Writer with signature `ErrorCode(void* buffer, size_t chunk_count)`

Returns: 成功返回 ErrorCode::OK；空间不足返回 ErrorCode::FULL；否则返回写入器错误码 / Returns ErrorCode::OK on success, ErrorCode::FULL when free space is insufficient, otherwise returns the writer error code

Note: 写入器每次收到一段连续 payload 存储区，字节长度为 chunk_count * element_size / The writer receives one contiguous payload storage chunk each time, with byte length chunk_count * element_size

Definition at line 234 of file spsc_queue_base.hpp.

  {
    if (count == 0U)
    {
      return ErrorCode::OK;
    }
 
    const auto current_tail = tail_.load(std::memory_order_relaxed);
    const auto current_head = head_.load(std::memory_order_acquire);
    const size_t capacity = RingCapacity();
    const size_t free_space = (current_tail >= current_head)
                                  ? (capacity - (current_tail - current_head) - 1)
                                  : (current_head - current_tail - 1);
 
    if (free_space < count)
    {
      return ErrorCode::FULL;
    }
 
    const size_t first_chunk = std::min(count, capacity - current_tail);
    Writer& writer_ref = writer;
    const ErrorCode first_ec = writer_ref(PayloadPtr(current_tail), first_chunk);
    if (first_ec != ErrorCode::OK)
    {
      return first_ec;
    }
 
    if (count > first_chunk)
    {
      const ErrorCode second_ec = writer_ref(PayloadPtr(0), count - first_chunk);
      if (second_ec != ErrorCode::OK)
      {
        return second_ec;
      }
    }
 
    tail_.store((current_tail + count) % capacity, std::memory_order_release);
    return ErrorCode::OK;
  }

◆ Reset()

void LibXR::SPSCQueueBase::Reset ( )

inline

从 consumer 侧丢弃当前可见的全部 payload / Discard all currently available payloads from the consumer side

本方法读取 producer 已发布的 tail 快照，并将 consumer head 推进到该位置；不会写入 tail，因此单个 producer 可以继续运行。与该快照并发发布的 payload 可能被丢弃，也可能保留。

This method snapshots the producer-published tail and advances the consumer head to it. It never writes tail, so the single producer may continue. Payloads published concurrently with the snapshot may be discarded or retained.

Definition at line 526 of file spsc_queue_base.hpp.

  {
    const auto current_tail = tail_.load(std::memory_order_acquire);
    head_.store(current_tail, std::memory_order_release);
  }

◆ Size()

size_t LibXR::SPSCQueueBase::Size ( ) const

inline

获取当前已用元素数 / Get the current element count

Returns: 当前元素个数 / Current number of stored payloads

Definition at line 536 of file spsc_queue_base.hpp.

  {
    const auto current_head = head_.load(std::memory_order_acquire);
    const auto current_tail = tail_.load(std::memory_order_acquire);
    return (current_tail >= current_head)
               ? (current_tail - current_head)
               : (RingCapacity() - current_head + current_tail);
  }

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

src/structure/queue/spsc_queue_base.hpp

Public Types

Public Member Functions

Detailed Description

Member Typedef Documentation

◆ IndexType

Constructor & Destructor Documentation

◆ SPSCQueueBase() [1/2]

◆ SPSCQueueBase() [2/2]

◆ ~SPSCQueueBase()

Member Function Documentation

◆ ConsumeWithReader()

◆ EmptySize()

◆ MaxSize()

◆ PeekBatchBytes()

◆ PeekBytes()

◆ PopBatchBytes()

◆ PopBytes()

◆ PopBytesWithReader()

◆ ProduceWithWriter()

◆ PushBatchBytes()

◆ PushBytes()

◆ PushBytesWithWriter()

◆ Reset()

◆ Size()