槽池的无类型内核：槽状态机、发布序号与扫描逻辑。 Typeless core of the slot pool: slot state machine, publish sequence and scan logic. More...

#include <slot_pool.hpp>

Inheritance diagram for LibXR::SlotPoolBase:

Public Member Functions
	SlotPoolBase (size_t slot_size, size_t slot_align, size_t slot_count)
	构造槽池内核 / Construct the slot-pool core.

	~SlotPoolBase ()
	析构槽池内核 / Destroy the slot-pool core.

ErrorCode	Acquire (size_t &index)
	认领一个空闲槽（空闲 → 写入中） / Claim one free slot (free -> writing).

void	Publish (size_t index)
	发布已写好的槽（写入中 → 就绪），并取得发布序号 / Publish a fully written slot (writing -> ready) and take a publish sequence number.

ErrorCode	Take (size_t &index)
	认领序号最小的就绪槽（就绪 → 读取中） / Claim the oldest ready slot (ready -> reading).

void	Release (size_t index)
	释放自己持有的槽（写入中或读取中 → 空闲） / Release an owned slot (writing or reading -> free).

size_t	Size () const
	获取槽总数 / Get the total number of slots.

Protected Types
enum	SlotState : uint32_t { SLOT_FREE = 0 , SLOT_WRITING = 1 , SLOT_READY = 2 , SLOT_READING = 3 }
	槽状态 / Slot state. More...

Protected Member Functions
void *	SlotPtr (size_t index)
	获取指定槽位的存储地址 / Get the storage address of one slot.

const void *	SlotPtr (size_t index) const
	获取指定槽位的存储地址（只读）/ Get the storage address of one slot (const).

Protected Attributes
const size_t	slot_size_
	单个槽的字节数 / Byte size of one slot.

const size_t	slot_count_
	槽总数 / Total number of slots.

const size_t	slot_stride_
	相邻槽位之间的步长 / Byte stride between slots.

const size_t	alloc_align_

std::atomic< uint32_t > *	states_
	每槽状态字 / Per-slot atomic state word.

std::atomic< uint32_t > *	seqs_
	每槽发布序号 / Per-slot publish sequence.

std::byte *	slots_
	槽存储缓冲区 / Byte buffer storing slots.

std::atomic< uint32_t >	publish_seq_

std::atomic< uint32_t >	release_count_

Static Protected Attributes
static constexpr size_t	INVALID_INDEX = static_cast<size_t>(-1)
	无效槽位下标 / Sentinel for "no slot chosen".

Detailed Description

槽池的无类型内核：槽状态机、发布序号与扫描逻辑。 Typeless core of the slot pool: slot state machine, publish sequence and scan logic.

内核只管理每个槽的原子状态与发布序号，以及按槽位下标访问的原始存储； payload 类型语义由上层的 SlotPool<T> 负责。所有状态转换都用 CAS 完成，同一时刻一个槽只属于一个执行者。

The core only manages the per-slot atomic state, the publish sequence and the raw per-index storage; payload type semantics live in SlotPool<T> above it. Every state transition is a CAS, so at any moment a slot belongs to exactly one execution context.

Definition at line 35 of file slot_pool.hpp.

Member Enumeration Documentation

◆ SlotState

enum LibXR::SlotPoolBase::SlotState : uint32_t

protected

槽状态 / Slot state.

Enumerator
SLOT_FREE	空闲，可被 `Acquire` 认领 / Free, claimable by `Acquire`
SLOT_WRITING	写入中，已被生产者持有 / Writing, owned by a producer
SLOT_READY	就绪，等待 `Take` / Ready, waiting for `Take`
SLOT_READING	读取中，已被消费者持有 / Reading, owned by a consumer

Definition at line 125 of file slot_pool.hpp.

  {
    SLOT_FREE = 0,     
    SLOT_WRITING = 1,  
    SLOT_READY = 2,    
    SLOT_READING = 3   
  };

