3.11 并发与同步

前面十章里，「锁」这个词出现了几十次：zone->lock、mmap_lock、rcu_read_lock、node->list_lock。这一章把它讲透。

内核并发比用户态难，因为「同时」有五个来源，而且它们可以叠加：

1. 多 CPU：真并行，两个核同时执行内核代码；
2. 中断：打断当前代码，在同一个 CPU 上「插队」；
3. 软中断/tasklet：在中断返回路径上执行，仍可能抢占进程上下文；
4. 抢占：CONFIG_PREEMPT 下，内核代码随时可能被换出；
5. 内核线程：workqueue、kswapd 等在各 CPU 上并发跑。

一个 spin_lock 只挡住「其他 CPU 上的 spin_lock」，挡不住本地中断（除非用 spin_lock_irqsave）；一个 mutex 只挡住「会睡眠的路径」，挡不住中断上下文（中断里根本不能拿 mutex）。选错锁的后果不是「慢一点」，而是死锁、数据损坏、随机 panic。

本章的主线是按场景选工具：先讲并发来源与竞争的本质，再讲原子操作与屏障（最底层），然后自旋锁、睡眠锁、RCU（三种典型场景），最后是调试。奔跑吧第 4 章是主要母本，OSTEP 第 28–32 章给理论对照。

3.11.1 竞争从哪里来：一个计数器引发的血案

先看一个「看起来没问题」的代码：

/* 错误示范：无保护的引用计数 */
static int refcnt;

static void get_obj(void)
{
	refcnt++;          /* 编译成 load; add; store 三条指令 */
}

static void put_obj(void)
{
	if (--refcnt == 0)
		free_obj();
}

refcnt++ 不是原子操作。两个 CPU 同时执行时可能交错：

CPU0: load  refcnt (=1)          CPU1: load  refcnt (=1)
CPU0: add   1        (=2)        CPU1: add   1        (=2)
CPU0: store refcnt = 2           CPU1: store refcnt = 2
结果：两次 get，计数只加了 1 → 对象被提前释放 → use-after-free

这类错误在内核里叫 race condition，它的特点是「平时不出现，压力大时随机崩」。修复有三种层次：

选型的第一个问题是：临界区里能不能睡眠？能，就用睡眠锁；不能（中断上下文、持 spinlock 时），只能用自旋锁或原子操作。

3.11.2 原子操作与内存屏障：CPU 的重排序

原子操作

atomic_t 是一组「读-改-写」原子接口，底层是带 lock 前缀的指令（x86）或 ldxr/stxr 循环（ARM64）：

atomic_t counter = ATOMIC_INIT(0);

atomic_inc(&counter);              /* counter++，原子 */
int v = atomic_read(&counter);     /* 普通读，不加锁 */
atomic_set(&counter, 10);
int old = atomic_xchg(&counter, 20);
int prev = atomic_cmpxchg(&counter, 20, 30);  /* 相等则换，返回旧值 */

refcount_t 是「带溢出保护的引用计数」，2017 年之后内核强制用它替代 atomic_t 做引用计数：

refcount_t refs = REFCOUNT_INIT(1);
refcount_inc(&refs);               /* 从 0 增加会 WARN 并拒绝 */
if (refcount_dec_and_test(&refs))
	free_obj();                /* 归零返回 true */

它比 atomic_t 多做了两件事：溢出时饱和（不会回绕到 0）、从 0 增加时报警（说明有 use-after-free）。代价是多几条指令，值得。

内存屏障

原子性解决「不丢更新」，但解决不了「顺序」。CPU 和编译器都会重排内存访问：

/* 生产者 */
data = 42;
ready = 1;          /* 消费者看到 ready==1 时，data 一定是 42 吗？ */

/* 消费者 */
while (!ready)
	;
print(data);        /* 可能是 0！ */

在 x86 的 TSO 模型下，store-store 不会重排，所以这段在 x86 上「通常」正确；在 ARM64 的弱内存模型下，ready = 1 可能先于 data = 42 可见，消费者读到 0。编译器也可能把 data 的写挪到 ready 之后（它不知道别的 CPU 在看）。

正确的写法：

/* 生产者 */
WRITE_ONCE(data, 42);
smp_store_release(&ready, 1);      /* 保证之前的写先可见 */

/* 消费者 */
while (!smp_load_acquire(&ready))
	;
print(READ_ONCE(data));            /* 一定看到 42 */

READ_ONCE/WRITE_ONCE 只阻止编译器优化（把访问合并、删除、重排），不产生硬件屏障；smp_load_acquire/smp_store_release 同时提供编译器与硬件语义（x86 上 release 就是普通 store，acquire 就是普通 load，因为 TSO 已经保证；ARM64 上是 ldar/stlr）。

屏障家族：

3.11.3 自旋锁：短临界区的忙等

spinlock_t 是「拿不到就原地转」的锁，临界区必须极短，且绝对不能睡眠。

/* include/linux/spinlock.h 的典型用法 */
spinlock_t lock;

spin_lock(&lock);
/* 临界区：几条指令，不调用可能睡眠的函数 */
spin_unlock(&lock);

/* 如果临界区会被中断处理程序访问，必须关本地中断 */
unsigned long flags;
spin_lock_irqsave(&lock, flags);
...
spin_unlock_irqrestore(&lock, flags);

/* 如果只和软中断竞争 */
spin_lock_bh(&lock);
...
spin_unlock_bh(&lock);

三种变体的区别：

选错的经典后果是死锁：进程上下文拿了 spin_lock，本地中断来了，中断处理程序也去拿同一把锁——同一个 CPU 上自旋，永远拿不到。spin_lock_irqsave 就是为这个场景准备的。

6.x 的 spinlock_t 底层是 qspinlock（队列自旋锁，4.2+）：每个等待者在 per-CPU 的 MCS 队列节点上自旋，而不是所有人抢同一个 cache line。这解决了「锁释放时所有 CPU 同时抢总线」的惊群问题。CONFIG_QUEUED_SPINLOCKS 默认开。

spin_lock 内部会 preempt_disable()，所以临界区里不能调用可能睡眠的函数（might_sleep() 会报警）。等待期间 CPU 空转（cpu_relax() 里是 pause 指令），所以临界区超过几微秒就该考虑换锁。

一个用 seqlock 的读侧：

/* 读侧：无锁，检查序号重试 */
u64 read_counter(struct my_dev *dev)
{
	unsigned int seq;
	u64 val;

	do {
		seq = read_seqbegin(&dev->seq);
		val = dev->counter;
	} while (read_seqretry(&dev->seq, seq));
	return val;
}

/* 写侧：加锁、更新、递增序号 */
void update_counter(struct my_dev *dev, u64 v)
{
	write_seqlock(&dev->seq);
	dev->counter = v;
	write_sequnlock(&dev->seq);
}

