本文共 14814 字，大约阅读时间需要 49 分钟。

本文讲述的是linux 4.7.1下的RCU实现(tree-RCU)。

在kernel中，rcu有tiny rcu和tree rcu两种实现，tiny rcu更加简洁，通常用在小型嵌入式系统中，tree rcu则被广泛使用在了server, desktop以及android系统中.

RCU(Read-Copy Update)，顾名思义就是读-拷贝修改，它是基于其原理命名的。对于被RCU保护的共享数据结构，读者不需要获得任何锁就可以访问它，但写者在访问它时首先拷贝一个副本，然后对副本进行修改，最后使用一个回调（callback）机制在适当的时机把指向原来数据的指针重新指向新的被修改的数据。这个时机就是所有引用该数据的CPU都退出对共享数据的操作。

因此RCU实际上是一种改进的rwlock，读者几乎没有什么同步开销，它不需要锁，不使用原子指令，而且在除alpha的所有架构上也不需要内存栅（Memory Barrier），因此不会导致锁竞争，内存延迟以及流水线停滞。不需要锁也使得使用更容易，因为死锁问题就不需要考虑了。写者的同步开销比较大，它需要延迟数据结构的释放，复制被修改的数据结构，它也必须使用某种锁机制同步并行的其它写者的修改操作。读者必须提供一个信号给写者以便写者能够确定数据可以被安全地释放或修改的时机。有一个专门的垃圾收集器来探测读者的信号，一旦所有的读者都已经发送信号告知它们都不在使用被RCU保护的数据结构，垃圾收集器就调用回调函数完成最后的数据释放或修改操作。 RCU与rwlock的不同之处是：它既允许多个读者同时访问被保护的数据，又允许多个读者和多个写者同时访问被保护的数据（注意：是否可以有多个写者并行访问取决于写者之间使用的同步机制），读者没有任何同步开销，而写者的同步开销则取决于使用的写者间同步机制。但RCU不能替代rwlock，因为如果写比较多时，对读者的性能提高不能弥补写者导致的损失。

读者在访问被RCU保护的共享数据期间不能被阻塞，这是RCU机制得以实现的一个基本前提，也就说当读者在引用被RCU保护的共享数据期间，读者所在的CPU不能发生上下文切换，spinlock和rwlock都需要这样的前提。写者在访问被RCU保护的共享数据时不需要和读者竞争任何锁，只有在有多于一个写者的情况下需要获得某种锁以与其他写者同步。写者修改数据前首先拷贝一个被修改元素的副本，然后在副本上进行修改，修改完毕后它向垃圾回收器注册一个回调函数以便在适当的时机执行真正的修改操作。等待适当时机的这一时期称为grace period，而CPU发生了上下文切换称为经历一个quiescent state，grace period就是所有CPU都经历一次quiescent state所需要的等待的时间。垃圾收集器就是在grace period之后调用写者注册的回调函数来完成真正的数据修改或数据释放操作的。

下面，我们以tree-RCU为例，来学习相关代码。

1. rcu_read_lock()

include/linux/rcupdate.h

rcu_read_lock()做的事就是设置禁抢占flag，没有锁的开销。

2. rcu_read_unlock()

include/linux/rcupdate.h

rcu_read_unlock()做的事是使能内核抢占。

rcu lock的使用如下:

故，当rcu lock后，当前进程就不会被schedule调度出cpu，即会一直霸占cpu；而当rcu unlock后，当前进程在下一次调度时可能会被调度出cpu。即当rcu lock后，如果发现当前cpu发生过schedule程序运行，说明当前进程已经结果对临界区的读访问。在rcu write进程发起read-copy-update修改副本数据后，检测当前所有在临界区的cpu(disable preempt)，如果所有这些cpu都发生过了schedule程序的调用，那么可以认为在修改数据期间，所有读取到旧数据的进程都已经结束了对旧数据的引用，而这之后的数据访问，都将引用到新数据，旧数据可以删除了。

