Linux RCU 源码分析

RCU (Rred Copy Update) 是内核实现并发的核心机制之一，本文通过分析 Linux-6.18.19 的源码总结了 Tree RCU 与 SRCU (Sleepable RCU) 的核心工作原理。

1. 同步机制对比

2. RCU API

2.1 读取侧

rcu_dereference：保证本次操作稳定读取一个指针值；

rcu_read_lock：保证这个指针指向的旧对象暂时不会被释放；

struct foo *p;

rcu_read_lock();

p = rcu_dereference(global_foo);
if (p)
  do_something(p);

rcu_read_unlock();

2.2 发布侧

rcu_assign_pointer：保证指针的初始化对所有读者可见；

struct foo *new;

new = kmalloc(sizeof(*new), GFP_KERNEL);
if (!new)
  return -ENOMEM;

new->a = 1;
new->b = 2;

/* 对象全部初始化完成后再发布。 */
rcu_assign_pointer(global_foo, new);

2.3 同步回收

进程阻塞在 synchronize_rcu 上等待 GP 结束后被唤醒释放资源；

struct foo *old;
struct foo *new;

new = alloc_foo();

mutex_lock(&foo_mutex);

old = rcu_replace_pointer(global_foo, new,
          lockdep_is_held(&foo_mutex));

mutex_unlock(&foo_mutex);

synchronize_rcu();
kfree(old);

2.4 异步回收

进程注册释放资源的回调函数，等到 GP 结束后在软中断中被执行；

struct foo {
  int value;
  struct rcu_head rcu;
};

static void foo_free_rcu(struct rcu_head *head)
{
  struct foo *foo;

  foo = container_of(head, struct foo, rcu);
  kfree(foo);
}

struct foo *old;

mutex_lock(&foo_mutex);

old = rcu_dereference_protected(
  global_foo, lockdep_is_held(&foo_mutex));

rcu_assign_pointer(global_foo, NULL);

mutex_unlock(&foo_mutex);

if (old)
  call_rcu(&old->rcu, foo_free_rcu);

3. Tree RCU

每个 GP 有8个状态：

RCU_GP_WAIT_GPS  (1)  →  等待有人请求 GP
RCU_GP_DONE_GPS  (2)  →  收到请求，准备初始化
RCU_GP_ONOFF     (3)  →  应用 CPU 热插拔变更
RCU_GP_INIT      (4)  →  初始化所有叶节点的 qsmask
RCU_GP_WAIT_FQS  (5)  →  等待 QS 或超时
RCU_GP_DOING_FQS (6)  →  正在强制执行 QS
RCU_GP_CLEANUP   (7)  →  GP 结束，清理
RCU_GP_CLEANED   (8)  →  清理完成，回到 1

3.1 关键数据结构

3.1.1 rcu_state

该结构体保存RCU全局状态，管理RCU节点、维护状态机、保存重要标识字段等；

   rcu_state
   │
   ├── gp_seq ★              全局 GP 序号（偶=空闲 奇=活跃）
   ├── gp_kthread ★           GP 内核线程（全局唯一）
   ├── gp_wq ★                GP kthread 睡在这里等活
   ├── gp_flags ★             RCU_GP_FLAG_INIT / FQS
   ├── gp_state ★             WAIT_GPS → DONE_GPS → ... → CLEANED
   ├── gp_activity ★          GP kthread 最近活跃时刻（stall 检测用）
   ├── cbovld ★               回调是否积压（加速 FQS）
   ├── jiffies_force_qs ★     下次 FQS deadline
   ├── node[] ★               rcu_node 数组（N叉树）
   ├── level[]                每层首节点指针
   │
   ├── srs_next ★             synchronize_rcu 新请求（llist）
   ├── srs_wait_tail ★        等当前 GP 的请求链表
   ├── srs_done_tail          GP 完成待唤醒的请求
   │
   ├── barrier_cpu_count      rcu_barrier 还剩几个 CPU
   ├── expedited_need_qs      加速 GP 还剩几个 CPU
   │
   ├── gp_start              本轮 GP 开始时刻
   └── jiffies_stall          下次 stall 检查时刻

3.1.2 rcu_node

RCU节点以多叉树的形式组织，每个叶节点使用qsmask的bitmap标识一组CPU是否已经QS，同时还保存在当前GP期间阻塞的进程；

   rcu_node
   │
   ├── lock ★                 保护本节点
   ├── gp_seq                 本节点跟踪的 gp_seq
   ├── qsmask ★               未报 QS 的子节点/CPU 位掩码
   ├── expmask                 加速 GP 独立掩码
   ├── parent                  父节点（根=NULL）
   ├── level                   层级（根=0）
   ├── grpmask                 在父节点 qsmask 中的位
   │
   ├── blkd_tasks ★            被抢占读者的链表
   ├── gp_tasks ★              当前 GP 等的第一个阻塞任务
   ├── exp_tasks               加速 GP 等的第一个阻塞任务
   │
   └── completedqs             本节点 QS 全完成的 gp_seq

3.1.3 rcu_data

每个 rcu_data 对应一个CPU，使用分段的回调链表管理何时能够进入GP；

rcu_data (per-CPU)
   │
   ├── mynode ★               指向所属叶 rcu_node
   ├── grpmask ★              在叶节点 qsmask 中的位
   ├── gp_seq                  本 CPU 看到的 gp_seq
   ├── cpu_no_qs ★             本 CPU 是否还未报 QS
   │   ├── .b.norm             普通 GP
   │   └── .b.exp              加速 GP
   ├── defer_qs_pending        延迟 QS 排队中？
   │
   ├── cblist ★                分段回调链表
   │   └── DONE / WAIT / NEXT_READY / NEXT
   │
   ├── rcu_cpu_kthread_task    rcuc 内核线程
   ├── rcu_cpu_has_work        有活要干
   │
   ├── rcu_urgent_qs           GP 太久 → tick 中强制 QS
   └── ticks_this_gp           本轮 GP 本 CPU 处理的 tick 数

3.1.4 task_struct

每个进程中，会标识进入RCU临界区的嵌套深度，以及在临界区中被阻塞的标志；

task_struct (per-task，相关字段)
   │
   ├── rcu_read_lock_nesting ★ 嵌套深度（0=不在临界区）
   ├── rcu_read_unlock_special ★ need_qs / blocked 标记
   ├── rcu_blocked_node ★      被抢占时所属叶节点
   └── rcu_node_entry          挂在 blkd_tasks 的链表节点

3.2 核心流程

全局的 rcu_state 中的 gp_seq 为奇数时表示目前正在 GP 内，为偶数时表示正在等待新的 GP；

3.2.1 读者进入临界区

无论 GP 是否开启，读者进入临界区的操作都是相同的，即调用 rcu_read_lock，而核心在于其调用的 __rcu_read_lock，该函数在非抢占内核内的实现十分简单，几乎无开销：

static inline void __rcu_read_lock(void)
{
	preempt_disable(); // 增加 preempt_count 引用计数
}

而在可抢占内核（CONFIG_PREEMPT = y）下，该函数的实现要复杂一些：

1. 首先在 rcu_preempt_read_enter 中会将 task_struct中的 nesting 字段加1，该字段用于标识当前进程正在临界区内，当临界区嵌套时该字段会大于1，同时会对嵌套的最大层数有一个检测，超过INT_MAX / 2 时会产生一个警告：
2. 如果内核配置了 CONFIG_RCU_STRICT_GRACE_PERIOD，表示开启了 strict 模式，那么会将 task_struct 中的 need_qs 置位，在退出临界区时会主动上报 QS；

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. */
}