read_seqbegin() 里含 smp_rmb()，read_seqretry() 里含 smp_mb()；写侧在修改前后各递增一次序号（变成奇数→偶数），读侧看到奇数或变化就重试。它的前提是「读侧可以重复读」——如果读操作有副作用，不能用。

3.11.4 睡眠锁：mutex、semaphore、completion

临界区长、或者需要睡眠（分配内存、等 I/O、拷贝用户数据）时，用睡眠锁。它们在内核里由 struct mutex、struct semaphore、struct completion 实现，共同点是拿不到就睡眠，因此只能在进程上下文使用。

mutex 的几个硬规则：

• 不能递归：同一个任务重复 mutex_lock() 会死锁（CONFIG_DEBUG_MUTEXES 会检测）。
• 不能用于中断上下文：mutex_lock() 可能睡眠。
• 持有期间不能 kfree() 掉锁本身，也不能让锁所在的内存失效。
• mutex_unlock() 必须由持有者调用（内核会检查 owner）。

mutex 在 6.x 有一个优化：CONFIG_MUTEX_SPIN_ON_OWNER 允许在「owner 正在别的 CPU 上跑」时先自旋一会儿，避免立即睡眠的上下文切换开销。这是「乐观自旋」：如果 owner 很快释放，自旋比睡眠快；如果 owner 被抢占，自旋会浪费 CPU，所以有阈值。

completion 是最轻的「等一个事件」：

struct completion done;
init_completion(&done);

/* 等待方 */
wait_for_completion(&done);                 /* 睡眠直到 complete */
wait_for_completion_timeout(&done, HZ);     /* 带超时，返回剩余 jiffies */

/* 通知方（可以在中断上下文） */
complete(&done);                            /* 唤醒一个等待者 */
complete_all(&done);                        /* 唤醒所有，且之后 wait 立即返回 */

complete() 可以在中断上下文调用（它只用 spinlock），这是它比 mutex 更适合「中断通知进程」的原因。

3.11.5 RCU：读侧零开销的同步

RCU（Read-Copy-Update）解决的是「读多写极少」的场景：链表、路由表、配置、struct file 的 f_op。它的核心思想是写侧复制、延迟释放：

1. 写侧不修改正在被读的数据，而是复制一份、改副本、把指针原子地换过去；
2. 旧数据不能立刻释放——可能还有读者在用；
3. 等所有「替换前就开始读」的读者都结束后，再释放旧数据。这段等待叫宽限期（grace period）。

/* 读侧：不需要锁，只需要 RCU 读锁（在 PREEMPT_RCU 下只是标记） */
rcu_read_lock();
p = rcu_dereference(global_ptr);
if (p)
	use(p);                 /* 不能睡眠，不能交出 p 的所有权 */
rcu_read_unlock();

/* 写侧：分配新对象、改副本、发布、等待、释放 */
new = kmalloc(...);
*new = *old;
new->field = value;
rcu_assign_pointer(global_ptr, new);   /* 发布：release 语义 */
synchronize_rcu();                      /* 等所有旧读者退出 */
kfree(old);

时间 ─────────────────────────────────────────────────────────►
       │ 写侧：rcu_assign_pointer(p, new)
       │           │
       │           ├── 宽限期开始（所有 CPU 记录当前状态）
       │           │
  读者 A ─────────┼──── 读 old ────┐
       │           │                │ 读锁释放 → quiescent state
  读者 B ─────────┼──────── 读 old ────────┐
       │           │                        │
  读者 C ─────────────────┼── 读 new（看到新指针）
       │           │                        │
       │           ▼                        ▼
       │      synchronize_rcu() 返回（所有旧读者都退出）
       │           │
       │           └── kfree(old) 安全

读图：宽限期不是「固定时长」，而是「等所有在替换前进入读侧的 CPU 都经过一次静止状态（quiescent state）」。静止状态包括：上下文切换、返回用户态、进入 idle、显式 rcu_quiescent_state()。所以一个「一直占着 CPU 不切换、也不进内核」的读者会拖住宽限期——这是 rcu stall 的常见原因。

RCU 的 API 分两类：

rcu_read_lock() 在 6.8 的 CONFIG_PREEMPT_RCU 下不关抢占，只是给当前任务加一个「在读侧」的标记；读侧可以被抢占，甚至可以在读侧睡眠吗？不行——被抢占可以，主动睡眠（schedule() 以外的阻塞）会让宽限期无法推进，且 rcu_read_unlock() 可能在不同 CPU 上执行。需要睡眠的读侧用 SRCU（sleepable RCU）。

RCU 的变体：

写侧要小心「发布顺序」：rcu_assign_pointer() 之前对 new 的所有初始化必须完成，否则读者可能看到半成品。rcu_assign_pointer() 内含 release 屏障，rcu_dereference() 内含 acquire 依赖屏障，两者配对。

3.11.6 per-CPU 与无锁编程

如果数据本来就是「每个 CPU 一份」，根本不需要锁：

/* 定义 per-CPU 变量 */
static DEFINE_PER_CPU(u64, packet_count);

/* 本 CPU 递增：单条指令，无锁 */
this_cpu_inc(packet_count);

/* 读本 CPU 的值：需要关抢占，防止迁移到别的 CPU */
u64 n = this_cpu_read(packet_count);

/* 读所有 CPU 的总和 */
u64 total = 0;
int cpu;
for_each_possible_cpu(cpu)
	total += per_cpu(packet_count, cpu);

this_cpu_* 系列在 x86 上编译成「带 %gs 段前缀的指令」，单条指令完成，比「关抢占 + 取指针 + 操作 + 开抢占」快得多。代价是「读总和」要遍历所有 CPU，且结果不是某一时刻的快照（各 CPU 并发更新）。

get_cpu_var()/put_cpu_var() 是「关抢占 + 取引用」的组合；6.x 推荐用 local_lock_t（5.16+）表达「per-CPU + 需要保护」的场景：

static DEFINE_PER_CPU(struct stats, stats);
static DEFINE_PER_CPU(local_lock_t, stats_lock);

local_lock(&stats_lock);
this_cpu_inc(stats.packets);
local_unlock(&stats_lock);

local_lock 在非 RT 内核上退化为「关抢占」，在 PREEMPT_RT 上变成真正的 per-CPU 睡眠锁——同一份代码在两种内核上都正确，这是它被推荐的原因。

无锁编程的其他工具：

无锁不等于「没有同步」：它把锁换成了「原子操作 + 内存序 + 重试」，正确性更难保证。内核的原则是「除非有明确收益（热路径、中断上下文），否则用锁」。

3.11.7 死锁、优先级反转与锁调试

四种典型死锁