实际上,linux在实现的时候，并不检测在修改数据时，处于临界区的cpu，而是检测所有Online cpu都发生过schedule程序的调用，作为判定可以删除旧数据的依据。

内核中，把rcu writer修改数据后到所有cpu都发生了进程上下文切换(schedule)的这段时间，称为grace period(GP); 把单个cpu在rcu writer修改数据后到发生进程上下文切换，叫做quiescent state.

从上面的描述，我们知道，在使用rcu lock/unlock时，临界区的代码不能引起schedule的调用。

3. synchronize_rcu()  //writer修改副本数据后阻塞调用，等所有对旧数据的引用结束了才返回。

include/linux/rcupdate.h

kernel/rcu/tree.c

这里主要关注3281行，wait_rcu_gp().

include/linux/rcupdate.h

kernel/rcu/update.c

356行，初始化rs_array[0].completion.done = 0, 以及wait queue: rs_array[0].completion.wait.

357行，调用call_rcu_sched().

kernel/rcu/tree.c

   

__call_rcu(), 设置rs_array[0].head.func = wakeme_after_rcu(), 将&rs_array[0].head赋值给由rdp->nxttail[RCU_NEXT_TAIL]指向的rdp->nxtlist, 同时修改rdp->nxttail[RCU_NEXT_TAIL]，使其指向&rs_array[0].head.next, 即 ** rdp->nxttail[RCU_NEXT_TAIL] == NULL。

注意，这里是放入当前cpu的rcu data中的。

最后，调用__call_rcu_core()后就返回了。

kernel/rcu/tree.c

这里rcu_is_watching()返回1.

kernel/rcu/tree.c

this_cpu_ptr(&rcu_dynticks.dynticks)初始值为1.

rcu_init() -> rcu_cpu_notify() -> rcu_prepare_cpu() -> rcu_init_percpu_data()

366行, wait_for_completion() -> wait_for_common() -> __wait_for_common() -> do_wait_for_common()

kernel/sched/completion.c

do_wait_for_common()将当前进程加到x->wait等待队列中，并设置当前进程的状态为UNINTERRUPTIBLE，之后调用schedule_timeout() -> schedule(). 这样后，当前进程就进入了休眠态了。当被唤醒后，检查x->done是否不为0，不为0，则当前进程唤醒了，说明旧数据没有人使用了。

那么，这个当前进程如何被唤醒呢？即系统如何判定grace period已经过去? 我们接着看。

在tick中断处理函数中:

kernel/time/timer.c

kernel/rcu/tree.c

rcu_check_callbacks()判断当前是否有pending的rcu，如果有，则调用invoke_rcu_core()触发RCU软中断RCU_SOFTIRQ.

kernel/rcu/tree.c

那么，这里的rcu_pending()的条件是什么？

kernel/rcu/tree.c

cpu_needs_another_gp()的651行 *rdp->nxttail[RCU_NEXT_READY_TAIL]就是rdp->nxtlist，我们在调用synchronize_rcu()的时候，注册了rcu_head到rdp->nxtlist上(当前cpu的rdp)，故这里*rdp->nxttail[RCU_NEXT_READY_TAIL] ！= NULL，652行返回true.

由于我们synchronize()时，将callback挂在了当时跑synchronize()的cpu对应的sdp->nxtlist上，故这个callback只能在这一个cpu上可见，其他cpu的sdp上是没有这个callback的。

 

当使用synchronize_rcu()后，当前cpu(挂载了callback的cpu)在下一次的时钟中断处理中会触发RCU软中断:

update_process_times() -> rcu_check_callbacks() -> invoke_rcu_call() -> raise_softirq(RCU_SOFTIRQ)

RCU软中断处理函数为: rcu_process_callbacks()

kernel/rcu/tree.c

在讲软中断rcu_process_callbacks()之前，我们先来看一下RCU-tree初始化构造的“世界”。

