Linux RCU 实战指南：用 rcu_barrier() 安全卸载使用了 call_rcu() 的可卸载模块

本文基于 Linux 内核文档 rcubarrier.rst 展开，聚焦一个具体而棘手的内核开发问题：当一个可卸载模块通过 call_rcu() 注册了 RCU 回调后，如何保证模块卸载时不会留下“悬空”的回调指针。读完全文，你将理解 synchronize_rcu() 为何不足以解决模块卸载问题、rcu_barrier() 与 srcu_barrier() 的确切语义与使用协议，并能结合 kernel/rcu/tree.c 中的当前实现看懂“屏障回调”这一经典并发技巧的完整演进脉络。

背景：call_rcu() 与非阻塞释放

RCU 的更新者（updater）经常使用 call_rcu() 发起对宽限期（grace period）结束的异步等待。这个原语接收两个参数：一个指向位于 RCU 保护数据结构内部的 struct rcu_head 的指针，以及一个稍后可能被调用的、用于释放该数据结构的函数指针。例如，从 IRQ 上下文删除链表元素 p 的代码可以写成：

list_del_rcu(p);
call_rcu(&p->rcu, p_callback);

由于 call_rcu() 从不阻塞，这段代码可以安全地在 IRQ 上下文中使用。回调函数 p_callback() 可以这样定义：

static void p_callback(struct rcu_head *rp)
{
	struct pstruct *p = container_of(rp, struct pstruct, rcu);

	kfree(p);
}

这正是 RCU 内存回收的基本范式：删除是立即完成的，内存释放被推迟到宽限期结束后的回调执行时刻。

问题：模块卸载时回调还挂在那里怎么办？

如果 p_callback() 定义在一个可卸载模块里，问题就来了：在仍有 RCU 回调挂起（pending）时卸载模块，之后执行这些回调的 CPU 在调用 p_callback() 时会跳转到已被释放的模块代码区，其结果不可预料。

一个直觉方案是在模块退出路径里调用 synchronize_rcu()，但这并不充分：synchronize_rcu() 确实等待宽限期结束，但它不等待回调完成。回调只有在宽限期结束后的某个时刻才会在某个 CPU 的软中断里执行，而 synchronize_rcu() 返回时这些回调可能只是“入队”了而已。

有人会想：那就连续调用多次 synchronize_rcu() 总行了吧？依然没有保证。如果系统存在很重的 RCU 回调负载，部分回调可能被延迟执行，以便让其他处理先进行——例如实时内核（PREEMPT_RT）为了避免过大的调度延迟，就要求对回调做这样的延迟。多次宽限期结束仍无法确保旧回调已经跑完。

rcu_barrier() 原语及其使用协议

能正确处理这一局面的是 rcu_barrier() 原语。它不等待宽限期结束，而是等待所有未决的 RCU 回调完成。这里有一个关键语义，必须牢记：

rcu_barrier() 并不蕴含 synchronize_rcu()。特别地，如果系统中任何地方都没有排队的 RCU 回调，rcu_barrier() 完全有资格立即返回，不等待任何东西，更谈不上等待一个宽限期。

在模块退出路径中，标准用法是三步走：

1. 阻止任何新的 RCU 回调被投递（prevent any new RCU callbacks from being posted）。
2. 执行 rcu_barrier()。
3. 允许模块卸载。

第一步不能省略：如果回调在 rcu_barrier() 运行期间继续产生，屏障永远等不到“全部完成”。

SRCU 对应物：srcu_barrier()

对 SRCU 存在对应的 srcu_barrier() 函数，且必须让它与 call_srcu() 的“风味”（flavor）匹配。如果模块使用了多个 srcu_struct，卸载时就必须对每个结构各调用一次 srcu_barrier()。例如，模块同时使用了 call_rcu()、作用于 srcu_struct_1 的 call_srcu() 和作用于 srcu_struct_2 的 call_srcu()，则卸载时需要这三行代码：

1  rcu_barrier();
2  srcu_barrier(&srcu_struct_1);
3  srcu_barrier(&srcu_struct_2);

文档还提示了一个延迟敏感场景下的技巧：如果延迟至关重要（latency is of the essence），可以用 workqueue 让这三个函数并发执行，而不是串行等待。

总结规则

