C++ 并发队列：从有界数组到安全内存回收
前言：先明确“无锁”的含义
在 C++ 并发编程中，实现多生产者、多消费者（MPMC）队列，需要同时考虑数据竞争、内存序、操作语义和对象生命周期。使用原子变量与 CAS，并不自动意味着算法满足无锁保证。
Lock-Free 是一种进展保证：在参与线程持续执行的条件下，系统整体能够持续完成操作，不依赖某个特定线程恢复执行。 它不保证每个线程都能在有限步内完成自己的操作；后者涉及更强的 Wait-Free 保证。
判断队列时，还必须说明操作的含义。例如，允许“暂时无法取得元素”时返回失败的尝试接口，与要求失败一定表示队列为空的接口，具有不同的语义。不能仅凭函数没有等待、能够返回，就断言它是严格语义下的无锁 FIFO 队列。
本文先讨论数组队列的常见问题，再介绍基于序列号的有界队列、Michael–Scott 链表队列，以及安全内存回收的基本方法。
一、简单的原子数组队列有哪些问题？
一种常见设计使用 head、tail 和 write 三个原子变量：生产者通过 CAS 预订 tail 对应的位置，写完数据后按顺序更新 write，消费者通过 head 取出已发布的数据。
这种设计不一定没有用途，但需要明确以下问题：
- 按顺序提交可能阻碍进展。 如果生产者必须等待前面的生产者推进
write，那么前者暂停后，后续生产者即使写完自己的数据，也无法完成提交。这种依赖不满足 Lock-Free 保证。 - 抢占位置和完成访问是两件事。 如果消费者读取数据前后只用一个
head表示状态，生产者可能在旧消费者仍然读取时复用槽位。并发只读本身不构成数据竞争；与覆盖、移动或析构重叠的无同步访问才是问题。其后果是未定义行为，可能表现为崩溃或静默的数据损坏。 - 伪共享可能影响性能。 不同线程频繁修改同一缓存行里的独立变量，会产生额外的缓存一致性开销。具体影响取决于数据布局、处理器和负载，需要测量。
因此，队列不仅要区分位置归谁所有，还要区分数据何时可读、槽位何时可以复用。
二、基于 Sequence 的有界 MPMC 队列
Dmitry Vyukov 提出了一种经典的有界 MPMC 队列：为环形数组中的每个槽位（Cell）增加独立的原子序列号（Sequence），表示槽位在当前轮次中的状态。
生产者先争夺入队位置，写完数据后发布序列号；消费者先确认数据已经发布，再争夺出队位置，读完后更新序列号，允许下一轮生产者复用槽位。
这种设计不需要生产者按顺序更新一个全局 write 指针，但不满足严格意义上的 Lock-Free 保证。算法作者也明确指出了这一点。[1]
为什么仍会受到暂停线程的影响？
假设容量为 4，队列最初为空：
- 生产者 P0 抢到位置 0，在发布数据之前暂停。
- P1 抢到位置 1，写入并发布数据，入队返回成功。
- 消费者检查位置 0，发现数据没有就绪，返回失败，不能跳过它去消费位置 1。
- 其他生产者可以继续填充剩余位置，但绕回位置 0 后，也无法继续成功入队。
此时，恢复有效的数据传递仍依赖 P0 继续执行。消费者暂停在“抢到位置、尚未归还槽位”的阶段，也会影响后续槽位复用。
这也意味着：出队返回失败，不一定表示没有已经完成入队的元素。 如果把失败解释为严格的“队列为空”，上面的执行顺序便不符合普通线性化 FIFO 队列的接口约定。
C++ 示例实现
下面的 C++17 示例保留原算法的结构，增加容量检查、资源管理和元素类型约束，并将接口命名为 try_push / try_pop，强调失败也可能由暂未就绪的槽位引起。
示例采用有限位宽的无符号序号。序号比较要求被比较的逻辑位置距离小于计数空间的一半；容量检查只是必要条件，还必须避免某个操作保留旧位置期间，队列推进超过这一范围。实际部署应结合计数器位宽、操作速率和最大线程停顿时间审查这一假设。
#include <atomic>
#include <cstddef>
#include <limits>
#include <memory>
#include <stdexcept>
#include <type_traits>
template <typename T>
class MPMCQueue {
static_assert(std::is_default_constructible_v<T>,
"T must be default constructible");
static_assert(std::is_nothrow_copy_assignable_v<T>,
"T must be nothrow copy assignable");
static_assert(std::is_nothrow_destructible_v<T>,
"T must be nothrow destructible");
struct Cell {
std::atomic<std::size_t> sequence;
T data;
};
// 示例布局参数，不代表所有处理器的缓存行大小。
static constexpr std::size_t counter_alignment = 64;
static constexpr std::size_t half_range =
std::size_t{1} << (std::numeric_limits<std::size_t>::digits - 1);
static std::size_t checked_capacity(std::size_t n) {
if (n < 2 || (n & (n - 1)) != 0 || n >= half_range) {
throw std::invalid_argument(
"capacity must be a power of two, >= 2 and < half range");
}
return n;
}
// 无符号减法按模运算，避免有符号减法溢出。
// 判断有效的前提：逻辑序号距离严格小于 half_range。
static bool is_behind(std::size_t seq, std::size_t expected) noexcept {
return (seq - expected) >= half_range;
}
const std::size_t capacity_;
const std::size_t buffer_mask_;
const std::unique_ptr<Cell[]> buffer_;
alignas(counter_alignment) std::atomic<std::size_t> enqueue_pos_{0};
alignas(counter_alignment) std::atomic<std::size_t> dequeue_pos_{0};
public:
explicit MPMCQueue(std::size_t capacity)
: capacity_(checked_capacity(capacity)),
buffer_mask_(capacity_ - 1),
buffer_(new Cell[capacity_]) {
for (std::size_t i = 0; i < capacity_; ++i) {
buffer_[i].sequence.store(i, std::memory_order_relaxed);
}
}
MPMCQueue(const MPMCQueue&) = delete;
MPMCQueue& operator=(const MPMCQueue&) = delete;
bool try_push(const T& data) noexcept {
Cell* cell;
std::size_t pos = enqueue_pos_.load(std::memory_order_relaxed);
for (;;) {
cell = &buffer_[pos & buffer_mask_];
const std::size_t seq =
cell->sequence.load(std::memory_order_acquire);
if (seq == pos) {
if (enqueue_pos_.compare_exchange_weak(
pos, pos + 1, std::memory_order_relaxed)) {
break;
}
} else if (is_behind(seq, pos)) {
// 队列已满，或下一入队槽位尚未归还。
return false;
} else {
pos = enqueue_pos_.load(std::memory_order_relaxed);
}
}
cell->data = data;
cell->sequence.store(pos + 1, std::memory_order_release);
return true;
}
bool try_pop(T& data) noexcept {
Cell* cell;
std::size_t pos = dequeue_pos_.load(std::memory_order_relaxed);
for (;;) {
cell = &buffer_[pos & buffer_mask_];
const std::size_t expected = pos + 1;
const std::size_t seq =
cell->sequence.load(std::memory_order_acquire);
if (seq == expected) {
if (dequeue_pos_.compare_exchange_weak(
pos, pos + 1, std::memory_order_relaxed)) {
break;
}
} else if (is_behind(seq, expected)) {
// 队列为空，或下一出队槽位尚未发布。
return false;
} else {
pos = dequeue_pos_.load(std::memory_order_relaxed);
}
}
// 与 is_nothrow_copy_assignable 检查的 const T& 赋值保持一致。
data = static_cast<const T&>(cell->data);
cell->sequence.store(pos + capacity_, std::memory_order_release);
return true;
}
};内存序与使用约束
位置计数器负责争夺所有权，序列号负责发布数据和归还槽位。 因此，位置计数器的 CAS 可以使用 memory_order_relaxed；槽位序列号需要承担两个方向的同步：
- 生产者写入数据后执行 release store，消费者通过对应的 acquire load 确认数据可读。
- 消费者读取完成后执行 release store，后续生产者通过对应的 acquire load 确认槽位可以覆盖。
这使普通的 T data 访问得到必要的先行发生关系。相比 seq_cst 是否更快、快多少，需要针对目标平台测量。
示例还具有以下边界：
- 不支持任意元素类型。
T必须可默认构造，且复制赋值不抛异常。出队时显式从const T&赋值，确保调用与类型约束检查的是同一个重载；否则，额外定义的operator=(T&)可能被优先选中并抛异常。抢占位置后的赋值异常会使槽位无法正常发布或归还；本例函数标记为noexcept，若异常逃出则会终止程序。std::string通常不满足这里的无异常复制赋值约束。 - 不分配槽位，不等于整个操作不分配内存。 队列在构造时分配数组，但
T的赋值仍可能分配内存或执行阻塞操作；noexcept也不代表非阻塞。 - 调用者仍需管理外部对象的同步。 入队源对象不能同时被其他线程无同步地修改；多个出队操作也不能无同步地写入同一个输出对象。
- 构造和析构不与队列操作并发。 对象应在构造完成后安全地交给其他线程使用，销毁前必须确保所有访问已经结束。
- 对齐只能缓解部分伪共享。 分开两个计数器不能消除相邻 Cell 的伪共享，也不能消除生产者、消费者交接同一 Cell 时必要的缓存一致性通信。
- 序列号不是无限整数。 固定槽位避免了动态节点释放后复用地址的问题，但仍需遵守前述回绕与序号距离假设，不能笼统宣称“彻底解决所有 ABA 问题”。
此外，std::atomic<std::size_t> 是否由无锁操作实现取决于平台，可通过 is_always_lock_free 或 is_lock_free() 检查。即使这些原子操作无锁，也不能改变该队列本身的进展限制。
三、动态容量与 Michael–Scott Queue
如果容量不能预先固定，可以考虑链表或动态分段数组等结构。“无界”通常指没有预设的固定容量上限，并不意味着内存无限，也不意味着只能使用逐节点链表。
Michael–Scott Queue（MS Queue）是一种经典的无锁链表队列，其核心算法包含以下步骤：[2]
- 维护虚拟头节点。
head指向一个不作为当前有效队列元素返回的节点。出队推进head后，后继节点成为新的虚拟头节点。 - 先链接新节点。 入队线程通过 CAS 将新节点链接到链尾的
next。这个成功的 CAS 是入队操作的线性化点，即该操作在抽象队列中生效的时刻。 - 再尝试推进
tail。 链接成功后，线程尝试将tail向后移动。这个更新允许失败，因为其他线程可能已经帮助完成。 - 帮助其他线程推进。 如果发现
tail落后，线程可以协助更新它。因此，原线程在链接节点后暂停，不会使后续操作只能等待它恢复。
与前面的有界队列相比，这里的关键差异是：一个线程已经完成的结构修改可以被其他线程识别，并继续推进后续步骤。
不过，核心算法的 Lock-Free 保证不自动覆盖完整 C++ 实现。动态内存分配、元素操作、底层原子指令和回收机制，都可能引入阻塞或其他限制。尤其是从链表中摘除节点，并不代表可以立即释放它。
四、安全内存回收（SMR）
并发链表需要区分两个时刻：节点从数据结构中摘除，以及不再有线程可能访问它。只有后者得到保证，才能释放节点。
例如，消费者 A 读取了旧 head，随后暂停；消费者 B 推进 head 并立即释放旧头节点。A 恢复后如果解引用旧指针，就会发生 use-after-free。这属于未定义行为，不一定立即崩溃。
如果地址被重新分配，另一个逻辑对象可能拥有相同的指针值，从而引入 ABA 风险。解决 ABA 和保证对象生命周期相关，但并不等价：只给指针增加版本号，并不能让已经释放的内存继续安全可读。
垃圾回收环境通常承担部分生命周期管理工作；在 C++ 中，则需要为节点设计相应的安全回收协议。Hazard Pointers 和 EBR 是两种常见方案，也存在 RCU、引用计数及其他方法。具体选择取决于数据结构和进展要求。
1. 风险指针（Hazard Pointers）
Hazard Pointers 通过发布保护记录，告诉回收线程哪些节点暂时不能释放。记录通常由某个线程持有并更新，但需要能被回收线程扫描，不能是其他线程不可见的普通局部变量。
保护一个从共享原子指针中取得的节点，基本步骤是：
- 从共享指针读取节点地址。
- 将该地址发布到风险指针记录中。
- 重新读取共享指针，验证保护的地址仍然有效；验证失败则重新尝试。
- 验证成功后才解引用节点，并在访问结束后解除保护。
第三步不能省略。否则，节点可能在“读取地址”和“发布保护”之间已经被摘除并释放。C++ 标准草案中的 protect / try_protect 明确包含这种发布后的验证过程。[3]
回收线程先将摘除的节点标记为待回收（retire），再按协议扫描保护记录、批量释放可以回收的节点。发布、验证和扫描之间还需要正确的内存序；仅仅使用原子指针，并不能自动保证安全。
对于 MS Queue 等需要访问多个关联节点的结构，还必须确认每个待解引用节点都受到相应保护，并验证所依赖的结构关系。一个保护记录不一定足以完成整个操作。
2. 基于世代的回收（Epoch-Based Reclamation，EBR）
EBR 通过读者的活动区间来判断旧节点是否仍可能被访问，常见设计包括：
- 维护全局世代以及每个参与线程的活动状态。
- 线程访问共享节点前，按协议进入受保护区间；结束访问后宣布退出。
- 摘除节点时记录相关世代，并将节点加入待回收集合。
- 只有确认可能持有这些节点的旧读者都已结束相应访问，才允许回收。
这里需要覆盖所有可能访问节点的读者，包括进入更早世代但尚未退出的线程。不能仅检查“世代 N 的线程已经离开”，也不能只凭全局计数器从 N 变成 N+1 就认定安全。具体的世代推进与回收规则取决于实现。
EBR 通常避免逐节点发布保护记录，读侧开销较低，但传统 EBR 对线程停顿敏感：某个线程长期停留在受保护区间，可能阻止回收，使待回收内存持续增长。[4] 因此，评估时除了吞吐量，还应检查长时间停顿、内存占用以及批量回收的延迟。
五、如何选择和评估队列
选型前，应先明确生产者与消费者数量、容量限制、FIFO 语义、失败返回的含义，以及是否需要严格的进展保证。
- 容量固定时，可考虑有界数组。 它通常具有较好的局部性，能够避免操作期间的节点分配，但需要核实具体算法的进展保证，不能把所有数组队列都称为 Lock-Free。
- 容量动态增长时，可考虑 MS Queue 或分段结构。 除了入队、出队算法，还要评估内存分配、回收及积压时的内存上限。
- 正确性与性能分别验证。 检查数据竞争、线性化语义、异常路径、计数器回绕和对象生命周期，再通过目标负载下的测试评估吞吐量与尾延迟。
CAS 解决的是单次原子更新问题。一个可靠的并发队列，还需要清晰的操作契约、可解释的进展保证，以及完整的资源生命周期设计。
参考资料
- Dmitry Vyukov：Bounded MPMC queue。
- Maged M. Michael、Michael L. Scott：Simple, Fast, and Practical Non-Blocking and Blocking Concurrent Queue Algorithms：原始伪代码。
- C++ 工作草案：Hazard pointer member functions。
- Trevor Brown：Reclaiming memory for lock-free data structures: there has to be a better way。
Comments NOTHING