# pipeline_ringbuffer

**Repository Path**: deaglebear/pipeline_ringbuffer

## Basic Information

- **Project Name**: pipeline_ringbuffer
- **Description**: pipeline_ringbuffer
- **Primary Language**: Unknown
- **License**: Apache-2.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2025-11-09
- **Last Updated**: 2026-09-03

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# Pipeline RingBuffer

[![License: Apache 2.0](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](LICENSE)
[![C++ Standard](https://img.shields.io/badge/C%2B%2B-20-blue.svg)](https://en.cppreference.com/w/cpp/20)

一个高性能、基于原子操作的多阶段流水线环形缓冲区，使用 C++20 实现。适用于数据需要流经多个处理阶段的场景，提供精细的并发控制和伪共享防护。

## 目录

- [概述](#概述)
- [适用场景](#适用场景)
- [核心概念](#核心概念)
- [编译与安装](#编译与安装)
- [快速入门](#快速入门)
- [Step 类型详解](#step-类型详解)
  - [ExclusiveStep（独占步骤）](#exclusivestep独占步骤)
  - [ShareStep（共享步骤）](#sharestep共享步骤)
- [单调时钟与 ABA 防护](#单调时钟与-aba-防护)
- [Consumer API](#consumer-api)
- [进阶用法](#进阶用法)
- [伪共享防护](#伪共享防护)
- [线程安全](#线程安全)
- [API 参考](#api-参考)
- [测试](#测试)

---

## 概述

`PipelineRingBuffer` 实现了一个**跨多个处理阶段共享的固定大小环形缓冲区**。缓冲区中的每个元素携带用户数据（`T`）以及一个内部状态机，用于控制哪个阶段可以获取该元素。

核心特性：

- **多阶段流水线**：数据自动按照 阶段 0 → 阶段 1 → … → 阶段 N → 回到阶段 0 的顺序流转
- **两种步骤类型**：Exclusive（每个元素由一个消费者独占，基于 CAS 抢占）和 Share（多消费者，基于提交计数协同）
- **伪共享防护**：每个缓冲区槽位都有填充，避免缓存行竞争
- **纯头文件**：`#include` 即可使用，无需编译链接
- **无互斥锁热路径**：拓扑封存后，领取和提交只使用原子操作，但不承诺严格的 lock-free 进度保证

## 适用场景

本库适用于以下场景：

- 需要**按阶段顺序处理数据流**（如：解析 → 转换 → 校验 → 输出）
- 每个阶段运行在独立的线程中
- 需要**限制内存使用**（固定缓冲区大小）
- 需要**低延迟、高吞吐**的阶段间数据传递
- 多个消费者需要**并行协作处理**同一阶段的数据（ShareStep）

如果只需要简单的单生产者-单消费者队列，使用 `std::queue` + 互斥锁可能更简单。本库的优势在于**流水线拓扑结构**——多个阶段串联处理。

## 核心概念

```
   阶段 0           阶段 1            阶段 2           阶段 3
 (Exclusive)      (Exclusive)        (Share)        (Exclusive)
      │                │                 │                │
  ┌───▼───┐       ┌───▼───┐       ┌───▼───┐       ┌───▼───┐
  │Consumer│       │Consumer│       │Consumer│       │Consumer│
  │ (输入)  │  ──▶  │ (解析)  │ ──▶  │ (处理)  │ ──▶  │ (输出)  │
  └───────┘       └───────┘       │Consumer│       └───────┘
                                  │Consumer│
                                  └───────┘
                                       ▲
                           多个消费者并行处理同一阶段
```

**RingBuffer（环形缓冲区）** — 共享容器。它持有一个 `Item<T>` 数组和一个 `Step` 对象列表。

**Step（步骤）** — 表示流水线中的一个处理阶段。每个步骤有类型（Exclusive 或 Share）和一组消费者。

**Consumer（消费者）** — 工作线程用于从所在步骤获取元素、处理元素、并将元素提交到下一步骤的句柄。

**Item（元素）** — 缓冲区中的一个槽位。包含用户数据（`T`）和内部状态字段（`clock`、`commit_count`、`index`）。

数据在流水线中的流转过程：

```
[阶段 0 可用] → Consumer 0 获取 → 处理中 → Consumer 0 提交
                                                  ↓
[阶段 1 可用] → Consumer 1 获取 → 处理中 → Consumer 1 提交
                                                  ↓
                      … 更多阶段 …
                                                  ↓
[阶段 0 可用] ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ← ┘
```

## 编译与安装

### 环境要求

- **C++20** 编译器（GCC 11+、Clang 16+）
- **CMake** >= 3.24

### 编译项目

```bash
git clone <仓库地址>
cd pipeline_ringbuffer
cmake -S . -B build
cmake --build build
```

### 在项目中使用（CMake）

```cmake
# 方式一：add_subdirectory
add_subdirectory(path/to/pipeline_ringbuffer)
target_link_libraries(your_target PRIVATE pipeline_ringbuffer)

# 方式二：install + find_package
# （先执行 cmake --install build --prefix /your/prefix）
find_package(pipeline_ringbuffer REQUIRED)
target_link_libraries(your_target PRIVATE PipelineRingBuffer::pipeline_ringbuffer)
```

本库是**纯头文件库**。`pipeline_ringbuffer` 这个 CMake target 会提供正确的头文件路径和 C++20 编译特性。

### 依赖项

核心库**无外部依赖**——仅使用 C++ 标准库。测试代码使用了 [rapidjson](https://github.com/Tencent/rapidjson)、[spdlog](https://github.com/gabime/spdlog) 和 [magic_enum](https://github.com/Neargye/magic_enum) 用于基准测试和日志（均为可选依赖）。

## 快速入门

最简单的流水线：一个输入阶段 + 一个输出阶段。

```cpp
#include "pipeline_ringbuffer/pipeline_ringbuffer.h"
#include <iostream>
#include <thread>

using namespace PipelineRingBuffer;

struct MyData {
    int value;
};

int main() {
    // 创建环形缓冲区：2 个阶段，类型为 Exclusive → Exclusive
    // buffer_size 必须是 2 的幂
    RingBuffer<MyData> ring_buffer(8, {EStepType::Exclusive, EStepType::Exclusive});

    // 为每个阶段创建消费者
    auto producer = ring_buffer.create_consumer(0);  // 阶段 0
    auto consumer = ring_buffer.create_consumer(1);  // 阶段 1
    if (ring_buffer.freeze_topology() != ERingBufferError::None) {
        return 1;                                    // 每个阶段必须至少有一个 Consumer
    }

    // 生产者线程：写入数据
    std::thread producer_thread([&]() {
        for (int i = 0; i < 100; i++) {
            auto item = producer->claim_scoped();      // 阻塞直到有可用元素
            item->data.value = i;
        } // move-only guard 析构时自动提交，不进行堆分配
    });

    // 消费者线程：读取数据
    std::thread consumer_thread([&]() {
        for (int i = 0; i < 100; i++) {
            auto item = consumer->claim_scoped();
            std::cout << "收到: " << item->data.value << std::endl;
        }
    });

    producer_thread.join();
    consumer_thread.join();
    return 0;
}
```

## Step 类型详解

### ExclusiveStep（独占步骤）

**独占**步骤保证每个元素**只有一个消费者**能够获取。多个消费者可以绑定到同一个独占步骤，但它们通过 **CAS 竞争**——每个元素只有一个消费者能抢到。

领取过程：

1. 读取单调递增的 64 位 `m_sequence`，通过 `sequence & cap_mask` 计算槽位索引。
2. 读取槽位的单调 `clock`，若 `clock % (2*step_count) == 2*step_no` 说明该槽位正处于本步可认领态，再用一次 CAS 将 `clock` +1（available → claimed）预占槽位。
3. 再用 CAS 推进 `m_sequence`。成功后返回携带 `{item, clock+1}` 的 `Claim<T>`；失败则把 `clock` 条件回滚（只有时钟仍停在 claimed 时才回退）。
4. 提交时用领取时保存的 clock 做一次 CAS：`clock → clock + 1`（claimed → 下一阶段 available）。槽位被复用后时钟对不上，CAS 自然失败。

`clock` CAS 是槽位所有权的线性化点，`sequence` CAS 用于确认预占针对的仍是当前游标。`clock` 既是可用性状态又是提交凭证，单调不回绕，因此 sequence 从 `UINT64_MAX` 回绕到 0 时也无需额外推导。

该实现保证不会有两个 ExclusiveConsumer 同时获得同一槽位。不过，如果线程在成功预占 `clock` 后暂停，当前阶段最终会停止推进，因此这里的“无互斥锁”不等同于严格的 lock-free。

适用场景：
- 每个阶段只有一个工作线程
- 少量工作线程竞争处理任务

```cpp
// 三个工作线程都从同一个 ExclusiveStep 消费
auto w1 = ring_buffer.create_consumer(1);
auto w2 = ring_buffer.create_consumer(1);
auto w3 = ring_buffer.create_consumer(1);
// 它们通过 CAS 竞争，每个元素只会被一个工作线程获取
```

### ShareStep（共享步骤）

**共享**步骤允许多个消费者**同时处理同一个元素**。每个消费者有自己独立的单调序列，因此它们独立遍历缓冲区。元素只有在**所有消费者都提交后**才会以 release 语义推进到下一阶段。

适用场景：
- 需要对同一个元素进行并行处理（如：多个线程并发解析同一个 JSON 的不同字段）
- 需要**栅栏语义**——等待所有工作线程完成后再进入下一阶段

```cpp
// 三个工作线程都处理相同的元素
auto w1 = ring_buffer.create_consumer(2);  // ShareStep
auto w2 = ring_buffer.create_consumer(2);
auto w3 = ring_buffer.create_consumer(2);
// 三个工作线程都能看到每个元素；只有全部提交后元素才会进入下一阶段
```

**注意事项**：ShareStep 的消费者各自独立跟踪位置。它们必须以大致相同的速度消费元素——**最慢的消费者决定了整体吞吐量**。每个 ShareConsumer 同时最多持有一个元素；提交前再次调用 `try_claim()` 会返回空 `Claim<T>`。

## 单调时钟与 ABA 防护

### 什么是 ABA 问题

无锁编程中，一个线程读取共享值 A 后被暂停，期间其他线程把值改成 B 又改回 A；该线程恢复后用 CAS 比较，发现"还是 A"便误以为状态从未变化。环形缓冲区会反复复用同一槽位，因此必须同时验证槽位状态和游标。

### 槽位状态：一个单调 64 位时钟

每个槽位只有一个单调递增的 64 位 `clock`，同时编码「当前阶段」和「绕环代数」：

```
clock = lap * period + phase
period = 2 * step_count   // 一个完整 lap 需要的时钟步数
phase  = clock % period   // 0 .. 2*step_count - 1，当前阶段状态
lap    = clock / period   // 已完成的 lap 数
```

- `phase == 2 * step_no`：本步可认领（available）；
- `phase` 为奇数：Exclusive 认领后的 claimed 态；
- `lap`：绕环代数。

`clock` 每次状态转移只做 +1（Exclusive 认领/提交）或 +2（Share 最后一位提交者），**从不回退、永不回绕（2^64 足够）**。因此旧游标或旧 token 复活时，它保存的时钟值必然与当前不符，CAS 自然失败——ABA 防护由单调性结构内建，无需额外的轮次字段。

### 与 sequence 的分工

`sequence` 和 `clock` 是两个层级的东西：

- `sequence` 是**步级游标**，回答「下一个该看哪个槽位」，按缓冲区容量绕环（`sequence & cap_mask`）；
- `clock` 是**槽级状态机**，回答「这个槽位现在轮到谁、能否认领/提交」，按阶段数绕环（`clock % period`）。

生产者把槽位的 `clock` 推进到下一步，下一步用 `sequence` 找到它、再用 `clock` 确认「轮到我」，提交时再把 `clock` 推给下下步。

以「2 步 Exclusive → Exclusive，容量 4」为例，槽 0 的 `clock` 随时间变化：

```
初始                 clock=0  （phase 0，step0 available，lap 0）
step0 认领 (0→1)     clock=1  （phase 1，claimed）
step0 提交 (1→2)     clock=2  （phase 2，step1 available）
step1 认领 (2→3)     clock=3  （phase 3，claimed）
step1 提交 (3→4)     clock=4  （phase 0，step0 available，lap 1）
```

`period = 2 × 2 = 4`，所以 lap 每 4 步 +1。生产者游标走到第 5 个元素时 `4 & 3 = 0` 又回到槽 0，但槽 0 的 `clock` 已是 4（lap 1）而非最初的 0——状态看似一样，值不同了，这正是 ABA 防护的依据。

### 状态转移：每次只需一次 CAS

ExclusiveStep 每个元素的生命周期推进两次 `clock`（认领 +1、提交 +1），ShareStep 认领不改 `clock`、由最后一个提交者一次性 +2 跳到下一阶段：

1. `clock CAS: c -> c + 1`，先取得槽位的排他预占权；
2. `sequence CAS: s -> s + 1`，确认预占基于当前游标；
3. 第二次 CAS 失败时，将 `clock` **条件回滚**（只有时钟仍停在 claimed 时才回退，避免误伤）。

提交阶段：Exclusive 用一次 `clock CAS: c -> c + 1`（claimed → 下一阶段 available），Share 的最后一位提交者用一次 `clock CAS: c -> c + 2`。提交凭证就是认领时保存的 clock 值，重复或陈旧的提交会因时钟对不上而失败。release 语义由这次 CAS 承担，把 payload 发布给下一阶段。

## Consumer API

每个消费者提供以下几种获取元素的策略：

| 方法 | 行为 | 返回值 |
|------|------|--------|
| `try_claim()` | 非阻塞：无可用元素时返回空 token | `Claim<T>` |
| `claim()` | 阻塞：自旋直到可用 | `Claim<T>` |
| `claim(timeout)` | 带超时的获取 | `Claim<T>` |
| `try_claim_scoped()` | `try_claim()` 的兼容别名 | `ScopedClaim`（即 `Claim<T>`） |
| `claim_scoped()` | `claim()` 的兼容别名 | `ScopedClaim`（即 `Claim<T>`） |
| `claim_scoped(timeout)` | `claim(timeout)` 的兼容别名 | `ScopedClaim`（即 `Claim<T>`） |
| `try_claim_guard()` | 非阻塞 RAII：析构时自动提交 | `shared_ptr<Item<T>>` 或 `nullptr` |
| `claim_guard()` | 阻塞 RAII：自旋直到可用，自动提交 | `shared_ptr<Item<T>>` |
| `claim_guard(timeout)` | 超时 RAII：超时返回 `nullptr` | `shared_ptr<Item<T>>` 或 `nullptr` |
| `commit(claim)` | 提前提交并消费 token；重复、陈旧或错误 owner 返回 false | bool |

`Claim<T>` 不可复制但可以移动，支持 `->`、`*` 和 `get()`。所有 Claim 都会在析构时自动提交；显式 `commit()` 用于提前发布，成功或失败后 token 都会失效。`_guard` 版本为需要共享所有权的旧代码保留。

```cpp
// RAII 风格 —— 推荐
{
    auto item = consumer->claim_scoped();  // 阻塞直到可用
    item->data.do_work();
}  // 离开作用域，自动提交

// 显式提交 —— 需要提前发布时使用
auto claim = consumer->try_claim();
if (claim) {
    claim->data.do_work();
    const bool committed = consumer->commit(claim);
}
```

## 进阶用法

### 混合步骤类型的多阶段流水线

以下示例展示一个 4 阶段流水线：输入 → 字段切分（并行）→ 数据解析（并行）→ 输出。

```cpp
#include "pipeline_ringbuffer/pipeline_ringbuffer.h"

using namespace PipelineRingBuffer;

struct QuoteItem {
    std::string raw_json;
    double price;
    double volume;
    // ... 其他字段
};

// 流水线布局：Exclusive → Exclusive → Share → Exclusive
RingBuffer<QuoteItem> ring_buffer(16, {
    EStepType::Exclusive,  // 阶段 0: 输入（1 个 worker）
    EStepType::Exclusive,  // 阶段 1: 预处理（3 个 worker 竞争）
    EStepType::Share,      // 阶段 2: 并行解析（3 个 worker，每个都处理同一元素）
    EStepType::Exclusive   // 阶段 3: 输出（1 个 worker）
});

// 阶段 0: 单个生产者
auto input = ring_buffer.create_consumer(0);

// 阶段 1: 多个 worker 竞争获取元素
auto preproc_1 = ring_buffer.create_consumer(1);
auto preproc_2 = ring_buffer.create_consumer(1);
auto preproc_3 = ring_buffer.create_consumer(1);

// 阶段 2: 并行处理 —— 所有 worker 都看到每个元素
auto parser_1 = ring_buffer.create_consumer(2);
auto parser_2 = ring_buffer.create_consumer(2);
auto parser_3 = ring_buffer.create_consumer(2);

// 阶段 3: 单个消费者
auto output = ring_buffer.create_consumer(3);

// 在启动所有工作线程前校验并固定拓扑
const auto topology_result = ring_buffer.freeze_topology();
// 确认 topology_result == ERingBufferError::None 后再启动线程
```

### 配合停止标志使用 try_claim

当需要优雅地停止工作线程时：

```cpp
std::atomic<bool> running{false};

void worker_thread(Consumer<MyData>* consumer) {
    while (!running) { /* 自旋等待启动信号 */ }

    while (running) {
        auto item = consumer->try_claim();
        if (!item) continue;  // 当前无可用元素，重试

        // 处理元素 ...
        consumer->commit(item);
    }
}

// 在主线程中：
running = true;
std::this_thread::sleep_for(std::chrono::seconds(5));
running = false;  // 所有 worker 退出循环
```

## 伪共享防护

当多个 CPU 核心访问恰好共享同一缓存行的相邻内存位置时，性能会因**伪共享**（false sharing）而下降。本库通过两种方式防止伪共享：

### 1. 元素级对齐

每个 `Item<T>` 整体按 64 字节对齐，内部同步字段集中在控制缓存行，`data` 从下一条缓存行开始：

```
┌────────────── Item[0] ──────────────┬────────────── Item[1] ──────────────┐
│    clock/count/index    │ data…  │    clock/count/index    │ data…  │
└──────── control line ──────┴ payload┴──────── control line ──────┴ payload┘
```

等待线程只读取控制行，不会因为当前 owner 修改 payload 而反复失效缓存；Item 大小按缓存行向上取整，相邻槽位也不会共享缓存行。该布局会增加小型 `T` 的内存占用，以换取更稳定的多核访问行为。

### 2. 步骤级对齐

`ExclusiveStep::m_sequence` 和 `ShareStep::m_head_sequence` 原子游标都按 `hardware_destructive_interference_size`（固定为 64 字节）对齐，防止步骤内部状态与相邻数据发生伪共享。

填充大小在编译期确定：
- 固定为 **64 字节**，避免不同 `-mtune` / `-mcpu` 编译单元产生不兼容的公开类型布局
- x86-64 上与缓存行大小一致；128B 缓存行平台上属于偏保守配置

## 线程安全

- **先创建拓扑再启动线程**：应在工作线程启动前创建全部 Consumer；首次领取会原子封存拓扑，之后调用 `create_consumer()` 会返回 `nullptr`
- **完整拓扑**：封存时会校验每个阶段至少有一个 Consumer；也可以在线程启动前显式调用 `freeze_topology()`，提前发现漏配阶段
- **ExclusiveStep**：多个 ExclusiveConsumer 可以并发竞争同一个步骤；`clock` CAS 预占槽位，`m_sequence` CAS 确认游标
- **ShareStep**：每个工作线程应使用独立的 ShareConsumer；不要跨线程共享，也不要在提交当前元素前领取下一个元素
- **Claim**：领取结果不可复制；提交会消费 token，析构时会自动提交。不要在 Claim 销毁后继续使用 `get()` 得到的指针
- **阶段间可见性**：所有 ShareConsumer 提交后，其写入对下一阶段可见；多个 ShareConsumer 同时写同一字段时仍需自行同步
- **生命周期**：RingBuffer 必须晚于所有尚未析构的 `Claim<T>` 和 claim guard 销毁；Consumer 由 RingBuffer 自身持有
- **对象所有权**：RingBuffer 不可复制、不可移动；Consumer 由 RingBuffer 持有，不要手动释放

## API 参考

### `RingBuffer<T>`

```cpp
RingBuffer(uint64_t buffer_size, const std::vector<EStepType>& step_type_list)
```
构造环形缓冲区。`buffer_size` 必须是正的 2 的幂（包括 1），`step_type_list` 至少包含一个阶段。

```cpp
Consumer<T>* create_consumer(int step_no)
```
创建绑定到指定阶段索引的消费者。消费者由 RingBuffer 持有，**不要手动 delete**；失败时返回 `nullptr`，不会抛异常。

```cpp
ConsumerCreateResult<T> try_create_consumer(int step_no)
```
需要区分失败原因时使用。返回值包含 `consumer` 和 `ERingBufferError`：`InvalidStep`、`TopologyFrozen` 或 `ResourceUnavailable`。

```cpp
ERingBufferError freeze_topology()
bool topology_frozen() const
```
显式校验并封存消费者拓扑，以及查询拓扑是否已经封存。成功返回 `None`，阶段缺少 Consumer 时返回 `IncompleteTopology`；若未显式调用，第一次领取元素时会自动尝试封存。

```cpp
const Item<T>* peer(uint64_t index) const
```
按索引查看任意缓冲区槽位（调试用）。

### `Consumer<T>`（基类）

| 方法 | 说明 |
|------|------|
| `Claim<T> try_claim()` | 非阻塞获取，失败时返回空 Claim |
| `Claim<T> claim()` | 阻塞获取（自旋） |
| `Claim<T> claim(milliseconds)` | 带超时的获取，超时返回空 Claim |
| `ScopedClaim try_claim_scoped()` | 非阻塞、无堆分配的 RAII 获取 |
| `ScopedClaim claim_scoped()` | 阻塞、无堆分配的 RAII 获取 |
| `ScopedClaim claim_scoped(milliseconds)` | 带超时、无堆分配的 RAII 获取 |
| `shared_ptr<Item<T>> try_claim_guard()` | 非阻塞 RAII 获取 |
| `shared_ptr<Item<T>> claim_guard()` | 阻塞 RAII 获取 |
| `shared_ptr<Item<T>> claim_guard(milliseconds)` | 带超时的 RAII 获取 |
| `bool commit(Claim<T>&)` | 提前提交并消费 Claim，校验 owner 和 clock |
| `Step<T>* step()` | 获取此消费者所属的步骤 |
| `IRingBuffer<T>* ringbuffer()` | 获取父级环形缓冲区 |

### `Claim<T>`

| 方法 | 说明 |
|------|------|
| `operator bool()` | token 是否仍持有一个有效领取 |
| `get()` / `operator*()` / `operator->()` | 访问所领取的 `Item<T>` |
| `clock()` | 返回领取时捕获的时钟值 |
| `commit()` | 提交并消费 token；重复或陈旧提交返回 `false` |
| `reset()` | 提交并消费 token，忽略提交结果 |

`Claim<T>` 不可复制、可以移动；析构会调用 `reset()`。它和通过 `get()` 取得的指针都不能比所属 RingBuffer 存活更久。

### `Item<T>`

| 成员 | 类型 | 说明 |
|------|------|------|
| `data` | `T` | 用户自定义数据 |
| `index` | `uint64_t` | 槽位在缓冲区中的索引（只读） |
| `clock` | `std::atomic_uint64_t` | 单调时钟：`phase = clock % (2*step_count)` 为阶段状态，`lap = clock / (2*step_count)` 为绕环代数（内部同步用） |
| `commit_count` | `std::atomic_int32_t` | Share 步骤的提交计数（内部同步用） |
| `claimed()` | `bool` | 派生投影：clock 为奇数即 claimed 态（仅 Exclusive 有意义；调试/断言用） |
| `step_no(period)` | `uint64_t` | 派生投影：`(clock % period) / 2`，当前阶段索引（调试/断言用） |
| `lap(period)` | `uint64_t` | 派生投影：`clock / period`，绕环代数（调试/断言用） |

### `EStepType`

| 值 | 含义 |
|----|------|
| `Exclusive` | 每个元素只有一个消费者获取（基于 CAS） |
| `Share` | 多个消费者共享；所有消费者都能看到每个元素 |

### `Step<T>`（基类）

| 方法 | 说明 |
|------|------|
| `step_type()` | 返回 `EStepType::Exclusive` 或 `EStepType::Share` |
| `step_no()` | 此步骤在流水线中的索引 |
| `ringbuffer()` | 父级环形缓冲区 |
| `consumer_list()` | 绑定到此步骤的所有消费者 |

### 填充工具

```cpp
// 可导出到你的结构体中使用
using PipelineRingBuffer::PaddingArray;  // char[64] —— 手动填充用

// 示例：填充自定义数据以规避伪共享
struct MyData {
    double price;
    double volume;
    PaddingArray _pad;  // 64 字节填充
};
```

## 测试

构建并运行测试：

```bash
cmake -S . -B build -DPIPELINE_RINGBUFFER_BUILD_TESTS=ON
cmake --build build
cd build && ctest
```

| 测试名称 | 说明 |
|----------|------|
| `pipeline_ringbuffer.basic` | 验证数据在 4 阶段流水线中的流转（Exclusive → Exclusive → Share → Exclusive） |
| `pipeline_ringbuffer.rapidjson` | 使用 rapidjson 基准测试 JSON 解析吞吐量 |
| `pipeline_ringbuffer.slice_parallel_parser` | 集成测试：基于字段切片的并行 JSON 解析 |
| `pipeline_ringbuffer.rapidjson_parallel_parser` | 集成测试：基于 rapidjson 的并行 JSON 解析 |
| `pipeline_ringbuffer.sequence_cursor` | 验证 tail、单槽位超时以及 `UINT64_MAX -> 0` 的真实 sequence/clock 回绕 |
| `pipeline_ringbuffer.exclusive_contention` | 验证多个 ExclusiveConsumer 高竞争下不会同时持有同一槽位 |
| `pipeline_ringbuffer.regression` | 验证拓扑封存、极端超时、Share 单槽、重复领取保护、跨阶段可见性和所有权类型约束 |
| `pipeline_ringbuffer.slot_reuse` | 小容量缓冲区高频复用压力测试：多生产者竞争 + Share 栅栏 + 多输出竞争，验证恰好一次投递和无重复认领 |

## 许可证

Apache-2.0。详见 [LICENSE](LICENSE)。