优先级反转是实时系统的经典问题：低优先级任务 L 持有锁，高优先级任务 H 等锁；此时中优先级任务 M 就绪，抢占了 L，导致 H 被 M 间接阻塞。解法是优先级继承：H 等锁时，L 临时提升到 H 的优先级，尽快执行完释放锁。rt_mutex 实现这套机制，PREEMPT_RT 把所有 spinlock 换成 rt_mutex。

lockdep：在死锁发生前报警

lockdep（CONFIG_PROVE_LOCKING）不需要真的死锁就能发现「可能死锁」：它记录每把锁的获取顺序，构建一张「锁类依赖图」，一旦发现环就报警。

# 打开 lockdep 后，内核会打印类似报告：
======================================================
WARNING: possible circular locking dependency detected
------------------------------------------------------
kworker/0:1/123 is trying to acquire lock:
ffff888012345678 (&b->lock){+.+.}-{3:3}, at: func_b+0x1a/0x40

but task is already holding lock:
ffff888087654321 (&a->lock){+.+.}-{3:3}, at: func_a+0x2b/0x50

other info that might help us debug this:
 Possible unsafe locking scenario:
       CPU0                    CPU1
       ----                    ----
  lock(&a->lock);
                               lock(&b->lock);
                               lock(&a->lock);
  lock(&b->lock);

读 lockdep 报告的三步：① 看「谁在等谁」（holding / trying to acquire）；② 看 Possible unsafe locking scenario 的交叉顺序；③ 找代码里对应的加锁点，统一锁序或改用 ww_mutex。

配套的调试设施：

hung task 报告长这样：

INFO: task java:1234 blocked for more than 120 seconds.
      Not tainted 6.8.0 #1
"echo 0 > /proc/sys/kernel/hung_task_timeout_secs" disables this message.
task:java            state:D stack:0     pid:1234  ppid:1      flags:0x00004002
Call Trace:
 __schedule+0x2c3/0x8a0
 schedule+0x5e/0xe0
 io_schedule+0x46/0x70
 ...

state:D（不可中断睡眠）+ 栈顶是 io_schedule，说明卡在 I/O；栈顶是 mutex_lock，说明等锁。sysrq-w 能一次性打印所有 D 状态任务，是「机器卡住但还能响应 sysrq」时的第一手材料。

mm 的锁矩阵：一把锁保护什么

顺序规则：先 mmap_lock 再 pte_lock，先 PG_locked 再 zone->lock；反向持锁是 lockdep 报告的典型 ABBA。mmap_lock 本身仍是 rwsem；6.4 引入的 per-VMA lock 是缺页路径的补充，读缺页可以只锁 VMA，但不是替换 mmap_lock。

3.11.8 锁的结构体：字段、缓存行与所有权

3.11.3–3.11.5 讲了“用哪种锁”，这一节讲“锁里面是什么”。字段布局决定了三件事：锁占几个字节、争用时谁在哪个 cache line 上自旋、调试信息从哪里来。

spinlock：从 ticket 到 qspinlock

/* include/linux/spinlock_types.h —— 6.8 节选 */
typedef struct spinlock {
	union {
		struct raw_spinlock rlock;
#ifdef CONFIG_DEBUG_LOCK_ALLOC
		struct {
			u8 __padding[LOCK_PADSIZE];
			struct lockdep_map dep_map;
		};
#endif
	};
} spinlock_t;

typedef struct raw_spinlock {
	arch_spinlock_t raw_lock;
#ifdef CONFIG_DEBUG_SPINLOCK
	unsigned int magic, owner_cpu;
	void *owner;
#endif
} raw_spinlock_t;

x86 的 arch_spinlock_t 是 4 字节的 qspinlock：

/* arch/x86/include/asm/qspinlock_types.h —— 6.8 */
typedef struct qspinlock {
	union {
		atomic_t val;
		struct {
			u8 locked;
			u8 pending;
		};
		struct {
			u16 locked_pending;
			u16 tail;
		};
	};
} arch_spinlock_t;

qspinlock 的关键是 tail：每个等待者在自己的 per-CPU MCS 节点上自旋，而不是所有人抢同一个 cache line。锁释放时只需把 locked 清掉并唤醒队头，避免“惊群”式总线风暴。4.2 之前是 ticket spinlock，所有人在同一个 next/owner 上自旋；核数一多，释放锁的那次写会让所有等待者的 cache line 同时失效。

CONFIG_DEBUG_SPINLOCK 会在锁后面追加 magic/owner_cpu/owner，用来抓“未初始化”“重复解锁”“非持有者解锁”。这些字段让 spinlock_t 从 4 字节膨胀到几十字节，生产内核通常关闭。

mutex：owner + 等待队列 + 乐观自旋

/* include/linux/mutex.h —— 6.8 节选 */
struct mutex {
	atomic_long_t		owner;		/* 低 3 位是 flags，高位是 task 指针 */
	raw_spinlock_t		wait_lock;
#ifdef CONFIG_MUTEX_SPIN_ON_OWNER
	struct optimistic_spin_queue osq;	/* 乐观自旋队列 */
#endif
	struct list_head	wait_list;
#ifdef CONFIG_DEBUG_MUTEXES
	void			*magic;
#endif
};

owner 的低位编码是 mutex 实现里最绕的部分：MUTEX_FLAG_WAITERS 表示有人在等，MUTEX_FLAG_HANDOFF 表示解锁时直接把锁交给队头，MUTEX_FLAG_PICKUP 表示队头已接手。mutex_unlock() 看到 WAITERS 会走慢路径，唤醒队头并可能设置 HANDOFF。读懂这三个 flag，才能解释“为什么 mutex 的解锁有时比加锁还慢”。

rwsem、semaphore、completion

rw_semaphore 的 count 不是简单的“读者数”：最高位表示写者持有/等待，低位是读者计数。down_read() 在写者等待时会排队（RWSEM_WAITING_BIAS），这是“写者优先”的实现方式。读多写少且写者可能饿死的场景，应该用 RCU 或 seqlock，而不是 rwsem。

锁与 cache line

一个 spinlock_t 只有 4 字节，但它所在的 cache line 可能还放着别的热数据。两个 CPU 争同一把锁时，cache line 在两者之间来回弹（cache line bouncing），每次弹的成本是几十到几百纳秒。两个工程手段：

1. __cacheline_aligned：把锁单独放到一个 cache line，避免与数据互相失效。
2. per-CPU 分片：每个 CPU 一把锁 + 一份数据，读总和时再合并（3.11.6 的 per-CPU 思路）。

pahole -C mutex vmlinux 能看到锁的实际大小与对齐；perf c2c 能抓到 cache line 争用（需要硬件支持）。

3.11.9 原子操作与内存序：从 API 到 litmus test

3.11.2 给了 READ_ONCE/smp_store_release 的最小用法。这一节把原子 API 补齐，并用两个 litmus test 说明“为什么 x86 上能跑、ARM64 上会挂”。

原子 API 的分层

