Lesson 64: 无锁环形缓冲区
练习任务
难度：难
实现一个无锁（lock-free）的 SPSC（Single Producer Single Consumer）环形缓冲区，使用 C11 _Atomic 类型和原子操作。完成五个核心任务：
init_buffer()— 将所有 buffer 槽位初始化为 -1（哨兵值表示"空"），重置 write_idx 和 read_idx 为 0print_buffer(label)— 遍历缓冲区：值 >= 0 打印数字，否则打印 "."；格式[label] buf=[...]producer_write(value)— 忙等待至缓冲区非满（w - r < BUFFER_SIZE），写入到buffer[w % 8]，发布递增 write_idxconsumer_read()— 忙等待至缓冲区非空（r < w），从buffer[r % 8]读取并标记 -1，发布递增 read_idx，返回读到的值main()— 按交错时间线执行：生产 5 个 → 消费 3 个 → 生产 5 个（含回绕）→ 消费 7 个
验证方式：
make testmake test 编译后运行，通过管道 | diff 比对 expected_output.txt。
本课共有 1 组测试用例——交错时间线追踪的完整输出（20 步操作 + 最终结果汇总），见 exercises.toml 的 stdout 字段。
提示：五个 TODO 中最容易出错的是：索引永远递增不取模（只在访问
buffer[]时% BUFFER_SIZE）、满判断用>= BUFFER_SIZE而非>、消费后把槽位标记回 -1、用atomic_store_explicit/atomic_load_explicit而非=直接赋值。
核心知识点
- 无锁编程的动机 — 锁的四个核心代价：阻塞、死锁、优先级反转、锁竞争；原子操作如何绕过它们
- SPSC 环形缓冲区索引设计 —
write_idx和read_idx单调递增永不取模，只在访问 buffer 时% BUFFER_SIZE - C11
_Atomic语义 — 原子类型的读写不可分割；atomic_store_explicit/atomic_load_explicit提供显式内存顺序控制 - memory_order 分级 — 从
relaxed（仅原子性）到seq_cst（全序屏障）的六级体系；acquire-release 配对建立 happens-before - 回绕与哨兵值 — 当
w % 8绕回 0 时覆盖已消费的旧数据；-1 标记空槽位用于可视化和状态判断 - 缓存行伪共享 — 两个变量共享同一 64 字节缓存行时，一个核心的写入导致另一个核心缓存行失效；padding 对齐解决方案
- 忙等待 vs 阻塞等待 — spin-wait（唤醒延迟 ~0，浪费 CPU）vs
pthread_cond_wait（不浪费 CPU， 唤醒延迟 ~1-10μs） - lock-free vs wait-free — lock-free 保证整体进度（至少一个线程前进），wait-free 保证单个线程进度（每个线程都有界完成）
- 从教学 SPSC 到工程缓冲写 — Node.js
fs.writeFileSync的底层缓冲机制、Linux kfifo、DPDK 无锁环的工程连接
代码框架
c
#include <stdatomic.h>
#include <stdio.h>
#define BUFFER_SIZE 8
#define TOTAL_ITEMS 10
/* ─── Ring buffer data structure (provided) ─── */
static _Atomic int buffer[BUFFER_SIZE];
static _Atomic int write_idx = 0; /* producer advances this */
static _Atomic int read_idx = 0; /* consumer advances this */
/* ─── TODO 1: 初始化缓冲区 ─── */
static void init_buffer(void) {
// ① 将所有 BUFFER_SIZE 个槽位初始化为 -1（哨兵值，表示"空"）
// 使用 atomic_store_explicit(&buffer[i], -1, memory_order_relaxed)
// ② 将 write_idx 和 read_idx 从 0
// 使用 atomic_store_explicit(&write_idx, 0, memory_order_relaxed)
// 使用 atomic_store_explicit(&read_idx, 0, memory_order_relaxed)
}
/* ─── TODO 2: 打印缓冲区状态 ─── */
static void print_buffer(const char *label) {
// ① 打印 label 前缀: printf("[%s] buf=[", label);
// ② 遍历 i=0..BUFFER_SIZE-1:
// val = atomic_load_explicit(&buffer[i], memory_order_relaxed)
// if (val >= 0) printf("%d ", val) else printf(". ")
// (最后一个元素后不加空格，但此处为了格式一致可以保留)
// ③ 打印结尾: printf("]")
}
/* ─── TODO 3: 生产者写入 ─── */
static void producer_write(int value) {
// ① 忙等待直到缓冲区非满：
// while (1) {
// int w = atomic_load_explicit(&write_idx, memory_order_relaxed);
// int r = atomic_load_explicit(&read_idx, memory_order_acquire);
// if (w - r < BUFFER_SIZE) break;
// }
// 空: 读 read_idx 用 acquire 确保看到消费者最新释放
// ② 获取当前 write_idx (w) 和槽位 slot = w % BUFFER_SIZE
// ③ 写入数据：
// atomic_store_explicit(&buffer[slot], value, memory_order_relaxed)
// printf("[P] write %d at slot %d", value, slot)
// ④ 发布递增的 write_idx：
// atomic_store_explicit(&write_idx, w + 1, memory_order_release)
// 发布用 release 确保数据写入在索引更新前完成
// ⑤ 打印操作后状态：print_buffer(" after write")
// printf(" w=%d r=%d\n", w+1, r)
}
/* ─── TODO 4: 消费者读取 ─── */
static int consumer_read(void) {
// ① 忙等待直到缓冲区非空：
// while (1) {
// int r = atomic_load_explicit(&read_idx, memory_order_relaxed);
// int w = atomic_load_explicit(&write_idx, memory_order_acquire);
// if (r < w) break;
// }
// 注意: 读 write_idx 用 acquire 确保看到生产者最新释放
// ② 获取当前 read_idx (r) 和槽位 slot = r % BUFFER_SIZE
// ③ 读取数据：
// int val = atomic_load_explicit(&buffer[slot], memory_order_relaxed)
// 标记槽位为空：atomic_store_explicit(&buffer[slot], -1, memory_order_relaxed)
// printf("[C] read %2d at slot %d", val, slot)
// ④ 发布递增的 read_idx：
// atomic_store_explicit(&read_idx, r + 1, memory_order_release)
// ⑤ 打印操作后状态：print_buffer(" after read ")
// printf(" w=%d r=%d\n", w, r+1)
// ⑥ 返回读取的值
}
/* ─── TODO 5: 主函数 — 交错时间线 ─── */
int main(void) {
// ① init_buffer()
// ② 打印头部信息
// printf("=== Lock-Free Ring Buffer (SPSC) ===\n")
// printf("Capacity: %d, Total items: %d\n\n", BUFFER_SIZE, TOTAL_ITEMS)
// printf("--- Interleaved Timeline ---\n")
// ③ Phase 1: 生产 5 个 (0..4)
// for (i = 0; i < 5; i++) producer_write(i)
// ④ Phase 2: 消费 3 个 (0..2)
// for (i = 0; i < 3; i++) { int v = consumer_read(); (void)v; }
// ⑤ Phase 3: 生产 5 个 (5..9) — 含回绕！
// for (i = 5; i < 10; i++) producer_write(i)
// ⑥ Phase 4: 消费 7 个 (3..9)
// for (i = 0; i < 7; i++) { int v = consumer_read(); (void)v; }
// ⑦ 打印最终结果
// printf("\n--- Final Results ---\n")
// printf("Produced: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]\n")
// printf("Consumed: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]\n")
// printf("Final write_idx: %d\n", w)
// printf("Final read_idx: %d\n", r)
// int w = atomic_load(&write_idx)
// int r = atomic_load(&read_idx)
// printf("Buffer empty: %s\n", (w == r) ? "yes" : "no")
return 0;
}阅读骨架后，尝试自己填充 // ① 标记的部分。核心挑战在于：memory_order_relaxed 和 memory_order_release/acquire 分别用于什么操作？为什么读对方的索引要用 acquire、写自己的索引要用 release？忙等待循环里 re-load 对方的索引为什么要放在循环体内而非循环前？
TIP
先不要往下翻看参考解答。在纸上画出 8 个槽位 + write_idx/read_idx，手动追踪 20 步操作。重点关注第 12-13 步的回绕——write_idx=8 时 8%8=0，覆盖槽位 0 的旧数据。这 20 步追踪就是 expected_output.txt 的全部内容。
深度讲解
1. 无锁编程的动机——锁的四个代价
1.1 传统互斥锁的工作方式
c
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
// 线程 A
pthread_mutex_lock(&mutex);
buffer[slot] = data; // 临界区
pthread_mutex_unlock(&mutex);
// 线程 B — 同时到达时阻塞等待
pthread_mutex_lock(&mutex);
int v = buffer[slot]; // 临界区
pthread_mutex_unlock(&mutex);这就是课堂上最常见的并发同步方式。mutex 可以工作——但代价是什么？
1.2 四个代价——从微观到宏观
| 代价 | 机制 | 影响 |
|---|---|---|
| 阻塞等待 | 一个线程持锁时，其他线程调用 lock() 进入内核睡眠 | 上下文切换开销 ~1-10μs，即便临界区只有 5 条指令 |
| 死锁 | 线程 A 持有锁 L1 等 L2，线程 B 持有 L2 等 L1 | 程序永久挂起，排查困难 |
| 优先级反转 | 低优先级线程持锁阻塞高优先级线程 | 实时系统灾难（火星探路者事故） |
| 锁竞争 | 高并发下多个线程争抢同一把锁 | 吞吐量随核心数增加反而下降 |
上下文切换的真实代价：
用户态 → 内核态切换： ~0.5μs
保存/恢复寄存器： ~0.2μs
TLB/cache 冲突： ~2-5μs
─────────────────────────────────
总计： ~3-10μs
而一int buffer[slot]=data 只需 ~5ns (L1 cache hit)。
比例：10μs / 5ns = 2000:1！
锁的开销是临界区本身的 2000 倍。1.3 无锁的替代路径
无锁编程用 CPU 硬件提供的原子指令（CAS、FAA、LL/SC）直接在接在共享数据上操作：
c
// 无需 mutex！直接原子操作共享数据
_Atomic int write_idx = 0;
// 生产者（无锁）
int w = atomic_load_explicit(&write_idx, memory_order_relaxed);
atomic_store_explicit(&buffer[w % N], data, memory_order_relaxed);
atomic_store_explicit(&write_idx, w + 1, memory_order_release);
// 没有 lock/unlock，没有上下文切换！IMPORTANT
无锁编程并非"零开销"——原子指令本身比普通指令慢（x86 上 lock cmpxchg ~20 cycles vs mov ~1 cycle），内存屏障也有开销。但相比上下文切换的微秒级开销，原子操作仍快 2-3 个数量级。
锁 vs 无锁的适用场景：
临界区长（ >1μs）→ 锁更合适（阻塞睡眠释放 CPU）
临界区极短（ <100ns）→ 无锁更合适（避免上下文切换）
低竞争 → 锁和无锁差距不大
高竞争 → 锁急剧退化，无锁（特别是 SPSC）不受影响
需要公平性 → 锁（FIFO mutex）
追求吞吐 → 无锁2. SPSC 环形缓冲区的索引设计——数据结构核心
2.1 为什么是环形？
线性队列（每次出队移动所有元素）:
[A][B][C][D][_] → 出队A → [B][C][D][_][_] (数据移动 O(n))
环形缓冲区（移动指针，数据不动）:
[A][B][C][D] write_idx=4, read_idx=0
出队A → read_idx=1 (O(1)，数据仍在原位)
[.][B][C][D] (逻辑上 A 已出队)三个核心优势：
- O(1) 入队和出队——只需移动索引指针
- 固定内存——分配一次，无需
malloc/free - 缓存友好——连续内存访问，CPU 预取机制可以提前加载相邻槽位
2.2 单调递增索引——关键设计决策
c
/* 错误做法：取模索引 */
_Atomic int write_idx = 0; // 始终在 [0, BUFFER_SIZE-1]
// → write_idx=0, read_idx=0 → 是空？还是满（刚好绕了一圈）？
// → 无法区分！
/* 正确做法：单调递增索引 */
_Atomic int write_idx = 0; // 生产者已写入的元素总数（永不减少）
_Atomic int read_idx = 0; // 消费者已读取的元素总数（永不减少）
// 状态判断基于差值——差值是绝对的，不依赖"绕了几圈"
// 空：write_idx == read_idx
// 满：write_idx - read_idx == BUFFER_SIZE
// （注意：用 >= BUFFER_SIZE 更安全，>= 防止多等一轮）索引不取模的完整论证：
write_idx 和 read_idx 永远递增（代表已处理的总数），
只在访问 buffer[] 时用 % BUFFER_SIZE 计算物理槽位。
这带来两个关键好处：
1. 满/空判断基于差值 → 精确可靠
2. 索引值的绝对大小直接反映吞吐量
→ write_idx=1000000 表示生产者已写入了 100 万个元素
副作用：索引需要更大的数据类型（int 足够本课，生产代码用 size_t）2.3 SPSC 的结构图
BUFFER_SIZE = 8
┌─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┐
│ [0] │ [1] │ [2] │ [3] │ [4] │ [5] │ [6] │ [7] │
└─────┴─────┴─────┴─────┴─────┴─────┴─────┴─────┘
↑ ↑
read_idx % 8 write_idx % 8
(下一个要读的槽位) (下一个要个要写的槽位)
SPSC 语义：
┌──────────┐ write_idx ┌──────────────┐ read_idx ┌──────────┐
│ Producer │ ────────────────→ │ Ring Buffer │ ────────────────→ │ Consumer │
│ (1 线程) │ │ (capacity=8) │ │ (1 线程) │
└──────────┘ └──────────────┘ └──────────┘
为什么 SPSC 最简单？
- 只有一个线程写 write_idx → 无需 CAS（多生产者才需要 CAS 竞争）
- 只有一个线程读 read_idx → 无需 CAS（多消费者才需要 CAS 竞争）
- 索引各自演进，互不干扰
- SPSC 是 lock-free 最安全的教学入口NOTE
扩展到 MPMC（多生产者多消费者）需要 atomic_compare_exchange_weak 循环来争夺索引。那时复杂性急剧上升——需要处理 ABA 问题、内存回收（RCU/Hazard Pointers）。本课聚焦 SPSC——在理解 SPSC 之前不要碰 MPMC。
3. C11 _Atomic 的语义与存储——编译器和 CPU 的契约
3.1 什么是"原子"？
c
/* 非原子操作——可能被拆分为多条 CPU 指令 */
int x = 0;
x = 42; // 在 32 位平台上是一条指令
// 在 16 位平台上可能是两条（高位/低位分别写入）
// 在没有对齐的情况下，可能被拆分为两个总线传输
/* 原子操作——保证"不可分割" */
_Atomic int y = 0;
atomic_store(&y, 42); // 保证任何观察者要么看到 0，要么看到 42
// 不会看到"写了一半"的中间状态
/* 在 x86-64 上，对齐的 int 读写通常是原子的——但不要依赖这一点！ */
/* _Atomic 是编译器层面保证（跨平台），而非特定 CPU 的偶然行为。 */3.2 两种 API 风格
c
#include <stdatomic.h>
_Atomic int counter = 0;
/* 风格 1：隐式内存顺序（默认 seq_cst） */
atomic_store(&counter, 42); // 等价于 seq_cst
int v = atomic_load(&counter); // 等价于 seq_cst
counter++; // atomic_fetch_add，seq_cst
/* 风格 2：显式内存顺序（推荐——让意图更清晰） */
atomic_store_explicit(&counter, 42, memory_order_release);
int v = atomic_load_explicit(&counter, memory_order_acquire);
/* 本课统一使用显式风格 */CAUTION
不要直接用 = 赋值 _Atomic 变量！buffer[slot] = value 在有些编译器上会拒绝编译（C 禁止对原子类型直接赋值），即使能编译也可能丢失原子性保证。始终使用 atomic_store/load_explicit。
3.3 _Atomic 数组的特殊性
c
static _Atomic int buffer[BUFFER_SIZE];
// 每个元素都是独立原子对象
atomic_store_explicit(&buffer[3], 42, memory_order_relaxed);
int v = atomic_load_explicit(&buffer[3], memory_order_relaxed);
// 数组本身不是原子的——不能原子地交换整个 buffer
// buffer = other_buffer; // 编译错误_Atomic int buffer[8] 声明了一个包含 8 个原子整数的数组。每个 buffer[i] 的读写是原子的，但整体数组不是原子对象。
4. memory_order 分级——从 relax 到 seq_cst
4.1 C11 六级体系全景
弱 ────────────────────────────────────────────────→ 强
relaxed consume acquire release acq_rel seq_cst
↑ ↑ ↑ ↑ ↑ ↑
无顺序 已弃用 读屏障 写屏障 读写屏障 全序屏障
约束 不推荐 (单体)本课使用三种：
| memory_order | 保证 | 典型用途 | 本题场景 |
|---|---|---|---|
relaxed | 仅原子性，无顺序约束 | 独立计数器 | buffer 读写、打印 |
acquire | 后续操作不重排到此之前 | 读取共享标志 | consumer 读 write_idx |
release | 之前操作不重排到此之后 | 发布数据 | producer 写 write_idx |
4.2 acquire-release 配对——如何建立 happens-before
Producer (写入线程):
┌────────────────────────────────────────────┐
│ 1. 写入数据到 buffer[slot] │ ← relaxed
] (普通写入，可能还在 store buffer 中) │
│ │
│ 2. atomic_store(&write_idx, w+1, release) │ ← release 屏障
│ ★ 保证：步骤1的所有写入对后续可见 │
└────────────────────────────────────────────┘
│
│ happens-before 关系
▼
┌────────────────────────────────────────────┐
│ 3. atomic_load(&write_idx, acquire) │ ← acquire 屏障
│ ★ 保证：看到 write_idx 更新后， │
│ 步骤1的写入也一定可见 │
│ │
│ 4. 读取 buffer[slot] 的数据 │ ← relaxed
│ (一定能读到完整的数据！) │
└────────────────────────────────────────────┘
Consumer (读取线程):如果不用 release/acquire 而全部用 relaxed：
可能的 CPU 重排（弱内存模型架构上）：
CPU 将 atomic_store(&write_idx) 重排到 buffer[slot]=data 之前
→ 消费者看到 write_idx 已更新但 buffer[slot] 还是旧数据
→ 数据竞争！
在 x86 上不太可能发生（x86 有较强内存模型），
但在 ARM/PowerPC 等弱内存模型架构上会发生！4.3 内存顺序速查决策树
场景："我写入数据，然后告诉别人数据准备好了"
→ 写数据用 relaxed，, 写标志用 release
场景："我看到标志，然后读取对应的数据"
→ 读标志用 acquire，读数据用 relaxed
场景："只是一个计数器，不保护其他数据"
→ 用 relaxed 即可
场景："不确定该用什么"
→ 用 seq_cst（最安全但最慢，适合调试阶段）
场景："两个标志互相依赖，需要全局顺序"
→ 用 seq_cst（保证所有线程看到一致的操作顺序）4.4 为什么 acquire 要 re-load？
c
/* 错误：只在循环前加载一次 */
void producer_write_bad(int value) {
int r = atomic_load_explicit(&read_idx, memory_order_acquire);
while (1) {
int w = atomic_load_explicit(&write_idx, memory_order_relaxed);
if (w - r < BUFFER_SIZE) break; // r 永远不变！
}
// ...
}
/* 正确：在循环内重新加载 */
void producer_write_good(int value) {
while (1) {
int w = atomic_load_explicit(&write_idx, memory_order_relaxed);
int r = atomic_load_explicit(&read_idx, memory_order_acquire);
if (w - r < BUFFER_SIZE) break; // r 每次迭代都可能变化
}
// ...
}消费者在另一侧递增 read_idx——如果不在每次迭代时重新加载，生产者就看不到消费者的最新进度，可能永远阻塞！
5. 回绕与哨兵值——当索引绕回 0
5.1 回绕的完整步进
容量 8，写入 10 个元素（但中间消费了 3 个腾空间）：
Phase 1: 生产 5 个 (0..4)
w=0: 写槽0→buf=[0 . . . . . . .]
w=1: 写槽1→buf=[0 1 . . . . . .]
w=2: 写槽2→buf=[0 1 2 . . . . .]
w=3: 写槽3→buf=[0 1 2 3 . . . .]
w=4: 写槽4→buf=[0 1 2 3 4 . . .]
Phase 2: 消费 3 个 (0..2), r 从 0→3
r=0: 读槽0→buf=[. 1 2 3 4 . . .]
r=1: 读槽1→buf=[. . 2 3 4 . . .]
r=2: 读槽2→buf=[. . . 3 4 . . .]
Phase 3: 生产 5 个 (5..9)
w=5: 写槽5→buf=[. . . 3 4 5 . .]
w=6: 写槽6→buf=[. . . 3 4 5 6 .]
w=7: 写槽7→buf=[. . . 3 4 5 6 7] ← 满了: w-r=7-3=4 < 8 OK
w=8: 写槽0→buf=[8 . . 3 4 5 6 7] ← 8%8=0，回绕！覆盖已读的旧数据
w=9: 写槽1→buf=[8 9 . 3 4 5 6 7] ← 9%8=1，回绕！
此时：w=10, r=3, w-r=7, 可写槽位=15.2 哨兵值 -1 的工程价值
c
/* -1 的四个作用 */
// 1. 视觉标记——print_buffer 中 -1 显示为 "."
if (val >= 0) printf("%d ", val);
else printf(". ");
// 2. 逻辑标记——消费后清空槽位
atomic_store_explicit(&buffer[slot], -1, memory_order_relaxed);
// 3. 调试辅助——意外读到 -1 说明逻辑错误
// （正常不应在 read_idx 指向的位置读到 -1）
// 4. 防止幽灵数据——避免看到已消费的旧值
// （虽然 SPSC 的索引保证不会，但在更复杂的队列中很重要）NOTE
生产代码中通常使用 size_t 或 uint32_t 作为索引类型，用额外标志位区分空/满（例如 write_idx 的高位作为 wrap flag）。本课使用 int + -1 哨兵是为了教学简洁。
6. 缓存行伪共享——隐形的性能杀手
6.1 什么是伪共享？
CPU 缓存以缓存行（cache line，通常 64 字节）为单位操作：
CPU0 缓存行:
┌──────────────────────────────────────────────────────────┐
│ buffer[0..7] (32B) │ write_idx (4B) │ read_idx (4B) │ ...│
└────────────────────────────────────────────────────────────┘
如果 write_idx 和 read_idx 在同一个缓存行：
1. 生产者更新 write_idx → CPU0 的缓存行标记为 Modified
2. CPU1（消费者）的对应缓存行变为 Invalid
3. 消费者下次读 read_idx → Cache Miss → 从 CPU0 重新加载整个缓存行
4. 即使消费者从不访问 write_idx！
单个 4 字节的写入导致 64 字节的缓存行无效化！6.2 为什么 SPSC 场景影响较小？
两个线程的访问模式互补：
- 生产者：写 buffer[] + 读 read_idx + 写 write_idx
- 消费者：读 buffer[] + 读 write_idx + 写 read_idx
缓冲区的槽位本身就是共享数据——cache line 需要在两个核心间迁移
这是"真共享"（true sharing）——不可避免
write_idx 和 read_idx 的伪共享在 SPSC 中影响相对较小：
因为每个线程本来就交替读写 buffer 槽位，
cache line 的迁移是"计划内"的。
但在 MPMC 场景中，多个生产者竞争 write_idx，
伪共享会导致逐出风暴——吞吐量下降 20-50%。6.3 缓解方案
c
/* 方案 1：alignas 将索引放在不同缓存行 */
_Atomic int write_idx __attribute__((aligned(64))) = 0;
char _pad1[60];
_Atomic int read_idx __attribute__((aligned(64))) = 0;
char _pad2[60];
/* 方案 2：显式填充 */
struct ring_buffer {
_Atomic int buffer[8];
_Atomic int write_idx;
char _pad1[60]; // 64 - 4 = 60
_Atomic int read_idx;
char _pad2[60];
};
/* 本课不强制 padding（SPSC 影响小，代码简洁优先），
但你应该知道这个概念——面试可能问到 */7. 忙等待 vs 阻塞等待——工程中的权衡
7.1 两种等待机制对比
忙等待 (busy-wait / spin-wait):
while (condition_not_met) {
// 反复检查，消耗 CPU 周期，不进入内核
}
优点: 唤醒延迟 ~0（条件满足立即继续，无上下文切换）
缺点: 浪费 CPU（"空转"期原子 CPU 做无用功）
适 适合: 等待时间极短（< 几微秒）的场景
本题: 生产者等待非满、消费者等待非空（单线程交错无实际等待）
阻塞等待 (blocking wait):
pthread_cond_wait(&cond, &mutex);
// 线程进入睡眠状态，被唤醒后继续
优点: 不浪费 CPU（等待期间 CPU 可用于其他线程）
缺点: 唤醒延迟 ~1-10μs（上下文切换 + 调度延迟）
适合: 等待时间可能较长的场景忙等待的 CPU 使用模式：
时间轴: ═════■■■════════■■■■════════════■■■■════════
idle spin idle spin idle spin idle
阻塞等待的 CPU 使用模式：
时间轴: ═════════════════■■■■══════════════════════■■■■
sleep awake sleep awake7.2 为什么本题在单线程中也用忙等待？
本课是教学演示——在单线程中交错执行生产者和消费者操作，模拟多线程的行为。忙等待循环在单线程中实际上永远不会自旋（条件总是满足或总是在下一次操作时满足），但代码结构展示了完整的无锁同步模式。
在真实多线程场景中，while 循环会在条件不满足时真正自旋——等对方线程推进索引。
TIP
在 Linux 内核中，kfifo 是一个生产级的无锁环形缓冲区实现。它使用类似的设计（两个单调递增的无符号索引），但对满/空的处理更精妙——通过让索引持续增长（直到溢出回绕到 0）来区分空/满状态。有兴趣可以阅读 include/linux/kfifo.h。
8. lock-free vs wait-free——术语辨析
8.1 定义
lock-free (无锁):
系统整体保证至少有一个线程能在有限步骤内完成。
其他线程可能被阻塞，但整体进度不受影响。
wait-free (无等元素):
每个线程都保证在有限步骤内完成自己的操作。
| 持锁线程会因为其他线程而永远等待。8.2 SPSC 环形缓冲区的级别
SPSC 环形缓冲区 = lock-free，但非 wait-free
生产者视角：
写入操作本身 O(1)（一次 store），但可能忙等待缓冲区非满
等待时间取决于消费者速度——理论上可能无限等待
消费者视角：
读取操作本身 O(1)（一次 load），但可能忙等待缓冲区非空
等待时间取决于生产者速度——理论上可能无限等待
整体保证：
只要生产者和消费者都在运行，整个系统在推进
→ lock-free ✓
单线程保证：
如果消费者永远不读，生产者可能永远等不到空闲槽位
→ wait-free ✗进阶概念层次：
┌─────────────────────────────────────────┐
│ wait-free ← 最强，但实现最困难 │
│ e.g. RCU (Read-Copy-Update) 的读端 │
├─────────────────────────────────────────┤
│ lock-free ← 本课级别，实际中最常见 │
│ e.g. SPSC 环形缓冲区 │
├─────────────────────────────────────────┤
│ obstruction-free ← 排除其他线程后完成 │
│ e.g. 某些 CAS 重试循环 │
├─────────────────────────────────────────┤
│ blocking ← mutex 级别，最熟悉 │
│ e.g. pthread_mutex_lock │
└─────────────────────────────────────────┘
越往上实现越复杂，但延迟可预测性越好。
在实时系统中，wait-free 是硬需求。9. 从教学 SPSC 到工程缓冲写——Node.js 与 Linux 内核中的隐藏连接
9.1 环形缓冲无处不在
本课的 SPSC 环形缓冲区是教学版本——容量 8、单线程交错执行、打印每步状态。但在工程中，同样的设计模式使用在：
- Linux 内核
kfifo：内核空间无锁环形缓冲区 - DPDK 无锁环：高性能网络包处理，每秒数百万包
- LMAX Disruptor：Java 高性能线程间消息传递，吞吐量达每秒千万级
- 音频/视频流缓冲区：实时音视频处理中的生产者（解码器）到消费者（渲染器）
- 异步日志系统：日志写入线程作为生产者，文件落盘线程作为消费者
- Node.js
fs.writeFileSync的底层缓冲，见下一节
9.2 Node.js fs.writeFileSync——缓冲写的本质
Node.js 中的 fs.writeFileSync(path, data):
┌──────────┐ writeFileSync ┌─────────────┐ syscall (write) ┌───────┐
│ JS 线程 │ ──────────────→ │ libuv 缓冲区 │ ────────────────→ │ 磁盘 │
│ (生产者) │ │ (环形缓冲) │ │ (消费) │
└──────────┘ └─────────────┘ └───────┘
关键观察：
- writeFileSync 在 JS 层是"同步"的——调用后阻塞 JS 线程
- 但在底层 libuv 实现中，数据先写入缓冲区（生产者 = JS 环境）
- 然后由操作系统内核消费者通过 syscall 异步落盘
- 缓冲区就是环形缓冲——固定大小的内存区域，指针循环推进c
/*
* 概念示意：writeFileSync 底层的缓冲机制（简化版）
*
* 真实实现涉及 libuv 的 uv_fs_t、uv__fs_work 和线程池调度——
* 这里展示的是"环形缓冲"这一共享概念。
*/
#include <stdio.h>
#include <string.h>
#define BUF_SIZE 4096 // 典型 DIO block size
typedef struct {
char data[BUF_SIZE];
int len; // 缓冲了多少数据
} write_buffer_t;
/* 环形缓冲：应用层 JS → libuv 缓冲 → 内核 write() syscall */
void writefile_ringbuffer_concept(const char *path, const char *data) {
write_buffer_t wbuf; // 环形缓冲（教学简化版）
size_t total = strlen(data);
size_t written = 0;
while (written < total) {
size_t chunk = total - written;
if (chunk > BUF_SIZE) chunk = BUF_SIZE;
// "生产者"：数据写入缓冲
memcpy(wbuf.data, data + written, chunk);
wbuf.len = chunk;
// "消费者"：缓冲写入磁盘（实际是 write syscall）
// 真实代码中这是另一个线程或异步 I/O
// write_to_disk(path, wbuf.data, wbuf.len);
written += chunk;
}
}
/*
* 核心对应关系：
*
* 本课 SPSC 环 → Node.js writeFileSync 底层
* ──────────────────────────────────────────────────────
* buffer[8] → 内核页缓存 (page cache)
* producer_write(value) → 应用数据写入缓冲区
* consumer_read() → sync/flush 将缓冲数据落盘
* write_idx 递增 → 缓冲偏移递增
* read_idx 递增 → 已刷盘偏移递增
* 忙等待到非满 → 缓冲区满时阻塞（背压）
* 忙等待到非空 → 无数据可刷时等待
*
* Node.js writeFileSync 的"sync"含义：
* - JS 层面：调用后阻塞 JS 线程直到数据落到内核缓冲区
* - 不保证数据已物理写入磁盘（除非用 fsync）
* - 内核缓冲区 → 物理磁盘 是另一个异步过程
*/9.3 从本课到生产级实践的三个台阶
台阶 1: 本课 SPSC 教学环
- 容量 8，只验证 10 个元素的交错时间线
- 单线程交错执行，无真正的并发
- 用 int + -1 哨兵做教学简化
台阶 2: 生产级 SPSC 环
- 容量可配置（通常 2^n 以加速取模运算 & (N-1)）
- 真正的多线程（pthread_create 分别运错执行生产者和消费者）
- 用 size_t 索引 + wrap flag 区分空/满
- padding 消除伪共享
- 配合 pthread_cond_wait 降低空转 CPU 消耗
## 3: MPMC 无锁队列
- 多个生产者竞争 write_idx（需要 CAS 循环）
- 多个消费者竞争 read_idx（需要 CAS 循环）
- 需要解决 ABA 问题（hazard pointers / RCU）
- 内存回收策略（epoch-based reclamation）
- 参考：concurrencykit (C)、boost::lockfree (C++)、crossbeam (Rust)IMPORTANT
本课是 lock-free 编程的"第一级台阶"。理解 SPSC 的索引设计、memory_order 语义 acquire-release 配对之后，才能安全地进入 MPMC 和更复杂的并发数据结构。建议在生产代码中直接使用成熟的库（而非自己实现），但在学习阶段必须亲手写一遍才能理解底层机制。
参考解答
TODO 1-2: init_buffer 和 print_buffer
c
#include <stdatomic.h>
#include <stdio.h>
#define BUFFER_SIZE 8
static _Atomic int buffer[BUFFER_SIZE];
static _Atomic int write_idx = 0;
static _Atomic int read_idx = 0;
/* TODO 1: 初始化环形缓冲区 */
static void init_buffer(void) {
for (int i = 0; i < BUFFER_SIZE; i++)
atomic_store_explicit(&buffer[i], -1, memory_order_relaxed);
atomic_store_explicit(&write_idx, 0, memory_order_relaxed);
atomic_store_explicit(&read_idx, 0, memory_order_relaxed);
}
/* TODO 2: 打印缓冲区状态 */
static void print_buffer(const char *label) {
printf("[%s] buf=[", label);
for (int i = 0; i < BUFFER_SIZE; i++) {
int val = atomic_load_explicit(&buffer[i], memory_order_relaxed);
if (i > 0) printf(" ");
if (val >= 0)
printf("%d", val);
else
printf(".");
}
printf("]");
}
int main(void) {
init_buffer();
print_buffer("init");
printf("\n"); /* 预期: [init] buf=[. . . . . . . .] */
return 0;
}要点：init_buffer 中所有初始化为 memory_order_relaxed——初始化发生在主线程开始之前，不存在并发竞争。print_buffer 中读 buffer 用 relaxed——打印不需要与任何其他线程的操作建立 happens-before 关系。
TODO 3: producer_write——生产者写入
c
#include <stdatomic.h>
#include <stdio.h>
#define BUFFER_SIZE 8
static _Atomic int buffer[BUFFER_SIZE];
static _Atomic int write_idx = 0;
static _Atomic int read_idx = 0;
/* TODO 3: 生产者写入 */
static void producer_write(int value) {
int w, r, slot;
/* 忙等待直到缓冲区非满 */
while (1) {
w = atomic_load_explicit(&write_idx, memory_order_relaxed);
r = atomic_load_explicit(&read_idx, memory_order_acquire);
if (w - r < BUFFER_SIZE) break;
}
slot = w % BUFFER_SIZE;
/* 写入数据到缓冲区（relaxed——稍后 release 保证可见） */
atomic_store_explicit(&buffer[slot], value, memory_order_relaxed);
/* 打印操作信息 */
printf(" [P] write %d at slot %d", value, slot);
/* 发布递增的 write_idx release 确保数据写入先于索引更新） */
atomic_store_explicit(&write_idx, w + 1, memory_order_release);
/* 打印操作后状态 */
print_buffer(" after write");
printf(" w=%d r=%d\n", w + 1, r);
}
/* 需要先定义 print_buffer（见 TODO 2 解答） */核心逻辑：
- 忙等待循环：每次迭代都重新加载
read_idx（用acquire），确保看到消费者最新释放 - 数据写入 relaxed：不依赖顺序——后面的
release会兜底保证可见性 - 索引更新 release：保证
buffer[slot] = value在write_idx = w+1之前对消费者可见
TODO 4: consumer_read——消费者读取
c
#include <stdatomic.h>
#include <stdio.h>
#define BUFFER_SIZE 8
static _Atomic int buffer[BUFFER_SIZE];
static _Atomic int write_idx = 0;
static _Atomic int read_idx = 0;
/* TODO 4: 消费者读取 */
static int consumer_read(void) {
int r, w, slot, val;
/* 忙等待直到缓冲区非空 */
while (1) {
r = atomic_load_explicit(&read_idx, memory_order_relaxed);
w = atomic_load_explicit(&write_idx, memory_order_acquire);
if (r < w) break;
}
slot = r % BUFFER_SIZE;
/* 读取数据 */
val = atomic_load_explicit(&buffer[slot], memory_order_relaxed);
/* 标记槽位为空 */
atomic_store_explicit(&buffer[slot], -1, memory_order_relaxed);
/* 打印操作信息 */
printf(" [C] read %2d at slot %d", val, slot);
/* 发布递增的 read_idx（release 确保消费完成先于索引更新） */
atomic_store_explicit(&read_idx, r + 1, memory_order_release);
/* 打印操作后状态 */
print_buffer(" after read ");
printf(" w=%d r=%d\n", w, r + 1);
return val;
}
/* 需先定义 print_buffer（见 TODO 2 解答） */核心逻辑：
- 忙等待循环：每次迭代都重新加载
write_idx（用acquire），确保看到生产者最新释放 - 读取 relaxed：因为用
acquire读了write_idx，生产者写入的数据已保证可见 - 清空槽位：消费后设为 -1，这是 print_buffer 正确输出的关键
- 索引更新 release：通知生产者"这个槽位已释放，可以写入了"
TODO 5: main——完整交错时间线
c
#include <stdatomic.h>
#include <stdio.h>
#define BUFFER_SIZE 8
#define TOTAL_ITEMS 10
static _Atomic int buffer[BUFFER_SIZE];
static _Atomic int write_idx = 0;
static _Atomic int read_idx = 0;
/* ─── 前面的 TODO 1-4 函数定义（略，见上方代码） ─── */
/* TODO 5: 主函数 — 交错时间线 */
int main(void) {
init_buffer();
printf("=== Lock-Free Ring Buffer (SPSC) ===\n");
printf("Capacity: %d, Total items: %d\n\n", BUFFER_SIZE, TOTAL_ITEMS);
printf("--- Interleaved Timeline ---\n");
/* Phase 1: 生产 5 个 (0..4) */
for (int i = 0; i < 5; i++)
producer_write(i);
/* Phase 2: 消费 3 个 (0..2) */
for (int i = 0; i < 3; i++) {
int v = consumer_read();
(void)v;
}
/* Phase 3: 生产 5 个 (5..9) — 8 和 9 回绕到槽 0 和 1 */
for (int i = 5; i < 10; i++)
producer_write(i);
/* Phase 4: 消费 7 个 (3..9) */
for (int i = 0; i < 7; i++) {
int v = consumer_read();
(void)v;
}
/* 最终结果汇总 */
int w = atomic_load_explicit(&write_idx, memory_order_relaxed);
int r = atomic_load_explicit(&read_idx, memory_order_relaxed);
printf("\n--- Final Results ---\n");
printf("Produced: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]\n");
printf("Consumed: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]\n");
printf("Final write_idx: %d\n", w);
printf("Final read_idx: %d\n", r);
printf("Buffer empty: %s\n", (w == r) ? "yes" : "no");
return 0;
}
/*
* 完整输出见 exercises.toml 中的 stdout 字段。
*
* 预期行为：
* - 20 步操作（生产 10 次 + 消费 10 次）
* - 第 12-13 步回绕：write_idx=8,9 写入 buffer[0],buffer[1]
* - 最终 write_idx=10, read_idx=10, 缓冲区空
*/核心要点：
- 交错顺序：先有生产才能消费——主函数中是单线程串行操作，不存在等待
- Phase 3 的关键：write_idx=8 时
8%8=0，写入 buffer[0]——这就是回绕 - 最终验证：write_idx == read_idx == 10 → 缓冲区空 ✓
对照检查：
init_buffer中用memory_order_relaxed了吗？print_buffer中val >= 0的判断正确吗（-1 打印 "."）？producer_write的满判断用w - r < BUFFER_SIZE（而非<=）了吗？consumer_read的释放用的memory_order_release（而非relaxed）吗？main 的 Phase 3 中 i 从 5 到 9（而非 0 到 4）了吗？
课堂讨论
- 如果把 BUFFER_SIZE 改为 4，在当前交错时间线（生产 5 个再消费 3 个）下程序会发生什么？为什么？
- 全部用
memory_order_seq_cst替代relaxed/release/acquire，程序行为会改变吗？性能会如何变化？ - 为什么 SPSC 只需要 load/store 而 MPMC 需要 CAS（compare-and-swap）？CAS 引入了什么新的复杂度？
- 消费者重复读
write_idx在弱内存模型（ARM）和强内存模型（x86）上的行为有何不同？如果写成while (r >= w)而不是while (r < w)会怎样？ - 假设生产者和消费者在不同的 CPU 核心上运行，
write_idx和read_idx都在同一缓存行——为什么即使它们从不互相访问也会降低性能？ - Node.js 的
fs.writeFileSync说是"同步"写入，但底层使用了异步缓冲——这和本课的环形缓冲有何相通之处？
讨论答案
Q1: BUFFER_SIZE=4 的死锁陷阱
把 BUFFER_SIZE 改为 4 后：
Phase 1: 生产 5 个 (0..4)
w=0: 写槽0, w=1
w=1: 写槽1, w=2
w=2: 写槽2, w=3
w=3: 写槽3, w=4
此时：w=4, r=0, w-r=4 == BUFFER_SIZE → 缓冲区满
w=4: 尝试写入 → 忙等待！等待 r 前进以腾出空间
但是！r=0 且 Phase 2（消费）还未开始！
→ 生产者在等待永远不可能发生的消费
→ 你的程序（在本课单线程模式下）进入死循环！
教训：
SPSC 的时序必须正确——生产者不能比消费者快太多。
在本课的交错设计中，Phase 3 之前必须消费足够的元素腾空间。
真实多线程场景中，这表现为"消费跟不上生产"——需要背压机制。Q2: seq_cst 的代价
用 seq_cst 替代所有 relaxed/release/acquire：
c
/* 版本 A: 本课推荐的 release/acquire */
atomic_store_explicit(&write_idx, w+1, memory_order_release);
// x86: 编译为普通 mov（x86 已有强内存模型）
// ARM: 编译为 stlr（store-release 指令）
// 开销: ~1-5 cycles
/* 版本 B: 全部用 seq_cst */
atomic_store_explicit(&write_idx, w+1, memory_order_seq_cst);
// x86: 可能需要 mfence 指令（全屏障）→ ~100 cycles
// ARM: 编译为 stl + dmb → ~30-50 cycles
// 开销: 10-100× 更高
// 程序正确性不变，但吞吐量可能下降 2-5×。
// 在本课教学场景（单线程交错）中无实际影响。seq_cst 是"安全但慢"的选择——适合调试阶段确认逻辑正确后再优化。
Q3: SPSC 为何不需要 CAS
c
/* SPSC：只有一个生产者写 write_idx → 无竞争 */
void producer_write_sp(int value) {
int w = atomic_load(&write_idx, relaxed);
buffer[w % N] = value; // 无竞争——只有我在写
atomic_store(&write_idx, w+1, release); // 无竞争——只有我在写
}
/* MPMC：多个生产者竞争 write_idx → 需要 CAS */
void producer_write_mp(int value) {
int w, new_w;
do {
w = atomic_load(&write_idx, relaxed);
new_w = w + 1;
} while (!atomic_compare_exchange_weak(
&write_idx, &w, new_w,
memory_order_release,
memory_order_relaxed
));
// ↑ 如果另一个线程在你读取 w 后抢先更新了 write_idx，
// CAS 会失败并重试——这是多生产者竞争的核心
buffer[w % N] = value; // 终于抢到了槽位
}
// MPMC 额外引入的问题：
// 1. ABA 问题：w 从 A→B→A，CAS 误以为"没变"
// 2. 内存回收：多个线程可能同时读同一个槽位——需要 RCU/Hazard Pointers
// 3. 公平性：CAS 重试循环可能导致某些线程饥饿Q4: 弱/强内存模型的行为差异
c
/* 消费者忙等待环 */
while (1) {
r = atomic_load_explicit(&read_idx, memory_order_relaxed);
w = atomic_load_explicit(&write_idx, memory_order_acquire);
if (r < w) break;
}
/*
* x86 (强内存模型 TSO):
* - acquire 读 ≈ 普通 mov 指令（编译器屏障即可）
* - 几乎不可能出现数据可见性问题（但不要依赖！）
*
* ARM (弱内存模型):
* - acquire 读 → ldar 指令（真正硬件屏障）
* - 如果没有 acquire，CPU 可能：
* 1. 先从缓存读到旧的 buffer[slot]
* 2. 再看到更新的 write_idx
* → 读到"未来的索引 + 过去的数据" → 数据竞争！
*
* 如果写成 while (r < w) 而非 while (r >= w):
* r < w 是正确条件（有数据可读）
* r >= w 是空的条件，然后 break —— 逻辑完全反了
*/Q5: 伪共享的量化影响
实验（伪共享影响依赖硬件，以下是典型值）：
SPSC 环，每秒 1000 万次操作：
- write_idx 和 read_idx 在同一缓存行： 8.5M ops/s
- write_idx 和 read_idx 在不同缓存行： 9.8M ops/s
- 差异：~15%
MPMC 环，4 生产者 + 4 消费者：
- 索引在同一缓存行： 3.2M ops/s
- 索引分离 + padding： 6.1M ops/s
- 差异：~90%
结论：SPSC 场景下伪共享有影响但不大（15%），
MPMC 场景下影响剧烈（90%）——因为每个生产者都写 write_idx，
缓存行频繁在所有核心间"弹跳"。Q6: fs.writeFileSync 与环形缓冲
fs.writeFileSync 的"同步"在 JS 层面——调用后阻塞 JS 事件循环直到完成。但在底层 C/C++ 实现中：
- 数据被拷贝到 libuv 的内部缓冲区（libuv 使用固定大小的 buffer 池，类似环形缓冲）
- libuv 在工作线程中执行实际的
write()syscall - JS 层同步等待工作线程完成
环形缓冲在这里的角色：
- 生产者：JS 线程写入数据到缓冲区
- 消费者：libuv 工作线程从缓冲区读取并通过 syscall 落盘
- 同步语义：JS 层"等待"就是忙等待/条件变量的工程化封装
本课的 SPSC 环展示了这种模式的最底层原理——去掉 JS、libuv、工作线程池等抽象层后，核心就是"一个写索引，一个读索引，环形内存"。
c
/*
* 概念连接：
*
* 本课 SPSC 环 Node.js writeFileSync
* ────────────────── ──────────────────────
* buffer[8] libuv req->bufs[] (缓冲区)
* producer_write(v) uv_buf_t 填入数据 (写入数据)
* consumer_read() uv__fs_work() (消费数据落盘)
* 忙等待 uv_cond_wait (同步等待)
* write_idx 递增 buf 偏移递增
* read_idx 递增 已写入字节数递增
*
* 本质相同——只是工程版本多了错误处理、线程管理、跨平台抽象。
*/课后练习
修改容量验证。将 BUFFER_SIZE 从 8 改为 4，同时调整主函数的交错时间线，确保程序不死锁——消费发生在生产填满缓冲区之前。
知识点提示：确保 Phase 1 的生产数量不超过 BUFFER_SIZE，或者在 Phase 1 和 Phase 3 之间插入足够的消费。
参考解答
c#define BUFFER_SIZE 4 #define TOTAL_ITEMS 8 int main(void) { init_buffer(); printf("=== Lock-Free Ring Buffer (SPSC) ===\n"); printf("Capacity: %d, Total items: %d\n\n", BUFFER_SIZE, TOTAL_ITEMS); printf("--- Interleaved Timeline ---\n"); /* Phase 1: 生产 3 个（不超过容量！） */ for (int i = 0; i < 3; i++) producer_write(i); /* Phase 2: 消费 2 个——腾空间 */ for (int i = 0; i < 2; i++) { int v = consumer_read(); (void)v; } /* Phase 3: 再生产 5 个（含回绕） */ for (int i = 3; i < 8; i++) producer_write(i); /* Phase 4: 消费剩余 6 个 */ for (int i = 0; i < 6; i++) { int v = consumer_read(); (void)v; } /* 最终状态 */ int w = atomic_load_explicit(&write_idx, memory_order_relaxed); int r = atomic_load_explicit(&read_idx, memory_order_relaxed); printf("\nFinal: w=%d r=%d empty=%s\n", w, r, (w == r) ? "yes" : "no"); return 0; }关键调整：Phase 1 只生产 3 个（< BUFFER_SIZE），Phase 2 消费 2 个腾空间，Phase 3 再生产 5 个。
用
seq_cst对比。将所有memory_order_relaxed、memory_order_release、memory_order_acquire替换为memory_order_seq_cst。程序行为改变了吗？在 x86 和 ARM 上分别会有什么不同？知识点提示：seq_cst 保证所有操作的全局总序——在正确性上等价于本课的 release/acquire 方案，但性能更差。实现两个版本并用
time命令对比（在 x86 上差距可能很小）。参考解答
c/* 将三个宏统一为 seq_cst */ /* 修改前 */ #define MO_RLX memory_order_relaxed #define MO_ACQ memory_order_acquire #define MO_REL memory_order_release /* 修改后——全部 seq_cst */ #define MO_ALL memory_order_seq_cst /* 代码中对 buffer 的读写都用 MO_ALL */ // atomic_store_explicit(&buffer[slot], value, MO_ALL); // atomic_load_explicit(&buffer[slot], MO_ALL); /* * 正确性：完全一致（seq_cst 是更强的保证） * 性能：在 x86 上差异约 5-10%（x86 本身内存模型较强）， * 在 ARM 上差异可能达 30-50%（seq_cst 需要完整 dmb 屏障） * * 验证方法： * $ gcc -std=c11 -O2 ringbuffer_seqcst.c -o rb_seq * $ time ./rb_seq > /dev/null * (单线程交错无意义——需要在真多线程 + 大批量下测试) */实现
is_empty()和is_full()辅助函数。编写两个函数，基于 write_idx 和 read_idx 判断缓冲区状态。考虑：在真多线程场景中，这些函数返回的结果是瞬时快照，可能在下一行代码执行时就过期了——为什么？知识点提示：在多线程环境中，
is_empty()返回true后，生产者在下一个指令周期可能就写入了新数据。返回值的有效期为零——只能在"此刻"信任它。参考解答
c/* 瞬时快照——返回值在返回时可能已过期 */ static int is_empty(void) { int w = atomic_load_explicit(&write_idx, memory_order_acquire); int r = atomic_load_explicit(&read_idx, memory_order_relaxed); return w == r; } static int is_full(void) { int w = atomic_load_explicit(&write_idx, memory_order_acquire); int r = atomic_load_explicit(&read_idx, memory_order_relaxed); return (w - r) >= BUFFER_SIZE; } /* * 为什么是瞬时快照？ * * 时间线： * T1: is_empty() → true （真多线程：消费者刚读完最后一个元素） * T2: producer_write(42) → 写入 （另一线程：生产者写入新数据） * T3: 回到 T1 的调用者，用 true 做决策 → 已过期！ * * 结论：这类函数只能用于调试/日志，不适用于同步控制逻辑。 * 同步逻辑必须用忙等待循环（每次迭代重新加载） * 或 atomic_compare_exchange 的 CAS 语义。 */实现
buffer_count()。返回当前缓冲区中可读元素的数量（write_idx - read_idx）。同理，返回的是瞬时快照——在多线程环境下，返回值可能立即过期。知识点提示：对两个原子变量做差值运算本身不是原子的——读取 write_idx 和 read_idx 之间存在时间窗口。
参考解答
cstatic int buffer_count(void) { int w = atomic_load_explicit(&write_idx, memory_order_acquire); int r = atomic_load_explicit(&read_idx, memory_order_relaxed); return w - r; } /* * 非原子性的根源： * * 读取 write_idx（时刻 T1）→ w = 5 * [时间窗口：消费者读了 2 个元素，r 从 3 → 5] * 读取 read_idx （时刻 T2）→ r = 5 * 返回值：5 - 5 = 0 * 而实际缓冲区在 T1 时刻有 2 个可读，T2 时刻为 0 * * 由于对两个独立原子变量的读取不是原子的， * 这个复合操作\"看到\"的是两个不同瞬间的状态混合。 * * 这在实际中通常可接受——只要能容忍瞬时误差。 */真多线程版。将生产者和消费者分别放入两个线程（
pthread_create）。生产 1000 个元素，消费 1000 个。验证：消费到的序列是 0..999 吗？如果有乱序，是什么原因？知识点提示：用
pthread创建线程，用pthread_join等待完成。注意：多线程下printf本身不是线程安全的——输出可能交错。如果需要干净输出，用pthread_mutex_t保护printf（这把锁用于保护外围 I/O，不影响核心环形缓冲的无锁设计）。参考解答
c#include <stdatomic.h> #include <stdio.h> #include <pthread.h> #define BUFFER_SIZE 16 #define TOTAL_ITEMS 1000 static _Atomic int buffer[BUFFER_SIZE]; static _Atomic int write_idx = 0; static _Atomic int read_idx = 0; static void producer_write(int value) { while (1) { int w = atomic_load_explicit(&write_idx, memory_order_relaxed); int r = atomic_load_explicit(&read_idx, memory_order_acquire); if (w - r < BUFFER_SIZE) break; /* 可选：sched_yield() 或短 sleep 降低 CPU 空转 */ } int slot = atomic_load_explicit(&write_idx, memory_order_relaxed) % BUFFER_SIZE; atomic_store_explicit(&buffer[slot], value, memory_order_relaxed); atomic_fetch_add_explicit(&write_idx, 1, memory_order_release); } static int consumer_read(void) { while (1) { int r = atomic_load_explicit(&read_idx, memory_order_relaxed); int w = atomic_load_explicit(&write_idx, memory_order_acquire); if (r < w) break; } int slot = atomic_load_explicit(&read_idx, memory_order_relaxed) % BUFFER_SIZE; int val = atomic_load_explicit(&buffer[slot], memory_order_relaxed); atomic_store_explicit(&buffer[slot], -1, memory_order_relaxed); atomic_fetch_add_explicit(&read_idx, 1, memory_order_release); return val; } static void *producer_thread(void *arg) { (void)arg; for (int i = 0; i < TOTAL_ITEMS; i++) producer_write(i); return NULL; } static void *consumer_thread(void *arg) { (void)arg; for (int i = 0; i < TOTAL_ITEMS; i++) { int v = consumer_read(); if (v != i) { printf("ORDER ERROR: expected %d, got %d\n", i, v); } } return NULL; } int main(void) { pthread_t prod, cons; pthread_create(&prod, NULL, producer_thread, NULL); pthread_create(&cons, NULL, consumer_thread, NULL); pthread_join(prod, NULL); pthread_join(cons, NULL); printf("Done. write_idx=%d read_idx=%d\n", atomic_load(&write_idx), atomic_load(&read_idx)); return 0; }SPSC 保证顺序（单生产者单消费者），消费者应该收到
0, 1, 2, ..., 999的严格顺序。如果出现乱序，检查是否存在数据竞争（忘了用 acquire/release）。
参考资料
- ISO/IEC 9899:2011, Section 7.17 —
<stdatomic.h>specification - Preshing, J. An Introduction to Lock-Free Programming — https://preshing.com/20120612/an-introduction-to-lock-free-programming/
- Preshing, J. Memory Ordering at Compile Time — https://preshing.com/20120625/memory-ordering-at-compile-time/
- Preshing, J. Memory Barriers: a Hardware View for Software Hackers — https://preshing.com/20120710/memory-barriers-are-like-source-control-operations/
- McKenney, P. E. Is Parallel Programming Hard, And, If So, What Can You Do About It? — Appendix C (Memory Barriers)
- Williams, A. C++ Concurrency in Action, 2nd ed., Chapter 5 (The C++ Memory Model)
- Linux kernel:
include/linux/kfifo.h— lock-free ring buffer implementation - LMAX Disruptor: https://lmax-exchange.github.io/disruptor/
- DPDK Ring Library: https://doc.dpdk.org/guides/prog_guide/ring_lib.html
- libuv
fs.c— Node.js file system operations backingfs.writeFileSync man 3 atomic— C11 atomic operations overview
"If you don't understand memory ordering, your lock-free code is wrong. If you think you do, it's probably still wrong." — Anonymous Concurrency Developer