static void rcu_preempt_read_enter(void)
{
	WRITE_ONCE(current->rcu_read_lock_nesting, READ_ONCE(current->rcu_read_lock_nesting) + 1);
}

3.2.2 写者发布

写者通过该宏修改指针：

1. 首先调用 rcu_check_sparse 在编译期检查要修改的指针是否被 __rcu 所修饰；
2. 然后会判断传入的 v 是否为 NULL，如果是则表示取消发布，此时直接写入 p 即可；
3. 否则，smp_store_release 会使用内存屏障，使得在该指令之前的 对 v 的 store 都已经完成；

#define rcu_assign_pointer(p, v)					      \
do {									      \
	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)); \
} while (0)

3.2.3 写者等待

写者完成发布后需要等待在发布之前就已经在临界区中读到旧内容的读者退出临界区：

1. lockdep检查，禁止在 RCU 的读临界区中调用该函数，否则会导致死锁；
2. 然后调用 rcu_blocking_is_gp 判断目前调度器是否可用，如果可用则返回 false，根据当前 GP 是否为 expedited 调用不同的函数；
3. 如果当前调度器不可用，那么应该只有单核在执行，也就不需要任何的同步，这里会手动将 gp_seq 加2，直接跳过当前 GP，最后修改树结构中的所有节点的 gp_seq；
4. 内核初始化完成后，调度器都是可用的，如果是正常的 GP 则会来到 synchronize_rcu_xxx，

void synchronize_rcu(void)
{
	unsigned long flags;
	struct rcu_node *rnp;

	RCU_LOCKDEP_WARN(lock_is_held(&rcu_bh_lock_map) ||
			 lock_is_held(&rcu_lock_map) ||
			 lock_is_held(&rcu_sched_lock_map),
			 "Illegal synchronize_rcu() in RCU read-side critical section");
	if (!rcu_blocking_is_gp()) {
		if (rcu_gp_is_expedited())
			synchronize_rcu_expedited();
		else
			synchronize_rcu_normal();
		return;
	}

	// 手动开关 GP，供 polling API 使用
	rcu_poll_gp_seq_start_unlocked(&rcu_state.gp_seq_polled_snap);
	rcu_poll_gp_seq_end_unlocked(&rcu_state.gp_seq_polled_snap);

	// Update the normal grace-period counters to record
	// this grace period, but only those used by the boot CPU.
	// The rcu_scheduler_starting() will take care of the rest of
	// these counters.
	local_irq_save(flags);
	WARN_ON_ONCE(num_online_cpus() > 1);
	rcu_state.gp_seq += (1 << RCU_SEQ_CTR_SHIFT);
	for (rnp = this_cpu_ptr(&rcu_data)->mynode; rnp; rnp = rnp->parent)
		rnp->gp_seq_needed = rnp->gp_seq = rcu_state.gp_seq;
	local_irq_restore(flags);
}

3.2.3.1 普通GP

synchronize_rcu_normal：首先判断 rcu_normal_wake_from_gp 是否小于1，该变量会在 rcu_init 中进行赋值，如果 CPU 数量不超过16的话会被置1，否则为-1；

当 CPU 数不超过16时，会初始化 rs 结构体，然后调用 rcu_sr_normal_add_req 将其添加到ru_state 的 srs_next 链表中，然后会调用 start_poll_synchronize_rcu 计算出一个gp_seq_req 并传入 rcu_start_this_gp 中，尝试开启新的 GP；然后该函数就会阻塞等待 rs.completion 完成；

如果CPU数超过16，则会调用 wait_rcu_gp，最终通过复用 call_rcu ，在 rcu_data 上的 cblist的RCU_NEXT 段上挂载一个 wakeme_after_rcu 回调，那么等到 GP 结束后最终就会通过软中断执行该回调；

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);

	/*
	 * This code might be preempted, therefore take a GP
	 * snapshot before adding a request.
	 */
	if (IS_ENABLED(CONFIG_PROVE_RCU))
		get_state_synchronize_rcu_full(&rs.oldstate);

	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"));
}

static bool rcu_start_this_gp(struct rcu_node *rnp_start, struct rcu_data *rdp,
			      unsigned long gp_seq_req)
{
	bool ret = false;
	struct rcu_node *rnp;

	raw_lockdep_assert_held_rcu_node(rnp_start);
	trace_rcu_this_gp(rnp_start, rdp, gp_seq_req, TPS("Startleaf"));
	// 从叶节点开始向上遍历
    for (rnp = rnp_start; 1; rnp = rnp->parent) {
		if (rnp != rnp_start)
			raw_spin_lock_rcu_node(rnp);
        // 如果：
        // 1. 当前节点的gp_seq_needed 大于请求的gp_seq_req，无需写入；
        // 2. 当前请求的gp_seq_req对应的GP已经开始，无需写入；
        // 3. 当前节点非叶 并且 该节点正在GP内，不再向上传递，等待cleanup处理；
        // 则会跳过节点写入
		if (ULONG_CMP_GE(rnp->gp_seq_needed, gp_seq_req) ||
		    rcu_seq_started(&rnp->gp_seq, gp_seq_req) ||
		    (rnp != rnp_start &&
		     rcu_seq_state(rcu_seq_current(&rnp->gp_seq)))) {
			trace_rcu_this_gp(rnp, rdp, gp_seq_req,
					  TPS("Prestarted"));
			goto unlock_out;
		}
        // 写入当前节点，同时判断当前节点是否正在GP内，这个处理是为了至少有一个节点被写入请求；
		WRITE_ONCE(rnp->gp_seq_needed, gp_seq_req);
		if (rcu_seq_state(rcu_seq_current(&rnp->gp_seq))) {

			trace_rcu_this_gp(rnp_start, rdp, gp_seq_req,
					  TPS("Startedleaf"));
			goto unlock_out;
		}
		if (rnp != rnp_start && rnp->parent != NULL)
			raw_spin_unlock_rcu_node(rnp);
		if (!rnp->parent)
			break;  /* At root, and perhaps also leaf. */
	}

	// 遍历完成
    // GP已经开始，无需唤醒；
	if (rcu_gp_in_progress()) {
		trace_rcu_this_gp(rnp, rdp, gp_seq_req, TPS("Startedleafroot"));
		goto unlock_out;
	}
	trace_rcu_this_gp(rnp, rdp, gp_seq_req, TPS("Startedroot"));
	// 向rcu_state中写入标志位，后续会到rcu_state.gp_wq中唤醒内核线程
    WRITE_ONCE(rcu_state.gp_flags, rcu_state.gp_flags | RCU_GP_FLAG_INIT);
	WRITE_ONCE(rcu_state.gp_req_activity, jiffies);
	if (!READ_ONCE(rcu_state.gp_kthread)) {
		trace_rcu_this_gp(rnp, rdp, gp_seq_req, TPS("NoGPkthread"));
		goto unlock_out;
	}
	trace_rcu_grace_period(rcu_state.name, data_race(rcu_state.gp_seq), TPS("newreq"));
	ret = true;  /* Caller must wake GP kthread. */
unlock_out:
	// 如果当前请求比遍历到的某节点的gp_seq_needed还小，则写入叶子节点；
	if (ULONG_CMP_LT(gp_seq_req, rnp->gp_seq_needed)) {
		WRITE_ONCE(rnp_start->gp_seq_needed, rnp->gp_seq_needed);
		WRITE_ONCE(rdp->gp_seq_needed, rnp->gp_seq_needed);
	}
	if (rnp != rnp_start)
		raw_spin_unlock_rcu_node(rnp);
	return ret;
}