kernel/rcu/tree.c

首先是rcu_spawn_gp_kthread()，它给每个rcu flavor都创建一个内核线程rcu_gp_kthread()，当前支持的rcu flavor有rcu_bh_state和rcu_sched_state。

其中rcu_bh_state是用于softirq中的，其他的情况下用rcu_sched_state.

kernel/rcu/tree.c

rcu_init()做rcu-tree的初始化，后面，我们只关注rcu_sched_state.

rcu_init_one()构造了rcu-tree，结构如下图：

我们再来看rcu_init_one()的代码。

4567行,levelcnt[]记录每一级的rcu_node的数目，levelcnt[0]为1，levelcnt[1]为32，levelcnt[2]为32*32, levelcnt[3]为32*32*32.

4569行,将rsp->level[]指向每一级的第一个rcu_node.

4570行，计算每一级rcu_node的child的的数目，我们简单点，就是32.

4576开始的for循环，初始化每一个rcu_node，其中grplo, grphi表示当前rcu_node的child的编号范围，同一level的所有rcu_node的grplo, grphi连续编号，即前一个rcu_node的grphi + 1为当前rcu_node的grplo.

4617行开始初始化leaf rcu_node(叶子节点)的rcu_data, 每个leaf rcu_node有编号为grplo – grphi这些的rcu_data, 每个rcu_data对应一个cpu。

rcu_init_one()大致就这些内容，我们接着看rcu_init()剩下的部分。

4741行，注册RCU_SOFTIRQ软中断。

4750-4751行，调用rcu_cpu_notify() 初始化每个cpu的rcu_data，并在每个cpu上启一个内核线程。

我们先来看rcu_prepare_cpu()，这个函数调用rcu_init_percpu_data()来初始化cpu的rcu_data.

4314行，调用Init_callback_list()初始化callback list.

下面接着看rcu_init_percpu_data().

4326行开始，取当前cpu的rcu_data的父节点rcu_node，将自己的编号rdp->grpmask记录到父节点的qsmaskinitnext,以及expmaskinitnext, 设置rdp的gpnum和completed编号为父节点对应的(初始值为UL(0-300)), 并设置当前cpu的rcu_data的core_needs_qs为false，表示当前cpu不需要qs(quiescent state), 还初始化rdp->cpu_no_qs.b.norm为true，表示当前cpu没有qs需要传送给父节点。

至此，rcu_prepare_cpu()也大致结束。

下面来看rcu_cpu_notify()另外一个函数rcu_prepare_kthreads().

rcu_prepare_kthreads()给每一个叶子节点leaf rcu_node创建一个内核线程rcu_boost_kthread.

至此，rcu_init()也大致结束了。

还有一个初始化函数rcu_spawn_boost_kthreads()。

rcu_spawn_boost_kthreads() -> smpboot_register_percpu_thread() -> smpboot_register_percpu_thread_cpumask() 给每个cpu的rcu_data创建一个内核线程, 内核线程的入口为rcu_cpu_kthread().

好了，初始化就看到这里。大致来说，初始化就是构造了rcu-tree，并给这个rcu-tree的每个叶子节点创建一个内核线程rcu_boost_kthread()，以及给每个rcu_data创建一个内核线程rcu_cpu_kthread().

我们现在回到RCU软中断处理函数。

当我们通过调用synchronize()将rcu callback挂到当前cpu的rcu_data的callback list上后，当前cpu在下一次时钟中断处理函数中将raise一个RCU_SOFTIRQ中断，这个软中断的处理函数为rcu_process_callbacks()。

kernel/rcu/tree.c

对于系统第一个rcu callback，rdp->gpnum == rnp->gpnum, rdp->completed == rnp->completed, 且rdp->gpwrap == 0, note_gp_changes()在1887行返回。

而由于rdp->core_needs_qs == false，故，rcu_check_quiescent_state()在2507行返回。