• 模块使用了 call_rcu() → 卸载前必须调用 rcu_barrier()；
• 模块使用了 call_srcu() → 卸载前必须在同一个 srcu_struct 上调用 srcu_barrier()；
• 两者都用 → rcu_barrier() 与 srcu_barrier() 都要调用。

一个容易踩坑的变体是从定时器回调里投递 call_rcu() 的模块：此时必须先停止投递新定时器、取消（或等待）所有已投递的定时器，然后才调用 rcu_barrier() 等待剩余回调完成。否则定时器会在屏障运行期间不断产生新回调。

实例：rcutorture 模块的退出路径

Documentation/RCU/rcubarrier.rst 引用了一个早期版本 rcutorture 模块退出函数的完整代码，它是上述三步协议的标准示范：

1  static void
2  rcu_torture_cleanup(void)
3  {
4    int i;
5
6    fullstop = 1;
7    if (shuffler_task != NULL) {
8      VERBOSE_PRINTK_STRING("Stopping rcu_torture_shuffle task");
9      kthread_stop(shuffler_task);
10    }
11    shuffler_task = NULL;
12
13    if (writer_task != NULL) {
14      VERBOSE_PRINTK_STRING("Stopping rcu_torture_writer task");
15      kthread_stop(writer_task);
16    }
17    writer_task = NULL;
18
19    if (reader_tasks != NULL) {
20      for (i = 0; i < nrealreaders; i++) {
21        if (reader_tasks[i] != NULL) {
22          VERBOSE_PRINTK_STRING(
23            "Stopping rcu_torture_reader task");
24          kthread_stop(reader_tasks[i]);
25        }
26        reader_tasks[i] = NULL;
27      }
28      kfree(reader_tasks);
29      reader_tasks = NULL;
30    }
31    rcu_torture_current = NULL;
32
33    if (fakewriter_tasks != NULL) {
34      for (i = 0; i < nfakewriters; i++) {
35        if (fakewriter_tasks[i] != NULL) {
36          VERBOSE_PRINTK_STRING(
37            "Stopping rcu_torture_fakewriter task");
38          kthread_stop(fakewriter_tasks[i]);
39        }
40        fakewriter_tasks[i] = NULL;
41      }
42      kfree(fakewriter_tasks);
43      fakewriter_tasks = NULL;
44    }
45
46    if (stats_task != NULL) {
47      VERBOSE_PRINTK_STRING("Stopping rcu_torture_stats task");
48      kthread_stop(stats_task);
49    }
50    stats_task = NULL;
51
52    /* Wait for all RCU callbacks to fire. */
53    rcu_barrier();
54
55    rcu_torture_stats_print(); /* -After- the stats thread is stopped! */
56
57    if (cur_ops->cleanup != NULL)
58      cur_ops->cleanup();
59    if (atomic_read(&n_rcu_torture_error))
60      rcu_torture_print_module_parms("End of test: FAILURE");
61    else
62      rcu_torture_print_module_parms("End of test: SUCCESS");
63  }

逐段解读：

• 第 6 行设置全局变量 fullstop，阻止 RCU 回调重新投递自己。大多数场景下不需要这一步，因为 RCU 回调很少包含对 call_rcu() 的调用；rcutorture 是例外，它的回调会再次投递回调，所以必须显式设置。
• 第 7–50 行依次停止 shuffler、writer、reader、fakewriter、stats 等所有内核任务。执行到第 53 行时，保证不会有新的 rcutorture RCU 回调被投递——这正是三步协议的第一、二步之间的“阻断新回调”环节。
• 第 53 行的 rcu_barrier() 等待所有先前已存在的回调完成。
• 第 55–62 行打印状态并做与具体操作（ops）相关的清理，随后返回，模块卸载流程得以完成。

映射到当前源码，rcutorture 的退出逻辑如今在 kernel/rcu/rcutorture.c 的 rcu_torture_cleanup() 中。从源码结构看，新实现把屏障抽象成了每套操作集的一个函数指针：退出时先调用 cur_ops->cb_barrier（各操作集分别映射到 rcu_barrier、rcu_barrier_tasks、rcu_barrier_tasks_trace 等，见 kernel/rcu/rcutorture.c 中 cb_barrier 字段的赋值），再逐个 torture_stop_kthread() 停止各任务线程，与文档中“先阻断、再屏障、后卸载”的协议完全一致。该模块中同样存在文档所述“回调重新投递”的特殊性，因此在 read-exit 压测循环里也会调用 rcu_barrier() 来等待 task_struct 的释放回调完成、避免 OOM（见 kernel/rcu/rcutorture.c 中 rcu_barrier(); // Wait for task_struct free, avoid OOM. 的注释）。