3.2.3.2 加速GP

而加速 GP 的处理则会调用 synchronize_rcu_expedited，使用 IPI 代替普通 GP 内核线程的复杂管理，能够降低QS延迟，但代价是代价是 CPU 开销大：

1. 加速 GP 使用单独的gp_seq：rcu_state.expedited_sequence；
2. 所有进程尝试同一个漏斗锁，如果获取不到说明已经有人在做了；
3. 拿到锁的进程调用 synchronize_rcu_expedited_queue_work 向工作队列中添加一个任务；
4. rcu_exp_gp_kworker 调用 wait_rcu_exp_gp，最终会使得每个节点的 rnp->exp_kworker 向自己管理的 CPU 发送 IPI 来加速 QS；
5. 直接阻塞等待工作队列完成，释放漏斗锁；

void synchronize_rcu_expedited(void)
{
	unsigned long flags;
	struct rcu_exp_work rew;
	struct rcu_node *rnp;
	unsigned long s;

	RCU_LOCKDEP_WARN(lock_is_held(&rcu_bh_lock_map) ||
			 lock_is_held(&rcu_lock_map) ||
			 lock_is_held(&rcu_sched_lock_map),
			 "Illegal synchronize_rcu_expedited() in RCU read-side critical section");

	/* Is the state is such that the call is a grace period? */
	if (rcu_blocking_is_gp()) {
		// Note well that this code runs with !PREEMPT && !SMP.
		// In addition, all code that advances grace periods runs
		// at process level.  Therefore, this expedited GP overlaps
		// with other expedited GPs only by being fully nested within
		// them, which allows reuse of ->gp_seq_polled_exp_snap.
		rcu_poll_gp_seq_start_unlocked(&rcu_state.gp_seq_polled_exp_snap);
		rcu_poll_gp_seq_end_unlocked(&rcu_state.gp_seq_polled_exp_snap);

		local_irq_save(flags);
		WARN_ON_ONCE(num_online_cpus() > 1);
		rcu_state.expedited_sequence += (1 << RCU_SEQ_CTR_SHIFT);
		local_irq_restore(flags);
		return;  // Context allows vacuous grace periods.
	}

	/* If expedited grace periods are prohibited, fall back to normal. */
	if (rcu_gp_is_normal()) {
		synchronize_rcu_normal();
		return;
	}

	/* 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_ctr(s) & 0x3],
		   sync_exp_work_done(s));

	/* Let the next expedited grace period start. */
	mutex_unlock(&rcu_state.exp_mutex);
}

3.2.4 CPU上报QS

3.2.4.1 进程切换

每次进程切换时，在 __schedule 中会调用如下函数上报 QS，不过普通 gp 调用的 rcu_qs 只会清理 rcu_data.cpu_no_qs.b.norm，而 node 上的 qsmask 要在 tick 中断中的软中断中进行上报：

void rcu_note_context_switch(bool preempt)
{
	struct task_struct *t = current;
	struct rcu_data *rdp = this_cpu_ptr(&rcu_data);
	struct rcu_node *rnp;

	trace_rcu_utilization(TPS("Start context switch"));
	lockdep_assert_irqs_disabled();
	WARN_ONCE(!preempt && rcu_preempt_depth() > 0, "Voluntary context switch within RCU read-side critical section!");
    // 当前进程在读临界区内被阻塞，需要将自己挂到阻塞链表中
	if (rcu_preempt_depth() > 0 &&
	    !t->rcu_read_unlock_special.b.blocked) {

		/* Possibly blocking in an RCU read-side critical section. */
		rnp = rdp->mynode;
		raw_spin_lock_rcu_node(rnp);
		t->rcu_read_unlock_special.b.blocked = true;
		t->rcu_blocked_node = rnp;

		/*
		 * Verify the CPU's sanity, trace the preemption, and
		 * then queue the task as required based on the states
		 * of any ongoing and expedited grace periods.
		 */
		WARN_ON_ONCE(!rcu_rdp_cpu_online(rdp));
		WARN_ON_ONCE(!list_empty(&t->rcu_node_entry));
		trace_rcu_preempt_task(rcu_state.name,
				       t->pid,
				       (rnp->qsmask & rdp->grpmask)
				       ? rnp->gp_seq
				       : rcu_seq_snap(&rnp->gp_seq));
		rcu_preempt_ctxt_queue(rnp, rdp);
	} else {
        // 处理 加速gp / special.b.need_qs / special.b.blocked
		rcu_preempt_deferred_qs(t);
	}

	// 普通gp，本地 qs 完成
	rcu_qs();
	if (rdp->cpu_no_qs.b.exp)
		rcu_report_exp_rdp(rdp);
	rcu_tasks_qs(current, preempt);
	trace_rcu_utilization(TPS("End context switch"));
}

非抢占内核的处理无需前面的判断，因为每次进程切换时必然 qs，所以直接调用 rcu_qs：

void rcu_note_context_switch(bool preempt)
{
	trace_rcu_utilization(TPS("Start context switch"));
	rcu_qs();
	/* Load rcu_urgent_qs before other flags. */
	if (!smp_load_acquire(this_cpu_ptr(&rcu_data.rcu_urgent_qs)))
		goto out;
	this_cpu_write(rcu_data.rcu_urgent_qs, false);
	if (unlikely(raw_cpu_read(rcu_data.rcu_need_heavy_qs)))
		rcu_momentary_eqs();
out:
	rcu_tasks_qs(current, preempt);
	trace_rcu_utilization(TPS("End context switch"));
}

3.3.4.2 tick中断

tick 中断处理会调用 rcu_sched_clock_irq ：

1. 首先调用下面的 rcu_flavor_sched_clock_irq，判断当前 CPU 中的 QS 情况；
2. 调用 rcu_pending，如果 core_needs_qs ==1 并且 rdp->cpu_no_qs.b.norm == 0 则说明本CPU 已经 QS，调用 invoke_rcu_core 触发软中断；

static void rcu_flavor_sched_clock_irq(int user)
{
	struct task_struct *t = current;

	lockdep_assert_irqs_disabled();
    // 在临界区 或者 关闭抢占/关闭BH -> 延后处理
	if (rcu_preempt_depth() > 0 ||
	    (preempt_count() & (PREEMPT_MASK | SOFTIRQ_MASK))) {
		/* No QS, force context switch if deferred. */
		if (rcu_preempt_need_deferred_qs(t)) {
			set_tsk_need_resched(t);
			set_preempt_need_resched();
		}
    // 不再临界区 并且 被标记 need_qs -> 处理欠账
	} else if (rcu_preempt_need_deferred_qs(t)) {
		rcu_preempt_deferred_qs(t); /* Report deferred QS. */
		return;
    // 不再临界区 并且 未被标记 need_qs -> 上报 qs
	} else if (!WARN_ON_ONCE(rcu_preempt_depth())) {
		rcu_qs(); /* Report immediate QS. */
		return;
	}

    //在临界区 并且 未完成QS 并且 未被标记 need_qs 并且 超时 -> 标记 need_qs
	/* If GP is oldish, ask for help from rcu_read_unlock_special(). */
	if (rcu_preempt_depth() > 0 &&
	    __this_cpu_read(rcu_data.core_needs_qs) &&
	    __this_cpu_read(rcu_data.cpu_no_qs.b.norm) &&
	    !t->rcu_read_unlock_special.b.need_qs &&
	    time_after(jiffies, rcu_state.gp_start + HZ))
		t->rcu_read_unlock_special.b.need_qs = true;
}