atomic_t 的实现是 typedef struct { int counter; } atomic_t;，所有操作通过 arch_atomic_* 落到 lock 前缀指令（x86）或 ldxr/stxr 循环（ARM64）。atomic_read() 是普通读，不带屏障；需要 acquire 语义时用 atomic_read_acquire() 或读后加 smp_mb()。

refcount 的饱和语义

refcount_t refs = REFCOUNT_INIT(1);

refcount_inc(&refs);            /* 1 -> 2 */
if (refcount_dec_and_test(&refs))
	free_obj();             /* 2 -> 1 -> 0，返回 true */
if (!refcount_inc_not_zero(&refs))
	/* 已经是 0：对象正在销毁，不能再拿引用 */

refcount_t 比 atomic_t 多做的保护：

代价是每条操作多几条指令与分支。引用计数一律用 refcount_t，不要用 atomic_t——这是 2017 年之后内核的硬规则。

CAS 循环的写法

无锁更新用 cmpxchg 循环，模板是“读—算—试”：

/* 把 obj->state 从 OLD 改成 NEW，失败则重读重试 */
static bool try_transition(struct obj *o, int old, int new)
{
	int cur = atomic_read(&o->state);

	do {
		if (cur != old)
			return false;
	} while (!atomic_try_cmpxchg(&o->state, &cur, new));
	return true;
}

atomic_try_cmpxchg() 与 atomic_cmpxchg() 的区别：前者在失败时把 cur 更新为当前值，循环里不用再读一次；后者只返回旧值。x86 上两者都编译成 lock cmpxchg，差别在循环体的指令数。

CAS 循环的三个陷阱：

1. ABA：值从 A 变 B 又变回 A，CAS 会成功，但中间状态已经丢了。解法是加版本号/标记位（3.11.11）。
2. 活锁：高争用时所有线程都在重试，CPU 空转。解法是退避（cpu_relax()、指数退避）或换锁。
3. 内存序：cmpxchg 默认是 full barrier；atomic_try_cmpxchg_acquire/release 可以降序，但要与另一侧配对。

两个 litmus test

消息传递（store buffer）：

/* CPU0 */                    /* CPU1 */
data = 42;                    while (!flag)
smp_store_release(&flag, 1);      ;
                              r = smp_load_acquire(&data);

IRIW（独立读独立写）：

/* CPU0: */ x = 1; smp_wmb(); 
/* CPU1: */ y = 1; smp_wmb();
/* CPU2: */ r1 = y; smp_rmb(); r2 = x;
/* CPU3: */ r3 = x; smp_rmb(); r4 = y;

在弱内存模型下，CPU2 可能看到 y=1, x=0，CPU3 同时看到 x=1, y=0——两个读者对写顺序的判断相反。x86 的 TSO 不允许这种结果；ARM64 需要 smp_mb() 而不是 smp_rmb()/smp_wmb() 才能排除。

屏障矩阵

“x86 不需要屏障”是半真半假：TSO 确实保证 LoadLoad、LoadStore、StoreStore，但 StoreLoad 仍可重排。所以“写标志 → 读数据”的 Dekker 型代码在 x86 上也需要 mfence 或 lock 前缀。

KCSAN 与 data_race

数据竞争不一定会崩，可能只是读到撕裂值。CONFIG_KCSAN 用编译器插桩检测“无标记的并发访问”，报告里会指出两个访问点。对于“故意不加锁、但能接受撕裂”的统计字段，用 data_race() 标注，告诉 KCSAN 这是有意为之：

/* 统计字段：允许撕裂，只要最终一致 */
data_race(stat->packets++);

data_race() 不是屏障，也不保证原子性；它只是让工具闭嘴。真正的共享状态仍要用 READ_ONCE/WRITE_ONCE 或锁。

3.11.10 RCU 的宽限期：等谁、怎么等、等不到怎么办

3.11.5 讲了 RCU 的“复制—替换—延迟释放”。这一节回答三个更底层的问题：宽限期（grace period, GP）由谁推进？call_rcu() 的回调在哪里执行？宽限期卡住时内核怎么报警？

Tree RCU 的三层结构

rcu_state（全局）
├── rcu_gp_kthread      推进 GP 的内核线程
├── rcu_gp_seq          GP 序号（每完成一次 +4）
├── node[]              rcu_node 树（按 CPU 数分层）
│    └── rcu_node       每个节点负责一组 CPU 的 QS 汇总
└── rcu_data（per-CPU）
     ├── cblist          回调链表（分段：NORMAL/WAIT/DONE）
     ├── gp_seq          本 CPU 看到的 GP 序号
     └── qs_pending      是否还未报告静止状态

GP 的推进不是“等固定时间”，而是自底向上收集静止状态（quiescent state, QS）：每个 CPU 报告一次 QS，rcu_node 汇总，根节点收齐后 rcu_gp_kthread 推进序号。

什么算静止状态

关键：QS 不是“读锁释放”。rcu_read_unlock() 只是减少嵌套计数；真正让 GP 前进的是“这个 CPU 经过了一个不可能持有读锁的状态”。所以一个在用户态死循环、从不进内核的进程，不会阻止 GP；一个在内核里关抢占死循环的 CPU，会。

call_rcu 的回调路径

/* kernel/rcu/tree.c: call_rcu() —— 6.8 节选，已简化 */
void call_rcu(struct rcu_head *head, rcu_callback_t func)
{
	unsigned long flags;
	struct rcu_data *rdp;

	head->func = func;
	head->next = NULL;
	local_irq_save(flags);
	rdp = this_cpu_ptr(&rcu_data);
	rcu_segcblist_enqueue(&rdp->cblist, head);   /* 入本 CPU 回调链表 */
	if (!rcu_segcblist_is_offloaded(&rdp->cblist))
		invoke_rcu_core();                   /* 唤醒 GP 线程/软中断 */
	local_irq_restore(flags);
}

回调执行：

rcu_core() / rcu_do_batch()
 ├─ 取 rdp->cblist 里已经过 GP 的回调
 ├─ 逐个调用 head->func(head)
 └─ 超过预算（blimit）就放回，等下一轮

call_rcu() 可以在原子上下文调用（它只关本地中断、入链表），这是它比 synchronize_rcu() 更适合中断/软中断的原因。synchronize_rcu() 会睡眠等 GP 完成，只能在进程上下文用。

宽限期的四个变体

rcu_barrier() 不是“再等一个 GP”，而是“等所有 CPU 的回调链表清空”。模块卸载时如果只调 synchronize_rcu()，仍可能有 call_rcu() 的回调在排队并引用模块代码；必须 rcu_barrier()。

回调卸载（NOCB）

默认回调在软中断/rcuc/N 线程里执行。实时场景希望把回调从隔离核上挪走：

rcu_nocbs=2,3        # 2/3 号 CPU 的回调交给 rcuo/N 线程
rcu_nocb_poll        # rcuo 线程轮询，而不是等 IPI