我们继续看__rcu_process_callbacks().

2986行，调用cpu_needs_another_gp()来判断是否要有一个新的grace period。由于我们之前加入了callback，故cpu_needs_another_gp()在652行返回true，表示需要一个新的grace period。

然后，调用rcu_start_gp()启动一个新的grace period.

我们来看这个rcu_start_gp()函数。

在调用synchronize()后，rcu_data的callback list关系如下:

rcu_start_gp()函数处理后，变为:

rcu_start_gp()最后设置rcu_state的gp_flags为RCU_GP_FLAG_INIT。

wakeup内核线程rcu_gp_kthread()，被唤醒的内核线程运行在唤醒它的cpu上。

rcu_gp_kthread()启动后，设置当前的gp_state为RCU_GP_WAIT_GPS(等待grace period start), 就一直在等待rsp->gp_flags为RCU_GP_FLAG_INIT。

当rcu_gp_kthread()唤醒后，设置gp_state为RCU_GP_DONE_GPS，然后调用rcu_gp_init()来初始化一个新的grace period. 我们来分析下这个rcu_gp_init()，看它都做了什么。

1918行，清除gp_flags。

1932行, rcu_state的gpnum加1。

1942行开始，将所有叶子节点rcu_node的qsmaskinit设置为qsmaskinitnext，qsmaskinitnext的每一个bit表示这个rcu_node的child的情况，1表示存在，0表示不存在。

之后调用rcu_init_new_rnp()初始化tree上所有的rcu_node的qsmaskinit。这样，每个rcu_node通过qsmaskinit的bit位就知道child的情况，比如bit0位1，表示其编号位0的child存在。

我们回到rcu_gp_init()，继续看代码:

1998行开始，将所有rcu_node的qsmask初始化位qsmaskinit，并且将gpnum和completed初始化位rcu_state的gpnum，我们知道，在这之前rcu_state的gpnum加1了，所以，这之后，所有rcu_node的gpnum都比他们自己的completed的要大1. 即表明grace period开始了(gpnum != completed)

2008行，对于当前cpu所在的rcu_node，调用__note_gp_changes()。

rdp和rnp的completed都没有增长，此时他们是相等的。故__note_gp_changes()调用rcu_accelerate_cbs(rsp, rnp, rdp)。注意，这里的rnp为rdp对应的rnp，非root_rnp。

由于rnp为非root_rnp，故1750行,c = completed + 2.

1754行break时，i = RCU_WAIT_TAIL.

1762行，i 变为RCU_NEXT_READY_TAIL.

1770到1773行，将RCU_NEXT_READY_TAIL和RCU_NEXT_TAIL的nxtcompleted加2. 即如下图所示。

最后调用rcu_start_future_gp()。

1604行，这里rnp ！= root_rnp，故c = completed + 2。

1606行，之前设置need_future_gp[c & 0x1]时，c = completed + 1, 所以，这里不满足。

1624行这里,gpnum != completed满足了，因为之前将所有的rcu_node的gpnum都加1了，completed不变。故1628行返回false.

返回之后，我们回到__note_gp_changes().

1857行，由于rcu_node的gpnum加1了，rcu_data的gpnum还没有，故rdp->gpnum != rnp->gpnum满足，然后执行1863行到1869行，设置rdp->gpnum为rnp->gpnum，同时，设置cpu_no_qs.b.norm为true, 设置core_needs_qs为true， 设置gpwrap为false。

结束后，我们返回rcu_gp_init()。rcu_gp_init()不关心__ note_gp_changes()返回值(实际上，这次返回的是false).

rcu_gp_init()最后返回true. rcu_gp_init()返回后，rcu_gp_kthread()就等待rsp->gp_flags & RCU_GP_FLAG_FQS后或者grace period完成后，才wakeup。

rcu_gp_kthread()醒来的主要判断依据为函数rcu_gp_fqs_check_wake()。