3.2.4.3 EQS

CPU 在用户态 / idle / 离线时 ，会修改percpu 变量 context-tracking state 中的状态，后续 gp kthread 根据此状态就可以替该 CPU 上报 QS；

3.2.4.4 软中断完成上报

在 tick 硬中断中，如果是非 rt 内核会使用软中断完成后续的上报任务，如果是 rt 内核则会启动一个 rcuc 内核线程：

/* Perform RCU core processing work for the current CPU.  */
static __latent_entropy void rcu_core(void)
{
	unsigned long flags;
	struct rcu_data *rdp = raw_cpu_ptr(&rcu_data);
	struct rcu_node *rnp = rdp->mynode;

	if (cpu_is_offline(smp_processor_id()))
		return;
	trace_rcu_utilization(TPS("Start RCU core"));
	WARN_ON_ONCE(!rdp->beenonline);

	/* Report any deferred quiescent states if preemption enabled. */
	if (IS_ENABLED(CONFIG_PREEMPT_COUNT) && (!(preempt_count() & PREEMPT_MASK))) {
		rcu_preempt_deferred_qs(current);
	} else if (rcu_preempt_need_deferred_qs(current)) {
		set_tsk_need_resched(current);
		set_preempt_need_resched();
	}

	/* Update RCU state based on any recent quiescent states. */
	rcu_check_quiescent_state(rdp);

	/* No grace period and unregistered callbacks? */
	if (!rcu_gp_in_progress() &&
	    rcu_segcblist_is_enabled(&rdp->cblist) && !rcu_rdp_is_offloaded(rdp)) {
		local_irq_save(flags);
		if (!rcu_segcblist_restempty(&rdp->cblist, RCU_NEXT_READY_TAIL))
			rcu_accelerate_cbs_unlocked(rnp, rdp);
		local_irq_restore(flags);
	}

	rcu_check_gp_start_stall(rnp, rdp, rcu_jiffies_till_stall_check());

	/* If there are callbacks ready, invoke them. */
	if (!rcu_rdp_is_offloaded(rdp) && rcu_segcblist_ready_cbs(&rdp->cblist) &&
	    likely(READ_ONCE(rcu_scheduler_fully_active))) {
		rcu_do_batch(rdp);
		/* Re-invoke RCU core processing if there are callbacks remaining. */
		if (rcu_segcblist_ready_cbs(&rdp->cblist))
			invoke_rcu_core();
	}

	/* Do any needed deferred wakeups of rcuo kthreads. */
	do_nocb_deferred_wakeup(rdp);
	trace_rcu_utilization(TPS("End RCU core"));

	// If strict GPs, schedule an RCU reader in a clean environment.
	if (IS_ENABLED(CONFIG_RCU_STRICT_GRACE_PERIOD))
		queue_work_on(rdp->cpu, rcu_gp_wq, &rdp->strict_work);
}

3.2.5 kthread 推进状态

内核通过一个内核线程管理 GP 状态机：

1. 最初在 WAIT_GPS 状态，该内核线程阻塞在 rcu_state.gp_wq上等待 flag 上被设置 RCU_GP_FLAG_INIT;
2. 被唤醒后进入 DONE_GPS 状态，调用 rcu_gp_init 进行新 GP 的初始化，然后调用到 rcu_gp_fqs_loop 收集 QS；
3. 收集完成后进入 CLEANUP 状态，调用 rcu_gp_cleanup 进行收尾工作；
4. 收尾结束后进入 CLEANED 状态，然后进入下一轮循环，回到 WAIT_GPS 状态；

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);
	}
}

rcu_gp_init 的核心功能如下：

1. 调用 rcu_sr_normal_gp_init 收割 src_next，将当前该链表中的节点拿到，使用 wait_head指向；
2. 调用 rcu_seq_start将全局的 gp_seq 加1；
3. 调用 rcu_poll_gp_seq_start 通知 polling API；
4. 进入 ONOFF 状态，处理 CPU 热插拔引起的节点变化；
5. 进入 INIT 状态，初始化所有 rcu_node，初始化每个节点的 gp_seq 与 qsmask；调用 rcu_preempt_check_blocked_tasks 将目前在临界区中被阻塞的进程挂载到 gp_tasks 上；
6. 如果开启了 strict 模式，通过 IPI 告知所有 CPU，新的GP开启；

static noinline_for_stack bool rcu_gp_init(void)
{
	unsigned long flags;
	unsigned long oldmask;
	unsigned long mask;
	struct rcu_data *rdp;
	struct rcu_node *rnp = rcu_get_root();
	bool start_new_poll;
	unsigned long old_gp_seq;

	WRITE_ONCE(rcu_state.gp_activity, jiffies);
	raw_spin_lock_irq_rcu_node(rnp);
	if (!rcu_state.gp_flags) {
		/* Spurious wakeup, tell caller to go back to sleep.  */
		raw_spin_unlock_irq_rcu_node(rnp);
		return false;
	}
	WRITE_ONCE(rcu_state.gp_flags, 0); /* Clear all flags: New GP. */

	......
        
	start_new_poll = rcu_sr_normal_gp_init();
	/* Record GP times before starting GP, hence rcu_seq_start(). */
	old_gp_seq = rcu_state.gp_seq;

	rcu_seq_start(&rcu_state.gp_seq);
	
    ......
        
	rcu_poll_gp_seq_start(&rcu_state.gp_seq_polled_snap);
	raw_spin_unlock_irq_rcu_node(rnp);

	if (start_new_poll)
		(void) start_poll_synchronize_rcu();

	// 处理热插拔
	WRITE_ONCE(rcu_state.gp_state, RCU_GP_ONOFF);
	/* Exclude CPU hotplug operations. */
	rcu_for_each_leaf_node(rnp) {
		local_irq_disable();
		
        ......

		raw_spin_unlock_rcu_node(rnp);
		arch_spin_unlock(&rcu_state.ofl_lock);
		local_irq_enable();
	}
	rcu_gp_slow(gp_preinit_delay); /* Races with CPU hotplug. */

	// 初始化rcu_node
	WRITE_ONCE(rcu_state.gp_state, RCU_GP_INIT);
	rcu_for_each_node_breadth_first(rnp) {
		rcu_gp_slow(gp_init_delay);
		raw_spin_lock_irqsave_rcu_node(rnp, flags);
		rdp = this_cpu_ptr(&rcu_data);
		rcu_preempt_check_blocked_tasks(rnp);
		rnp->qsmask = rnp->qsmaskinit;
		WRITE_ONCE(rnp->gp_seq, rcu_state.gp_seq);
		if (rnp == rdp->mynode)
			(void)__note_gp_changes(rnp, rdp);
		rcu_preempt_boost_start_gp(rnp);
		trace_rcu_grace_period_init(rcu_state.name, rnp->gp_seq,
					    rnp->level, rnp->grplo,
					    rnp->grphi, rnp->qsmask);
		/*
		 * Quiescent states for tasks on any now-offline CPUs. Since we
		 * released the ofl and rnp lock before this loop, CPUs might
		 * have gone offline and we have to report QS on their behalf.
		 * See Requirements.rst > Hotplug CPU > Concurrent QS Reporting.
		 */
		mask = rnp->qsmask & ~rnp->qsmaskinitnext;
		rnp->rcu_gp_init_mask = mask;
		if ((mask || rnp->wait_blkd_tasks) && rcu_is_leaf_node(rnp))
			rcu_report_qs_rnp(mask, rnp, rnp->gp_seq, flags);
		else
			raw_spin_unlock_irq_rcu_node(rnp);
		cond_resched_tasks_rcu_qs();
		WRITE_ONCE(rcu_state.gp_activity, jiffies);
	}

	// If strict, make all CPUs aware of new grace period.
	if (IS_ENABLED(CONFIG_RCU_STRICT_GRACE_PERIOD))
		on_each_cpu(rcu_strict_gp_boundary, NULL, 0);

	return true;
}