对应的内核线程：rcuop/N（offload 回调）、rcuog/N（GP）、rcuoc/N（callback）。它们默认绑定到 housekeeping CPU，隔离核上不再跑 RCU 回调。

stall：宽限期为什么卡住

rcu_cpu_stall_timeout 默认 21 秒。超过后打印：

INFO: rcu_preempt detected stalls on CPUs/tasks:
	Tasks blocked on level-0 rcu_node (CPUs 0-7): P1234
	(detected by 0, t=21002 jiffies, g=12345, q=678)

常见原因：

rcu stall 不是“RCU 坏了”，而是“有 CPU/任务没有经过静止状态”。先找那个 CPU，而不是调 stall_timeout。

3.11.11 无锁队列与内存回收：ABA、hazard pointer 与 kfifo

“无锁”不是“没有同步”，而是把锁换成了“原子操作 + 内存序 + 重试 + 回收协议”。这一节讲四种常见结构，以及它们各自的安全边界。

四种无锁结构

kfifo 是内核里最常用的无锁队列：生产者只写 in，消费者只写 out，两者用 smp_wmb()/smp_rmb() 配对，避免索引先于数据可见。它不支持多生产者——多生产者要先加锁或用 kfifo 的 record 接口。

llist（lock-less list）只保证“头插”的原子性：llist_add() 用 cmpxchg 把新节点接到头部，llist_del_all() 一次性摘下整条链。它适合“生产者只增、消费者批量取”的场景（如 task_work、irq_work），不适合任意删除。

ABA：CAS 循环的经典漏洞

Treiber 栈的 pop：

old = head;                 // 读到 A
next = old->next;           // 读到 B
CAS(&head, old, next);      // 期望 A，写入 B

如果在这三步之间，另一个线程把 A、B 都弹出，又把 A 压回，head 又等于 A，CAS 会成功——但 B 已经不在栈里（甚至已被释放）。结果 head 指向一个游离节点，栈结构损坏。这就是 ABA。

三种防护：

内核里 kfifo 不需要 ABA 防护（索引只增不减），llist 靠“不删除”回避，Treiber 栈在用户态库（如 liburcu）里才常见。写内核代码时，优先用 kfifo/llist/RCU，而不是自己实现一个带回收的无锁栈。

回收协议对比

3.11.12 优先级反转与优先级继承：rt_mutex 的链式调整

优先级反转不是“低优先级任务变慢”，而是“高优先级任务被中优先级任务间接阻塞”。这一节走一遍 PI 链的调整过程，并说明为什么只有 rt_mutex 能解决它。

反转的时间线

时刻    L（低优先级，持锁）   M（中优先级）        H（高优先级）
t0      持锁运行
t1                           就绪，抢占 L
t2                                              就绪，等锁 → 阻塞
t3                           继续运行（H 被 M 间接阻塞）
t4                           让出
t5      L 继续，释放锁
t6                                              H 拿到锁

H 的等待时间 = L 剩余临界区 + M 的运行时间。M 与这把锁无关，却决定了 H 的延迟——这就是“反转”。

PI 的调整过程

优先级继承（priority inheritance, PI）的规则：当 H 等 L 持有的锁时，L 临时提升到 H 的优先级，直到释放锁。

rt_mutex_adjust_prio_chain(task, ...)
 ├─ 1. 找到 task 正在等的 rt_mutex（task->pi_blocked_on）
 ├─ 2. 找到该锁的 owner
 ├─ 3. 若 owner 的优先级低于 waiter，提升 owner
 │      （rt_mutex_setprio()，参数是 owner 与 waiter 任务）
 ├─ 4. owner 自己也可能在等另一把锁 → 沿链继续
 └─ 5. 链上出现环 → 死锁，打印/返回 -EDEADLK

相关结构：

/* kernel/locking/rtmutex.c —— 6.8 节选 */
struct rt_mutex_base {
	raw_spinlock_t		wait_lock;
	struct rb_root_cached	waiters;	/* 按优先级排序 */
	struct task_struct	*owner;
};

/* include/linux/sched.h —— task_struct 里的 PI 字段 */
struct rb_root_cached	pi_waiters;	/* 等我持有的锁的任务 */
struct task_struct	*pi_top_task;	/* 链上最高优先级的等待者 */
struct rt_mutex_waiter	*pi_blocked_on;	/* 我正在等的锁 */

pi_waiters 是“谁在等我”，pi_blocked_on 是“我在等谁”。两者构成一张有向图，rt_mutex_adjust_prio_chain() 沿着 blocked_on → owner 走，边走边提升优先级。链的长度通常很短（1–3 层），但每层都要拿 wait_lock，所以 PI 不是免费的。

为什么 mutex 不能 PI

普通 mutex 的 owner 只记录“谁持有”，没有 pi_waiters 树，也没有“等锁任务按优先级排序”的等待队列。它无法在 H 阻塞时反向找到 L 并提升 L。CONFIG_DEBUG_MUTEXES 能检测“递归/非持有者解锁”，但不做 PI。

rt_mutex 的等待队列是按优先级排序的红黑树（waiters），不是 FIFO 链表。这保证高优先级等待者先被唤醒，也保证 PI 链调整时能快速找到“最高优先级的等待者”。

ww_mutex：多资源死锁的 wound/wait

两个任务以不同顺序拿 A、B 会 ABBA 死锁。ww_mutex（wound/wait）的做法：

1. 每个事务有一个 ww_acquire_ctx，记录已获得的锁与顺序；
2. 拿锁时若发现“自己比持有者更老/更重要”，可以“wound”（迫使持有者回滚）；
3. 否则等待，并在等待时释放已持有的锁，避免死锁。

struct ww_acquire_ctx ctx;
ww_acquire_init(&ctx, &reservation_ww_class);
ww_mutex_lock(&res->lock, &ctx);
...
ww_mutex_unlock(&res->lock);
ww_acquire_fini(&ctx);

GPU 驱动（drm_gem）大量使用 ww_mutex：一个 ioctl 可能要锁多个 buffer，顺序由用户态决定，必须靠 wound/wait 打破环。

PREEMPT_RT 的锁语义

所以 RT 上“spin_lock 里不能睡眠”不再是绝对规则，但“raw_spin_lock 里不能睡眠”仍是。写可移植代码时，问自己：这段代码在 RT 上会被调度吗？如果会，临界区必须短且可睡眠安全。

3.11.13 lockdep 深入：锁类、子类与报告字段

3.11.7 给了 lockdep 报告的读法。这一节讲它凭什么“提前”发现死锁，以及为什么有时会误报。

锁类与子类

lockdep 不跟踪“这一把锁”，而是跟踪锁类（lock class）：同一代码位置、同一类型的锁视为一类。这样它不需要枚举所有实例，就能建立“类之间的获取顺序图”。