经典实现：把屏障挂进每个 CPU 的回调队列

rcu_barrier() 最初由 Dipankar Sarma 实现（Nikita Danilov 在文件系统中使用 RCU 时提出了原始需求，详见后文 Quiz 1）。其实现利用了一个核心事实：一旦 RCU 回调入队到某个每 CPU 回调队列中，它们之间永远不会乱序执行。因此，只要在每个 CPU 的回调队列尾部各挂一个“屏障回调”，那么当所有屏障回调都开始执行时，它们之前入队的所有回调必然已经完成。

原始代码大致如下：

1  void rcu_barrier(void)
2  {
3    BUG_ON(in_interrupt());
4    /* Take cpucontrol mutex to protect against CPU hotplug */
5    mutex_lock(&rcu_barrier_mutex);
6    init_completion(&rcu_barrier_completion);
7    atomic_set(&rcu_barrier_cpu_count, 1);
8    on_each_cpu(rcu_barrier_func, NULL, 0, 1);
9    if (atomic_dec_and_test(&rcu_barrier_cpu_count))
10      complete(&rcu_barrier_completion);
11    wait_for_completion(&rcu_barrier_completion);
12    mutex_unlock(&rcu_barrier_mutex);
13  }

逐行解读：

• 第 3 行验证调用者处于进程上下文；
• 第 5、12 行用 rcu_barrier_mutex 互斥锁，保证同一时刻只有一个 rcu_barrier() 在使用那组全局的 completion 和计数器；
• 第 6、7 行初始化 completion 和计数器（注意初始值设为 1，原因见 Quick Quiz #2）；
• 第 8 行让每个 CPU 都执行 rcu_barrier_func()。on_each_cpu() 参数列表中的最后一个 1 是 wait 标志，保证所有 rcu_barrier_func() 调用完成后 on_each_cpu() 才返回；
• 第 9 行从 rcu_barrier_cpu_count 中减掉初始计数，若减到零则第 10 行直接完成 completion，从而让第 11 行的 wait_for_completion() 免于阻塞；
• 无论哪种路径，第 11 行都在需要时等待 completion。

每个 CPU 上执行的 rcu_barrier_func() 定位 RCU 内部的每 CPU rcu_data 结构，并向当前 CPU 的队列投递屏障回调：

1  static void rcu_barrier_func(void *notused)
2  {
3    int cpu = smp_processor_id();
4    struct rcu_data *rdp = &per_cpu(rcu_data, cpu);
5    struct rcu_head *head;
6
7    head = &rdp->barrier;
8    atomic_inc(&rcu_barrier_cpu_count);
9    call_rcu(head, rcu_barrier_callback);
10  }

第 3、4 行找到含 struct rcu_head 的每 CPU rcu_data，第 7 行取到该 rcu_head 指针，第 8 行递增全局计数器（回调稍后负责递减），第 9 行把 rcu_barrier_callback() 注册到当前 CPU 的队列尾部。

而 rcu_barrier_callback() 只做一件事——原子递减计数器，减到零时完成 completion：

1  static void rcu_barrier_callback(struct rcu_head *notused)
2  {
3    if (atomic_dec_and_test(&rcu_barrier_cpu_count))
4      complete(&rcu_barrier_completion);
5  }

这套代码 2008 年被重写，此后又多次重写，但整体思路不变。

当前实现：rcu_seq 序列号与“计数初值为 2”

文档指出，当前实现更复杂，动机有二：避免打扰空闲 CPU（尤其是电池供电系统），以及在实时系统上最小化对非空闲 CPU 的干扰；此外还叠加了大量优化。对照当前源码 kernel/rcu/tree.c 中的 rcu_barrier()，可以看到经典思路的完整继承：