rcu_gp_fqs_loop 函数用于回收QS：

1. 每轮循环计算 jiffies_force_qs 强制 QS 的时间戳与 jiffies_kick_kthreads 强制唤醒的时间戳；
2. 切换状态到 WAIT_FQS，阻塞等待超时或被主动唤醒；
3. 被唤醒后，切换到 DOING_FQS 状态，判断 QS 是否已经全部完成（根节点qsmask 全0 并且无阻塞进程），如果完成，直接跳出循环；
4. 如果没有完成，会首先判断如果已经超时 或者 FLAGS被设置了 FQS / OVLD，则会调用 rcu_gp_fqs 进行一次强制 QS 的催促，否则是被虚假唤醒，进入下一轮循环；
5. rcu_gp_fqs 最终会调用到 force_qs_rnp：遍历所有叶子节点：
   1. 如果 qsmask 已经归零，但是有阻塞进程，如果配置了 CONFIG_RCU_BOOST，则会提高被阻塞进程的优先级；
   2. 如果 qsmask 不为0,那么会遍历该叶子节点下的所有CPU，对其调用 rcu_watching_snap_save / rcu_watching_snap_recheck（根据是否第一次被调用）判断该CPU目前的状态，如果返回值>0，那么可以调用 rcu_report_qs_rnp 代替该 CPU 上报 QS，如果返回值<0，强制对该CPU执行 resched，返回0则留给该 CPU 自行处理；

static noinline_for_stack void rcu_gp_fqs_loop(void)
{
	bool first_gp_fqs = true;
	int gf = 0;
	unsigned long j;
	int ret;
	struct rcu_node *rnp = rcu_get_root();

	j = READ_ONCE(jiffies_till_first_fqs);
	if (rcu_state.cbovld)
		gf = RCU_GP_FLAG_OVLD;
	ret = 0;
	for (;;) {
		if (rcu_state.cbovld) {
			j = (j + 2) / 3;
			if (j <= 0)
				j = 1;
		}
        // 计算强制QS的时间
		if (!ret || time_before(jiffies + j, rcu_state.jiffies_force_qs)) {
			WRITE_ONCE(rcu_state.jiffies_force_qs, jiffies + j);

			smp_wmb();
			WRITE_ONCE(rcu_state.jiffies_kick_kthreads,
				   jiffies + (j ? 3 * j : 2));
		}
		trace_rcu_grace_period(rcu_state.name, rcu_state.gp_seq,
				       TPS("fqswait"));
        // 超时等待
		WRITE_ONCE(rcu_state.gp_state, RCU_GP_WAIT_FQS);
		(void)swait_event_idle_timeout_exclusive(rcu_state.gp_wq,
				 rcu_gp_fqs_check_wake(&gf), j);
		rcu_gp_torture_wait();
		WRITE_ONCE(rcu_state.gp_state, RCU_GP_DOING_FQS);

		if (!READ_ONCE(rnp->qsmask) &&
		    !rcu_preempt_blocked_readers_cgp(rnp))
			break;
		/* If time for quiescent-state forcing, do it. */
		if (!time_after(rcu_state.jiffies_force_qs, jiffies) ||
		    (gf & (RCU_GP_FLAG_FQS | RCU_GP_FLAG_OVLD))) {
			trace_rcu_grace_period(rcu_state.name, rcu_state.gp_seq,
					       TPS("fqsstart"));
			rcu_gp_fqs(first_gp_fqs);
			gf = 0;
			if (first_gp_fqs) {
				first_gp_fqs = false;
				gf = rcu_state.cbovld ? RCU_GP_FLAG_OVLD : 0;
			}
			trace_rcu_grace_period(rcu_state.name, rcu_state.gp_seq,
					       TPS("fqsend"));
			cond_resched_tasks_rcu_qs();
			WRITE_ONCE(rcu_state.gp_activity, jiffies);
			ret = 0; /* Force full wait till next FQS. */
			j = READ_ONCE(jiffies_till_next_fqs);
		} else {
			/* Deal with stray signal. */
			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("fqswaitsig"));
			ret = 1; /* Keep old FQS timing. */
			j = jiffies;
			if (time_after(jiffies, rcu_state.jiffies_force_qs))
				j = 1;
			else
				j = rcu_state.jiffies_force_qs - j;
			gf = 0;
		}
	}
}

static void force_qs_rnp(int (*f)(struct rcu_data *rdp))
{
	int cpu;
	unsigned long flags;
	struct rcu_node *rnp;

	rcu_state.cbovld = rcu_state.cbovldnext;
	rcu_state.cbovldnext = false;
	rcu_for_each_leaf_node(rnp) {
		unsigned long mask = 0;
		unsigned long rsmask = 0;

		cond_resched_tasks_rcu_qs();
		raw_spin_lock_irqsave_rcu_node(rnp, flags);
		rcu_state.cbovldnext |= !!rnp->cbovldmask;
		if (rnp->qsmask == 0) {
			if (rcu_preempt_blocked_readers_cgp(rnp)) {
				/*
				 * No point in scanning bits because they
				 * are all zero.  But we might need to
				 * priority-boost blocked readers.
				 */
				rcu_initiate_boost(rnp, flags);
				/* rcu_initiate_boost() releases rnp->lock */
				continue;
			}
			raw_spin_unlock_irqrestore_rcu_node(rnp, flags);
			continue;
		}
		for_each_leaf_node_cpu_mask(rnp, cpu, rnp->qsmask) {
			struct rcu_data *rdp;
			int ret;

			rdp = per_cpu_ptr(&rcu_data, cpu);
			ret = f(rdp);
			if (ret > 0) {
				mask |= rdp->grpmask;
				rcu_disable_urgency_upon_qs(rdp);
			}
			if (ret < 0)
				rsmask |= rdp->grpmask;
		}
		if (mask != 0) {
			/* Idle/offline CPUs, report (releases rnp->lock). */
			rcu_report_qs_rnp(mask, rnp, rnp->gp_seq, flags);
		} else {
			/* Nothing to do here, so just drop the lock. */
			raw_spin_unlock_irqrestore_rcu_node(rnp, flags);
		}

		for_each_leaf_node_cpu_mask(rnp, cpu, rsmask)
			resched_cpu(cpu);
	}
}

rcu_gp_cleanup 函数进行GP结束后的清理工作：

1. 调用rcu_poll_gp_seq_end 通知 polling API GP 结束；
2. 遍历所有节点，更新所有节点的 gp_seq（此时应该为GP开始时的 gp_seq + 3），根据每个节点的gp_seq_needed 判断是否需要开启新的GP，唤醒每个节点的nocb回调执行的内核线程；
3. 如果当前节点与内核线程在同一个CPU，那么会调用 _note_gp_changes 执行 cblist 中的回调；其他CPU在下一个tick 来临时，在软中断中调用 note_gp_changes 发现 rdp->gp_seq != rnp->gp_seq，会将cblist中注册的回调从 WAIT 移动到 DONE，随后调用 rcu_do_batch 执行回调；
4. 修改rcu_state.gp_seq，GP正式终结，进入 IDLE 状态；
5. 根据 needgp 重新设置 INIT flag；
6. 调用 rcu_sr_normal_gp_cleanup 唤醒 srs 等待者：

