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

Linux RCU 实战指南用 rcu_barrier() 安全卸载使用了 call_rcu() 的可卸载模块【免费下载链接】linuxLinux kernel source tree项目地址: https://gitcode.com/GitHub_Trending/li/linux本文基于 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 内部的每 CPUrcu_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的每 CPUrcu_data第 7 行取到该rcu_head指针第 8 行递增全局计数器回调稍后负责递减第 9 行把rcu_barrier_callback()注册到当前 CPU 的队列尾部。而rcu_barrier_callback()只做一件事——原子递减计数器减到零时完成 completion1 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()可以看到经典思路的完整继承串行化与早退mutex_lock(rcu_state.barrier_mutex)序列化并发的屏障请求随后检查rcu_seq_done(rcu_state.barrier_sequence, s)——如果别的调用者已经完成了同样的屏障工作序列号已翻转完成直接返回。rcu_state.barrier_sequence是rcu_seq序列状态量用一次 CAS 即可标记屏障开始/结束。计数初值设为 2atomic_set(rcu_state.barrier_cpu_count, 2)。源码注释解释了为什么是 2 而不是 0这是为了防止在刚入队的回调被立即执行、或本任务被抢占时计数过早归零而提前返回。这与原始代码里“初值为 1 事后减 1”的写法是同一思想的现代版本——屏障回调全部入队之前计数永远不可能归零。逐 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)。等待完成所有屏障回调入队后atomic_sub_and_test(2, ...)减去初始计数归零则complete()最后wait_for_completion(rcu_state.barrier_completion)。收尾rcu_seq_end()标记屏障结束并把所有 CPU 的barrier_seq_snap更新为本轮gseq让后续调用能通过快速路径直接早退。此外kernel/rcu/tree.c 还提供了rcu_barrier_throttled()对rcu_barrier()加上“全系统至多每秒一次”的护栏专为测试套件在测试间隙冲刷上一轮回调、避免 OOM 而设计对应 rcutree 的do_rcu_barrier模块参数。对于小型配置单 CPU/简单 RCUkernel/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()的正确性论证都依然成立。【免费下载链接】linuxLinux kernel source tree项目地址: https://gitcode.com/GitHub_Trending/li/linux创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考