1. 串行化与早退：mutex_lock(&rcu_state.barrier_mutex) 序列化并发的屏障请求；随后检查 rcu_seq_done(&rcu_state.barrier_sequence, s)——如果别的调用者已经完成了同样的屏障工作（序列号已翻转完成），直接返回。rcu_state.barrier_sequence 是 rcu_seq 序列状态量，用一次 CAS 即可标记屏障开始/结束。
2. 计数初值设为 2：atomic_set(&rcu_state.barrier_cpu_count, 2)。源码注释解释了为什么是 2 而不是 0：这是为了防止在刚入队的回调被立即执行、或本任务被抢占时，计数过早归零而提前返回。这与原始代码里“初值为 1 + 事后减 1”的写法是同一思想的现代版本——屏障回调全部入队之前，计数永远不可能归零。
3. 逐 CPU 挂载屏障回调：for_each_possible_cpu() 循环中，通过 smp_load_acquire(&rdp->barrier_seq_snap) 快速路径跳过已经响应过本轮屏障的 CPU；对空队列的 CPU 直接写入 barrier_seq_snap 标记跳过；对离线 CPU 直接本地执行 rcu_barrier_entrain(rdp)；对在线 CPU 则用 smp_call_function_single(cpu, rcu_barrier_handler, (void *)cpu, 1) 发起跨 CPU 调用——注意最后一个参数 1 即 wait 标志，正是 Quick Quiz #3 中讨论的 on_each_cpu() 第四参数的继承。rcu_barrier_handler() 在目标 CPU 的跨 CPU IRQ 上下文中加 barrier_lock 自旋锁后调用 rcu_barrier_entrain()：将 rdp->barrier_head.func 设为 rcu_barrier_callback，用 rcu_segcblist_entrain() 挂到该 CPU 回调队列，成功则 atomic_inc(&rcu_state.barrier_cpu_count)。
4. 等待完成：所有屏障回调入队后，atomic_sub_and_test(2, ...) 减去初始计数，归零则 complete()，最后 wait_for_completion(&rcu_state.barrier_completion)。
5. 收尾：rcu_seq_end() 标记屏障结束，并把所有 CPU 的 barrier_seq_snap 更新为本轮 gseq，让后续调用能通过快速路径直接早退。

此外，kernel/rcu/tree.c 还提供了 rcu_barrier_throttled()：对 rcu_barrier() 加上“全系统至多每秒一次”的护栏，专为测试套件在测试间隙冲刷上一轮回调、避免 OOM 而设计，对应 rcutree 的 do_rcu_barrier 模块参数。

对于小型配置（单 CPU/简单 RCU），kernel/rcu/tiny.c 中的实现退化为 wait_rcu_gp(call_rcu_hurry)——等待一个宽限期并催促回调尽快执行。由于 tiny RCU 的回调语义是宽限期结束后立即批量执行，这个实现是保守但安全的。

srcu_barrier() 的当前实现在 kernel/rcu/srcutree.c：结构上与 rcu_barrier() 同源（rcu_seq 序列号 + 互斥锁 + completion + 计数器），但作用于指定 srcu_struct 上每 CPU 的 srcu_data 队列，回调为 srcu_barrier_cb()，且入队前持有一段 __srcu_read_lock_nmisafe() 读锁以保证 srcu_data 尺寸状态稳定。文档中“每个 srcu_struct 都要单独屏障一次”的要求，正对应它按 ssp 参数独立维护序列号与计数器的设计。

快速小测（Quick Quizzes）与解析

原文档附带三道快速小测，是理解这套机制并发细节的最佳材料，这里完整收录并给出答案。

Quick Quiz #1：还有哪里可能需要 rcu_barrier()？

问：除了模块卸载，还有别的场景需要 rcu_barrier() 吗？

答：饶有意味的是，rcu_barrier() 最初根本不是为模块卸载而实现的。Nikita Danilov 在某文件系统中使用 RCU，导致文件系统**卸载（unmount）**时刻出现同样的困境：还有回调挂着，而承载回调代码的文件系统要下线了。Dipankar Sarma 应此需求编写了 rcu_barrier()，让 Nikita 能在文件系统卸载流程中调用它。后来，文档作者本人在实现 rcutorture 时撞上了 RCU 模块卸载问题，发现 rcu_barrier() 同样解决了它。

Quick Quiz #2：为什么计数初值不是 0？