static noinline void rcu_gp_cleanup(void)
{
	int cpu;
	bool needgp = false;
	unsigned long gp_duration;
	unsigned long new_gp_seq;
	bool offloaded;
	struct rcu_data *rdp;
	struct rcu_node *rnp = rcu_get_root();
	struct swait_queue_head *sq;

	WRITE_ONCE(rcu_state.gp_activity, jiffies);
	raw_spin_lock_irq_rcu_node(rnp);
	rcu_state.gp_end = jiffies;
	gp_duration = rcu_state.gp_end - rcu_state.gp_start;
	if (gp_duration > rcu_state.gp_max)
		rcu_state.gp_max = gp_duration;

	rcu_poll_gp_seq_end(&rcu_state.gp_seq_polled_snap);
	raw_spin_unlock_irq_rcu_node(rnp);

	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);
		if (WARN_ON_ONCE(rcu_preempt_blocked_readers_cgp(rnp)))
			dump_blkd_tasks(rnp, 10);
		WARN_ON_ONCE(rnp->qsmask);
		WRITE_ONCE(rnp->gp_seq, new_gp_seq);
		if (!rnp->parent)
			smp_mb(); // Order against failing poll_state_synchronize_rcu_full().
		rdp = this_cpu_ptr(&rcu_data);
		if (rnp == rdp->mynode)
			needgp = __note_gp_changes(rnp, rdp) || needgp;
		/* smp_mb() provided by prior unlock-lock pair. */
		needgp = rcu_future_gp_cleanup(rnp) || needgp;
		// Reset overload indication for CPUs no longer overloaded
		if (rcu_is_leaf_node(rnp))
			for_each_leaf_node_cpu_mask(rnp, cpu, rnp->cbovldmask) {
				rdp = per_cpu_ptr(&rcu_data, cpu);
				check_cb_ovld_locked(rdp, rnp);
			}
		sq = rcu_nocb_gp_get(rnp);
		raw_spin_unlock_irq_rcu_node(rnp);
		rcu_nocb_gp_cleanup(sq);
		cond_resched_tasks_rcu_qs();
		WRITE_ONCE(rcu_state.gp_activity, jiffies);
		rcu_gp_slow(gp_cleanup_delay);
	}
	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);
	ASSERT_EXCLUSIVE_WRITER(rcu_state.gp_seq);
	WRITE_ONCE(rcu_state.gp_state, RCU_GP_IDLE);
	/* Check for GP requests since above loop. */
	rdp = this_cpu_ptr(&rcu_data);
	if (!needgp && ULONG_CMP_LT(rnp->gp_seq, rnp->gp_seq_needed)) {
		trace_rcu_this_gp(rnp, rdp, rnp->gp_seq_needed,
				  TPS("CleanupMore"));
		needgp = true;
	}
	/* Advance CBs to reduce false positives below. */
	offloaded = rcu_rdp_is_offloaded(rdp);
	if ((offloaded || !rcu_accelerate_cbs(rnp, rdp)) && needgp) {

		WRITE_ONCE(rcu_state.gp_flags, RCU_GP_FLAG_INIT);
		WRITE_ONCE(rcu_state.gp_req_activity, jiffies);
		trace_rcu_grace_period(rcu_state.name, rcu_state.gp_seq, TPS("newreq"));
	} else {

		WRITE_ONCE(rcu_state.gp_flags, rcu_state.gp_flags & RCU_GP_FLAG_INIT);
	}
	raw_spin_unlock_irq_rcu_node(rnp);

	// Make synchronize_rcu() users aware of the end of old grace period.
	rcu_sr_normal_gp_cleanup();

	// If strict, make all CPUs aware of the end of the old grace period.
	if (IS_ENABLED(CONFIG_RCU_STRICT_GRACE_PERIOD))
		on_each_cpu(rcu_strict_gp_boundary, NULL, 0);
}

4. SRCU

Tree RCU 要求读者禁止在临界区内睡眠，如果有一个进程在临界区内长时间休眠那么 GP 就会被延长，导致大量的资源无法被释放。SRCU 被设计用于解决此问题，其核心实现有两点：

1. 不再使用全局的 rcu_state，而是在每个子系统区域内使用独立的结构体进行管理；
2. 使用使用双槽位计数器来取代 CPU 上报 QS 的方式管理 GP；

4.1 关键数据结构

4.1.1 srcu_struct

srcu_struct（每域一个，用户看到的"锁"，类似 rcu_state 的句柄）
      │
      ├── srcu_ctrp ★            当前读索引指针
      ├── sda ★                  per-CPU srcu_data 数组
      ├── dep_map                lockdep 域（每个 srcu_struct 独立）
      └── srcu_sup → srcu_usage  真正的更新侧状态（下面这棵）

4.1.2 srcu_usage

srcu_usage（每域，对标 rcu_state）
      │
      ├── srcu_gp_seq ★          GP 序号，低 2 位即状态：IDLE(0)/SCAN1(1)/SCAN2(2)
      │                            （没有 rcu_state.gp_state 枚举，状态直接编码在 seq 里）
      ├── srcu_gp_seq_needed ★   需要推进到的最远 GP（"有没有活干"）
      ├── srcu_gp_seq_needed_exp ★ 需要 expedited 的最远 GP（对标 gp_flags 的加速意图）
      ├── work ★                 宽限期驱动 delayed_work（对标 gp_kthread + gp_wq！
      │                            没有 kthread，process_srcu 挂共享 rcu_gp_wq 上自循环）
      ├── srcu_cb_mutex ★        串行化"结束 GP + 触发回调"（防两轮 GP 同时发回调）
      ├── srcu_gp_mutex ★        串行化状态机推进（一个时刻只有一个推进者）
      ├── lock                   spinlock，保护 gp_seq/size_state 等
      ├── srcu_gp_start          本轮 GP 开始 jiffies（退避按 GP 年龄增长用）
      ├── srcu_last_gp_end       上次 GP 结束时刻 ns（srcu_should_expedite 判定空闲用）
      ├── node[] ★               srcu_node 组合树数组
      ├── level[]                每层首节点指针
      ├── srcu_size_state ★      SMALL→ALLOC→WAIT_BARRIER→…→BIG 状态机
      │                            （RCU 没有；解决"实例多 + 建树贵"的内存问题）
      ├── srcu_barrier_seq        srcu_barrier 专用 GP 序号
      ├── srcu_barrier_mutex      串行化 srcu_barrier
      ├── srcu_barrier_completion 等所有 CPU 回调做完
      ├── srcu_barrier_cpu_cnt ★  srcu_barrier 还剩几个 CPU（对标 barrier_cpu_count）
      ├── reschedule_jiffies/…   自适应重排节流（srcu_max_nodelay，对标 FQS 节流）
      ├── srcu_n_exp_nodelay      expedited 阶段连续"无延迟"次数
      ├── srcu_size_jiffies/…     SMALL→BIG 竞争检测窗口
      └── sda_is_static           sda 能否 free_percpu

4.1.3 srcu_node

