1 lost update 与有锁并发
在多线程场景下，不同线程对同一数据进行操作，即容易导致 lost update 的问题——以简单的加法计数器为例，考虑
int counter = 0;
void increment() {
++counter;
}直觉上，++counter 可理解为一次操作，然而从 CPU 的执行角度看，在执行加法运算的先后，需要处理数据的搬运。我们可以先将其理解为下面三个步骤：
int old = counter;
int desired = old + 1;
counter = desired;在多线程的场景下，则很容易出现如图的问题：
A 与 B 都读到了 7，各自算出 8，随后又先后将 8 写回。对于同一个计数器，两次自增最终只留下了一次更新，这就是 lost update。在 C++ 中，这构成了 data race（数据竞争）的问题，而这一行为是未定义的，会产生不受约束的结果。
为了解决上述问题，一个很自然的解决思路就是在一个线程计算时上锁，使得在某个时刻，只有一个线程有权修改这块状态。
#include <mutex>
std::mutex m;
int counter = 0;
void increment() {
std::lock_guard lock(m); // 离开作用域时自动解锁
++counter;
}现代互斥锁在没有竞争时，常可通过用户态原子操作完成加解锁，无须每次进入内核。以 Linux 上基于 futex 的锁为例，只有出现竞争、需要等待或唤醒时，才可能进入内核。因此，不能简单地将锁的开销理解为“每次操作都发生一次系统调用”。
1.1 锁的依赖性问题
锁带来的进展问题核心在于持锁线程可能停止运行：一个线程的暂停，可以阻止所有需要这把锁的其他线程取得进展。假设 A 已经获得锁，并在修改共享状态的中途被操作系统抢占，此时 B、C、D 即使有机会在其他核心上运行，也必须等到 A 恢复执行并释放锁。如果改为自旋等待，只是将等待方式从休眠改成了反复检查，仍然没有消除对 A 的依赖。无锁算法尝试改变这种依赖关系：A 暂停时，其他持续运行的线程仍然能够提交自己的操作，或者帮助某个已发布的操作完成，使系统整体继续前进。
需要注意的是，这里的“整体继续前进”讨论的是当线程继续执行算法步骤时，算法本身会不会因为某个暂停的线程而让所有操作都无法完成。
无锁并发的出发点就是为了解决上述问题，即一个并发数据结构至少要讨论以下两个独立的方面：
- Safety：确保程序并发执行时结果的正确性。
- Progress / Liveness：明确程序在什么条件下能够完成操作。
对于第一个方面，我们使用 Linearizability（线性化）作为并发对象的重要正确性条件；对于第二个方面，则区分不同层次的进展保证。
1.2 线性化 Linearizability
线性化的核心要求是：并发操作虽然发生了各种交错，但对外应该看起来像是每个操作在调用和返回之间的某一个瞬间原子发生。这个瞬间称为 linearization point（线性化点）。
以栈为例，A 正在 push(X)，B 同时开始 pop()。如果 A 先把 X 发布为栈顶，随后 B 成功把它移走，那么就可以将这段执行理解为先 push(X)，后 pop() → X，即使两个函数的执行区间明显重叠。
1.3 Progress
我们可以通过建立一下模型来区分并发数据结构不同层次的进展保证：
| 模型 | 保证 | 没有保证的部分 |
|---|---|---|
| Mutex / blocking | 依靠互斥保护状态；等待者的进展可能依赖持锁者 | 持锁线程暂停时，其他等待者可能全部停住 |
| Obstruction-free | 一个操作若独自执行足够多的步骤，就能完成 | 多线程持续竞争时，可能互相干扰而始终无法完成 |
| Lock-free | 持续执行的系统中，总有某个操作能在有限步骤内完成 | 某个特定线程可能反复失败，发生 starvation |
| Wait-free | 每个持续执行的操作都能在有界的自身步骤内完成，界可以依赖线程数等参数 | 不直接保证现实时间上限，也不意味着没有竞争 |
2 原子操作
锁的思路是先获得一段时间内的独占访问权，再修改状态；另一种思路则是直接让某一次状态更新不可分割。这就引出了原子操作。
对于一个原子对象，load 与 store 分别提供原子读取和原子写入，但若我们先 load、在局部变量中计算、再 store，这三个步骤合起来仍然不是一次原子更新。若需要表达以当前值为基础，计算并提交新值，就需要 Read-Modify-Write（RMW，读—修改—写） 的原子操作。
2.1 CAS 无锁算法 Compare-And-Swap
原子操作中很重要的一种操作即 CAS（Compare-And-Swap），其核心思想就是将比较和修改视为不可分割的原子操作。在 C++ 中，对应两个方法 compare_exchange_weak 与 compare_exchange_strong，我们可以将其逻辑抽象为
#include <atomic>
bool CAS(std::atomic<int>& address, int& expected, int desired) noexcept {
return address.compare_exchange_strong(
expected, desired,
std::memory_order_seq_cst,
std::memory_order_seq_cst);
}expected 同时承担两种功能——比较成功前，它保存在原子对象应当被观察到的值；比较失败后，它更新了原子对象实际上的值。
以计数器为例，我们可以利用 CAS 将“读取—计算—检查并提交”组织成一个重试循环：
#include <atomic>
std::atomic<unsigned> counter{7};
unsigned increment_and_get() noexcept {
unsigned expected = counter.load(std::memory_order_relaxed); // 观察
for (;;) {
unsigned desired = expected + 1; // 根据当前观察计算
if (counter.compare_exchange_weak(
expected,
desired,
std::memory_order_relaxed,
std::memory_order_relaxed)) { // 检查并提交
return desired;
}
// 失败时 expected 已更新；下一轮必须重新计算 desired。
}
}
假设 A、B 都读到了 7，A 首先成功提交 7 → 8。此时 B 再提交 7 → 8 就会失败，其 expected 被更新为 8；随后 B 重算 desired = 9，再次尝试提交。两次成功的 CAS 就分别对应两次自增的线性化点。
当然，对于简单的加法计数器，标准库已经提供了 fetch_add 方法：
#include <atomic>
std::atomic<unsigned> counter{7};
unsigned increment_and_get() noexcept {
unsigned old = counter.fetch_add(1, std::memory_order_relaxed);
return old + 1;
}2.1.1 compare_exchange_weak 与 compare_exchange_strong
C++ 的 compare_exchange 是语言层面的抽象，这也就意味着不同 CPU 架构实现原子 RMW 的方式不同。很多处理器使用 Load-Linked 与 Store-Conditional 来实现原子更新，以 RISC-V 为例，可以简单理解为：
- LL：读取内存，同时建立一个 reservation，其中 reservation 可以理解为 CPU 内部维护的一种过渡状态。
- SC：如果这个 reservation 仍然有效则写入，否则写入失败。
对于一个正常的原子操作流程：
然而，例如中间发生了其它 CPU 的访问，reservation 可能失效，此时，即便 CAS 的判断逻辑没有问题，原子操作依然可能失败。
对于 compare_exchange_weak 而言，这次底层失败可以直接视为一种 spurious failure（伪失败），可将其当做比较失败一样做处理；而对于 compare_exchange_strong 而言，遭遇底层失败时，直接对整个 CAS 流程重试。
对于 CAS 本身就在重试循环里的情形，通常直接选择 compare_exchange_weak；而对于只尝试一次，false 必须意味着真的不匹配的情形，必须选择 compare_exchange_strong 。例如，
#include <atomic>
std::atomic<int> state{0};
bool try_start() {
int expected = 0;
return state.compare_exchange_strong(
expected,
1,
std::memory_order_relaxed,
std::memory_order_relaxed);
}2.2 cache line 与竞争
从硬件角度讲，线程对共享变量的操作通常会经过各个 CPU 核心的缓存。缓存一致性一般以 cache line（缓存行） 为单位协调，而不是只协调某个变量的几个字节。换言之，哪怕线程修改的是不同变量，但只要这些变量碰巧落在同一缓存行，仍然互相影响。一个核心要对某条缓存行执行写入或原子更新时，需要取得相应的写入权限，并使其他核心中的冲突副本得到处理。
以典型的 x86 缓存内存访问为例，带 LOCK 前缀的原子指令通常通过缓存一致性机制实现原子性，而不是直接给整个内存总线上锁。因此，lock-free 并没有消除共享热点。如果很多线程都修改同一个计数器或同一个 head，仍然会发生缓存行写入权限的反复迁移，同时伴随 CAS 重试。
3 内存序 Memory Ordering
到这里，我们已经能够保证某一次更新不可分割。然而，一个数据结构通常不只有一个原子变量：例如我们考虑一个生产者先准备数据，再设置“已经准备好”的标志的情形：
#include <atomic>
int data = 0;
std::atomic<bool> ready{false};
void producer() {
data = 42;
ready.store(true, std::memory_order_relaxed);
}
int consumer() {
while (!ready.load(std::memory_order_relaxed)) {
// 此时还没有获得读取 data 的同步依据。
}
return data;
}这段代码中，producer 先写 data，再写 ready ，在预期的情况下，consumer 在读取到 ready 的条件变更后读取已修改过的 data 的值。然而，对于现实的场景，CPU 执行内存写入的顺序，与其他 CPU 核心观察到这些写入的顺序，不一定相同——对于现代 CPU， CPU 在执行写入指令时，写入先进入 Store Buffer（存储缓冲区），之后核心通过缓存层级和一致性协议处理这次写入，此时其他核心随后才能按照相应的内存序约束观察到该写入。那么在弱内存序处理器上，则可能出现以下问题：
- 两个写入的可见顺序发生变化
弱内存序处理器并不一定要求不同地址的写入按照严格的 FIFO 顺序对外可见，因此 consumer 可能先读取到 ready = true，但 data 仍为旧值。
- 读取指令自身也可能乱序执行
虽然从源代码上看， consumer 先读取 ready ，再读取 data ，然而现代 CPU 支持 Out-of-Order Execution（乱序执行）、Speculative Execution（推测执行）、Branch Prediction（分支预测）等等，CPU 可以预测循环会退出，并尝试提前执行后面的内存读取，从而导致先读取 data ，再读取 ready 的非预期情形。
也就是说，我们需要引入内存序来为同步操作之间建立跨线程关系。
3.1 happens-before
我们通过三个术语来说明前面的关系：
| 关系 | 含义 | 前面例子中的对应 |
|---|---|---|
| Sequenced-before | 同一线程内，语言规定的求值先后关系 | 写 data 先于 release store |
| Synchronizes-with | 满足条件的同步操作之间建立的跨线程关系 | 写入 true 的 release 与读到该值的 acquire |
| Happens-before | 将相应的线程内顺序和同步关系连接起来的先行关系 | 写 data 先行于消费者读 data |
所谓内存序，就是为原子操作选择一组顺序与同步要求，用来规定：这个操作除了保证自身不可分割以外，还需要怎样约束周围的内存访问，以及它能够与另一线程的哪些操作建立关系。
我们可以把前面的例子分成两条线程内的链：生产者先写 data，再写 ready；消费者先读 ready，再读 data。只有这两条链还不够，真正缺少的是一条跨线程的连接。release store 与读到它发布值的 acquire load，就可以提供这条连接，使线程内的先后关系拼成跨线程的 happens-before。
3.2 常用的 memory_order
| 内存序 | 可以用在哪类操作 | 核心含义 |
|---|---|---|
relaxed | load、store、RMW | 保留原子性及该对象的修改顺序，不额外发布或获取周围数据 |
acquire | load、RMW | 读到相应发布后，为后续访问建立同步依据 |
release | store、RMW | 发布当前线程之前的访问 |
acq_rel | RMW | 同一次更新既获取已有发布，又发布本线程之前的访问 |
seq_cst | load、store、RMW | 在适用的 acquire / release 语义之外，将 seq_cst 操作纳入共同的全序 |
3.3.1 relaxed
relaxed 只保留原子性及该对象的修改顺序，不额外发布或获取周围数据。以统计完成次数为例，多个线程都向同一个计数器加一，我们只要求这些加法不会互相覆盖，并不通过计数器通知其他共享数据已准备好：
#include <atomic>
std::atomic<unsigned> completed{0};
void record_completion() {
completed.fetch_add(1, std::memory_order_relaxed);
}
需要注意的一点是，内存序是针对具体原子操作选择的，不是要求整个程序采用统一的同步策略。若要在所有工作结束后汇总结果，还可以使用线程 join 等其他同步方式，而 relaxed 并不要求整个程序都不需同步。
3.3.2 acquire
acquire 要求后面的内存访问不能以破坏 acquire 语义的方式，被提前到 acquire 之前，换言之，acquire 确保后续的内存访问相对于它受到正确排序。同样考虑一个生产者先准备数据，再设置“已经准备好”的标志的情形——对于消费者一侧， ready 的 acquire load 读到生产者 release store 发布的 true 后，消费者才能依据这次同步读取 data：
#include <atomic>
int data = 0;
std::atomic<bool> ready{false};
void producer() {
data = 42;
ready.store(true, std::memory_order_release);
}
int consumer() {
while (!ready.load(std::memory_order_acquire)) {
// 此时还没有获得读取 data 的同步依据。
}
return data;
}
3.3.3 release
release 为此前的内存访问与这次发布操作建立必要的排序约束，即确保之前的内存访问相对于它受到正确排序。对于生产者一侧，先完成普通数据的初始化，再用 release 发布标志。
更进一步的，release 和 acquire 实际承担着内存访问的排序约束和跨线程传递同步关系的职责的两个不同工作：
- Ordering：约束本线程中的操作顺序，保证了 release 前与 acquire 后的内存访问不会越界
- Synchronization：在两个线程间建立关系，使两个线程通过同一个原子对象完成了数据传递。
3.3.4 acq_rel
acq_rel 使得一次 RMW 操作同时具有 acquire 和 release 语义，既能接收此前发布的数据，也能向其他线程发布数据。如果一个线程处于中间位置，既要接收前一个线程的发布，又要把自己准备的内容交给后一个线程，那么同一次 RMW 可以使用 acq_rel。考虑只有一个生产者、一个中继线程和一个消费者，stage 按 0 → 1 → 2 变化：
#include <atomic>
int data = 0;
int relay_data = 0;
std::atomic<int> stage{0};
void produce() {
data = 42;
stage.store(1, std::memory_order_release);
}
int relay() {
while (stage.load(std::memory_order_relaxed) != 1) {}
relay_data = 84;
stage.exchange(2, std::memory_order_acq_rel);
return data;
}
int consume_relay() {
while (stage.load(std::memory_order_acquire) != 2) {}
return data + relay_data;
}
3.3.5 seq_cst
acquire / release 可以连接实际发生通信的操作，但并不要求所有线程对不同原子对象的访问都排入同一个顺序。如果两个线程分别操作不同的原子变量，彼此之间可能根本没有形成同步关系。例如：
#include <atomic>
std::atomic<int> x{0};
std::atomic<int> y{0};
int r1, r2;
void thread_a() {
x.store(1, std::memory_order_release);
r1 = y.load(std::memory_order_acquire);
}
void thread_b() {
y.store(1, std::memory_order_release);
r2 = x.load(std::memory_order_acquire);
}一种可能的执行时序是：
- Core 0 执行
x.store(1)，写入暂留在自己的 Store Buffer。 - Core 1 执行
y.store(1)，写入暂留在自己的 Store Buffer。 - Core 0 继续读取
y，发现 Core 1 的写入尚未对自己可见，因此得到0。 - Core 1 继续读取
x，同样得到0。 - 之后，两边 Store Buffer 中的写入完成对外可见。
这也就导致了经典的 Store → Load 重排序效应。仅按源代码的执行顺序来看，r1 与 r2 不能同时为 0。导致这一问题的主要原因在于 release 主要约束本线程位于该操作之前的内存访问，acquire 主要约束本线程位于该操作之后的内存访问，但不能约束 store 与 load 之间的内存访问顺序。
而 seq_cst 则为相应原子操作增加共同全序，同时保留适用的 acquire / release 语义：
#include <atomic>
std::atomic<int> x{0}, y{0};
int r1 = -1, r2 = -1;
void thread_a() {
x.store(1, std::memory_order_seq_cst);
r1 = y.load(std::memory_order_seq_cst);
}
void thread_b() {
y.store(1, std::memory_order_seq_cst);
r2 = x.load(std::memory_order_seq_cst);
}
3.3 atomic_thread_fence
我们可以发现，上述操作本质上就是在约束内存访问排序，尤其是防止内存访问重排序造成非预期的行为，我们可以利用 std::atomic_thread_fence 来进一步说明这一点。fence 的本质即建立一种围栏，使得围栏前后的内存访问不会越界。
上述四种 memory_order 都可转化为对应的 fence 操作，例如
#include <atomic>
#include <cassert>
int data1 = 0;
int data2 = 0;
std::atomic<bool> ready1{false};
std::atomic<bool> ready2{false};
void producer() {
data1 = 42;
data2 = 99;
std::atomic_thread_fence(
std::memory_order_release
);
ready1.store(true, std::memory_order_relaxed);
ready2.store(true, std::memory_order_relaxed);
}
void consumer() {
while (!ready1.load(std::memory_order_relaxed)
&& !ready2.load(std::memory_order_relaxed)) {
}
std::atomic_thread_fence(
std::memory_order_acquire
);
// 此时可以安全读取两份数据
assert(data1 == 42);
assert(data2 == 99);
}
这里连接两个线程的仍然是原子标志：消费者退出循环，说明至少有一次 relaxed load 读到了生产者写入的 true；生产者的 release fence 位于这次标志写入之前，消费者的 acquire fence 位于读到该值之后，于是两道 fence 建立同步关系。data1 与 data2 都写在 release fence 前，因此消费者在 acquire fence 后可以读取两份数据，不要求两个标志都已经读到 true。这里假设只有这一次发布，数据发布后不再被改写。两道 fence 若没有这样的原子读写关系连接，就不能仅凭位置建立跨线程同步。C++ 的 fence 同步条件
我们将一个线程中的连续内存操作进一步抽象：
#include <atomic>
template <class OperationA, class OperationB>
void with_fence(OperationA operation_A,
OperationB operation_B,
std::memory_order order) {
operation_A();
std::atomic_thread_fence(order);
operation_B();
}根据 A、B 是读取（Load）还是写入（Store），一共只有四种组合，我们据此可以得出不同 Fence 禁止重排序的类型：
| 访问顺序 | Acquire Fence | Release Fence | Acq_rel Fence | Seq_cst Fence |
|---|---|---|---|---|
| Load → Load | ✓ | - | ✓ | ✓ |
| Load → Store | ✓ | ✓ | ✓ | ✓ |
| Store → Load | - | - | - | ✓ |
| Store → Store | - | ✓ | ✓ | ✓ |
3.4 CAS 中的内存序
CAS 成功时执行 RMW，失败时只是读取原子对象并更新 expected，因此可以分别指定成功与失败的内存序。
在成功时，执行 RMW，所以既包含 Load，也包含 Store，这与上述说明过的情形基本一致；而在失败时，CAS 只读取原子对象而不做 Store 操作，因此不能使用 release 与 acq_rel 内存序。
特别地，在 C++ 中，也允许指定一个内存序，此时编译器会自动确定失败路径的内存序。
| 单个参数 | 成功内存序 | 失败内存序 |
|---|---|---|
| relaxed | relaxed | relaxed |
| acquire | acquire | acquire |
| release | release | relaxed |
| acq_rel | acq_rel | acquire |
| seq_cst | seq_cst | seq_cst |
4 lock-free 数据结构
前面的计数器只需要更新一个数值。接下来我们考虑链式数据结构，它需要同时处理结构关系、状态发布以及节点生命周期。
4.1 Treiber Stack
栈的特点是后进先出（LIFO）。如果用单向链表表示栈，只需要一个共享的栈顶指针 head，每个节点保存自己的值和下一个节点指针。
先看 push。假设当前栈顶为 A，我们希望将 X 放在它前面：首先设置 X.next = A，然后尝试 CAS，将 head 从 A 改为 X。若其他线程先改变了 head，则需要根据新观察重新设置 X.next，再重试；再看 pop。先观察当前栈顶 A，并读出 A.next = B，然后尝试 CAS，将 head 从 A 改为 B。成功说明我们移走了这次观察到的栈顶；失败则重新观察并尝试。
push 与非空 pop 的线性化点，都是成功更新 head 的 CAS；空栈 pop 的线性化点，则是观察到 head == nullptr 的那次原子读取。
假设节点不重新入栈，代码可写为：
#include <atomic>
#include <optional>
struct Node {
const int value;
Node* next = nullptr;
explicit Node(int v) : value(v) {}
};
class IntStack {
std::atomic<Node*> head_{nullptr};
public:
// node 已初始化，并且只由当前线程执行这唯一一次 push。
// node 从这里开始直到所有访问线程结束，都不能销毁或复用。
void push(Node& node) noexcept {
Node* expected = head_.load(std::memory_order_relaxed);
do {
node.next = expected;
} while (!head_.compare_exchange_weak(
expected, &node,
std::memory_order_release,
std::memory_order_relaxed));
}
std::optional<int> pop() noexcept {
Node* expected = head_.load(std::memory_order_acquire);
while (expected != nullptr) {
Node* desired = expected->next;
if (head_.compare_exchange_weak(
expected, desired,
std::memory_order_acquire, // C++17 也可以写作 std::memory_order_relaxed
std::memory_order_acquire)) {
return expected->value;
}
}
return std::nullopt;
}
bool is_lock_free() const noexcept {
return head_.is_lock_free();
}
};push 只需要获取当前栈顶的地址，并将其保存为新节点的 next，不需要访问旧节点的内容，因此初始读取和 CAS 失败时使用 relaxed 就足够了。而 pop 不同，它需要通过栈顶指针读取节点的 next 和 value。这些数据可能由其他线程初始化，因此需要使用 acquire，确保能够看到节点在入栈前完成的写入。
这里还有一个细节：即使栈顶经过其他线程多次修改，只要这些修改都是 CAS 等读—修改—写（RMW）操作，先前通过 release 建立的发布关系仍可以通过 release sequence（释放序列） 传递，使后续的 acquire 能够正确读取更早入栈的节点。
4.1.2 ABA Problem
现在放开“节点不重新入栈”的限制，就会出现另一个问题：CAS 只比较这次观察到的值，并不会自动记录值变化的历史。
假设原来的链表是 A → B → C。T1 读到 head = A，又保存了 A.next = B，随后暂停。T2 将 A 弹出，再将 B 弹出，然后把 A 重新入栈；此时链表是 A → C，head 又回到了 A。
T1 恢复后提交 CAS(head, A, B)，仍然可以成功，因为 head 的值确实又是 A。然而 B 已经被移走，T1 却把这个基于旧结构得到的 B 重新接回，破坏了栈的语义。这就是 ABA：共享值经历 A → … → A，数值恢复了，但旧观察依赖的结构关系已经发生变化。
这里为了突出逻辑问题，先假设 A、B 都没有释放。现实中还有更危险的地址复用：A 被释放后，分配器把同一地址分给了另一个节点，CAS 仍可能只看到相同的地址。
常见处理思路是给指针附加版本号，使 CAS 比较 (pointer, version)，而不是只有 pointer。这样，回到同一地址也会因为版本不同而比较失败。具体来说，将 head 从一个指针扩展成一对状态 (pointer, version)，并要求每次成功修改 head 时，版本号也一同增加。两部分必须通过同一次原子操作读取或更新，不能分别维护一个原子指针和一个原子计数器，否则线程仍可能读到两次不同更新拼出的组合。
以前面的交错为例，初始状态为 (A, 12)，T1 据此准备将它改为 (B, 13)。T2 先后弹出 A、弹出 B、重新压入 A，使共享状态经历 (A, 12) → (B, 13) → (C, 14) → (A, 15)。T1 恢复后，虽然地址仍是 A，但完整的比较对象已不同，因此旧 CAS 会失败，不能把保存的 B 接回。
失败后，要在安全保护节点的前提下，根据新观察重新读取后继并计算 desired，不能只更新版本号、继续使用旧的 B。实现时可以使用平台支持的宽 CAS，或者在明确的地址位宽、对齐及容量限制下编码指针或索引与版本；无论哪种方式，都要确认整个组合原子类型确实是 lock-free，并处理版本回绕的问题。
4.1.3 Memory Reclamation
即使禁止把同一个节点重新入栈，如果 pop 成功后马上 delete，仍然不安全。考虑以下交错：
T1 手里还保存着 A 的地址，但 T2 已经把 A 释放了。因此，T1 恢复后读取 A->next 时，就访问了一个已经不存在的节点。所以，pop 成功只说明节点已经离开栈，并不说明所有线程都不再使用它。我们可以先把节点放进待回收集合，这一步叫 retire；等到回收机制确认其他线程已不可能再访问它，才真正 delete。这样就把“从栈中移走”与“释放内存”分成了两步。
两种常见方法是 Hazard Pointer 与 Epoch-based Reclamation（EBR）：
| 方法 | 核心思路 | 需要注意的代价 |
|---|---|---|
| Hazard Pointer | 读者发布自己正在保护的具体节点；回收者检查保护记录 | 保护记录的发布、验证与扫描都有成本 |
| EBR | 读者声明进入某个 epoch 的读侧临界区；等待可能持有旧节点的读者退出后回收 | 一个长时间暂停且尚未退出的读者，可能长期推迟回收 |
4.1.3.1 Hazard Pointer
Hazard Pointer 的思路是：既然回收者不知道其他线程手里还拿着哪些指针，就让每个线程把自己正在使用的节点地址公开出来。每个线程拥有自己的保护槽，回收者可以读取所有线程的保护槽。
不过，不能简单地“读指针，然后登记保护，就开始解引用”。假设 T1 读到 head = A 后暂停，T2 可能在 T1 发布保护前就移走并释放 A。因此，T1 发布保护后，还需要重新读取 head，确认它仍然等于 A；如果不相等，就放弃这次观察，重新开始。在验证成功前，不能读取 A->next 或 A->value。
对于这里的栈，读者的协议轮廓可以写成：
#include <atomic>
#include <optional>
struct Node {
const int value;
Node* next = nullptr;
};
std::optional<int> read_top(std::atomic<Node*>& head,
std::atomic<Node*>& hazard) noexcept {
Node* p;
do {
p = head.load(std::memory_order_seq_cst);
hazard.store(p, std::memory_order_seq_cst);
} while (head.load(std::memory_order_seq_cst) != p);
std::optional<int> value;
if (p != nullptr) {
value = p->value;
}
hazard.store(nullptr, std::memory_order_seq_cst);
return value;
}保护槽发布的是“我还可能访问 A”，并不是给 A 加锁。其他线程仍然可以把 A 从栈中移走，只是不能立刻释放它。移走 A 的线程先把它放进 retire 集合，之后按照回收协议扫描所有保护槽：如果 A 仍出现在某个槽里，就继续保留；没有被保护的节点才可以回收。扫描通常会成批进行，以分摊检查保护槽的成本。
例如，T1 的保护槽里只有 A，即使 T1 长时间暂停，回收者仍可以处理未受保护的 B、C。
4.1.3.2 Epoch-based Reclamation
EBR 换了一个角度：不再逐个登记正在访问的节点，而是记录“这个线程从哪一轮开始访问共享结构，目前有没有退出”。这里的一轮叫 epoch，可以理解为回收协议维护的逻辑代数，它不是时间戳，也不表示经过了多少毫秒。
线程在读取共享节点之前，先按照协议进入读侧临界区，发布自己的活跃状态与所处 epoch，并完成必要的验证。只要仍在临界区里，就可能持有这一轮观察到的旧节点；退出时，线程声明自己已经不再访问这些节点。移走的节点也不会马上释放，而是记入相应的 retire 批次。
#include <atomic>
#include <optional>
struct Node {
const int value;
Node* next = nullptr;
};
class EpochDomain {
public:
virtual ~EpochDomain() = default;
virtual void enter() noexcept = 0;
virtual void leave() noexcept = 0;
virtual void retire(Node* node) noexcept = 0;
virtual void collect() noexcept = 0;
};
class EpochGuard {
EpochDomain& domain_;
public:
explicit EpochGuard(EpochDomain& domain) noexcept : domain_(domain) {
domain_.enter();
}
~EpochGuard() {
domain_.leave();
}
EpochGuard(const EpochGuard&) = delete;
EpochGuard& operator=(const EpochGuard&) = delete;
};
std::optional<int> pop(std::atomic<Node*>& head, EpochDomain& domain) {
EpochGuard guard(domain);
Node* expected = head.load(std::memory_order_acquire);
while (expected != nullptr) {
Node* desired = expected->next;
if (head.compare_exchange_weak(
expected, desired,
std::memory_order_acquire,
std::memory_order_acquire)) {
int value = expected->value;
domain.retire(expected);
return value;
}
}
return std::nullopt;
}
void reclaim(EpochDomain& domain) noexcept {
domain.collect();
}回收者真正要确认的是：在节点移走之前就可能拿到它的读者，都已经结束那次访问。 这段等待通常称为 grace period。新进入的读者只能沿共享结构中仍然有效的路径取节点，不能再从 head 找到已经移走的 A；而旧读者必须先退出，才能排除它们手里还保存着 A 的可能。
图中 T1 进入得早，退出得晚，它的访问区间跨过了这批节点的退役位置。即使其他读者都已经退出，回收者仍然需要等 T1；T1 退出后，旧批次才可能满足回收条件。不同 EBR 实现会用不同的代数推进与分批规则来完成这个判断，不能只给全局 epoch 加一，就认为上一轮节点都可以释放。
这样一来，读者不必为每个节点单独发布保护，但保护范围也更大：一个停在旧临界区中的线程，即使只实际访问过 A，也可能让整批旧节点持续等待。所以 EBR 更依赖读者及时退出；如果把节点指针带出临界区继续使用，退出声明就失去了意义。
在禁止活节点重新入栈的算法中，安全回收可以阻止仍被保护的地址提前释放和复用，从而处理相应的地址复用 ABA。但它不自动解决任意结构中的所有 ABA。前面的教学实现采用更直接的限制：整个并发期间都不回收，也不复用节点，代价是空间只能在所有线程结束后统一释放。
还需要区分“操作可以继续完成”和“内存可以继续回收”。某些 EBR 方案里，一个暂停读者不阻止队列操作，却会阻止旧节点回收；若最终耗尽可用空间，系统运行仍会受到影响。因此，回收策略是实际无锁设计的一部分。
4.2 Michael–Scott Queue
栈只围绕一个 head 提交更新，而队列要实现先进先出（FIFO），通常需要同时维护 Head 与 Tail。一个很自然的问题是：既然单字 CAS 只能改一个位置，怎样避免某个线程在更新到一半时暂停，把整个队列卡住？
Michael–Scott Queue 的思路是让中间状态仍然可以被其他线程识别和推进。它使用一个单向链表，并保留一个 dummy node（哨兵节点）：Head 指向当前哨兵，第一个有效元素位于 Head.next；Tail 指向尾部节点，但可以暂时落后。
初始空队列只有一个哨兵 D，即 Head = Tail = D，D.next = null。哨兵让“没有前驱的第一个节点”与后续节点使用同一种链接方式。
4.2.1 Enqueue：先接链，再推进 Tail
假设现在是 D → X，Tail 指向 X。入队 Y 需要先初始化 Y 的值，并令 Y.next = null，然后尝试 CAS，将 X.next 从 null 改为 Y。这次接链成功的 CAS 是入队的线性化点：从此 Y 已经属于队列。
随后再尝试把 Tail 从 X 推进到 Y。但这个第二次更新只是帮助后续操作更快找到尾部，不决定 Y 是否已经入队。
假设 A 接上 Y 后立刻暂停，Tail 仍指向 X。B 读到 Tail = X，却发现 X.next = Y，就能知道 Tail 落后了，于是帮助将 Tail 推进到 Y，再继续自己的入队。B 无须等 A 恢复执行。
这就是 helping（帮助机制）：将操作分成可以被其他线程识别的状态，让暂时停下来的线程不独占剩余步骤。队列不要求 Head 和 Tail 每一瞬间都精确描述同一个快照，而要求可观察到的中间状态有合法的推进方式。
4.2.2 Dequeue
出队时，先读取旧哨兵 D，再读取 D.next = X。如果有有效元素，就读取 X.value，然后尝试 CAS，将 Head 从 D 改为 X。成功的 CAS 是非空出队的线性化点。
这时返回的是 X 的值，但 X 节点本身没有立即离开链表，它成为新的哨兵；真正移走的是旧哨兵 D。不要因为“返回了 X 的值”，就把 X 节点直接释放。
由于多个消费者可能同时观察 X，不能让失败的消费者提前 move 或破坏 X.value。后面的 C++ 代码使用不可变且可复制的值。要支持有副作用的移动、异常或更一般的元素类型，还需要额外的接口与生命周期设计。
出队时还会遇到三种典型情况：
| 已验证的观察 | 说明 | 后续动作 |
|---|---|---|
Head == Tail 且 Head.next == null | 队列为空 | 返回 empty |
Head == Tail 且 Head.next != null | 元素已接上，但 Tail 落后 | 帮助推进 Tail，然后重试 |
Head != Tail 且有后继 | 队列存在元素 | 尝试 CAS 推进 Head |
这些条件中的指针值不是天然一致的快照，读完后还要验证 Head 是否仍等于最初观察到的值；验证失败就重试。
#include <atomic>
#include <optional>
class IntQueue {
struct Node {
const int value;
std::atomic<Node*> next{nullptr};
Node* retired_next = nullptr;
explicit Node(int v) : value(v) {}
};
std::atomic<Node*> head_;
std::atomic<Node*> tail_;
std::atomic<Node*> retired_{nullptr};
void retire(Node* node) noexcept {
Node* expected = retired_.load(std::memory_order_relaxed);
do {
node->retired_next = expected;
} while (!retired_.compare_exchange_weak(
expected, node,
std::memory_order_relaxed,
std::memory_order_relaxed));
}
public:
IntQueue() {
Node* dummy = new Node(0);
head_.store(dummy, std::memory_order_relaxed);
tail_.store(dummy, std::memory_order_relaxed);
}
IntQueue(const IntQueue&) = delete;
IntQueue& operator=(const IntQueue&) = delete;
~IntQueue() {
Node* node = head_.load(std::memory_order_relaxed);
while (node != nullptr) {
Node* next = node->next.load(std::memory_order_relaxed);
delete node;
node = next;
}
node = retired_.load(std::memory_order_relaxed);
while (node != nullptr) {
Node* next = node->retired_next;
delete node;
node = next;
}
}
void enqueue(int value) {
Node* node = new Node(value);
for (;;) {
Node* tail = tail_.load(std::memory_order_acquire);
Node* next = tail->next.load(std::memory_order_acquire);
if (tail != tail_.load(std::memory_order_acquire)) {
continue;
}
if (next == nullptr) {
if (tail->next.compare_exchange_weak(
next, node,
std::memory_order_release,
std::memory_order_relaxed)) {
tail_.compare_exchange_strong(
tail, node,
std::memory_order_release,
std::memory_order_relaxed);
return;
}
} else {
tail_.compare_exchange_weak(
tail, next,
std::memory_order_release,
std::memory_order_relaxed);
}
}
}
std::optional<int> dequeue() noexcept {
for (;;) {
Node* head = head_.load(std::memory_order_acquire);
Node* tail = tail_.load(std::memory_order_acquire);
Node* next = head->next.load(std::memory_order_acquire);
if (head != head_.load(std::memory_order_acquire)) {
continue;
}
if (head == tail) {
if (next == nullptr) {
return std::nullopt;
}
tail_.compare_exchange_weak(
tail, next,
std::memory_order_release,
std::memory_order_relaxed);
} else {
int value = next->value;
if (head_.compare_exchange_weak(
head, next,
std::memory_order_acq_rel,
std::memory_order_acquire)) {
retire(head);
return value;
}
}
}
}
};