Constructor & Destructor Documentation

◆ SlotPoolBase()

LibXR::SlotPoolBase::SlotPoolBase	(	size_t	slot_size,
		size_t	slot_align,
		size_t	slot_count )

构造槽池内核 / Construct the slot-pool core.

Parameters

slot_size	单个槽的字节数 / Byte size of one slot
slot_align	槽存储的对齐要求 / Alignment requirement of the slot storage
slot_count	槽总数 / Total number of slots

Note: 仅构造时分配内存，运行期不再分配 / Allocates only at construction; no allocation at run time.

Definition at line 8 of file slot_pool.cpp.

    : slot_size_(slot_size),
      slot_count_(slot_count),
      slot_stride_(AlignUp(slot_size_, slot_align)),
      alloc_align_(std::max(alignof(std::max_align_t), slot_align)),
      states_(nullptr),
      seqs_(nullptr),
      slots_(nullptr),
      publish_seq_(0),
      release_count_(0)
{
  ASSERT(slot_size_ > 0);
  ASSERT(slot_count_ > 0);
  ASSERT(slot_align > 0);
  ASSERT(slot_count_ <= std::numeric_limits<size_t>::max() / slot_stride_);
  ASSERT(slot_stride_ <= std::numeric_limits<size_t>::max() / slot_align);
 
  states_ = new (std::align_val_t(LibXR::CONCURRENCY_ALIGNMENT))
      std::atomic<uint32_t>[slot_count_];
  seqs_ = new (std::align_val_t(LibXR::CONCURRENCY_ALIGNMENT))
      std::atomic<uint32_t>[slot_count_];
 
  slots_ = static_cast<std::byte*>(
      ::operator new[](slot_stride_ * slot_count_, std::align_val_t(alloc_align_)));
 
  for (size_t index = 0; index < slot_count_; ++index)
  {
    states_[index].store(SLOT_FREE, std::memory_order_relaxed);
    seqs_[index].store(0, std::memory_order_relaxed);
  }
}

◆ ~SlotPoolBase()

LibXR::SlotPoolBase::~SlotPoolBase ( )

析构槽池内核 / Destroy the slot-pool core.

Definition at line 40 of file slot_pool.cpp.

{
  ::operator delete[](slots_, std::align_val_t(alloc_align_));
  // 与构造时显式对齐的 new[] 配对 / Pairs with the explicitly aligned new[] of
  // the constructor.
  ::operator delete[](states_, std::align_val_t(LibXR::CONCURRENCY_ALIGNMENT));
  ::operator delete[](seqs_, std::align_val_t(LibXR::CONCURRENCY_ALIGNMENT));
}

Member Function Documentation

◆ Acquire()

ErrorCode LibXR::SlotPoolBase::Acquire ( size_t & index )

认领一个空闲槽（空闲 → 写入中） / Claim one free slot (free -> writing).

Parameters

index 成功时写入认领的槽位下标 / Receives the claimed slot index on success

Returns: 成功返回 ErrorCode::OK；没有空闲槽返回 ErrorCode::FULL Returns ErrorCode::OK on success; ErrorCode::FULL when no slot is free

扫描全部槽位，跳过被其他上下文占用的槽（写入中、就绪、读取中）。一个被打断的操作只占住它自己的槽，不会挡住其他上下文认领剩余的空闲槽。 The full slot array is scanned and slots owned by other contexts (writing, ready, reading) are skipped. An interrupted operation only occupies its own slot and never blocks other contexts from claiming the remaining free slots.

Definition at line 49 of file slot_pool.cpp.