srcu_node（组合树节点，每域一棵，对标 rcu_node[]）
      │
      ├── srcu_have_cbs[4] ★     每 GP 槽（%4）记录的 GP 序号——"本子树是否已有人
      │                            为这个 GP 请求过"，漏斗锁去重靠它
      ├── srcu_data_have_cbs[4] ★ 对应槽位里哪些 CPU(sdp) 有回调（位图，决定
      │                            回调调度到哪几个 CPU）
      ├── srcu_gp_seq_needed_exp  子树的 expedited 需求（向根传播）
      ├── srcu_parent             父指针
      └── grplo/grphi             覆盖的 CPU 区间

4.1.4 srcu_data

srcu_data（per-CPU，每域一套，对标 rcu_data）
      │
      ├── srcu_ctrs[2] ★★       双索引双计数：每索引 {atomic_long srcu_locks,
      │                            srcu_unlocks}，只增不减——前面聊的 flip/双扫描
      │                            全部作用在它上面，这是 SRCU 最核心的字段
      ├── srcu_reader_flavor ★   本 CPU 用的读者类型（NORMAL/NMI/FAST，
      │                            决定检查时用 smp_mb 还是 synchronize_rcu）
      ├── srcu_cblist ★          4 段回调链表 DONE/WAIT/NEXT_READY/NEXT
      │                            （对标 rcu_segcblist，加速机制一样）
      ├── srcu_gp_seq_needed      本 CPU 需要的最远 GP（提升后走漏斗锁）
      ├── srcu_gp_seq_needed_exp  本 CPU 需要的最远 expedited GP
      ├── work ★                 回调调用 work（挂 rcu_gp_wq、绑本 CPU；
      │                            对标 rcu_core softirq/nocb kthread）
      ├── delay_work ★           回调调用的延迟 timer（srcu_schedule_cbs_sdp）
      ├── srcu_cblist_invoking   防重入标志（同一时刻只发一轮回调，barrier 依赖它）
      ├── srcu_barrier_head      srcu_barrier 专用回调节点
      ├── mynode / grpmask       指向叶子 srcu_node + 在位图中的位
      ├── lock
      ├── cpu
      └── ssp                     回指所属域

4.2 核心流程

4.2.1 读者进入临界区

拿到计数器指针（idx0 / idx1），加1；

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);
}

4.2.2 写者发布

与普通 RCU 一致；

4.2.3 写者等待

当前内核的 SRCU 中加速 GP 的实现与普通 GP 是一样的，都调用 __call_srcu，然后阻塞等待，：

static void __synchronize_srcu(struct srcu_struct *ssp, bool do_norm)
{
	struct rcu_synchronize rcu;

	srcu_lock_sync(&ssp->dep_map);

	......
        
	init_completion(&rcu.completion);
	init_rcu_head_on_stack(&rcu.head);
	__call_srcu(ssp, &rcu.head, wakeme_after_rcu, do_norm);
	wait_for_completion(&rcu.completion);
	destroy_rcu_head_on_stack(&rcu.head);

	smp_mb();
}

__call_srcu 最终调用到 srcu_gp_start_if_needed 决定是否开启新的 GP：

1. 调用 __srcu_read_lock_nmisafe 进入读临界区，防止当前的 GP 突然结束；
2. 将回调加入 cblist 的 NEXT 段中；
3. 调用 rcu_seq_snap 获取下一轮 GP 等待标签；
4. 调用 rcu_segcblist_advance 与 rcu_segcblist_accelerate ，按照[DONE | WAIT | NEXT_READY | NEXT]，更新新一轮 cblist；
5. 判断是否更新 srcu_gp_seq_needed，调用 srcu_funnel_gp_start / srcu_funnel_exp_start 开 GP；

static unsigned long srcu_gp_start_if_needed(struct srcu_struct *ssp,
					     struct rcu_head *rhp, bool do_norm)
{
	unsigned long flags;
	int idx;
	bool needexp = false;
	bool needgp = false;
	unsigned long s;
	struct srcu_data *sdp;
	struct srcu_node *sdp_mynode;
	int ss_state;

	check_init_srcu_struct(ssp);
	/*
	 * While starting a new grace period, make sure we are in an
	 * SRCU read-side critical section so that the grace-period
	 * sequence number cannot wrap around in the meantime.
	 */
	idx = __srcu_read_lock_nmisafe(ssp);
	ss_state = smp_load_acquire(&ssp->srcu_sup->srcu_size_state);
	if (ss_state < SRCU_SIZE_WAIT_CALL)
		sdp = per_cpu_ptr(ssp->sda, get_boot_cpu_id());
	else
		sdp = raw_cpu_ptr(ssp->sda);
	spin_lock_irqsave_sdp_contention(sdp, &flags);
	if (rhp)
		rcu_segcblist_enqueue(&sdp->srcu_cblist, rhp);
	
	s = rcu_seq_snap(&ssp->srcu_sup->srcu_gp_seq);
	if (rhp) {
		rcu_segcblist_advance(&sdp->srcu_cblist,
				      rcu_seq_current(&ssp->srcu_sup->srcu_gp_seq));
		/*
		 * Acceleration can never fail because the base current gp_seq
		 * used for acceleration is <= the value of gp_seq used for
		 * advancing. This means that RCU_NEXT_TAIL segment will
		 * always be able to be emptied by the acceleration into the
		 * RCU_NEXT_READY_TAIL or RCU_WAIT_TAIL segments.
		 */
		WARN_ON_ONCE(!rcu_segcblist_accelerate(&sdp->srcu_cblist, s));
	}
	if (ULONG_CMP_LT(sdp->srcu_gp_seq_needed, s)) {
		sdp->srcu_gp_seq_needed = s;
		needgp = true;
	}
	if (!do_norm && ULONG_CMP_LT(sdp->srcu_gp_seq_needed_exp, s)) {
		sdp->srcu_gp_seq_needed_exp = s;
		needexp = true;
	}
	spin_unlock_irqrestore_rcu_node(sdp, flags);

	/* Ensure that snp node tree is fully initialized before traversing it */
	if (ss_state < SRCU_SIZE_WAIT_BARRIER)
		sdp_mynode = NULL;
	else
		sdp_mynode = sdp->mynode;

	if (needgp)
		srcu_funnel_gp_start(ssp, sdp, s, do_norm);
	else if (needexp)
		srcu_funnel_exp_start(ssp, sdp_mynode, s);
	__srcu_read_unlock_nmisafe(ssp, idx);
	return s;
}

srcu_funnel_gp_start 采用漏斗的形式从叶子节点开始向上上报下一轮的 gp_seq，如果发现上层节点已经被通知，则会退出循环，最终只有一个 CPU 能够通知到根节点并执行后续操作；

随后调用 srcu_gp_start 开启新的 GP，从 IDLE 状态切换到 SCAN1；

最后将一个调用 process_srcu 的 worker 加入到 srcu_usage 的 wq 中；

