Linux内核分析之内核同步-02
This language version is unavailable; showing the other language.
17.1 RCU 原理 —— 读侧无锁的优雅方案
RCU(Read-Copy-Update,读-拷贝-更新)是 Linux 内核中最独特的同步机制:读者不加锁、不用原子操作、甚至(在强序架构上)不用内存屏障,只做两件事——取指针、解引用;写者则复制一份新数据、原子地替换指针、然后等所有"可能还在读旧数据"的读者都离开后才真正释放旧数据。这个"等"的窗口就是宽限期(Grace Period),它把删除动作从关键路径上彻底摘走。3.5 节已在 start_kernel 流程中初识 RCU 的初始化,本节将结合 Linux 7.0.10 内核源码,逐行分析 RCU 的发布-订阅模型、rcu_read_lock() 在两种内核配置下的真实展开、以及 RCU 链表原语如何让"读侧无锁"落到实处。
17.1.1 为什么读者可以完全无锁
不变式:读者要么看不到,要么看得完整
考虑一个被频繁读取、极少修改的指针:
读者: p = gp; read p->a, p->b;
写者: 更新 gp 指向的对象
用锁的思路,读者必须防止写者在读取 p->a 与 p->b 之间改掉它们。RCU 换了一个思路:写者从不原地修改,只创建新副本并原子替换指针。于是读者拿到的指针要么指向旧对象(完整、不再变化),要么指向新对象(完整)——永远不存在"读到一半被改"的撕裂。唯一剩下的问题是:旧对象什么时候能删?答案是等所有可能持有旧指针的读者都读完——这就是宽限期的语义,17.2 节展开其实现。
由此推出 RCU 的三条使用纪律:
- 读者不得阻塞(睡眠)在 RCU 读侧临界区内(可抢占 RCU 下允许被抢占,但不允许自愿阻塞,见 17.1.4 节);
- 读者临界区内获得的指针在临界区外无效(写者可能在宽限期后释放它);
- 写者更新共享数据必须通过 RCU 发布原语(
rcu_assign_pointer()),保证读者不会看到"新指针 + 旧数据"的组合。
每种上下文都能找到的通用形态
RCU 不是一把锁,而是一套约定:任何能保证"读者与写者不并发破坏数据"的场景都可以用 RCU 模式改写。内核中 RCU 读侧临界区的典型形状:
rcu_read_lock();
p = rcu_dereference(gp); /* 读侧取指针 */
...只读访问 p->... ...
rcu_read_unlock();
/* 写侧 */
spin_lock(&gp_lock); /* 写者之间仍需互斥 */
new = kmalloc(...); ...初始化...
old = rcu_dereference_protected(gp, lockdep_is_held(&gp_lock));
rcu_assign_pointer(gp, new); /* 发布 */
spin_unlock(&gp_lock);
call_rcu(&old->rcu, free_fn); /* 宽限期后释放旧对象 */
写者之间的互斥仍由普通锁承担——RCU 协调的是读者与写者之间,不是写者与写者之间。
17.1.2 __rcu 注解与发布-订阅原语
__rcu:让编译器帮忙找漏
// include/linux/compiler_types.h:53
# define __rcu __attribute__((noderef, address_space(__rcu)))
__rcu 标记"此指针受 RCU 保护"。sparse 静态检查器(line 53)据此拒绝"裸解引用 RCU 指针"(必须经 rcu_dereference() 消除 noderef);普通编译下它退化为 BTF 类型标签(compiler_types.h:72),仅进入 BTF 调试信息。这层静态检查能在编译期抓住大量"绕过 RCU 原语直接读写"的错误。
rcu_assign_pointer() —— 发布
// include/linux/rcupdate.h:570-579
#define rcu_assign_pointer(p, v) \
context_unsafe( \
uintptr_t _r_a_p__v = (uintptr_t)(v); \
rcu_check_sparse(p, __rcu); \
\
if (__builtin_constant_p(v) && (_r_a_p__v) == (uintptr_t)NULL) \
WRITE_ONCE((p), (typeof(p))(_r_a_p__v)); \
else \
smp_store_release(&p, RCU_INITIALIZER((typeof(p))_r_a_p__v)); \
)
发布的内存序语义(line 576-577):smp_store_release() 保证指针发布之前的所有写(初始化新对象的那些 store)先于指针本身可见——读者一旦看到新指针,读到的一定是初始化完成的新对象。这就是 15.2 节 RELEASE 语义的直接应用。两个细节:赋 NULL 是编译期常量时退化为 WRITE_ONCE()(无需屏障,NULL 不携带数据);rcu_check_sparse() 配合 __rcu 注解做静态检查。初始化后从不更新的指针可用更省的 RCU_INIT_POINTER()(rcupdate.h:1041-1045)。
rcu_dereference() —— 订阅
// include/linux/rcupdate.h:511-518
#define __rcu_dereference_check(p, local, c, space) \
({ \
/* Dependency order vs. p above. */ \
typeof(*p) *local = (typeof(*p) *__force)READ_ONCE(p); \
RCU_LOCKDEP_WARN(!(c), "suspicious rcu_dereference_check() usage"); \
rcu_check_sparse(p, space); \
((typeof(*p) __force __kernel *)(local)); \
})
读侧为什么不需要屏障(line 513):READ_ONCE() 保证单次原子读取,而其后的字段访问与指针值之间形成地址依赖(address dependency)——CPU 必须先拿到指针值才知道去哪读字段,硬件因此保证"先读指针、后用指针"的顺序。绝大多数架构(包括 x86_64 与 ARM64)保证地址依赖的顺序性,因此订阅不需要 acquire 屏障;唯一例外是上古的 Alpha,由架构层在 rcu_dereference 展开中补齐屏障。lockdep 条件参数 c(rcu_dereference_check() 形式)允许在"持某把锁时也可调用"的场景下声明豁免,RCU_LOCKDEP_WARN 在 debug 构建下校验。
族中其他成员:rcu_dereference_protected()(rcupdate.h:742-752)用于写者侧——读者不可能并发,直接取值,lockdep_is_held() 声明持锁证据;rcu_replace_pointer()(rcupdate.h:592-597)原子地"取旧换新"。
17.1.3 rcu_read_lock() / rcu_read_unlock()
通用壳层的展开(include/linux/rcupdate.h):
// include/linux/rcupdate.h:845-853 与 876-884
static __always_inline void rcu_read_lock(void)
__acquires_shared(RCU)
{
__rcu_read_lock();
__acquire_shared(RCU);
rcu_lock_acquire(&rcu_lock_map);
RCU_LOCKDEP_WARN(!rcu_is_watching(),
"rcu_read_lock() used illegally while idle");
}
...
static inline void rcu_read_unlock(void)
__releases_shared(RCU)
{
RCU_LOCKDEP_WARN(!rcu_is_watching(),
"rcu_read_unlock() used illegally while idle");
rcu_lock_release(&rcu_lock_map); /* Keep acq info for rls diags. */
__release_shared(RCU);
__rcu_read_unlock();
}
真正的开销全部藏在 __rcu_read_lock() 里,它有两种截然不同的展开,由 CONFIG_PREEMPT_RCU 决定:
// include/linux/rcupdate.h:93-118 (非抢占分支)
static inline void __rcu_read_lock(void)
{
preempt_disable();
}
static inline void __rcu_read_unlock(void)
{
if (IS_ENABLED(CONFIG_RCU_STRICT_GRACE_PERIOD))
rcu_read_unlock_strict();
preempt_enable();
}
非抢占 RCU(CONFIG_PREEMPT_RCU=n,本树默认在非抢占内核下关闭,见 kernel/rcu/Kconfig:8-34 的推导链)下,读侧临界区就是一段不可抢占区间:preempt_disable() 使本 CPU 在临界区内不会跑别的任务,于是"本 CPU 处于读侧临界区"等价于"本 CPU 可抢占",宽限期检测直接复用抢占计数(17.2 节的静止状态机制)。读侧开销是两次 preempt_count 原子加减——这就是"RCU 读侧几乎免费"的真实成本。
可抢占 RCU(PREEMPT_RCU=y,SMP 抢占内核默认开启)下,__rcu_read_lock() 的实现在 kernel/rcu/tree_plugin.h:412-421:
// kernel/rcu/tree_plugin.h:386-446
/* limit value for ->rcu_read_lock_nesting. */
#define RCU_NEST_PMAX (INT_MAX / 2)
static void rcu_preempt_read_enter(void)
{
WRITE_ONCE(current->rcu_read_lock_nesting, READ_ONCE(current->rcu_read_lock_nesting) + 1);
}
static int rcu_preempt_read_exit(void)
{
int ret = READ_ONCE(current->rcu_read_lock_nesting) - 1;
WRITE_ONCE(current->rcu_read_lock_nesting, ret);
return ret;
}
...
void __rcu_read_lock(void)
{
rcu_preempt_read_enter();
if (IS_ENABLED(CONFIG_PROVE_LOCKING))
WARN_ON_ONCE(rcu_preempt_depth() > RCU_NEST_PMAX);
if (IS_ENABLED(CONFIG_RCU_STRICT_GRACE_PERIOD) && rcu_state.gp_kthread)
WRITE_ONCE(current->rcu_read_unlock_special.b.need_qs, true);
barrier(); /* critical section after entry code. */
}
...
void __rcu_read_unlock(void)
{
struct task_struct *t = current;
barrier(); // critical section before exit code.
if (rcu_preempt_read_exit() == 0) {
barrier(); // critical-section exit before .s check.
if (unlikely(READ_ONCE(t->rcu_read_unlock_special.s)))
rcu_read_unlock_special(t);
}
...
}
可抢占 RCU 下读侧临界区允许被抢占(这是抢占内核的硬需求——否则 RCU 读临界区会破坏实时性),"处于读侧"改为用任务自身字段 current->rcu_read_lock_nesting 计数表达。最外层解锁(nesting 减到 0)时若 rcu_read_unlock_special.s 非零——表明在读侧期间发生过抢占、宽限期在等待本任务等特殊情况——进入 rcu_read_unlock_special()(tree_plugin.h:725-773)补做静止状态上报。读者阻塞(而非仅被抢占)的情形由调度器钩子处理:13.2 节见过的 rcu_note_context_switch() 会在任务带读侧计数进入睡眠时把任务登记到叶节点的 blkd_tasks 链表,宽限期显式等待它——这正是 ch14 进程退出分析中 exit_rcu() 需要处理"任务可能死在读侧临界区"的根源。
两个 API 壳中的 rcu_is_watching() 检查值得注意:在 CPU 处于空闲(EQS,17.2.4 节)状态时使用 RCU 读侧是编程错误——空闲 CPU 按定义已是静止状态,读侧保护形同虚设,debug 构建下此处的 RCU_LOCKDEP_WARN 会报错。
17.1.4 RCU 链表原语
RCU 最密集的使用形态是链表。include/linux/rculist.h 提供全套变体,与普通链表 API 一一对应:
// include/linux/rculist.h:97-107 与 176-180
static inline void __list_add_rcu(struct list_head *new,
struct list_head *prev, struct list_head *next)
{
if (!__list_add_valid(new, prev, next))
return;
new->next = next;
new->prev = prev;
rcu_assign_pointer(list_next_rcu(prev), new);
next->prev = new;
}
...
static inline void list_del_rcu(struct list_head *entry)
{
__list_del_entry(entry);
entry->prev = LIST_POISON2;
}
插入(line 101-104)的顺序体现发布语义:先填好 new 自身的 next/prev,再用 rcu_assign_pointer() 发布 prev->next(list_next_rcu() 给指针加上 __rcu 注解),最后补 next->prev。读者沿 prev->next 遍历,在新节点可见时其自身字段必然已就绪。
删除(line 176-180)则与普通 list_del() 有本质区别:__list_del_entry() 把 entry 摘出链表,但不碰 entry 自身的 next(普通版会把它毒化为 LIST_POISON1)——因为此刻可能还有读者正握着这个指针沿旧链遍历,毒化会立即引发崩溃。entry->next 的最终清理由宽限期后的释放回调完成。这也解释了 8.3 节 PID 哈希表、9.4 节 net namespace 等处"删除后延迟释放"的统一模式。
遍历接口 list_for_each_entry_rcu()(rculist.h:475-479)内含 rcu_dereference_check(),支持第三参数声明"持某锁时也可遍历"(如 12.2 节 cgroup 遍历传 cgroup_mutex_is_held 类条件)。
读侧无锁的量化对比
| 原语 | 读者开销(x86_64,无争用) | 与写者协调 |
|---|---|---|
| spinlock | lock cmpxchg + preempt_count 两次加减 | 互斥 |
| rwlock 读侧 | lock add(计数+1)+ 检查 | 写者须等读者排空 |
| seqcount 读侧 | 普通 load + 末尾核对,可能重试 | 写者翻号 |
| RCU 读侧 | preempt_count 加减(非抢占)或任务计数(可抢占),零缓存行争用 | 写者等宽限期,读者完全无感 |
RCU 读者的伸缩性是线性的:一万个读者同时进入临界区,不存在任何共享缓存行上的写——这是锁类原语原理上做不到的。
17.1.5 写侧完整案例与 API 速查
kernel/audit.c 的审计连接指针
一个完整覆盖"初始化 → 发布 → 读 → 替换 → 延迟释放"的活例子:
// kernel/audit.c:115-122
struct auditd_connection {
struct pid *pid;
u32 portid;
struct net *net;
struct rcu_head rcu;
};
static struct auditd_connection __rcu *auditd_conn;
static DEFINE_SPINLOCK(auditd_conn_lock);
__rcu 注解声明保护关系;rcu_head rcu 是延迟释放的挂钩字段(17.2.3 节)。读侧:
// kernel/audit.c:230-241
int auditd_test_task(struct task_struct *task)
{
int rc;
struct auditd_connection *ac;
rcu_read_lock();
ac = rcu_dereference(auditd_conn);
rc = (ac && ac->pid == task_tgid(task) ? 1 : 0);
rcu_read_unlock();
return rc;
}
写侧替换与延迟释放:
// kernel/audit.c:562-566
spin_lock_irqsave(&auditd_conn_lock, flags);
ac_old = rcu_dereference_protected(auditd_conn,
lockdep_is_held(&auditd_conn_lock));
rcu_assign_pointer(auditd_conn, ac_new);
spin_unlock_irqrestore(&auditd_conn_lock, flags);
if (ac_old)
call_rcu(&ac_old->rcu, auditd_conn_free);
rcu_dereference_protected() + lockdep_is_held() 向 lockdep 证明"此时无读者并发";call_rcu() 把旧对象交给宽限期机制,回调 auditd_conn_free()(audit.c:516-524)在安全窗口后 kfree。这个五步模式是内核写侧 RCU 的标准模板。
延迟释放的快捷方式:kfree_rcu()
如果回调要做的只是 kfree,kfree_rcu(ptr, rhf) 省去手写回调:
// include/linux/rcupdate.h:1085-1086, 1117-1133
#define kfree_rcu(ptr, rhf) kvfree_rcu_arg_2(ptr, rhf)
#define kvfree_rcu(ptr, rhf) kvfree_rcu_arg_2(ptr, rhf)
...
#define kvfree_rcu_arg_2(ptr, rhf) \
do { \
typeof (ptr) ___p = (ptr); \
\
if (___p) { \
BUILD_BUG_ON(offsetof(typeof(*(ptr)), rhf) >= 4096); \
kvfree_call_rcu(&((___p)->rhf), (void *) (___p)); \
} \
} while (0)
BUILD_BUG_ON 限定挂钩字段必须落在对象前 4096 字节内(line 1122)——内核的延迟释放路径依赖偏移量编码技巧(无头版 kvfree_rcu_mightsleep 连 rcu_head 字段都不需要,rcupdate.h:1105-1106)。9.7 节 namespace 释放用的 kfree_rcu(ns, ns.ns_rcu) 即此 API。无头版允许睡眠等待(内部走 synchronize_rcu() 路径),两参版不可睡眠,适用场景不同。
API 速查
| 场景 | API |
|---|---|
| 读侧进入/退出 | rcu_read_lock() / rcu_read_unlock() |
| 读侧取指针 | rcu_dereference()(需持读锁)、rcu_dereference_check()(带 lockdep 豁免条件)、rcu_dereference_protected()(写侧/持锁) |
| 发布指针 | rcu_assign_pointer();初始化用 RCU_INIT_POINTER() |
| 原子替换 | rcu_replace_pointer() |
| 链表 | list_add_rcu() / list_del_rcu() / list_for_each_entry_rcu()(rculist.h) |
| 延迟释放 | call_rcu()、kfree_rcu()、kvfree_rcu() |
| 同步等待宽限期 | synchronize_rcu()(17.2.5 节) |
要点总结:
- RCU 把"读者与写者互斥"重构成"写者复制替换 + 等待旧读者退场",读者临界区内数据不可变,故无需任何互斥。
- 发布用
smp_store_release()(rcu_assign_pointer),订阅靠地址依赖(rcu_dereference内的READ_ONCE),一写一读构成完整的内存序配对。 __rcu注解 + sparse/BTF 双实现提供编译期防护;RCU_LOCKDEP_WARN提供运行时上下文校验。- 读侧开销只是抢占计数或任务字段自增,没有任何共享写——这是 RCU 与锁的本质分野,也是内核大量只读热路径选它的理由。
宽限期是这套机制的心脏:写者到底怎么知道"所有旧读者都走了"?下一节进入 tree RCU 的实现。
17.2 RCU 宽限期与回调机制
宽限期(Grace Period,GP)是 RCU 的心脏:写者把旧数据"挂在"宽限期机制上,机制保证宽限期结束时所有可能读到旧数据的读者都已离开,此后释放旧数据绝对安全。Linux 7.0.10 的 SMP 内核使用 tree RCU:一个 rcu_node 树聚合全 CPU 的静止状态(quiescent state,QS)上报,专用内核线程 rcu_gp 驱动宽限期从请求到收尾的状态机,per-CPU 的分段回调链表(四段 CBFS)则把"宽限期结束"转化为回调执行的批处理。本节将结合 Linux 7.0.10 内核源码,逐行分析 gp_seq 序号协议、宽限期状态机三阶段、QS 上报树、回调的排队与批量执行,以及 stall(迟滞)检测。
17.2.1 gp_seq —— 宽限期序号协议
低 2 位状态位
RCU 用一个 unsigned long gp_seq 跟踪宽限期的推进,约定在 kernel/rcu/rcu.h:16-61 的注释与宏中:
- 计数部分(高位):完成的宽限期数量,按 4 递增;
- 状态部分(低 2 位):
RCU_SEQ_CTR_SHIFT = 2、RCU_SEQ_STATE_MASK = 3(include/linux/rcupdate.h:47-48),0 表示无 GP 进行中,1/2/3 表示 GP 处于不同阶段。
启动与收尾的序号操作:
// kernel/rcu/rcu.h:94-113
static inline void rcu_seq_start(unsigned long *sp)
{
WRITE_ONCE(*sp, *sp + 1);
smp_mb(); /* Ensure update-side operation after counter increment. */
WARN_ON_ONCE(rcu_seq_state(*sp) != 1);
}
/* Compute the end-of-grace-period value for the specified sequence number. */
static inline unsigned long rcu_seq_endval(unsigned long *sp)
{
return (*sp | RCU_SEQ_STATE_MASK) + 1;
}
/* Adjust sequence number for end of update-side operation. */
static inline void rcu_seq_end(unsigned long *sp)
{
smp_mb(); /* Ensure update-side operation before counter increment. */
WARN_ON_ONCE(!rcu_seq_state(*sp));
WRITE_ONCE(*sp, rcu_seq_endval(sp));
}
rcu_seq_start()(line 94-99):+1 使状态位变为 1(GP 启动),屏障保证启动动作序号生效后才发生。rcu_seq_end()(line 108-113):rcu_seq_endval() 把值规格化为"状态位清零、计数部分 +4"(如 ...X1 → ...Y0,Y=X+1),完成一次宽限期的闭合。"判断一个快照是否已被当前宽限期覆盖"只需按此协议比较(rcu_seq_done() 等辅助,rcu.h:126-133)——每个 CPU、每个 rcu_node 都持有 gp_seq 的副本,宽限期的推进本质上是这个数字在树上的一致性传播。
树形结构:rcu_node / rcu_data / rcu_state
// kernel/rcu/tree.h:41-82 (节选 struct rcu_node)
struct rcu_node {
raw_spinlock_t __private lock; /* Root rcu_node's lock protects */
unsigned long gp_seq; /* Track rsp->gp_seq. */
unsigned long gp_seq_needed; /* Track furthest future GP request. */
unsigned long completedqs; /* All QSes done for this node. */
unsigned long qsmask; /* CPUs or groups that need to switch in */
/* order for current grace period to proceed.*/
/* In leaf rcu_node, each bit corresponds to */
/* an rcu_data structure, otherwise, each */
/* bit corresponds to a child rcu_node */
/* structure. */
unsigned long rcu_gp_init_mask; /* Mask of offline CPUs at GP init. */
unsigned long qsmaskinit;
unsigned long qsmaskinitnext;
unsigned long expmask; /* CPUs or groups that need to check in */
...
unsigned long grpmask; /* Mask to apply to parent qsmask. */
int grplo; /* lowest-numbered CPU here. */
int grphi; /* highest-numbered CPU here. */
...
qsmask 是树的核心:叶节点的每个 bit 对应一个 CPU,中间节点的每个 bit 对应一个子节点。宽限期启动时全树置位("所有人都要上报"),上报链自底向上清位(17.2.3 节),根节点清空即宽限期完成。grpmask 记录本节点在父节点中对应的那一位。树的形状由 CONFIG_RCU_FANOUT(64 位默认 64,中间节点扇出)与 CONFIG_RCU_FANOUT_LEAF(默认 16,叶节点扇出)推导(include/linux/rcu_node_tree.h:33-90),NUM_RCU_NODES、RCU_NUM_LVLS 在编译期算出——千级 CPU 的机器树高也只有 2-3 层。
每 CPU 一份的 rcu_data(tree.h:189-297)持有本 CPU 的 gp_seq 副本、QS 待报标志(cpu_no_qs)、回调链表 cblist、以及本 CPU 所在叶节点指针 mynode 与位 grpmask。全局的 rcu_state(tree.h:351-438)持有树的根:node[NUM_RCU_NODES] 数组、权威的 gp_seq、GP 内核线程指针 gp_kthread 与其等待队列 gp_wq、以及 synchronize_rcu() 等待者队列(srs_next 等,tree.h:427-432)。
17.2.2 宽限期状态机:rcu_gp_kthread() 三阶段
GP 由专用内核线程驱动,主循环(kernel/rcu/tree.c:2271-2304)清晰地分为三个阶段:
// kernel/rcu/tree.c:2271-2304
static int __noreturn rcu_gp_kthread(void *unused)
{
rcu_bind_gp_kthread();
for (;;) {
/* Handle grace-period start. */
for (;;) {
trace_rcu_grace_period(rcu_state.name, rcu_state.gp_seq,
TPS("reqwait"));
WRITE_ONCE(rcu_state.gp_state, RCU_GP_WAIT_GPS);
swait_event_idle_exclusive(rcu_state.gp_wq,
READ_ONCE(rcu_state.gp_flags) &
RCU_GP_FLAG_INIT);
rcu_gp_torture_wait();
WRITE_ONCE(rcu_state.gp_state, RCU_GP_DONE_GPS);
/* Locking provides needed memory barrier. */
if (rcu_gp_init())
break;
cond_resched_tasks_rcu_qs();
WRITE_ONCE(rcu_state.gp_activity, jiffies);
WARN_ON(signal_pending(current));
trace_rcu_grace_period(rcu_state.name, rcu_state.gp_seq,
TPS("reqwaitsig"));
}
/* Handle quiescent-state forcing. */
rcu_gp_fqs_loop();
/* Handle grace-period end. */
WRITE_ONCE(rcu_state.gp_state, RCU_GP_CLEANUP);
rcu_gp_cleanup();
WRITE_ONCE(rcu_state.gp_state, RCU_GP_CLEANED);
}
}
阶段一:等待 GP 请求(line 2279-2289)。gp_flags 的 RCU_GP_FLAG_INIT 由 rcu_start_this_gp() 在有回调需要新宽限期时设置(call_rcu() 路径,17.2.4 节),随后 swake_up 唤醒 GP 线程。rcu_gp_init()(tree.c:1804-1999)完成启动:rcu_seq_start() 推进序号(:1853)、自根向叶遍历全树把 qsmask 置满(:1954-1978)、处理离线 CPU 的掩码(:1887-1938)。init 返回 false 表示启动条件又变了(如 GP 已被合并),回到内层循环重新等待。
阶段二:静止状态收集(line 2293,rcu_gp_fqs_loop(),tree.c:2064-2145)。GP 线程以 RCU_GP_WAIT_FQS 状态定时醒来(间隔基准 RCU_JIFFIES_TILL_FORCE_QS,tree.h:305,约 3 个 jiffy),每轮调用 rcu_gp_fqs()(tree.c:2028-2059)扫描各叶节点 qsmask 未清的 CPU:首轮 rcu_watching_snap_save() 存下各 CPU 的"是否在观察"快照(tree.c:819-839),后续轮 rcu_watching_snap_recheck()(:855-893)比较快照——空闲中(EQS)或快照变化的 CPU 判定为已通过静止状态,由 GP 线程代为清位。这就是"CPU 睡着了也能完成上报"的 dynticks 机制(17.2.4 节)。
阶段三:收尾(line 2296-2299,rcu_gp_cleanup(),tree.c:2150-2266):
// kernel/rcu/tree.c:2150-2222
static noinline void rcu_gp_cleanup(void)
{
...
new_gp_seq = rcu_state.gp_seq;
rcu_seq_end(&new_gp_seq);
rcu_for_each_node_breadth_first(rnp) {
raw_spin_lock_irq_rcu_node(rnp);
...
WARN_ON_ONCE(rnp->qsmask);
WRITE_ONCE(rnp->gp_seq, new_gp_seq);
...
}
rnp = rcu_get_root();
raw_spin_lock_irq_rcu_node(rnp); /* GP before ->gp_seq update. */
/* Declare grace period done, trace first to use old GP number. */
trace_rcu_grace_period(rcu_state.name, rcu_state.gp_seq, TPS("end"));
rcu_seq_end(&rcu_state.gp_seq);
...
自根向叶(breadth-first)把新序号传播到全树(line 2161-2169),断言各节点 qsmask 已清空,最后对权威 gp_seq 执行 rcu_seq_end()。gp_state 的四阶段轨迹(RCU_GP_WAIT_GPS → RCU_GP_DONE_GPS → RCU_GP_WAIT_FQS → RCU_GP_CLEANUP)可通过 tracepoint 完整观测。
17.2.3 静止状态上报:从 tick 到根节点
什么算静止状态
静止状态(QS)是"该 CPU 不可能处于 RCU 读侧临界区"的时刻:进程上下文切换、进入空闲、进入用户态执行。非抢占 RCU 下读侧即不可抢占区间,任何调度事件都意味着离开;可抢占 RCU 下读者可被抢占,一个任务可能跨多个调度点持读锁,其判定改由 17.1.3 节的任务计数与 blkd_tasks 机制承担——但上报树的机械结构是共用的。
上报链
tick 与调度器钩子把"本 CPU 通过了静止状态"记入 per-CPU 变量,真正的树上传播由 rcu_core()(RCU_SOFTIRQ 或 rcuc 内核线程上下文)完成:
rcu_sched_clock_irq() tick 处理(tree.c:2696)
| 记 ticks_this_gp、检查 rcu_urgent_qs
v
rcu_sched_clock_irq -> rcu_qs() 等 清 rdp->cpu_no_qs.b.norm
v
rcu_core() -> rcu_check_quiescent_state() (tree.c:2497-2522)
| 发现 cpu_no_qs 已清 且 GP 序号匹配
v
rcu_report_qs_rdp(rdp) (tree.c:2443-2489)
| 清叶节点本 CPU 的 qsmask 位
v
rcu_report_qs_rnp(mask, rnp, ...) (tree.c:2339-2394)
| 沿树上溯,逐层清位
v
rcu_report_qs_rsp(flags) (tree.c:2315-2323)
根节点全零:置 RCU_GP_FLAG_FQS,唤醒 GP 线程
核心函数逐段看:
// kernel/rcu/tree.c:2443-2488
static void
rcu_report_qs_rdp(struct rcu_data *rdp)
{
...
rnp = rdp->mynode;
raw_spin_lock_irqsave_rcu_node(rnp, flags);
if (rdp->cpu_no_qs.b.norm || rdp->gp_seq != rnp->gp_seq ||
rdp->gpwrap) {
rdp->cpu_no_qs.b.norm = true; /* need qs for new gp. */
raw_spin_unlock_irqrestore_rcu_node(rnp, flags);
return;
}
mask = rdp->grpmask;
rdp->core_needs_qs = false;
if ((rnp->qsmask & mask) == 0) {
raw_spin_unlock_irqrestore_rcu_node(rnp, flags);
} else {
if (!rcu_rdp_is_offloaded(rdp)) {
WARN_ON_ONCE(rcu_accelerate_cbs(rnp, rdp));
}
rcu_disable_urgency_upon_qs(rdp);
rcu_report_qs_rnp(mask, rnp, rnp->gp_seq, flags);
/* ^^^ Released rnp->lock */
}
前置校验(line 2452-2458):cpu_no_qs.b.norm 仍为真(本 CPU 其实还没过 QS)或 gp_seq 不匹配(上报的是上一个 GP 的陈旧事件)都直接放弃——上报必须是"当前 GP、确已通过"的事件。随后沿树上溯:
// kernel/rcu/tree.c:2349-2394
/* Walk up the rcu_node hierarchy. */
for (;;) {
if ((!(rnp->qsmask & mask) && mask) || rnp->gp_seq != gps) {
raw_spin_unlock_irqrestore_rcu_node(rnp, flags);
return;
}
...
WRITE_ONCE(rnp->qsmask, rnp->qsmask & ~mask);
...
if (rnp->qsmask != 0 || rcu_preempt_blocked_readers_cgp(rnp)) {
raw_spin_unlock_irqrestore_rcu_node(rnp, flags);
return;
}
rnp->completedqs = rnp->gp_seq;
mask = rnp->grpmask;
if (rnp->parent == NULL)
break;
raw_spin_unlock_irqrestore_rcu_node(rnp, flags);
rnp_c = rnp;
rnp = rnp->parent;
raw_spin_lock_irqsave_rcu_node(rnp, flags);
oldmask = READ_ONCE(rnp_c->qsmask);
}
rcu_report_qs_rsp(flags); /* releases rnp->lock. */
两道止步条件(line 2351-2355 与 2362-2366):本层还有其他位未清(qsmask != 0——还有子树没上报完)或本 GP 有可抢占读者阻塞在该节点(rcu_preempt_blocked_readers_cgp(),17.1.3 节的 blkd_tasks 非空)→ 上报止于此层。全部清空到达根节点时 rcu_report_qs_rsp() 请求 GP 线程强制推进(FQS),宽限期就此闭合。
跨层加锁统一使用 raw_spin_lock_irqsave_rcu_node() 系列(kernel/rcu/rcu.h:460-515),内部追加 smp_mb__after_unlock_lock() 恢复树层级间的全序——这是树形聚合正确性的内存序基础。
dynticks:空闲 CPU 的代答
空闲或纯用户态执行的 CPU 不跑 tick 回调,永远不会主动上报。EQS(Extended Quiescent State)机制由 context tracking 实现:ct_kernel_exit()/ct_kernel_enter()(kernel/context_tracking.c:103-132, 142-170)在进入/退出空闲或用户态时翻转本 CPU 的 watching 状态(ct_state_inc() 对 CT_RCU_WATCHING 位的原子增减)。GP 侧不要求这类 CPU 做任何事——rcu_gp_fqs() 的快照比对(17.2.2 节)直接判定"进入 EQS 的 CPU 已通过静止状态"并代为清位。这避免了空闲机器被 RCU 反复唤醒,是 RCU 能耗设计的关键一环。
17.2.4 回调机制:call_rcu() 与分段回调链表
call_rcu 的路径
// kernel/rcu/tree.c:3101-3157
static void
__call_rcu_common(struct rcu_head *head, rcu_callback_t func, bool lazy_in)
{
...
if (debug_rcu_head_queue(head)) {
/* Probable double call_rcu(), so leak the callback. */
if (atomic_inc_return(&doublefrees) < 4) {
pr_err("%s(): Double-freed CB %p->%pS()!!! ", __func__, head, head->func);
mem_dump_obj(head);
}
WRITE_ONCE(head->func, rcu_leak_callback);
return;
}
head->func = func;
head->next = NULL;
kasan_record_aux_stack(head);
local_irq_save(flags);
rdp = this_cpu_ptr(&rcu_data);
RCU_LOCKDEP_WARN(!rcu_rdp_cpu_online(rdp), "Callback enqueued on offline CPU!");
lazy = lazy_in && !rcu_async_should_hurry();
/* Add the callback to our list. */
if (unlikely(!rcu_segcblist_is_enabled(&rdp->cblist))) {
...
}
check_cb_ovld(rdp);
if (unlikely(rcu_rdp_is_offloaded(rdp)))
call_rcu_nocb(rdp, head, func, flags, lazy);
else
call_rcu_core(rdp, head, func, flags);
local_irq_restore(flags);
}
call_rcu(head, func)(tree.c:3249-3253)与本树的 lazy 默认:lazy_in 传 enable_rcu_lazy,回调可能被扣在 bypass 列表中延迟很久以省电;需要立即启动宽限期的场景(如 rcu_barrier、synchronize_rcu 内部)用 call_rcu_hurry()(tree.c:3183-3187)。重复 call_rcu() 同一个 head 会被 debug_rcu_head_queue 抓住并故意泄漏(line 3108-3116,双免费检测)。
call_rcu_core()(tree.c:3009-3050)把回调经 rcutree_enqueue()(:2998)挂入本 CPU 的 rcu_segcblist,并在两个条件下立即踢 RCU 引擎(invoke_rcu_core(),tree.c:2915-2923:raise_softirq(RCU_SOFTIRQ) 或唤醒 rcuc 内核线程):CPU 正处于 EQS(!rcu_is_watching(),回调不能等下个 tick),或回调积压超过 qhimark(tree.c:418-423 的限流常量)。
四段 CBFS:回调的年级制
per-CPU 回调链表按"还要等几个宽限期"分为四段(include/linux/rcu_segcblist.h:34-64):
[head, tails[RCU_DONE_TAIL]) 已可执行(所属 GP 已结束)
[tails[DONE], tails[RCU_WAIT_TAIL]) 等当前 GP
[tails[WAIT], tails[NEXT_READY]) 等下一个 GP
[tails[NEXT_READY], tails[NEXT]) 刚到的(尚未关联 GP)
// kernel/rcu/rcu_segcblist.c:329-337
void rcu_segcblist_enqueue(struct rcu_segcblist *rsclp,
struct rcu_head *rhp)
{
rcu_segcblist_inc_len(rsclp);
rcu_segcblist_inc_seglen(rsclp, RCU_NEXT_TAIL);
rhp->next = NULL;
WRITE_ONCE(*rsclp->tails[RCU_NEXT_TAIL], rhp);
WRITE_ONCE(rsclp->tails[RCU_NEXT_TAIL], &rhp->next);
}
新回调永远进 RCU_NEXT_TAIL 段(line 334-336,尾插 O(1))。段的推进由两个操作完成:rcu_segcblist_advance()(rcu_segcblist.c:469-509)在每次 GP 推进时把"已等到"的段左移入 DONE 段;rcu_segcblist_accelerate()(:526-560)在 GP 启动时把 NEXT_READY/NEXT 段尽可能并入本次 GP 的等待名单——一个宽限期可以为多批回调服务,这是 RCU 吞吐的关键。
批量执行:rcu_do_batch()
// kernel/rcu/tree.c:2576-2605, 2624-2631
rcu_nocb_lock_irqsave(rdp, flags);
...
pending = rcu_segcblist_get_seglen(&rdp->cblist, RCU_DONE_TAIL);
div = READ_ONCE(rcu_divisor);
div = div < 0 ? 7 : div > sizeof(long) * 8 - 2 ? sizeof(long) * 8 - 2 : div;
bl = max(rdp->blimit, pending >> div);
...
rcu_segcblist_extract_done_cbs(&rdp->cblist, &rcl);
...
for (; rhp; rhp = rcu_cblist_dequeue(&rcl)) {
rcu_callback_t f;
count++;
debug_rcu_head_unqueue(rhp);
rcu_lock_acquire(&rcu_callback_map);
trace_rcu_invoke_callback(rcu_state.name, rhp);
f = rhp->func;
debug_rcu_head_callback(rhp);
WRITE_ONCE(rhp->func, (rcu_callback_t)0L);
f(rhp);
rcu_lock_release(&rcu_callback_map);
...
if (in_serving_softirq()) {
if (count >= bl && (need_resched() || !is_idle_task(current)))
break;
blimit 是下限而非上限(line 2581):bl = max(rdp->blimit, pending >> div)——回调洪峰时批大小按积压量比例放大(默认上限 DEFAULT_MAX_RCU_BLIMIT = 10000,tree.c:418),避免"每批只跑 10 个、积压雪崩"。软中断上下文里达到 bl 且调度器有需求即退出,剩余回调回插 DONE 段(rcu_segcblist_insert_done_cbs(),:2657 附近)下轮再跑——回调执行永远不会饿死调度。回调以 rcu_lock_acquire(&rcu_callback_map) 标注(lockdep 视角的一种"读上下文"),因此回调内不得睡眠、不得使用阻塞原语。
rcu_barrier():模块卸载的安全带
模块卸载前必须保证自己注册的所有回调都已执行,否则模块代码消失后回调跳进虚空。rcu_barrier()(tree.c:3817-3911)给每个有回调的 CPU 经 IPI 追加一个 rcu_barrier_callback(),用原子计数 + completion 等待全部执行——返回即保证既有回调清空。UP 内核的 tiny 版直接 wait_rcu_gp(call_rcu_hurry)(kernel/rcu/tiny.c:45-49)。
17.2.5 synchronize_rcu() 与 expedited 变体
普通路径:栈上回调 + completion
// kernel/rcu/tree.c:3277-3309
static void synchronize_rcu_normal(void)
{
struct rcu_synchronize rs;
trace_rcu_sr_normal(rcu_state.name, &rs.head, TPS("request"));
if (READ_ONCE(rcu_normal_wake_from_gp) < 1) {
wait_rcu_gp(call_rcu_hurry);
goto trace_complete_out;
}
init_rcu_head_on_stack(&rs.head);
init_completion(&rs.completion);
...
rcu_sr_normal_add_req(&rs);
/* Kick a GP and start waiting. */
(void) start_poll_synchronize_rcu();
/* Now we can wait. */
wait_for_completion(&rs.completion);
destroy_rcu_head_on_stack(&rs.head);
trace_complete_out:
trace_rcu_sr_normal(rcu_state.name, &rs.head, TPS("complete"));
}
默认实现就是 wait_rcu_gp(call_rcu_hurry)(line 3283-3285):在栈上放一个 rcu_head + completion(kernel/rcu/update.c:411-448),把 wakeme_after_rcu() 注册为回调,然后 wait_for_completion_state() 睡眠——同步等待宽限期被完全复用为异步回调机制,没有任何专门的等待代码。可选的 rcu_normal_wake_from_gp 路径改走 GP 线程直接唤醒(srs_next llist 队列,最多 SR_MAX_USERS_WAKE_FROM_GP = 5 个等待者每 GP,tree.h:333),减少回调调度延迟。
synchronize_rcu() 入口(tree.c:3349-3386)先做两个优化判定:rcu_blocking_is_gp()(tree.c:3265-3272)——若系统只有当前这一个可运行任务(如启动早期),调用者自己睡眠的瞬间就是静止状态,宽限期瞬时完成,直接返回;rcu_gp_is_expedited() 则转入加速路径。
expedited:IPI 驱动的加速宽限期
普通宽限期以 tick 为节拍,延迟在几十毫秒量级。需要低延迟的场景(模块卸载、网络命名空间销毁,9.4 节的 synchronize_rcu_expedited())用加速版:
// kernel/rcu/tree_exp.h:924-982
void synchronize_rcu_expedited(void)
{
unsigned long flags;
struct rcu_exp_work rew;
struct rcu_node *rnp;
unsigned long s;
...
/* Take a snapshot of the sequence number. */
s = rcu_exp_gp_seq_snap();
if (exp_funnel_lock(s))
return; /* Someone else did our work for us. */
/* Ensure that load happens before action based on it. */
if (unlikely((rcu_scheduler_active == RCU_SCHEDULER_INIT) || !rcu_exp_worker_started())) {
/* Direct call during scheduler init and early_initcalls(). */
rcu_exp_sel_wait_wake(s);
} else {
/* Marshall arguments & schedule the expedited grace period. */
rew.rew_s = s;
synchronize_rcu_expedited_queue_work(&rew);
}
/* Wait for expedited grace period to complete. */
rnp = rcu_get_root();
wait_event(rnp->exp_wq[rcu_seq_ctrl(s) & 0x3],
sync_exp_work_done(s));
...
加速宽限期维护独立的序号(expedited_sequence,tree_exp.h:20-24),由 per-CPU exp_kworker 并行下发,最终对每个在线 CPU 发 IPI:
// kernel/rcu/tree_exp.h:749-804
static void rcu_exp_handler(void *unused)
{
int depth = rcu_preempt_depth();
...
if (!depth) {
if (!(preempt_count() & (PREEMPT_MASK | SOFTIRQ_MASK)) ||
rcu_is_cpu_rrupt_from_idle())
rcu_report_exp_rdp(rdp);
else
rcu_exp_need_qs();
return;
}
...
if (depth > 0) {
raw_spin_lock_irqsave_rcu_node(rnp, flags);
if (rnp->expmask & rdp->grpmask) {
WRITE_ONCE(rdp->cpu_no_qs.b.exp, true);
t->rcu_read_unlock_special.b.exp_hint = true;
}
raw_spin_unlock_irqrestore_rcu_node(rnp, flags);
return;
}
WARN_ON_ONCE(1);
}
IPI 处理器的三分支:目标 CPU 不在读侧临界区(depth == 0)且抢占/软中断均未关闭 → 立即 rcu_report_exp_rdp() 上报,宽限期瞬间完成;在读侧临界区(depth > 0)→ 置 cpu_no_qs.b.exp 与 exp_hint,记账后放行,等该任务最外层 rcu_read_unlock() 时补报(17.1.3 节的 rcu_read_unlock_special() 路径)——expedited 不打断读者,只是催促。代价是 IPI 风暴(每 CPU 一发),因此内核以 rcu_gp_might_be_stalled() 等机制限制其滥用,频繁调用方应改用 rcu_async_should_hurry 类轮询接口。
17.2.6 stall 检测
宽限期卡住(某个 CPU 长期不上报)意味着所有 call_rcu 内存无法释放、synchronize_rcu 永久阻塞——RCU 自带看门狗。检测入口挂在 rcu_pending() 首行(tree.c:3670-3679 → kernel/rcu/tree_stall.h:772-867):
// kernel/rcu/tree_stall.h:818-857
gs1 = READ_ONCE(rcu_state.gp_seq);
smp_rmb(); /* Pick up ->gp_seq first... */
js = READ_ONCE(rcu_state.jiffies_stall);
smp_rmb(); /* ...then ->jiffies_stall before the rest... */
gps = READ_ONCE(rcu_state.gp_start);
smp_rmb(); /* ...and finally ->gp_start before ->gp_seq again. */
gs2 = READ_ONCE(rcu_state.gp_seq);
if (gs1 != gs2 ||
ULONG_CMP_LT(j, js) ||
ULONG_CMP_GE(gps, js) ||
!rcu_seq_state(gs2))
return; /* No stall or GP completed since entering function. */
rnp = rdp->mynode;
jn = jiffies + ULONG_MAX / 2;
self_detected = READ_ONCE(rnp->qsmask) & rdp->grpmask;
if (rcu_gp_in_progress() &&
(self_detected || ULONG_CMP_GE(j, js + RCU_STALL_RAT_DELAY)) &&
cmpxchg(&rcu_state.jiffies_stall, js, jn) == js) {
...
} else if (self_detected) {
print_cpu_stall(gs2, gps);
} else {
print_other_cpu_stall(gs2, gps);
}
四次读取 + 三道 smp_rmb() 的"序对校验"(line 818-829):首尾两次读 gp_seq 必须相同,否则采样期间 GP 推进了,中间读到的 jiffies 数据不可信——这是 15.2 节序号协议思想在检测器里的翻版。确认 GP 停滞后区分两种报告:本 CPU 自己未上报(print_cpu_stall,打印自身栈并请求调度)或别的 CPU 卡住(print_other_cpu_stall,打印肇事 CPU 的栈)。阈值由 RCU_JIFFIES_TILL_FORCE_QS 与 RCU_STALL_RAT_DELAY(tree.h:305-315)控制,rcutorture.stall_cpu 等参数可人为制造 stall 做测试。
要点总结:
- 宽限期是"全树 qsmask 清零"的过程:GP 线程三阶段状态机(等请求 → FQS 收集 → 清理)驱动,gp_seq 低 2 位状态位 + 高位计数在树上传播。
- QS 上报链
rcu_report_qs_rdp → rcu_report_qs_rnp → rcu_report_qs_rsp自叶向根逐层清位,可抢占读者的阻塞由 blkd_tasks 兜底,空闲 CPU 由 EQS 快照代答。 - 回调走 per-CPU 四段 CBFS:入 NEXT 段 O(1) 尾插,advance/accelerate 随 GP 推进左移,
rcu_do_batch按积压量自适应批量执行。 synchronize_rcu()复用回调机制(栈上 head + completion 等待),expedited 用 IPI 催促全 CPU;stall 检测以序对采样法防误报。
宽限期机制完成了"写者安全释放旧数据"的承诺。但它对读者有一条硬约束——不可阻塞。下一节的 SRCU 把这条约束也解除。
17.3 SRCU 与可睡眠 RCU
标准 RCU 的读者有一条铁律:读侧临界区内不得阻塞。这条约束在很多场景下过于苛刻——设备卸载时要等"所有正在读配置的路径"退出,而这些路径可能睡眠(拿别的锁、等 I/O);NMI 里崩溃转储要安全遍历模块链表,读者根本无法保证不阻塞。SRCU(Sleepable RCU,可睡眠 RCU)为此而生:读者可以在临界区内睡眠,代价是每个 srcu_struct 要维护自己独立的 per-CPU 计数器阵列、每对读写各付一次全屏障、宽限期由独立的驱动逻辑推进。本节将结合 Linux 7.0.10 内核源码,逐行分析 SRCU 的双计数器翻转协议、读侧实现(含本树新增的 SRCU-fast 变体)、宽限期状态机,以及它与标准 RCU 的选择判据。
17.3.1 SRCU 的问题域与设计取舍
为什么标准 RCU 不能睡
标准 RCU 判定"读者退场"的依据是全局的执行流事件:CPU 走了 tick、切了上下文、进了空闲——这些是每 CPU 独立且无需读者配合的信号。一旦允许读者睡在临界区,"退场"就变成了某个任务的私有状态:任务睡在某个等待队列上,CPU 可以照常 tick、照常切上下文,但那个持读锁的任务还没回来。全局的每 CPU 信号不再足够,必须改成显式计数:每个读者进入时计数 +1、退出时 -1,宽限期检查计数归零。
显式计数立刻带来三个连锁设计:
- 计数放哪? 放 per-CPU(避免读者间缓存行争用),但宽限期检查"总和为零"就要汇总所有 CPU——树形聚合,与 tree RCU 同构;
- 一个计数器够吗? 不够。宽限期进行中新读者还在 +1,永远分不清"旧读者没走完"还是"新读者又来了"。于是用双计数器:宽限期中途翻转指针,新旧读者记在两组计数上,宽限期只需等旧组归零;
- 内存序怎么保证? 计数器方案下读者不再经过抢占计数,读侧内存序必须显式补——每对进入/退出各一条
smp_mb()。
这三个决定就是 SRCU 全部实现的地基。
srcu_struct:热冷两层
// include/linux/srcutree.h:21-52
struct srcu_ctr {
atomic_long_t srcu_locks; /* Locks per CPU. */
atomic_long_t srcu_unlocks; /* Unlocks per CPU. */
};
...
struct srcu_data {
/* Read-side state. */
struct srcu_ctr srcu_ctrs[2]; /* Locks and unlocks per CPU. */
int srcu_reader_flavor; /* Reader flavor ... */
/* Update-side state. */
raw_spinlock_t __private lock ____cacheline_internodealigned_in_smp;
struct rcu_segcblist srcu_cblist; /* List of callbacks.*/
unsigned long srcu_gp_seq_needed; /* Furthest future GP needed. */
...
struct srcu_node *mynode; /* Leaf srcu_node. */
unsigned long grpmask; /* Mask for leaf srcu_node */
int cpu;
struct srcu_struct *ssp;
};
读侧热数据是 srcu_data.srcu_ctrs[2]——两套计数器,每套含 srcu_locks(进入总数)与 srcu_unlocks(退出总数)两个原子量。宽限期协议比较的是同组内 lock 与 unlock 的总和是否相等(而非差值),这样旧读者退出后旧组的两个计数会精确归等,且天然免疫计数回绕。其余字段:srcu_cblist(本 CPU 的回调,复用 17.2.4 节的 rcu_segcblist)、mynode/grpmask(SRCU 自己的树,结构同 rcu_node)。
本树把 srcu_struct 拆成了热冷两层(srcutree.h:71-100 与 105-111):读者高频触碰的指针(srcu_ctrp、sda、dep_map)留在 srcu_struct,宽限期推进所需的冷数据(gp_seq、树、锁、barrier 状态)集中到 srcu_usage——读者缓存行与更新侧元数据彻底分离。
17.3.2 读侧:计数、屏障与翻转
__srcu_read_lock() / __srcu_read_unlock()
// kernel/rcu/srcutree.c:790-810
int __srcu_read_lock(struct srcu_struct *ssp)
{
struct srcu_ctr __percpu *scp = READ_ONCE(ssp->srcu_ctrp);
this_cpu_inc(scp->srcu_locks.counter);
smp_mb(); /* B */ /* Avoid leaking the critical section. */
return __srcu_ptr_to_ctr(ssp, scp);
}
EXPORT_SYMBOL_GPL(__srcu_read_lock);
void __srcu_read_unlock(struct srcu_struct *ssp, int idx)
{
smp_mb(); /* C */ /* Avoid leaking the critical section. */
this_cpu_inc(__srcu_ctr_to_ptr(ssp, idx)->srcu_unlocks.counter);
}
读侧的完整动作只有四步:
- 读当前活跃计数器组指针
ssp->srcu_ctrp(0 或 1,指向srcu_ctrs[2]之一); this_cpu_inc()给本 CPU 的srcu_locks+1——per-CPU 单条 inc 指令,读者之间零争用;- 屏障 B:
smp_mb()把"临界区内的内存访问"与"进入动作"完全隔开——没有它,弱序架构上临界区读操作可能被重排到计数递增之前,读者"计了名"却已经在读旧数据; - 返回 idx(用的是哪组计数器),解锁时对称地先屏障 C 再给同组的
srcu_unlocks+1。
注意解锁的 idx 是进入时的快照——即使宽限期中途翻转了 srcu_ctrp,退出的读者依然给旧组计数 +1,这正是双计数器协议的要求:宽限期等的是"进入时的那批人"全部退出。
翻转:srcu_flip()
宽限期的推进逻辑在 srcu_flip()(srcutree.c:1169-1210):第一遍扫描(SCAN1)确认旧组 srcu_locks == srcu_unlocks 总和相等(try_check_zero(),"所有进入该组的读者都已退出"),然后翻转 ssp->srcu_ctrp——新读者从此记入另一组。第二遍扫描(SCAN2)再等翻转前那批"已在旧组但还没退"的读者清零。两个扫描都通过,两批读者都不在了,宽限期安全结束(srcu_gp_end())。状态机由 gp_seq 低位驱动(SRCU_STATE_SCAN1 → SRCU_STATE_SCAN2 → 空闲,srcutree.h:138-140):
// kernel/rcu/srcutree.c:847-859
static void srcu_gp_start(struct srcu_struct *ssp)
{
int state;
lockdep_assert_held(&ACCESS_PRIVATE(ssp->srcu_sup, lock));
WARN_ON_ONCE(ULONG_CMP_GE(ssp->srcu_sup->srcu_gp_seq, ssp->srcu_sup->srcu_gp_seq_needed));
WRITE_ONCE(ssp->srcu_sup->srcu_gp_start, jiffies);
WRITE_ONCE(ssp->srcu_sup->srcu_n_exp_nodelay, 0);
smp_mb(); /* Order prior store to ->srcu_gp_seq_needed vs. GP start. */
rcu_seq_start(&ssp->srcu_sup->srcu_gp_seq);
state = rcu_seq_state(ssp->srcu_sup->srcu_gp_seq);
WARN_ON_ONCE(state != SRCU_STATE_SCAN1);
}
每个 srcu_struct 有自己独立的宽限期——synchronize_srcu(A) 只等 A 的读者,与全局 tick、其他 srcu_struct、标准 RCU 的宽限期完全无关。这也是 SRCU 精确性的来源:标准 RCU 的宽限期要等全系统静下来,SRCU 只等这一个结构的计数归零,通常快得多。
推进由 GP 工作队列(workqueue 上下文,可以睡眠——SRCU 的更新侧本身允许睡眠)在 srcu_advance_state()(srcutree.c:1809-1871)中轮转:
// kernel/rcu/srcutree.c:1844-1870
if (rcu_seq_state(READ_ONCE(ssp->srcu_sup->srcu_gp_seq)) == SRCU_STATE_SCAN1) {
idx = !(ssp->srcu_ctrp - &ssp->sda->srcu_ctrs[0]);
if (!try_check_zero(ssp, idx, 1)) {
mutex_unlock(&ssp->srcu_sup->srcu_gp_mutex);
return; /* readers present, retry later. */
}
srcu_flip(ssp);
raw_spin_lock_irq_rcu_node(ssp->srcu_sup);
rcu_seq_set_state(&ssp->srcu_sup->srcu_gp_seq, SRCU_STATE_SCAN2);
ssp->srcu_sup->srcu_n_exp_nodelay = 0;
raw_spin_unlock_irq_rcu_node(ssp->srcu_sup);
}
if (rcu_seq_state(READ_ONCE(ssp->srcu_sup->srcu_gp_seq)) == SRCU_STATE_SCAN2) {
idx = !(ssp->srcu_ctrp - &ssp->sda->srcu_ctrs[0]);
if (!try_check_zero(ssp, idx, 2)) {
mutex_unlock(&ssp->srcu_sup->srcu_gp_mutex);
return; /* readers present, retry later. */
}
...
srcu_gp_end(ssp); /* Releases ->srcu_gp_mutex. */
}
读者还在 → try_check_zero() 返回 false,直接返回等下轮 work 重试(line 1851-1853 与 1865-1867 的 "readers present, retry later")——SRCU 宽限期是轮询式的,与 tree RCU 的 FQS 主动推进不同。
SRCU-fast:本树的新读侧变体
屏障 B/C 是 SRCU 读侧最大的开销(x86 上是两条 mfence)。本树引入了 SRCU-fast(__srcu_read_lock_fast(),srcutree.h:289-322):读侧去掉显式屏障,改为 this_cpu_inc + 依赖隐式的 RCU 读侧保护(读者短暂进入标准 RCU 读临界区,由全局宽限期保证内存序),要求以 DEFINE_SRCU_FAST()/init_srcu_struct_fast() 声明 flavor。它面向高频短临界区(如 NMI 安全的统计读取),是 SRCU 读开销向"接近标准 RCU"演进的最新一步。
17.3.3 SRCU 与标准 RCU 的对比与选择
| 维度 | 标准 RCU | SRCU |
|---|---|---|
| 读者可否睡眠 | 否(可抢占但不可阻塞) | 是 |
| 读侧开销 | preempt_count 或任务计数(近乎免费) | 2 次 smp_mb() + per-CPU 原子加减(SRCU-fast 减半) |
| 宽限期范围 | 全局(一棵树) | 每 srcu_struct 独立 |
| 宽限期驱动 | GP 内核线程 + tick FQS | workqueue 轮询重试 |
| 内存代价 | 全局树,共享 | 每结构独立的 per-CPU 计数阵列 + 树 |
| 实例生命周期 | 内建,无 init/cleanup | init_srcu_struct() / cleanup_srcu_struct() 必须显式管理 |
| 调试 | 全局 tracepoint/锁 | srcu_struct 级 trace |
选择规则可归纳为三条:
- 读者要睡眠 → 必须 SRCU;读者保证不睡 → 标准 RCU;
- 只有少量实例且读频不高 → SRCU 的内存代价可忽略;海量实例(如每 inode 一个)→ 独立计数阵列不现实,用标准 RCU;
- 模块/设备卸载路径推荐 SRCU——卸载方可以在持有引用等待读者时睡眠,
cleanup_srcu_struct()(srcutree.c:709-753)在仍有活跃读者或未清回调时直接 WARN 并泄漏(注释原话 "Forgot srcu_barrier(), so just leak it!"),把误用提前暴露。
17.3.4 使用模式与 API
/* 定义与初始化 */
static DEFINE_SRCU(my_srcu); /* 或 init_srcu_struct(&my_srcu) */
/* 读侧:可以睡眠! */
int idx = srcu_read_lock(&my_srcu);
p = srcu_dereference(gp, &my_srcu);
...可能睡眠的操作(mutex、I/O)... /* 标准 RCU 在此会 WARN */
srcu_read_unlock(&my_srcu, idx); /* 必须传入进入时的 idx */
/* 更新侧 */
spin_lock(&gp_lock);
old = rcu_dereference_protected(gp, lockdep_is_held(&gp_lock));
rcu_assign_pointer(gp, new);
spin_unlock(&gp_lock);
synchronize_srcu(&my_srcu); /* 或 call_srcu(&my_srcu, &old->rcu, free_fn) */
kfree(old);
与标准 RCU 的三个 API 差异:读侧返回 idx 且解锁必须带回(双计数器协议的要求);发布/订阅原语复用标准 RCU 的(rcu_assign_pointer/srcu_dereference),因为发布-订阅语义两者相同;更新侧等待是 synchronize_srcu()(srcutree.c:1574-1581,内部同样基于回调 + completion)或 call_srcu(),srcu_barrier()(:1696-1731)与 rcu_barrier() 同构(mutex 串行化 + 每 CPU 追加哨兵回调)。
典型用户:设备模型(device_del() 等待所有可能使用设备的路径退出)、sysfs/procfs 卸载同步、网络子系统的 netns 释放辅助(9.4 节 cleanup_net 路径的等待语义与 SRCU 的"可睡眠宽限等待"互为补充)、以及各驱动热插拔路径。
要点总结:
- SRCU 用"显式 per-CPU 计数 + 双组翻转"替代标准 RCU 的"全局执行流事件"判定,读者因此获得睡眠自由,代价是每对读写两次全屏障。
- 宽限期按
SCAN1 → flip → SCAN2 → end轮询推进,每 srcu_struct 独立,只等自己的读者。 - 热冷两层结构(srcu_struct/srcu_usage)隔离读者缓存行与更新侧元数据;SRCU-fast 用隐式 RCU 保护替代显式屏障,是读开销演进的最新形态。
- 实例需显式 init/cleanup,误用时 cleanup 阶段的 WARN+泄漏是有意的快速失败设计。
标准 RCU 与 SRCU 各自把"读侧开销"压到极限,但它们只解决同步本身。工程上还有一种常见形态:一个执行流做完工作后通知另一个执行流——下一节的 completion 就是这个最朴素的原语。
17.4 completion、fence 与 barrier
completion(完成量)是内核里最朴素的同步原语:一个执行流做完某件事后"喊一声",另一个等待这件事的执行流被唤醒。它比信号量语义更明确——等待的是一个一次性事件,而不是一单位的资源;比 wait_queue 更简单——内建的 done 计数消除了"先睡眠还是先唤醒"的经典竞态。本章前文已两次遇到它:17.2.5 节 synchronize_rcu() 用栈上 completion 等待回调,10.5 节 vfork 的父子同步、3.8 节 kthreadd_done 同样是它的用户。本节将结合 Linux 7.0.10 内核源码,逐行分析 completion 的结构与等待循环、complete_all 的"永久完成"语义、try_wait 的无阻塞探测,最后梳理内核中"屏障"一词的两种含义——内存屏障(15.2 节)与执行流屏障(rcu_barrier()、kthread_stop 系)——的完整图景。
17.4.1 struct completion 与核心语义
数据结构
// include/linux/completion.h:26-29
struct completion {
unsigned int done;
struct swait_queue_head wait;
};
只有两个字段:done 是"已完成但尚未被取走的事件"计数;wait 是简单等待队列(swait,10.5 节 vfork 分析中已见其轻量形态——单链表、每队列固定上限的唤醒扇出,区别于复杂的普通 wait_queue)。初始化与重置:
// include/linux/completion.h:84-88, 97-100
static inline void init_completion(struct completion *x)
{
x->done = 0;
init_swait_queue_head(&x->wait);
}
...
static inline void reinit_completion(struct completion *x)
{
x->done = 0;
}
done 与等待者的关系是理解 completion 的钥匙:
| done 状态 | complete() 的效果 | 等待者的效果 |
|---|---|---|
| 0 且无人等待 | done → 1 | 先到的等待者无需睡眠直接通过 |
| 0 且有人睡 | done → 1 并唤醒一个 | 队首被唤醒 |
| ≥1(已有未取走事件) | done → 2 | 后续两个等待者免睡通过 |
done 计数使 completion 天然免疫"complete 发生在 wait 之前"的竞态——事件先于等待发生也不会丢,这与 8.2 节 set_current_state() + 条件检查的睡眠契约本质相同,只是被封装成了不可误用的 API。它对应的需求是"一次性或整数次事件";需要"广播给任意多后来者"时用 complete_all()。
complete() 与 complete_all()
// kernel/sched/completion.c:21-31, 50-53, 72-82
static void complete_with_flags(struct completion *x, int wake_flags)
{
unsigned long flags;
raw_spin_lock_irqsave(&x->wait.lock, flags);
if (x->done != UINT_MAX)
x->done++;
swake_up_locked(&x->wait, wake_flags);
raw_spin_unlock_irqrestore(&x->wait.lock, flags);
}
...
void complete(struct completion *x)
{
complete_with_flags(x, WF_SYNC);
}
...
void complete_all(struct completion *x)
{
unsigned long flags;
lockdep_assert_RT_in_threaded_ctx();
raw_spin_lock_irqsave(&x->wait.lock, flags);
x->done = UINT_MAX;
swake_up_all_locked(&x->wait);
raw_spin_unlock_irqrestore(&x->wait.lock, flags);
}
complete()(line 50-53):done++ 后唤醒一个等待者(WF_SYNC 提示调度器"唤醒后很可能很快让出 CPU",见 13.2 节 wake affinity)。done != UINT_MAX 的判断是给 complete_all 让路。
complete_all()(line 72-82):把 done 置为 UINT_MAX 并唤醒全部等待者。UINT_MAX 是一个哨兵值——表示"永久完成,此后所有等待者立即通过、且不消耗计数"。注意与旧教材的差异:本树置 UINT_MAX(而非早期内核的 UINT_MAX/2),所有消费 done 的路径都改为与 UINT_MAX 比较(下文三处可见)。complete_all() 之后 completion 进入终态,复用前必须 reinit_completion()——hwrng 驱动(17.4.4 节)正是"init → complete_all → reinit"循环的实例。line 75 的 lockdep_assert_RT_in_threaded_ctx() 禁止 PREEMPT_RT 下在硬中断上下文调用 complete_all()(swait 唤醒全部在 RT 语义下不允许从中断发起)。
17.4.2 等待侧:do_wait_for_common() 主循环
// kernel/sched/completion.c:85-110
static inline long __sched
do_wait_for_common(struct completion *x,
long (*action)(long), long timeout, int state)
{
if (!x->done) {
DECLARE_SWAITQUEUE(wait);
do {
if (signal_pending_state(state, current)) {
timeout = -ERESTARTSYS;
break;
}
__prepare_to_swait(&x->wait, &wait);
__set_current_state(state);
raw_spin_unlock_irq(&x->wait.lock);
timeout = action(timeout);
raw_spin_lock_irq(&x->wait.lock);
} while (!x->done && timeout);
__finish_swait(&x->wait, &wait);
if (!x->done)
return timeout;
}
if (x->done != UINT_MAX)
x->done--;
return timeout ?: 1;
}
主循环逐段拆解:
- 免睡快路径(line 88-89 与 107-109):
done非零直接跳过循环——事件已发生,取走一个(done--)即返回。整个"已无竞争的等待"只有一次读判断。 - 信号检查(line 91-94):
signal_pending_state()按等待状态判断是否被信号打断——TASK_INTERRUPTIBLE变体响应任意信号(14.4 节信号唤醒路径),TASK_KILLABLE只响应致命信号,TASK_UNINTERRUPTIBLE完全不响应。 - 标准睡眠序列(line 95-101):入 swait 队列 →
__set_current_state(state)→ 先放锁再调度(action()即schedule_timeout()或io_schedule_timeout(),依据变体是否标记 io 等待)→ 醒来重新持锁检查条件。这正是 8.2.4 节"先设状态、再检查条件、再睡眠"三步契约的标准实现,持锁完成消除了"检查与睡眠之间丢事件"的窗口。 - 退出条件(line 102):
done非零(完成)或timeout耗尽(超时变体返回剩余时间,供调用方判断)。 - 消费事件(line 107-109):只有普通 complete 攒下的计数才被取走;
UINT_MAX永不消耗——complete_all 的广播语义。
各变体只是给 state 与 action 传不同值(completion.c:151-295):wait_for_completion()(UNINTERRUPTIBLE)、..._interruptible()、..._killable()(TASK_KILLABLE,10.5 节 vfork 用的正是它——父进程等待期间可被 SIGKILL 解救)、..._timeout()/..._io()(带超时与 io 权重统计)、wait_for_completion_state()(自定义状态,vfork 传 TASK_KILLABLE | TASK_FREEZABLE)。
非阻塞探测
// kernel/sched/completion.c:309-330, 342-357
bool try_wait_for_completion(struct completion *x)
{
unsigned long flags;
bool ret = true;
if (!READ_ONCE(x->done))
return false;
raw_spin_lock_irqsave(&x->wait.lock, flags);
if (!x->done)
ret = false;
else if (x->done != UINT_MAX)
x->done--;
raw_spin_unlock_irqrestore(&x->wait.lock, flags);
return ret;
}
...
bool completion_done(struct completion *x)
{
unsigned long flags;
if (!READ_ONCE(x->done))
return false;
raw_spin_lock_irqsave(&x->wait.lock, flags);
raw_spin_unlock_irqrestore(&x->wait.lock, flags);
return true;
}
try_wait_for_completion()(line 309-330):无阻塞地"取走一个已完成事件",先无锁快判再加锁确认——双检防止与 complete 并发时的计数撕裂。completion_done()(line 342-357)语义不同:它回答"是否已无未取走事件"(不消耗计数),用于卸载路径判断"还有没有人在等";内部的空临界区(拿锁即放)不是为了保护读,而是与 complete() 建立内存序——保证前一次 complete 的效果对本次判断可见。
17.4.3 使用模式
单次握手:vfork 与 kthreadd
内核中 completion 最常见的形态是单向一次性握手:10.5 节的 vfork——子进程 complete_vfork_done() 通知父进程"地址空间已释放";3.8 节 rest_init()——kernel_init 线程就绪后 complete(&kthreadd_done) 让启动线程继续。共同特征:事件只发生一次、等待者唯一,用 complete() 而非 complete_all。
终态广播 + 重初始化:hwrng 驱动
// drivers/char/hw_random/core.c:589-591, 620-641
if (rng->cleanup)
rng->cleanup(rng);
complete(&rng->cleanup_done);
...
init_completion(&rng->cleanup_done);
complete(&rng->cleanup_done);
init_completion(&rng->dying);
...
static void hwrng_unregister(struct hwrng *rng)
{
...
mutex_lock(&rng_mutex);
list_del(&rng->list);
complete_all(&rng->dying); /* 广播:本 rng 正在消亡 */
...
mutex_unlock(&rng_mutex);
wait_for_completion(&rng->cleanup_done); /* 等最后一次清理完成 */
}
这个文件是 completion 多种语义的同框教材:
cleanup_done初始化后立即 complete 一次(line 625-626)——预置"已发生"状态,使首次wait_for_completion()不被卡住,之后每次 cleanup 工作完成再 complete,形成"事件槽"模式;dying用complete_all()广播终态——任意多个读者(读操作打开 hwrng 设备的进程)都能立即看到"设备已注销",无需逐个唤醒;- 卸载路径
wait_for_completion(&rng->cleanup_done)与 kref(rng->ref,core.c:80-97)配合完成生命周期收尾——一个驱动同时使用 kref + completion + work,是第 16、17 章原语协同的缩影。
completion 与信号量的对比
| 维度 | completion | semaphore |
|---|---|---|
| 语义 | 一次性事件 | 可计数资源 |
| 唤醒公平性 | FIFO(swait) | 无保证 |
| done=0 时的 complete | 攒下一次事件(不丢) | up 后可能无人等待(不攒) |
| 典型误用 | —— | 用信号量当"锁+事件"双用途导致难查竞态 |
| 调试友好度 | 状态简单,一目了然 | 历史上大量误用,Documentation 建议仅用于资源 |
内核文档明确建议:新代码表示"等待某事完成"一律用 completion;信号量只保留给真正的多额度资源场景(且多为历史代码)。其根源就是 complete() 的"攒事件"语义与 semaphore up() 的"无主唤醒"语义在 done=0 时的分岔。
17.4.4 fence 与 barrier:两个层面的"同步墙"
本章标题里的"fence 与 barrier"在内核语境下必须分层理解——它们是两族完全不同的机制,只是中文译名相近。
第一层:内存 fence——单 CPU 的访存顺序
这是 15.2 节的领域:smp_mb()/smp_wmb()/smp_rmb() 与 ACQUIRE/RELEASE(smp_load_acquire()/smp_store_release())约束的是同一个 CPU 上访存指令的可见顺序,不涉及任何执行流等待。本章各处已见到它们作为同步原语的"地基":seqcount 的两道 smp_wmb()(16.3.5 节)、rcu_assign_pointer() 的 release(17.1.2 节)、SRCU 读侧的屏障 B/C(17.3.2 节)。fence 是原语的原料,不是原语本身——直接裸用 smp_mb() 协调两个执行流的代码在内核中属于底层设施(memory barrier 范式,Documentation/memory-barriers.txt 有完整规范),子系统层面几乎总是改用封装好的原语。
第二层:执行流 barrier——"等所有参与者到齐"
这是与 completion 同层的机制,内核中有两个代表:
rcu_barrier()(17.2.4 节):模块卸载前确保"所有已注册的 RCU 回调执行完毕"。实现是对每个有回调的 CPU 下发哨兵回调(IPI 追加),用原子计数 + completion 等待哨兵全部跑完——注意它等的是回调清空,不是宽限期(回调入队后所属宽限期可能尚未开始),这是卸载路径最常见的误解。
kthread 停止协议:kthread_stop() 置位停止标志、唤醒目标线程,然后 wait_for_completion(&k->exited) 等待线程真正退出(线程退出时 complete_kthread_done)——一次"请求-确认"双向握手,两个 completion(should_stop 事件的等待由线程侧轮询,退出事件用 completion 回传)。
三者共同点:等待的是"若干执行流/批处理到达某个点",而非内存可见性。选用判据一句话:要约束访存顺序用 fence(通常已藏在原语里);要等"人到齐"用 barrier 类;要等"一件事完成"用 completion。
17.4.5 顺带一提:kref 与 llist
作为本章(也是第三部分)的收尾,补充两个高频出现的轻量设施。
kref(include/linux/kref.h:19-21, 62-69)是 refcount_t 的薄语义包装(struct kref { refcount_t refcount; }):kref_get()/kref_put() 强制走 refcount 的饱和与 UAF 检查(15.1.3 节),kref_put() 归零时调用调用方提供的 release 回调。它确立的是命名约定——看到 kref 就知道这是引用计数管理生命周期,kref_put_mutex()/kref_put_lock() 还能把"减到零"与"持锁执行 release"原子绑定。hwrng 的 rng->ref(17.4.3 节)即标准用户。
llist(include/linux/llist.h:234-245):基于 cmpxchg 的无锁 LIFO 链表,llist_add_batch() 用一个 do-while cmpxchg 循环完成多生产者头插,单消费者 llist_del_all() 一次摘取全部——多生产者单消费者的零锁队列。真实用户:irq_work(kernel/irq_work.c:27-28 的 per-CPU raised_list/lazy_list,入队 llist_add() :107,执行侧 llist_del_all() :250——NMI 中也能安全排队工作的关键)、17.2.5 节 synchronize_rcu() 的等待者队列 srs_next。它与 per-CPU 变量(15.3 节)、RCU(17.1 节)共同构成内核"减少共享写"的三板斧。
要点总结:
- completion = done 计数 + swait 队列:complete 攒事件、complete_all 置 UINT_MAX 哨兵广播(本树语义,非旧版的 UINT_MAX/2)、等待循环是"持锁设状态-放锁睡眠-回来检查"的标准契约。
- 单次握手用 complete,终态广播用 complete_all(复用需 reinit),非阻塞探测用 try_wait/completion_done;"等待完成"语义新代码一律用 completion 而非信号量。
- fence(内存序,15.2 节)与 barrier(执行流集合点:rcu_barrier、kthread_stop)是两个层面的机制;fence 是原语原料,barrier 是集合点,completion 是单事件等待。
- kref 与 llist 作为引用计数与无锁链表的轻量设施,与 RCU、per-CPU 共同收束第三部分"消除共享写"的主题。
至此第三部分完结:从最底层的原子指令与内存屏障(15 章),到互斥的两大家族锁(16 章),再到读侧无锁的 RCU 家族与事件通知的 completion(17 章),内核的同步工具箱呈金字塔展开——越往上层,语义越特化、误用面越小、性能与正确性越难两全的权衡越早被固化进 API。理解每一层的适用边界,是读懂后续内存管理、文件系统与网络子系统中并发设计的前提。