问题：同一段代码可能以不同嵌套深度拿同一类锁（如递归遍历树）。这时要告诉 lockdep “这是不同子类”：

/* 同一把锁在不同嵌套层，用子类区分 */
mutex_lock_nested(&obj->lock, SINGLE_DEPTH_NESTING);
...
mutex_unlock(&obj->lock);

/* 或者给不同实例分配不同锁类 */
static struct lock_class_key a_key, b_key;
lockdep_set_class(&a->lock, &a_key);
lockdep_set_class(&b->lock, &b_key);

依赖图与环检测

lockdep 在每次 lock_acquire() 时记录“当前持有的锁 → 新锁”的有向边，在 lock_release() 时更新。它维护一张全局图，插入边时做环检测：

持锁栈: A → B → C
新请求: C → A
检查: A → B → C → A 形成环 → 报警

报警不需要真的发生死锁，只要“存在一条可能形成环的路径”。这就是它“提前”的原因，也是它误报的来源：两个实际上不会同时发生的上下文，可能在图上连成环。

报告里的关键字段

======================================================
WARNING: possible circular locking dependency detected
------------------------------------------------------
kworker/0:1/123 is trying to acquire lock:
ffff888012345678 (&b->lock){+.+.}-{3:3}, at: func_b+0x1a/0x40

but task is already holding lock:
ffff888087654321 (&a->lock){+.+.}-{3:3}, at: func_a+0x2b/0x50

Chain exists of:
  &c->lock --> &a->lock --> &b->lock

 Possible unsafe locking scenario:
       CPU0                    CPU1
       ----                    ----
  lock(&a->lock);
                               lock(&b->lock);
                               lock(&a->lock);
  lock(&b->lock);

{+.+.} 是 lockdep 的 usage bits（该锁在 hardirq/softirq 上下文的读/写使用标记），{3:3} 是 read/write 状态与子类。四位的精确顺序与含义以 Documentation/locking/lockdep-design.rst 为准；读报告时真正重要的是后面的 Chain exists of 与交叉加锁序列，不要凭记忆解读 usage bits。

统计与调试接口

/proc/lockdep_stats 的 debug_locks 为 0 表示 lockdep 已经关闭（通常在首次报警后自动关闭，避免刷屏）。debug_locks 没有受支持的运行时重置接口，通常只能重启内核；生产内核一般只在调试内核上开 lockdep。

误报与抑制

lockdep_off()/lockdep_on() 是全局开关，只能用于“确认无环但工具无法表达”的极少数场景；用它掩盖真实问题，下一个接手的人会付出更大代价。

3.11.14 实验：亲手制造 ABA，量一量锁争用

前两节讲了 ABA 与 PI 的原理，这一节给两个用户态就能跑的实验。它们不依赖内核模块，在 macOS/Linux 上都能编译运行；输出里的数值会变，趋势与结论不变。

实验一：让无标记 CAS“成功”地破坏栈

这个程序先用单线程重放一次 ABA 交错：t1 读到栈顶 A 与 A->next=B 后暂停；t2 把 A、B 都弹出，再把 A 压回；t1 的 CAS 以为“栈顶还是 A”，于是把栈顶改成 B——而 B 已经不在栈里。然后对比带 2 位 tag 的版本：同样的交错，tag 从 0 变 3，CAS 失败。

/* aba_stack.c — Treiber 栈的 ABA：无标记 CAS 会“成功地把头指到已摘除节点”
 * 编译: cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -pthread -o aba_stack aba_stack.c
 * 运行: ./aba_stack
 */
#define _GNU_SOURCE
#include <pthread.h>
#include <sched.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>

struct node {
	int value;
	struct node *next;
};

#define TAG_BITS 2
#define TAG_MASK ((uintptr_t)((1u << TAG_BITS) - 1))

static _Atomic(uintptr_t) head;
static _Atomic long pushed, popped;
static _Atomic int ready;

static struct node *untag(uintptr_t v)
{
	return (struct node *)(v & ~TAG_MASK);
}

static uintptr_t tag_inc(uintptr_t v)
{
	return (v & TAG_MASK) + 1;
}

/* 带标记的 push：每次修改都让 tag 递增，ABA 时旧 tag 对不上 */
static void push(struct node *n)
{
	uintptr_t old = atomic_load_explicit(&head, memory_order_relaxed);
	uintptr_t neu;

	do {
		n->next = untag(old);
		neu = (uintptr_t)n | (tag_inc(old) & TAG_MASK);
	} while (!atomic_compare_exchange_weak_explicit(
		&head, &old, neu, memory_order_release, memory_order_relaxed));
	atomic_fetch_add_explicit(&pushed, 1, memory_order_relaxed);
}

static struct node *pop(void)
{
	uintptr_t old = atomic_load_explicit(&head, memory_order_acquire);
	uintptr_t neu;

	do {
		struct node *n = untag(old);

		if (!n)
			return NULL;
		neu = (uintptr_t)n->next | (tag_inc(old) & TAG_MASK);
	} while (!atomic_compare_exchange_weak_explicit(
		&head, &old, neu, memory_order_acquire, memory_order_relaxed));
	atomic_fetch_add_explicit(&popped, 1, memory_order_relaxed);
	return untag(old);
}

/* 单线程重放 ABA 的交错：t1 读头后暂停，t2 做三次操作让头“看起来没变” */
static void demo_aba(void)
{
	struct node a = {.value = 1, .next = NULL};
	struct node b = {.value = 2, .next = NULL};
	struct node *uhead = &a;          /* 无标记版本 */
	uintptr_t snapshot;

	/* --- 无标记：指针相等就认为“没人动过” --- */
	b.next = NULL;
	a.next = &b;
	{
		struct node *t1_next = uhead->next;   /* t1 记下 a->next = b */
		struct node *p;

		p = uhead; uhead = p->next;           /* t2: pop a */
		p = uhead; uhead = p->next;           /* t2: pop b */
		a.next = uhead; uhead = &a;           /* t2: push a */
		(void)t1_next;

		/* t1 的 CAS：期望 &a，写入 &b */
		if (uhead == &a) {
			uhead = &b;
			printf("untagged: CAS succeeded -> head=b (b already popped!)\n");
		} else {
			printf("untagged: CAS failed\n");
		}
	}

	/* --- 带标记：同样的交错，tag 让 CAS 失败 --- */
	atomic_store(&head, (uintptr_t)&a);
	a.next = &b;
	snapshot = atomic_load(&head);
	(void)pop();                            /* t2: pop a */
	(void)pop();                            /* t2: pop b */
	push(&a);                               /* t2: push a，头指针又等于 &a */

	uintptr_t expected = snapshot;
	bool ok = atomic_compare_exchange_strong(&head, &expected, (uintptr_t)&b);
	printf("tagged:   CAS %s (tag %lu -> %lu)\n",
	       ok ? "succeeded" : "failed",
	       (unsigned long)(snapshot & TAG_MASK),
	       (unsigned long)(atomic_load(&head) & TAG_MASK));
}