static void srcu_funnel_gp_start(struct srcu_struct *ssp, struct srcu_data *sdp,
				 unsigned long s, bool do_norm)
{
	unsigned long flags;
	int idx = rcu_seq_ctr(s) % ARRAY_SIZE(sdp->mynode->srcu_have_cbs);
	unsigned long sgsne;
	struct srcu_node *snp;
	struct srcu_node *snp_leaf;
	unsigned long snp_seq;
	struct srcu_usage *sup = ssp->srcu_sup;

	/* Ensure that snp node tree is fully initialized before traversing it */
	if (smp_load_acquire(&sup->srcu_size_state) < SRCU_SIZE_WAIT_BARRIER)
		snp_leaf = NULL;
	else
		snp_leaf = sdp->mynode;

	if (snp_leaf)
		/* Each pass through the loop does one level of the srcu_node tree. */
		for (snp = snp_leaf; snp != NULL; snp = snp->srcu_parent) {
			if (WARN_ON_ONCE(rcu_seq_done(&sup->srcu_gp_seq, s)) && snp != snp_leaf)
				return; /* GP already done and CBs recorded. */
			spin_lock_irqsave_rcu_node(snp, flags);
			snp_seq = snp->srcu_have_cbs[idx];
			if (!srcu_invl_snp_seq(snp_seq) && ULONG_CMP_GE(snp_seq, s)) {
				if (snp == snp_leaf && snp_seq == s)
					snp->srcu_data_have_cbs[idx] |= sdp->grpmask;
				spin_unlock_irqrestore_rcu_node(snp, flags);
				if (snp == snp_leaf && snp_seq != s) {
					srcu_schedule_cbs_sdp(sdp, do_norm ? SRCU_INTERVAL : 0);
					return;
				}
				if (!do_norm)
					srcu_funnel_exp_start(ssp, snp, s);
				return;
			}
			snp->srcu_have_cbs[idx] = s;
			if (snp == snp_leaf)
				snp->srcu_data_have_cbs[idx] |= sdp->grpmask;
			sgsne = snp->srcu_gp_seq_needed_exp;
			if (!do_norm && (srcu_invl_snp_seq(sgsne) || ULONG_CMP_LT(sgsne, s)))
				WRITE_ONCE(snp->srcu_gp_seq_needed_exp, s);
			spin_unlock_irqrestore_rcu_node(snp, flags);
		}

	/* Top of tree, must ensure the grace period will be started. */
	spin_lock_irqsave_ssp_contention(ssp, &flags);
	if (ULONG_CMP_LT(sup->srcu_gp_seq_needed, s)) {
		/*
		 * Record need for grace period s.  Pair with load
		 * acquire setting up for initialization.
		 */
		smp_store_release(&sup->srcu_gp_seq_needed, s); /*^^^*/
	}
	if (!do_norm && ULONG_CMP_LT(sup->srcu_gp_seq_needed_exp, s))
		WRITE_ONCE(sup->srcu_gp_seq_needed_exp, s);

	/* If grace period not already in progress, start it. */
	if (!WARN_ON_ONCE(rcu_seq_done(&sup->srcu_gp_seq, s)) &&
	    rcu_seq_state(sup->srcu_gp_seq) == SRCU_STATE_IDLE) {
		srcu_gp_start(ssp);

		// And how can that list_add() in the "else" clause
		// possibly be safe for concurrent execution?  Well,
		// it isn't.  And it does not have to be.  After all, it
		// can only be executed during early boot when there is only
		// the one boot CPU running with interrupts still disabled.
		if (likely(srcu_init_done))
			queue_delayed_work(rcu_gp_wq, &sup->work,
					   !!srcu_get_delay(ssp));
		else if (list_empty(&sup->work.work.entry))
			list_add(&sup->work.work.entry, &srcu_boot_list);
	}
	spin_unlock_irqrestore_rcu_node(sup, flags);
}

4.2.4 kwoker 推进状态

SRCU 使用一个 kworker 推进 GP 完成：

1. 调用核心函数 srcu_advance_state 推进状态机；
2. 调用 srcu_get_delay 计算下次推进的时间；

static void process_srcu(struct work_struct *work)
{
	unsigned long curdelay;
	unsigned long j;
	struct srcu_struct *ssp;
	struct srcu_usage *sup;

	sup = container_of(work, struct srcu_usage, work.work);
	ssp = sup->srcu_ssp;

	srcu_advance_state(ssp);
	spin_lock_irq_rcu_node(ssp->srcu_sup);
	curdelay = srcu_get_delay(ssp);
	spin_unlock_irq_rcu_node(ssp->srcu_sup);
	if (curdelay) {
		WRITE_ONCE(sup->reschedule_count, 0);
	} else {
		j = jiffies;
		if (READ_ONCE(sup->reschedule_jiffies) == j) {
			ASSERT_EXCLUSIVE_WRITER(sup->reschedule_count);
			WRITE_ONCE(sup->reschedule_count, READ_ONCE(sup->reschedule_count) + 1);
			if (READ_ONCE(sup->reschedule_count) > srcu_max_nodelay)
				curdelay = 1;
		} else {
			WRITE_ONCE(sup->reschedule_count, 1);
			WRITE_ONCE(sup->reschedule_jiffies, j);
		}
	}
	srcu_reschedule(ssp, curdelay);
}

srcu_advance_state 中根据当前的不同状态进行操作：

1. 初始状态为IDLE：根据 srcu_gp_seq_needed 判断是否需要开启 GP 进入 SCAN1；
2. 进入SCAN1后，判断计数器的另一个槽位是否已经归零，如果是，则切换槽位，进入SCAN2；
3. 进入SCAN2后，同样判断计数器的另一个槽位是否已经归零，如果是，则结束当前 GP，回到 IDLE；
4. 宽限期结束时 srcu_gp_end 按树节点的 srcu_data_have_cbs 位图，将每个有回调的 CPU 的 srcu_data->work 放到全局 rcu_gp_wq 中，最终会被 kworker 执行回调；

static void srcu_advance_state(struct srcu_struct *ssp)
{
	int idx;

	mutex_lock(&ssp->srcu_sup->srcu_gp_mutex);

	/*
	 * Because readers might be delayed for an extended period after
	 * fetching ->srcu_ctrp for their index, at any point in time there
	 * might well be readers using both idx=0 and idx=1.  We therefore
	 * need to wait for readers to clear from both index values before
	 * invoking a callback.
	 *
	 * The load-acquire ensures that we see the accesses performed
	 * by the prior grace period.
	 */
	idx = rcu_seq_state(smp_load_acquire(&ssp->srcu_sup->srcu_gp_seq)); /* ^^^ */
	if (idx == SRCU_STATE_IDLE) {
		spin_lock_irq_rcu_node(ssp->srcu_sup);
		if (ULONG_CMP_GE(ssp->srcu_sup->srcu_gp_seq, ssp->srcu_sup->srcu_gp_seq_needed)) {
			WARN_ON_ONCE(rcu_seq_state(ssp->srcu_sup->srcu_gp_seq));
			spin_unlock_irq_rcu_node(ssp->srcu_sup);
			mutex_unlock(&ssp->srcu_sup->srcu_gp_mutex);
			return;
		}
		idx = rcu_seq_state(READ_ONCE(ssp->srcu_sup->srcu_gp_seq));
		if (idx == SRCU_STATE_IDLE)
			srcu_gp_start(ssp);
		spin_unlock_irq_rcu_node(ssp->srcu_sup);
		if (idx != SRCU_STATE_IDLE) {
			mutex_unlock(&ssp->srcu_sup->srcu_gp_mutex);
			return; /* Someone else started the grace period. */
		}
	}

	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);
		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;
		spin_unlock_irq_rcu_node(ssp->srcu_sup);
	}

	if (rcu_seq_state(READ_ONCE(ssp->srcu_sup->srcu_gp_seq)) == SRCU_STATE_SCAN2) {

		/*
		 * SRCU read-side critical sections are normally short,
		 * so check at least twice in quick succession after a flip.
		 */
		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. */
		}
		ssp->srcu_sup->srcu_n_exp_nodelay = 0;
		srcu_gp_end(ssp);  /* Releases ->srcu_gp_mutex. */
	}
}