{
  // 释放计数作为重扫的变化检测：扫描期间有槽被释放时，FULL 才允许重扫一次。
  // The release counter detects changes across the scan: a FULL result may only
  // trigger one rescan when some slot was released meanwhile.
  uint32_t release_begin = release_count_.load(std::memory_order_acquire);
  bool rescanned = false;
 
  for (;;)
  {
    for (size_t i = 0; i < slot_count_; ++i)
    {
      if (states_[i].load(std::memory_order_acquire) != SLOT_FREE)
      {
        continue;
      }
 
      uint32_t expected = SLOT_FREE;
      if (states_[i].compare_exchange_strong(expected, SLOT_WRITING,
                                             std::memory_order_acq_rel,
                                             std::memory_order_acquire))
      {
        index = i;
        return ErrorCode::OK;
      }
      // 该槽刚被其他上下文认领：继续扫剩余槽位。
      // The slot was just claimed by another context: keep scanning the rest.
    }
 
    const uint32_t release_end = release_count_.load(std::memory_order_acquire);
    if (release_end == release_begin || rescanned)
    {
      return ErrorCode::FULL;
    }
    release_begin = release_end;
    rescanned = true;
  }
}

◆ Publish()

void LibXR::SlotPoolBase::Publish ( size_t index )

发布已写好的槽（写入中 → 就绪），并取得发布序号 / Publish a fully written slot (writing -> ready) and take a publish sequence number.

Parameters

index 已通过 Acquire 认领的槽位下标 / Slot index claimed via Acquire

发布序号在 Publish 开始时由全局计数器 fetch_add 取得；Take 在已就绪的槽中按序号先后出队。调用方必须仍持有该槽（写入中），重复发布或发布未认领的槽是调用方错误。 The sequence number comes from a global fetch_add counter when Publish begins; Take dequeues ready slots in sequence order. The caller must still own the slot (writing); double publishing or publishing an unclaimed slot is a caller error.

Definition at line 88 of file slot_pool.cpp.

{
  ASSERT(index < slot_count_);
 
  // 先取序号再置就绪：重扫的发布计数变化先于就绪状态可见。
  // Take the sequence first, then publish: the counter change that drives the
  // rescan becomes visible before the ready state itself.
  const uint32_t seq = publish_seq_.fetch_add(1, std::memory_order_release);
  seqs_[index].store(seq, std::memory_order_relaxed);
 
  uint32_t expected = SLOT_WRITING;
  const bool ok = states_[index].compare_exchange_strong(
      expected, SLOT_READY, std::memory_order_release, std::memory_order_acquire);
  ASSERT(ok);
}

◆ Release()

void LibXR::SlotPoolBase::Release ( size_t index )

释放自己持有的槽（写入中或读取中 → 空闲） / Release an owned slot (writing or reading -> free).

Parameters

index 自己持有的槽位下标 / Slot index owned by the caller

释放不会失败；放弃一个只 Acquire 未发布的槽同样是合法用法（分配器模式）。释放其他执行者持有的槽是调用方错误。 Releasing cannot fail; abandoning a slot that was only acquired (allocator usage) is also valid. Releasing a slot owned by another context is a caller error.

Definition at line 184 of file slot_pool.cpp.

{
  ASSERT(index < slot_count_);
 
  // 释放计数先于状态置回空闲：已经扫过该槽的上下文能通过计数变化发现它。
  // Bump the release counter before the state returns to free: a context that
  // already scanned past this slot can detect it through the counter change.
  release_count_.fetch_add(1, std::memory_order_release);
 
  uint32_t expected = states_[index].load(std::memory_order_relaxed);
  // 只有持有者（写入中或读取中）才能释放 / Only the holder (writing or reading)
  // may release.
  ASSERT(expected == SLOT_WRITING || expected == SLOT_READING);
  const bool ok = states_[index].compare_exchange_strong(
      expected, SLOT_FREE, std::memory_order_release, std::memory_order_acquire);
  ASSERT(ok);
}

◆ Size()

size_t LibXR::SlotPoolBase::Size ( ) const

inlinenodiscard

获取槽总数 / Get the total number of slots.

Returns: 槽总数 / Total number of slots

Definition at line 113 of file slot_pool.hpp.