static void *worker(void *arg)
{
	long id = (long)arg;
	struct node *nodes = calloc(10000, sizeof *nodes);
	int i;

	if (!nodes)
		return NULL;
	for (i = 0; i < 10000; i++) {
		nodes[i].value = (int)(id * 10000 + i);
		push(&nodes[i]);
	}
	/* 等所有线程推完再开始弹，避免“栈空时 pop 到 NULL” */
	atomic_fetch_add(&ready, 1);
	while (atomic_load(&ready) < 4)
		sched_yield();

	for (i = 0; i < 10000; i++)
		(void)pop();
	/* 不 free：其他线程可能仍持有这些节点的 next 指针；
	 * 真实实现要用 hazard pointer / RCU / epoch 回收。 */
	return NULL;
}

int main(void)
{
	pthread_t th[4];
	long i;

	demo_aba();

	atomic_store(&head, 0);
	atomic_store(&pushed, 0);
	atomic_store(&popped, 0);
	for (i = 0; i < 4; i++)
		if (pthread_create(&th[i], NULL, worker, (void *)i) != 0) {
			perror("pthread_create");
			return 1;
		}
	for (i = 0; i < 4; i++)
		pthread_join(th[i], NULL);

	printf("stress: pushed=%ld popped=%ld head=%p (expect NULL)\n",
	       atomic_load(&pushed), atomic_load(&popped), (void *)untag(atomic_load(&head)));
	return 0;
}

编译与运行：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -pthread -o aba_stack aba_stack.c
./aba_stack

在 macOS/arm64（Apple clang）上的实际输出：

untagged: CAS succeeded -> head=b (b already popped!)
tagged:   CAS failed (tag 0 -> 3)
stress: pushed=40000 popped=40000 head=0x0 (expect NULL)

三行分别说明：

1. 无标记版本：CAS 成功，但 head=b 指向一个已经弹出的节点——这就是 ABA 的破坏力。
2. 带标记版本：tag 不匹配，CAS 失败，重试后读到正确的栈顶。
3. 并发压力测试：4 线程各 push/pop 10000 次，最终 head=NULL、push/pop 计数相等。

代码里的 TAG_BITS=2 只是演示；2 位 tag 每 4 次修改就回绕，真实实现要用 16 位以上，或 128 位 CAS（x86 cmpxchg16b、ARM64 casp）。tag 的本质是“让 ‘A 变回 A’ 这件事在 CAS 看来不再相同”。

注意压力测试没有 free 节点：一个线程 pop 到别的线程的节点后，原线程可能仍持有它的 next 指针；立刻 free 会让另一个线程读到已释放内存。真实的无锁栈必须配 hazard pointer、epoch 或 RCU 回收——这正是“无锁不等于无管理”的地方。

实验二：spinlock 与 mutex 的吞吐对比

同一个临界区（counter++），分别用 atomic_flag 自旋锁与 pthread_mutex，4 个线程各做 20 万次，比较吞吐。

/* lock_bench.c — 同一个临界区，spinlock 与 mutex 的吞吐对比
 * 编译: cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -pthread -o lock_bench lock_bench.c
 * 运行: ./lock_bench spin 4 200000
 *       ./lock_bench mutex 4 200000
 */
#define _GNU_SOURCE
#include <pthread.h>
#include <sched.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

static atomic_flag slock = ATOMIC_FLAG_INIT;
static pthread_mutex_t mlock = PTHREAD_MUTEX_INITIALIZER;
static long counter;
static long iters;
static int use_mutex;

static void spin_lock(void)
{
	while (atomic_flag_test_and_set_explicit(&slock, memory_order_acquire))
		; /* 真实内核用 cpu_relax()；用户态可用 sched_yield 降低争用 */
}

static void spin_unlock(void)
{
	atomic_flag_clear_explicit(&slock, memory_order_release);
}

static void *worker(void *arg)
{
	(void)arg;
	for (long i = 0; i < iters; i++) {
		if (use_mutex) {
			pthread_mutex_lock(&mlock);
			counter++;
			pthread_mutex_unlock(&mlock);
		} else {
			spin_lock();
			counter++;
			spin_unlock();
		}
	}
	return NULL;
}

static double now_sec(void)
{
	struct timespec ts;

	clock_gettime(CLOCK_MONOTONIC, &ts);
	return (double)ts.tv_sec + (double)ts.tv_nsec / 1e9;
}

int main(int argc, char *argv[])
{
	int nthreads = argc > 2 ? atoi(argv[2]) : 4;
	pthread_t *th;
	double t0, t1;
	long total;

	if (argc > 1 && strcmp(argv[1], "mutex") == 0)
		use_mutex = 1;
	else if (argc > 1 && strcmp(argv[1], "spin") == 0)
		use_mutex = 0;
	else {
		fprintf(stderr, "usage: %s spin|mutex [threads] [iters]\n", argv[0]);
		return 2;
	}
	iters = argc > 3 ? atol(argv[3]) : 200000;
	th = calloc((size_t)nthreads, sizeof *th);
	if (!th) {
		perror("calloc");
		return 1;
	}

	t0 = now_sec();
	for (int i = 0; i < nthreads; i++)
		if (pthread_create(&th[i], NULL, worker, NULL) != 0) {
			perror("pthread_create");
			return 1;
		}
	for (int i = 0; i < nthreads; i++)
		pthread_join(th[i], NULL);
	t1 = now_sec();

	total = counter;
	printf("%s threads=%d total=%ld sec=%.3f Mops/s=%.2f\n",
	       use_mutex ? "mutex" : "spin", nthreads, total, t1 - t0,
	       (double)total / (t1 - t0) / 1e6);
	free(th);
	return 0;
}

编译与运行：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -pthread -o lock_bench lock_bench.c
./lock_bench spin 4 200000
./lock_bench mutex 4 200000
./lock_bench spin 1 200000
./lock_bench mutex 1 200000

macOS/arm64 上的实际输出（Apple M 系列，4 线程）：

spin threads=4 total=800000 sec=0.114 Mops/s=7.01
mutex threads=4 total=800000 sec=0.017 Mops/s=46.73
spin threads=1 total=200000 sec=0.001 Mops/s=160.00
mutex threads=1 total=200000 sec=0.002 Mops/s=121.58

读法：

• 无争用：spinlock 160 Mops/s，mutex 122 Mops/s——自旋省掉了系统调用与睡眠，快约 30%。
• 4 线程争用：spinlock 掉到 7 Mops/s，mutex 反而 47 Mops/s——自旋锁的 cache line 在 4 个核之间来回弹，每次失败重试都在加剧争用；mutex 让拿不到锁的线程睡眠，把 CPU 让给持锁者。
• 这就是内核里“临界区极短才用 spinlock”的量化依据：临界区一长、争用一高，自旋的浪费会超过睡眠的开销。