问：为什么不把第 8 行的 on_each_cpu() 设计成把 rcu_barrier_cpu_count 初始化为零，从而省掉第 9、10 行？

答：假设 on_each_cpu() 被延迟，导致 CPU 0 的 rcu_barrier_func() 执行完毕、相应宽限期也走完了，这一切都发生在 CPU 1 的 rcu_barrier_func() 开始执行之前。此时计数器会被减到零，第 11 行的 wait_for_completion() 立即返回，完全没等到 CPU 1 的回调执行。

需要说明的是：2005 年这段代码刚加入时并没有这个问题，因为当时的 on_each_cpu() 会禁抢占，禁抢占区本身就是 RCU 读端临界区，阻止了 CPU 0 的宽限期完成。而 v4.20 前后的 RCU 风味整合（consolidation）之后，整合版 RCU 再次把不可抢占代码区视为读端临界区，这种可能性也被排除。文档强调：尽管如此，那个额外的初始计数仍可能是个好主意——依赖实现上的偶然巧合（accidents of implementation）会在实现变更时变成日后的意外 bug。当前 tree.c 中“初值设为 2”的注释，正是这条经验教训的直接体现。

Quick Quiz #3：会不会提前返回？

问：如果 CPU 0 的 rcu_barrier_func() 立刻执行（计数器加到 1），而另一个 CPU 的 rcu_barrier_func() 被延迟整整一个宽限期，rcu_barrier() 岂不是会提前返回？

答：不可能发生。原因是 on_each_cpu() 的最后一个参数（wait 标志）为 1，它一路传到 smp_call_function() 再到 smp_call_function_on_cpu()，使后者自旋等待直到跨 CPU 的 rcu_barrier_func() 调用完成。在 CONFIG_PREEMPTION 关闭的内核上，这本身就足以阻止宽限期完成，因为每个 CPU 在宽限期结束前都必须经历一次上下文切换（或其他静默状态 quiescent state）。

但在 CONFIG_PREEMPTION 内核上光靠 wait 标志不够：on_each_cpu() 因此在调用 smp_call_function() 期间以及本地调用 rcu_barrier_func() 期间都禁用了抢占。由于较新的 RCU 实现把禁抢占代码区视为 RCU 读端临界区，这阻止了宽限期完成。于是所有 CPU 都在第一个 rcu_barrier_callback() 可能执行之前跑完了 rcu_barrier_func()，计数器也就不可能过早归零。

文档最后补充：如果哪天 on_each_cpu() 出于实时延迟考虑放弃了禁抢占，那么“把计数器初始化为 1”这一手就会救场。当前 tree.c 将初值设为 2 的注释（“avoid a too-soon return to zero in case of an immediate invocation of the just-enqueued callback (or preemption of this task)”）与这段分析一脉相承。

小结：何时必须调用 rcu_barrier()

rcu_barrier() 的使用频率相对较低，因为绝大多数使用 RCU 的代码位于内核核心而非模块中。但只要你的可卸载模块用了 RCU，就必须用 rcu_barrier()（或对应风味的 srcu_barrier()）来保证安全卸载。实践清单：

场景	卸载/下线前的操作
模块使用 call_rcu()	阻断新回调 → rcu_barrier() → 卸载
模块使用 call_srcu()（多个 srcu_struct）	对每个 srcu_struct 各调用一次 srcu_barrier()
从定时器投递 RCU 回调	先停/取消所有定时器，再调用 rcu_barrier()
回调会重新投递自身	用全局标志（如 rcutorture 的 fullstop）阻止重投递，再屏障
文件系统 unmount 等“代码即将下线”场景	与模块卸载同理，调用 rcu_barrier()
延迟敏感	用 workqueue 并发执行多个屏障调用

从源码看，rcu_barrier() 的演化——从 on_each_cpu() + 禁抢占的“屏障回调排队”技巧，到 rcu_seq 序列号快速路径、空队列 CPU 跳过、离线 CPU 本地入队、rcu_barrier_throttled() 限流——始终围绕两个不变量展开：每 CPU 回调队列的执行顺序保证，以及计数在全部屏障回调入队前绝不允许归零。理解了这两点，无论实现如何重写，rcu_barrier() 的正确性论证都依然成立。