113{ return slot_count_; }

◆ SlotPtr() [1/2]

void * LibXR::SlotPoolBase::SlotPtr ( size_t index )

nodiscardprotected

获取指定槽位的存储地址 / Get the storage address of one slot.

Definition at line 202 of file slot_pool.cpp.

{
  ASSERT(index < slot_count_);
  return slots_ + index * slot_stride_;
}

◆ SlotPtr() [2/2]

const void * LibXR::SlotPoolBase::SlotPtr ( size_t index ) const

nodiscardprotected

获取指定槽位的存储地址（只读）/ Get the storage address of one slot (const).

Definition at line 208 of file slot_pool.cpp.

{
  ASSERT(index < slot_count_);
  return slots_ + index * slot_stride_;
}

◆ Take()

ErrorCode LibXR::SlotPoolBase::Take ( size_t & index )

认领序号最小的就绪槽（就绪 → 读取中） / Claim the oldest ready slot (ready -> reading).

Parameters

index 成功时写入认领的槽位下标 / Receives the claimed slot index on success

Returns: 成功返回 ErrorCode::OK；没有就绪槽返回 ErrorCode::EMPTY Returns ErrorCode::OK on success; ErrorCode::EMPTY when no slot is ready

槽之间按发布序号挑选（序号比较允许回绕）：同一发布者内部严格保序；一个还没发布完的槽不会挡住其他就绪槽，因此不同发布者之间不保证先进先出。 Slots are selected by publish sequence (wraparound-safe comparison): each publisher keeps its own order, and a slot that is not published yet never hides other ready slots, so there is no FIFO guarantee across publishers.

Definition at line 104 of file slot_pool.cpp.

{
  bool rescanned = false;
 
  for (;;)
  {
    // 本轮开始时的发布计数：本轮开始前就绪的条目，序号都早于它。
    // Publish count at the start of this pass: every item that was ready before
    // the pass began has a sequence before it.
    const uint32_t pass_begin = publish_seq_.load(std::memory_order_acquire);
    size_t best = INVALID_INDEX;
    uint32_t best_seq = 0;
 
    for (size_t i = 0; i < slot_count_; ++i)
    {
      if (states_[i].load(std::memory_order_acquire) != SLOT_READY)
      {
        continue;
      }
 
      const uint32_t seq = seqs_[i].load(std::memory_order_relaxed);
      if (best == INVALID_INDEX || SeqBefore(seq, best_seq))
      {
        best = i;
        best_seq = seq;
      }
    }
 
    if (best != INVALID_INDEX)
    {
      // 选中的条目在本轮开始后才取得序号时，同一发布者更早的条目可能在本轮
      // 扫过它的槽之后才就绪而被漏掉：重新扫描，直到选中的条目早于本轮开始。
      // 单核上中断里的扫描不会被低优先级上下文的发布插入，不会走到这里。
      // When the chosen item took its sequence after this pass began, an earlier
      // item of the same publisher may have become ready only after the pass had
      // scanned its slot: rescan until the chosen item predates the pass. On a
      // single core, a scan inside an interrupt cannot be interleaved with
      // publications from lower-priority contexts, so it never gets here.
      if (!SeqBefore(best_seq, pass_begin))
      {
        continue;
      }
 
      uint32_t expected = SLOT_READY;
      if (states_[best].compare_exchange_strong(expected, SLOT_READING,
                                                std::memory_order_acq_rel,
                                                std::memory_order_acquire))
      {
        // CAS 成功后复读序号：扫描到 CAS 之间该槽若被取走、释放并重新发布，
        // CAS 赢到的是新条目。撤销本次认领（读取中 → 就绪），把新条目留在
        // 池里，重新扫描，保证选中的仍是最早发布项。
        // Re-read the sequence after a winning CAS: if the slot was taken,
        // released and re-published between the scan and the CAS, the CAS won
        // the NEW item. Undo this claim (reading -> ready) to leave the new
        // item in the pool, then rescan so the oldest item still wins.
        if (seqs_[best].load(std::memory_order_relaxed) == best_seq)
        {
          index = best;
          return ErrorCode::OK;
        }
        uint32_t reading = SLOT_READING;
        const bool undone = states_[best].compare_exchange_strong(
            reading, SLOT_READY, std::memory_order_release, std::memory_order_acquire);
        ASSERT(undone);
        continue;
      }
      // 该槽刚被其他消费者取走：重新扫描。
      // Another consumer just claimed the slot: rescan.
      continue;
    }
 
    const uint32_t publish_end = publish_seq_.load(std::memory_order_acquire);
    if (publish_end == pass_begin || rescanned)
    {
      return ErrorCode::EMPTY;
    }
    rescanned = true;
  }
}