用户态的 atomic_flag 没有 cpu_relax()，也没有排队机制；内核的 qspinlock 用 MCS 队列把等待者分散到各自的 cache line，正是为了解决这里看到的“所有人抢同一个 cache line”。

观测：perf lock 与 bpftrace

用户态实验能说明趋势，定位线上锁热点要靠 perf lock：

# 全系统采集 5 秒，按争用时间排序
sudo perf lock contention -a --sleep 5

# 只看某个进程
sudo perf lock contention -p $PID --sleep 5

输出字段：

 contended   total wait     max wait     avg wait         type   caller
      1234      12.34 s      120.00 ms     10.00 ms     spinlock   func+0x1a
       567       1.23 s       50.00 ms      2.17 ms        mutex   other+0x2b

contended 是争用次数，total wait 是累计等待，max wait 是单次最长；caller 是加锁点。先看 max wait 找毛刺，再看 total wait 找吞吐瓶颈。

没有 perf lock 的内核（< 5.16）可以用 bpftrace 数争用事件：

sudo bpftrace -e '
tracepoint:lock:contention_begin
{
	@[ksym(args->lock_addr)] = count();
}
interval:s:5 { exit(); }'

args->lock_addr 是锁的地址，ksym() 转成符号；如果显示 0xffff...，说明该锁是动态分配的，需要结合调用栈（kstack）定位。

观测：lockdep 报告与统计

# 有没有环
sudo dmesg | grep -A30 "possible circular"

# lockdep 自身的统计
cat /proc/lockdep_stats | head -20

# 当前持有的锁（sysrq-d）
echo d | sudo tee /proc/sysrq-trigger

/proc/lockdep_stats 里先看 debug_locks 是否为 1；为 0 说明 lockdep 已停止报告。cyclic_chains 大于 0 表示检测到过环，chain_lookup_hits/misses 反映依赖图的规模与命中率。

3.11.15 陷阱清单：并发与同步篇

第 12 条需要展开：PREEMPT_RT 合并进主线是 6.12，6.8 上要打 RT 补丁。RT 的 spinlock_t 是 rt_mutex，spin_lock() 可能睡眠；但 raw_spinlock_t 仍是“关抢占、忙等”。所以“spinlock 里不能睡眠”这句话在 RT 上要改成“raw_spinlock 里不能睡眠”。

第 15 条是性能问题：smp_mb() 在 x86 上是 mfence（或 lock 前缀），在 ARM64 上是 dmb ish，都会阻止乱序执行。能用一个 release + 一个 acquire 解决的，不要用两个全屏障。

3.11.16 实验：seqlock、RCU 与锁争用

第一个模块用 seqlock 保护一个 64 位计数器，读侧无锁重试：

// seqlock_demo.c — seqlock 读写（Linux 6.8）
#include <linux/init.h>
#include <linux/module.h>
#include <linux/seqlock.h>
#include <linux/delay.h>

struct demo_data {
	seqlock_t seq;
	u64 counter;
	u64 timestamp;
};

static struct demo_data demo;

static int __init seqlock_demo_init(void)
{
	u64 v;

	seqlock_init(&demo.seq);
	for (v = 1; v <= 5; v++) {
		write_seqlock(&demo.seq);
		demo.counter = v * 1000;
		demo.timestamp = ktime_get_ns();
		write_sequnlock(&demo.seq);
		udelay(10);
	}

	/* 读侧：无锁，序号变化就重试 */
	{
		unsigned int seq;
		u64 c, t;
		do {
			seq = read_seqbegin(&demo.seq);
			c = demo.counter;
			t = demo.timestamp;
		} while (read_seqretry(&demo.seq, seq));
		pr_info("seqlock_demo: counter=%llu ts=%llu\n",
			(unsigned long long)c, (unsigned long long)t);
	}
	return 0;
}

static void __exit seqlock_demo_exit(void) { }

module_init(seqlock_demo_init);
module_exit(seqlock_demo_exit);
MODULE_LICENSE("GPL");
MODULE_DESCRIPTION("seqlock reader/writer demo");

第二个模块用 RCU 保护一个链表，写侧「复制-替换-等待-释放」：

// rcu_demo.c — RCU 发布与回收（Linux 6.8）
#include <linux/init.h>
#include <linux/module.h>
#include <linux/rcupdate.h>
#include <linux/slab.h>

struct item {
	int value;
	struct rcu_head rh;
};

static struct item __rcu *global_item;

static int __init rcu_demo_init(void)
{
	struct item *old, *new;

	/* 初始发布 */
	new = kmalloc(sizeof(*new), GFP_KERNEL);
	if (!new)
		return -ENOMEM;
	new->value = 1;
	rcu_assign_pointer(global_item, new);

	/* 读侧 */
	rcu_read_lock();
	old = rcu_dereference(global_item);
	pr_info("rcu_demo: read value=%d\n", old->value);
	rcu_read_unlock();

	/* 写侧：换一个新对象，旧对象延迟释放 */
	new = kmalloc(sizeof(*new), GFP_KERNEL);
	if (!new)
		return -ENOMEM;
	new->value = 2;
	old = rcu_dereference_protected(global_item, 1);
	rcu_assign_pointer(global_item, new);
	synchronize_rcu();             /* 等所有旧读者退出 */
	kfree(old);
	pr_info("rcu_demo: published value=2\n");
	return 0;
}

static void __exit rcu_demo_exit(void)
{
	struct item *p = rcu_dereference_protected(global_item, 1);

	RCU_INIT_POINTER(global_item, NULL);
	synchronize_rcu();
	kfree(p);
}

两个模块都在 QEMU 里加载，dmesg 看输出。注意 RCU 模块的写侧假设「只有一个写者」（用 rcu_dereference_protected(..., 1) 断言），多写者需要额外的互斥锁——RCU 只保证「读者看到一致的对象」，不保证「多个写者不互相覆盖」。

锁争用观测用 perf lock（5.16+）：

# 采 5 秒，按争用时间排序
perf lock contention -a --sleep 5 -b
# 只看某个进程
perf lock contention -p $PID -t 5
# 看某把锁的调用栈
perf lock contention -a -L &mmap_lock --sleep 3

bpftrace 也能数（需要 lock:contention_begin tracepoint）：

bpftrace -e 'tracepoint:lock:contention_begin { @[ksym(args->lock_addr)] = count(); } interval:s:5 { exit(); }'

3.11.17 本章出口检查

母本对应

下一站：3.12 VFS 与文件系统 把同步工具用到具体的子系统：路径解析怎么在 RCU 下走 dcache、inode 的锁怎么分工、ext4 的日志如何保证崩溃一致性。