rcu_gp_fqs_check_wake()判断rsp->gp_flags 是否有RCU_GP_FLAG_FQS(force quiescent state)或者root_rnp->qsmask为0(为0，表示rcu-tree完成了grace period)，返回true，返回true后，rcu_gp_kthread就回到2205行，设置gp_state为RCU_GP_DOING_FQS，然后继续等待下一个grace period start.

这一阶段结束。

之后，所有cpu的时钟中断发生后，在中断处理函数中

update_process_times() -> rcu_check_callbacks()

由于我们之前更新了所有RCU_NODE的gpnum, 使得所有leaf RCU_NODE(除了synchronize()的cpu外)的gpnum != 它的child(rcu data)的completed, rcu_pending()会返回true，即认为当前cpu需要一次quiescent state，于是调用invoke_rcu_core()，触发RCU软中断。

我们再回到RCU软中断处理函数中来。

rcu_process_callbacks() -> __rcu_process_callbacks()

这个时候，是除了调用synchronize()之外的所有cpu都会进来的。

__rcu_process_callbacks() -> rcu_check_quiescent_state() -> note_gp_changes()。

在其他cpu的rcu data中，rdp->gpnum != rnp->gpnum. 所以，会进入函数__note_gp_changes()。

而rdp->completed和rnp->completed是相等的，故，进入函数rcu_accelerate_cbs()。

其他cpu的rcu_data的callback list，是初始状态的，如下图:

1733行这里, rdp->nxttail[RCU_NEXT_TAIL]为&rdp->nxtlist, *rdp->nxttail[RCU_DONE_TAIL]就是rdp->nxtlist, 在所有除调用synchronize()的cpu之外的cpu, 都为NULL， 1734行返回false。

 

返回后，我们回到了__not_gp_changes().

返回后，由于rdp->gpnum != rnp->gpnum成立，故执行1863-1869行代码，将rdp->gpnum更新为rnp->gpnum，同时设置rdp->cpu_no_qs.b.norm = true, rdp->core_needs_qs = true.

__not_gp_changes()也就此返回，返回码为rcu_accelerate_cbs()的返回码，false.

__not_gp_changes()返回到note_gp_changes()，由于返回码为false，故1891-1892行不去唤醒rcu_gp_kthread内核线程。程序从note_gp_changes()返回到rcu_check_quiescent_state().

这个时候,rdp->core_needs_qs是true，表示当前cpu是需要一次qs上报的。

rdp->cpu_no_qs.b.norm == true， rdp->rcu_qs_ctr_snap == __this_cpu_read(rcu_qs_ctr)也成立，程序在2515返回，返回到__rcu_process_callbacks().

这里cpu_needs_another_gp()，由于rsp的gpnum != rsp的completed, 表示grace period正在处理，这个函数返回false，不会再start一个grace period.

__rcu_process_callbacks()中的2997-2998行，条件不成立，不执行。

之后，软中断处理结束。

到现在，我们所有cpu的rcu_data的gpnum 都更新了，且cpu_no_qs.b.norm = true, core_needs_qs = true.

当某个cpu发生了调度程序执行(任务切换)时，

__schedule() -> rcu_note_context_switch() -> rcu_sched_qs()

kernel/rcu/tree.c

由于rcu_sched_data.cpu_no_qs.s为true(之前设置了cpu_no_qs.b.norm = true, 这是个union)，故247行不会执行。程序执行251行，将cpu_no_qs.b.norm设置为false.

当所有cpu，包括调用synchronize()的cpu，都完成了一次schedule()，那么所有cpu的rcu_data的cpu_no_qs.b.norm都为false。

当cpu的cpu_no_qs.b.norm为false之后，再次发生的时钟中断处理函数中，rcu_pending()也是会返回true的。

update_process_times() -> rcu_check_callbacks() -> rcu_pending() -> __rcu_pending()