Member Data Documentation

◆ alloc_align_

const size_t LibXR::SlotPoolBase::alloc_align_

protected

槽存储的分配对齐 / Allocation alignment of the slot storage.

Definition at line 136 of file slot_pool.hpp.

◆ INVALID_INDEX

size_t LibXR::SlotPoolBase::INVALID_INDEX = static_cast<size_t>(-1)

staticconstexprprotected

无效槽位下标 / Sentinel for "no slot chosen".

Definition at line 122 of file slot_pool.hpp.

◆ publish_seq_

std::atomic<uint32_t> LibXR::SlotPoolBase::publish_seq_

protected

发布序号计数器，兼作 Take 重扫的发布计数 / Publish counter, also the change detector for the Take rescan.

Definition at line 144 of file slot_pool.hpp.

◆ release_count_

std::atomic<uint32_t> LibXR::SlotPoolBase::release_count_

protected

释放计数，Acquire 重扫的变化检测 / Release counter, the change detector for the Acquire rescan.

Definition at line 147 of file slot_pool.hpp.

◆ seqs_

std::atomic<uint32_t>* LibXR::SlotPoolBase::seqs_

protected

每槽发布序号 / Per-slot publish sequence.

Definition at line 139 of file slot_pool.hpp.

◆ slot_count_

const size_t LibXR::SlotPoolBase::slot_count_

protected

槽总数 / Total number of slots.

Definition at line 134 of file slot_pool.hpp.

◆ slot_size_

const size_t LibXR::SlotPoolBase::slot_size_

protected

单个槽的字节数 / Byte size of one slot.

Definition at line 133 of file slot_pool.hpp.

◆ slot_stride_

const size_t LibXR::SlotPoolBase::slot_stride_

protected

相邻槽位之间的步长 / Byte stride between slots.

Definition at line 135 of file slot_pool.hpp.

◆ slots_

std::byte* LibXR::SlotPoolBase::slots_

protected

槽存储缓冲区 / Byte buffer storing slots.

Definition at line 140 of file slot_pool.hpp.

◆ states_

std::atomic<uint32_t>* LibXR::SlotPoolBase::states_

protected

每槽状态字 / Per-slot atomic state word.

Definition at line 138 of file slot_pool.hpp.

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

src/structure/slot_pool.hpp
src/structure/slot_pool.cpp

Public Member Functions

Protected Types

Protected Member Functions

Protected Attributes

Static Protected Attributes

Detailed Description

Member Enumeration Documentation

◆ SlotState

Constructor & Destructor Documentation

◆ SlotPoolBase()

◆ ~SlotPoolBase()

Member Function Documentation

◆ Acquire()

◆ Publish()

◆ Release()

◆ Size()

◆ SlotPtr() [1/2]

◆ SlotPtr() [2/2]

◆ Take()

Member Data Documentation

◆ alloc_align_

◆ INVALID_INDEX

◆ publish_seq_

◆ release_count_

◆ seqs_

◆ slot_count_

◆ slot_size_

◆ slot_stride_

◆ slots_

◆ states_