此时，rdp->core_needs_qs仍旧为true, 而rdp->cpu_no_qs.b.norm在schedule()的时候被设置为false，故__rcu_pending()在4023行返回true. 当前cpu又会触发RCU软中断，从而进入软中断处理函数rcu_process_callbacks()。

也就是说，所有cpu在完成一次任务切换schedule()后，都会再次进入RCU软中断处理函数rcu_process_callbacks()中。

我们再来分析这个函数。

rcu_process_callbacks() -> __rcu_process_callbacks() -> rcu_check_quiescent_state() -> note_gp_changes().

按之前的分析rdp->gpnum == rnp->gpnum， rdp->completed == rnp->completed, note_gp_changes()中1887行直接返回，回到rcu_check_quiescent_state ().

由于rdp->core_needs_qs为true, 而rdp->cpu_no_qs.b.norm在调度程序中被设置为false，rcu_check_quiescent_state()会跑到2521行，调用rcu_report_qs_rdp().

rcu_report_qs_rdp()用于上报cpu的quiescent state到父节点RCU_NODE, RCU_NODE则会层层上报，直到root RCU_NODE.

我们来看这个rcu_report_qs_rdp()。

rcu_report_qs_rdp()中2455-2470行，条件不成立，不会执行(前面啰嗦了好多遍了)。直接来到2471行, rdp->grpmask中的含义: 假如当前cpu id = 3，则rdp->grpmask = 1 << (3 % 32), 而rdp的父节点RCU_NODE的rnp->grpmask= rdp0->grpmask | rdp1->grpmask|……|rdp31->grpmask.

所以，rcu_report_qs_rdp()会执行从2475行开始的代码，将rdp->core_needs_qs设为false，然后调用rcu_accelerate_cbs().

对于非调用synchronize()的cpu，由于*rdp->nxttail[RCU_DONE_TAILE] == rdp->nxtlist，errdp->nxtlist == NULL, 这些cpu都会在1734行返回false.

而对于调用了synchronize()的cpu，则继续执行rcu_accelerate_cbs()剩下的代码，我们接着看剩下的代码。

对于调用了synchronize()的cpu，则在rcu_start_future_gp ()中的1608行返回false，其callback list没有变化。

因此，所有cpu都会从rcu_accelerate_cbs()返回false，回到rcu_report_qs_rdp()中。

由于rcu_accelerate_cbs()返回false，2486行唤醒rcu_gp_kthread不会发生。程序只调用rcu_report_qs_rnp(), 这个函数用于rnp(RCU_NODE)层层上报qs事件, 直到root RCU_NODE.

rcu_report_qs_rnp()将父节点RCU_NODE的qsmask中自己占据的bit位清零，表示自己上报过了qs。2374行，如果父节点还有其他child还没有上报qs，则立即返回；否则，父节点就将自己在自己的父节点的qsmask占据的bit位清零，这样，直到最后一个cpu的qs上报后，rcu-tree就将qs上报到了rcu-tree的root RCU_NODE, 此时root rnp->grpmask == 0.

rcu_report_qs_rnp()将qs上报到root_rnp后，在2385行break跳出，进入函数rcu_report_qs_rsp()。

rcu_report_qs_rsp()要做的就是将之前在等待RCU_GP_FLAG_FQS而睡眠的rcu_gp_kthread内核线程唤醒。

这样，我么你的代码就跑到了rcu_gp_kthread()这个线程中了。

这个内核线程rcu_gp_kthread()从2203行醒来，由于此时root_rnp->qsmask == 0,程序在2210行break出去。最终进入rcu_gp_cleanup()处理。

2112-2128行，rcu_gp_cleanup()将所有RCU_NODE的completed更新为rsp-gpnum(completed + 1), 对于调用了synchronize()的cpu，还将调用__note_gp_changes()处理(2119行)。我们来看这次的__note_gp_changes()的调用。

由于rnp->completed更新过了(completed + 1)，而rdp->completed还没有，故__note_gp_changes()会在1850行调用rcu_advance_cbs()处理。

在rcu_advance_cbs()之前，callback list是这样的:

即RCU_WAIT_TAIL的completed + 1, 而RCU_NEXT_READY_TAIL的completed +2, 所以1808-1812行，循环2次，第二次在1810行break处理，i == RCU_NEXT_READY_TAIL, RCU_DONE_TAIL指向了RCU_WAIT_TAIL指向的，即，如下图：

最后，调用rcu_accelerate_cbs()处理。

由于*rdp->nxttail[RCU_DONE_TAIL] == cb->next, 而cb->next == NULL, 故rcu_accelerate_cbs()在1734行返回false，最终返回到__note_gp_changes().

程序从rcu_advance_cbs()返回false(rcu_accelerate_cbs()返回的false)后，更新rdp->completed为rnp->completed. 然后返回，返回到rcu_gp_cleanup()。

最后，更新rsp->completed为rsp->gpnum，将rsp状态改为idle。

2139行调用rcu_advance_cbs()时，由于*rdp->nxttail[RCU_DONE_TAIL]为NULL，返回false.

到这里rcu_gp_kthread内核线程也执行完毕了。

当调用了synchronize()的cpu在这之后的下一次时钟中断处理时，

update_process_times() -> rcu_check_callbacks() -> rcu_pending() -> __rcu_pending()

__rcu_pending()在4027行调用cpu_has_callbacks_ready_to_invoke()时，

由于rdp->nxttail[RCU_DONE_TAIL] == &cb->next, 故cpu_has_callbacks_ready_to_invoke()返回true，__rcu_pending()在4029行返回true，当前cpu之后就raise了RCU_SOFTIRQ，当前cpu随后进入了RCU软中断处理函数rcu_process_callbacks()。

rcu_process_callbacks() -> __rcu_process_callbacks()

__rcu_process_callbacks()在2998行，调用invoke_rcu_callbacks() -> invoke_rcu_callbacks_kthread().

invoke_rcu_callbacks_kthread()则设置当前cpu的rcu_cpu_has_work为1，同时，唤醒当前cpu对应的内核线程rcu_cpu_kthread(),这个线程是驻留在当前cpu的，不会被调度出去。

由于之前设置了rcu_cpu_has_work为1，故1222行进入rcu_kthread_do_work()处理。

rcu_kthread_do_work()调用rcu_do_batch()处理，rcu_do_batch()首先取出rdp->nxtlist中的callback，然后将rdp callback list还原(2750-2756行)，还原后的callback list如下图:

之后(2761行), while()循环，遍历cb list, 通过__rcu_reclaim()的118行，来调用注册的callback函数。本文例子中，这个callback函数只有一个，就是wakeme_after_rcu()

kernel/rcu/update.c

kernel/sched/completion.c

至此，一个流程讲完了。

从上面的代码实现来看，rcu-writer调用synchronize()等待所有cpu发生一次任务切换，当reader处在临界区时，由于禁了内核抢占，不会发生任务切换，所以，一旦任务切换了，就表明reader已经离开临界区了，当这个reader再次返回临界区时，此时它看到的数据是writer的新数据，就数据看不到了(相当于next由原先的指向就数据时，writer在synchronize()之前，将next指向新数据，这个点之后的所有reader，都将只能看到新数据，旧数据被writer隐藏起来了。)。当所有cpu发生了一次任务切换后，writer就可以从容地删除旧数据了。

对于rcu writer，从实现来看，目前同一个core地多个进程同时synchronize()(这里的同时，指的是一个进程还没有从synchronize()醒过来，另一个进程想要synchronize())，synchronize()只能设置一个callback，后面地callback会覆盖前面地callback。同一个core地多个进程的synchronize()还需要其他锁来辅助(互斥)。不同的core上进程synchronize()则没有关系，因为callback在不同cpu的rcu_data上。

转载地址：http://qtlci.baihongyu.com/