Linux2.6.11版本：classic RCU的实现

Posted 2020-08-16

 转载自：http://www.wowotech.net/kernel_synchronization/linux2-6-11-RCU.html

 

一、前言

无论你愿意或者不愿意，linux kernel的版本总是不断的向前推进，做为一个热衷于专研内核的工程师，最大的痛苦莫过于此：当你熟悉了一个版本的内核之后，内核已经推进到一个新的版本，你曾经熟悉的内容可能会变得陌生（这里主要说的是该模块的内部实现，实际上，内核中的每一个子系统都是会尽量保持接口API的不变）。怎么应对这种变化呢？一方面，具体的实现可能千差万别，但是基本的概念是一样的，无论哪一个版本的内核，总是能够理解一个内核子系统的基本概念和运作机理。另外一方面，不同版本之间的实现不同往往是有原因的，新版本中具体实现的不同往往是针对旧版本的问题而改进的，如果你能够理清不同版本之间的差异以及背后的原因，那么你其实也在不断的加深对计算机系统的理解（不断的迭代是一个不错的学习linux内核的方法）。

因此，在进入具体的Linux2.6.11版本内核RCU实现之前，我们首先描述why，也就是说为何修改RCU算法实现？旧内核有哪里不足，新的内核优点是什么？随后需要描述的是如何改进，最后，我们描述的是Linux2.6.11版本内核RCU模块的具体实现。

 

二、 Linux2.5.43版本的可扩展性（scalability）问题

1、原理

我们现在研究的并行运算系统大多是基于共享内存的：也就是说系统中有若干个CPU core，共享一个memory space。在这种情况下，各个CPU Core对memory space中的某个固定的变量数据的访问就很值得思考一下了。一个简单的例子：假设程序中有一个全局变量A，每个CPU core都运行一个线程对A进行访问。如果大家都是读，那么没有什么问题，除了第一个CPU core用于cache miss需要从main memory加载之外，其他的CPU core的第一次读操作都可以从通过snoop的操作，从其他cache中获取A的值，并将自己的A变量对应的cacheline设置成shared状态。之后，各个CPU core无论如何的发起多次read操作，都可以从自己的local cache中获取A值，因此，整个系统的性能非常好。

如果有写的操作会怎样呢？事情就不那么美妙了，写操作不能直接施加到shared状态的cache，必须invalidate其他CPU的local cache中A变量对应的cacheline，才能发起写的动作。因此，一次写操作，由于invalidate其他的local cache中的内容，结果会立刻重创其他所有CPU上的read操作的性能。特别是当CPU core数据增加的时候，共享变量的读写性能瓶颈也会逐步的显现出来。

对策是什么呢？主要有两种方法，一种是改成percpu变量。另外一种方法是减少对全局变量的访问。

2、Linux2.5.43版本的全局变量和访问分析

现在，我们先一起review一下Linux2.5.43版本的全局变量（percpu的那些全局变量不考虑）：

struct rcu_ctrlblk rcu_ctrlblk = 
    { .mutex = SPIN_LOCK_UNLOCKED, .curbatch = 1,  .maxbatch = 1, .rcu_cpu_mask = 0 };

struct rcu_ctrlblk { 
    spinlock_t    mutex; 
    long        curbatch;  
    long        maxbatch; 
    unsigned long    rcu_cpu_mask; 
};

对于rcu_ctrlblk数据结构中，rcu_cpu_mask成员修改异常频繁，每个CPU都会在自己经历Quiescent state之后，修改该变量。curbatch和maxbatch仅仅是在一次Grace period才修改，特别是如果cpu core增多，curbatch和maxbatch访问频率不变，但是rcu_cpu_mask成员的写操作呈线性增长。因此，如果RCU算法经常读rcu_cpu_mask，那么其读性能一定是非常的差。

下面，我们再一起过一下Linux2.5.43版本对对rcu_cpu_mask的访问情况。

（1）rcu_pending函数对rcu_cpu_mask的访问情况，代码如下：

static inline int rcu_pending(int cpu) 
{ 
    if ((……忽略其他检查 
        test_bit(cpu, &rcu_ctrlblk.rcu_cpu_mask)) 
        return 1; 
    else 
        return 0; 
}

在各个CPU的tick handler中，需要调用rcu_pending来检测是否进行后续RCU相关的操作（检查Quiescent state，调用callback什么的）。为什么做这个检查呢？逻辑思考是这样的：如果该CPU以及报告了Quiescent state，那么其rcu_cpu_mask必定被修改成0，因此，通过该bit可以知道其是否上报了Quiescent state，如果已经上报，那么后续的RCU相关的操作在该CPU上不需要进行（这时候，就需要等最后一个上报Quiescent state的CPU启动一次新的Grace Period检查过程之后，该CPU才有事情要做）。你知道的，CPU的资源有限，我们得省着点用，能少执行点代码就少执行点代码。

（2）rcu_check_quiescent_state函数对rcu_cpu_mask的访问情况，代码如下：

static void rcu_check_quiescent_state(void) 
{ 
    int cpu = smp_processor_id();

    if (!test_bit(cpu, &rcu_ctrlblk.rcu_cpu_mask)) {－－－－－－－－－A  
        return; 
    }

……

    spin_lock(&rcu_ctrlblk.mutex); 
    if (!test_bit(cpu, &rcu_ctrlblk.rcu_cpu_mask)) {－－－－－－－－－B  
        spin_unlock(&rcu_ctrlblk.mutex); 
        return; 
    } 
    clear_bit(cpu, &rcu_ctrlblk.rcu_cpu_mask);－－－－－－－－－－－C  
    RCU_last_qsctr(cpu) = RCU_QSCTR_INVALID; 
    if (rcu_ctrlblk.rcu_cpu_mask != 0) {－－－－－－－－－－－－－－－D 
        spin_unlock(&rcu_ctrlblk.mutex); 
        return; 
    } 
    rcu_ctrlblk.curbatch++; 
    rcu_start_batch(rcu_ctrlblk.maxbatch);－－－－－－－－－－－－－E 
    spin_unlock(&rcu_ctrlblk.mutex); 
}

在A点的访问，其思路和上一节描述的类似，也是本着解约CPU的MIPS的角度而增加的逻辑判断。B点的访问有必要吗？我没有看出有任何的必要。C和D的访问是必要的，在发现本CPU至少经历一次Quiescent state之后，当然要clear对应的cpu bit。如果rcu_cpu_mask等于0，当然要启动一个新的Grace period的检查过程（处理一批新的callback请求），这对应上面E处的代码。

（3）rcu_start_batch函数对rcu_cpu_mask的访问情况，代码如下：

static void rcu_start_batch(long newbatch) 
{ 
    if (rcu_batch_before(rcu_ctrlblk.maxbatch, newbatch)) { 
        rcu_ctrlblk.maxbatch = newbatch; 
    } 
    if (rcu_batch_before(rcu_ctrlblk.maxbatch, rcu_ctrlblk.curbatch) || 
        (rcu_ctrlblk.rcu_cpu_mask != 0)) {－－－－－－－－需要吗？ 
        return; 
    } 
    rcu_ctrlblk.rcu_cpu_mask = cpu_online_map;－－－重置CPU BITMASK，必须的 
}

如果rcu_start_batch只是在rcu_check_quiescent_state函数（5）处被调用，那么rcu_start_batch函数中对rcu_cpu_mask的非零检查是没有必要的。不过，在rcu_process_callbacks函数，各个CPU在第一次从nextlist链表摘下请求挂入curlist链表中的时候，也会调用该函数。

从上面的描述可以得出结论：Linux2.5.43版本对rcu_cpu_mask的读操作过于频繁，会导致cache line trashing，影响性能。

 

三、Linux2.5.43版本的其他问题

1、实时性问题

（1）原理。我们知道，对于linux kernel而言，中断上下文（包括softirq context，当然tasklet context是softirq context的一种）优先级总是高过进程优先级，也就是说，完成了中断上下文的执行，才会启动进程调度。RCU callback函数是在tasklet context中执行，本质上属于中断上下文，因此，一旦一个批次的Grace period过去，那么这个批次的callback函数会在中断上下文（bottom half）中被一一执行。

（2）Linux2.5.43版本代码review。具体代码如下：

static void rcu_process_callbacks(unsigned long unused) 
{ 
    int cpu = smp_processor_id(); 
    LIST_HEAD(list);

    if (!list_empty(&RCU_curlist(cpu)) && rcu_batch_after(rcu_ctrlblk.curbatch, RCU_batch(cpu))) { 
        list_splice(&RCU_curlist(cpu), &list); 
        INIT_LIST_HEAD(&RCU_curlist(cpu)); 
    }

…… 
    if (!list_empty(&list)) 
        rcu_do_batch(&list); 
}

在Linux2.5.43版本中，一旦一个批次的RCU callback请求完成了Grace Period，各个CPU会将curlist中的RCU callback请求从链表中摘下，并在tasklet上下文中，一次性的完成所有的callback函数的调用（参考rcu_do_batch函数）。在一般情况下，这样设计也没有什么问题，不过，在RCU callback比较多的时候，一次性完成所有的callback函数的执行是一个非常耗时的操作。当在tasklet上下文中调用成千个callback函数的时候，我想，此刻调度器的内心几乎是崩溃的，应该会有成千个草泥马奔腾而过。

2、接口封装问题

（1）原理。做为一种同步机制，RCU的接口应该能够屏蔽底层的细节，应该象其他的同步机制一样，隐含memory barrier的功能。如果其他模块在使用RCU模块提供的接口的时候，还需要调用memory barrier模块的接口API，那么这一点实际上是违背了软件抽象的基本原则。

（2）Linux2.5.43版本代码review。在该版本上，RCU模块的接口API有四个：

#define rcu_read_lock()        preempt_disable() 
#define rcu_read_unlock()    preempt_enable() 
extern void FASTCALL(call_rcu(struct rcu_head *head, void (*func)(void *arg), void *arg)); 
extern void synchronize_kernel(void);

前面两个用来标识RCU reader一侧的临界区的，后面两个是用来被updater调用，以便在适当的时候进行reclaimation的。这套接口函数有一个缺点，就是在RCU-protected pointer读写的时候需要一些额外的memory barrier的操作需要调用者自己来完成。

3、内存消耗问题

在Linux2.5.43这个版本中，检测Quiescent state是通过两个变量来完成的，如下：

    long        qsctr;          
    long        last_qsctr

qsctr是一个不断累加的值，last_qsctr是记录上次的qsctr的值，通过比对last_qsctr和qsctr的值就可以判断CPU的Quiescent state了。但是，我们可以再想想：判断Quiescent state需要两个变量吗？我们需要用一个counter来记录Quiescent state吗？经历3次Quiescent state和经历5次Quiescent state有区别吗？没有。因此，实际上用一个变量就可以解决的问题却使用两个变量。

另外，在Linux2.5.43这个版本中，RCU callback占用内存太多。将struct rcu_head 组织成双向链表是没有必要的，单链表足矣，此外，arg参数也没有必要，毕竟大多数情况下，struct rcu_head是嵌入在其他数据结构中，因此callback函数接收一个struct rcu_head *类型的指针就够了，在callback函数中，可以通过container_of还原嵌入struct rcu_head的数据结构，那么参数其实是没有必要了。

4、代码可读性问题

在Linux2.5.43这个版本中，对RCU请求批次的管理是通过下面的数据结构控制的：

struct rcu_ctrlblk {… 
    long        curbatch; 
    long        maxbatch;  
…};

curbatch和maxbatch语义不明晰，给代码阅读带来困难，特别是maxbatch，根本不知道要表达什么。通过代码分析可以知道，maxbatch主要用来确定在上一个Grace Period结束之后（即该批次的Grace period已经过去了）是否立刻启动一个新的Grace period批次。通过maxbatch这个成员可以确定是否启动。如果curbatch==maxbatch，则表示不启动，如果curbatch！=maxbatch则表示需要启动。相信知道真相的你眼泪就会掉下来：一个flag的事，干嘛搞这么复杂？

 

四、Linux2.6.11上的RCU算法改进

在控诉完Linux2.5.43的RCU实现之后，我们一起看看Linux2.6.11的RCU算法是如何进行改进的。

1、对scalability的改进。可以考虑先把Grace Period相关的数据独立出来，如下：

struct rcu_state { 
    spinlock_t    lock; 
    cpumask_t    cpumask; 
};

判断Grace Period的基本算法不变，还是依赖cpumask这个共享数据，每个bit表示一个CPU的Quiescent state，全0则表示渡过了Grace Period。write cpumask的策略不变，只要尽量减少对cpumask的读操作就OK了。

当然，程序逻辑上还是要判断各个CPU的Quiescent state，怎么办呢？Linux2.6.11内核的struct rcu_data引入了两个新的成员，如下：

struct rcu_data { 
    long        quiescbatch;  
    int        qs_pending;

…… };

这两个成员用来判断本CPU的Quiescent state。quiescbatch用来记录本CPU上，正在为哪一个批次号码的RCU callback请求进行Quiescent state的检测。qs_pending用来记录当前CPU是否已经上报了Quiescent state（对应quiescbatch那个批次号），如果等于1，表示处于pending状态，也就是说还没有上报Quiescent state。

在struct rcu_data数据结构中还有一个batch成员，这个成员又表示什么呢？当RCU callback请求从nextlist移动的curlist的时候，会分配一个batch ID，是current batch ID + 1，分配是一回事，能不能立刻启动该批次的Grace period检测是另外一回事。在current batch ID还没有通过Grace period之前，这个新分配的是不会成为current的，只能是pending的批次。因此，struct rcu_data数据结构中的batch成员是具体描述该CPU curlist中的批次号的。

2、检测是否经历至少一次Quiescent state。在Linux2.6.11版本中，只有一个变量来跟踪Quiescent state：

int        passed_quiesc;

这是个flag，要么是0，要么是1。如果等于1表示至少经历一次Quiescent state了。虽然无法记录CPU经历了多少次的Quiescent state，但是who care？

3、RCU callback数据结构的修改。struct rcu_head定义如下：

struct rcu_head { 
    struct rcu_head *next; 
    void (*func)(struct rcu_head *head); 
};

这个数据结构减少了50％的memory size，如果系统中大量使用了RCU，那么整体节省的memory不是一个小数目。

4、对实时性的改进。既然不能一次性完成所有的callback函数，那么总有一个限制吧，这就是maxbatch变量定义的：

static int maxbatch = 10;

缺省定义为10，也就是说如果该批次的callback请求有16个，那么完成10个callback调用之后，必须结束rcu tasklet的执行，等到下一个tick的时候完成剩余callback函数的执行。由于不能一次性执行完所有的callback，因此，需要在struct rcu_data 增加donelist链表，如下：

    struct rcu_head *donelist; 
    struct rcu_head **donetail;

如果不能一次处理完所有的请求，那么剩余的callback请求可以调用tasklet_schedule函数，在下次tasklet context中继续完成（参考rcu_do_batch函数）。由于使用了单向链表，因此struct rcu_data中的链表头不再是struct list_head，而是struct rcu_head 类型的指针，donetail指向链表的尾部。

5、对RCU控制块的改进。struct rcu_ctrlblk就是RCU模块的控制块，代码如下：

struct rcu_ctrlblk { 
    long    cur;        /* Current batch number.                      */ 
    long    completed;    /* Number of the last completed batch         */ 
    int    next_pending;    /* Is the next batch already waiting?         */ 
} ____cacheline_maxaligned_in_smp;

cur是当前正在处理的那个批次的callback请求（检测Grace Period进行中），这些请求会在各个CPU的curlist链表中（但是不表示各个CPU中的curlist中的RCU callback请求都属于当前批次，有些CPU的curlist中的请求属于next pending的批次）。对于当前正在处理的批次，它们正等待Grace Period过去，一旦过去，当前正在处理的批次会变成completed批次。

completed是上次完成的callback请求的批次。next_pending表示下一批次的callback请求是否ready了。

什么是当前（current batch number）？什么是上个批次（last batch number）？什么是下个批次（next batch number）？其实系统中能同时存在的批次最多就三个，不过全局RCU控制块只监控两个：cur等于current batch number，completed等于last completed batch number。一旦current batch完成了Grace Period，那么它就变成last completed batch number，同时自己加一，表示准备检测下一个batch ID的Grace Period。因此，通过比对各个CPU的rcu_data->quiescbatch和rcu_ctrlblk->cur，可以知道是否需要启动下一个批次的处理。

next batch只会出现在各个CPU的rcu_data->batch中。什么时候会出现这样的场景呢？如果某个特定的CPU处于下面的状态：

（1）没有当前批次的请求挂入本CPU的curlist中。虽然如此，但是Quiescent state仍然需要检测

（2）有新的请求挂入本CPU的nextlist链表中

当本CPU的tick到来的时候，会将nextlist链表中的请求移入curlist，同时分配batch ID。不过，由于本CPU还在检测Quiescent state，因此无法立刻启动新的批次。不过系统始终要记录这个标记，这就是next_pending标记的作用。

6、新功能的增加。在Linux2.6.11版本中，增加了两个新的功能：一个是对CPU HOTPLUG的支持，另外一个是针对reader side的临界区都是在softirq context中的。这种特殊的rcu使用场景需要一点特别的处理，那就是在检测是否经历至少一次Quiescent state的时候，除了context switch之外，执行完成一次softirq handler也是可以判断为是否经历至少一次Quiescent state的条件。

 

五、RCU数据流和控制流

1、初始状态

我们假设我们现在有一个双CPU的系统，状态定义如下：

rcu_state	rcu_ctrlblk	CPU0的rcu_data	CPU1的rcu_data
cpumask（0，1）	Cur（5）  Completed（4）  next_pending（0）	Quiescbatch（5）  passed_quiesc（1）  qs_pending（0）  batch（5）  nextlist（）  curlist（＝＝）  donelist（）	Quiescbatch（5）  passed_quiesc（0）  qs_pending（1）  batch（5）  nextlist（）  curlist（＝＝＝）  donelist（）

解释如下：

（1）无论CPU0还是CPU1都没有新申请的RCU callback，即nextlist为空，用nextlist（）表示

（2）CPU0已经通过QS（Quiescent State），即passed_quiesc成员已经被置位，用passed_quiesc（1）表示

（3）CPU1还没有通过QS，用passed_quiesc（0）

（4）当前批次是5，用cur（5）表示。上次完成的是批次4，用completed（4）表示。没有pending的批次，用next_pending（0）表示。

（5）curlist链表中有正在检测QS的请求，用curlist（＝＝＝）表示

2、CPU0的tick到来事件处理

我们来看看__rcu_pending函数的处理：

static inline int __rcu_pending(struct rcu_ctrlblk *rcp, struct rcu_data *rdp) 
{ 
    if (rdp->curlist && !rcu_batch_before(rcp->completed, rdp->batch))－－－－－（1） 
        return 1;  
    if (!rdp->curlist && rdp->nxtlist)－－－－－－－－－－－－－－－－－－－－－（2） 
        return 1;  
    if (rdp->donelist)－－－－－－－－－－－－－－－－－－－－－－－－－－－（3） 
        return 1;  
    if (rdp->quiescbatch != rcp->cur || rdp->qs_pending)－－－－－－－－－－－－（4） 
        return 1;  
    return 0;－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－（5） 
}

（1）如果rcp->completed大于或者等于rdp->batch，那么说明该CPU上的批次（rdp->batch）已经渡过GP（Grace Period），如果rdp->curlist非空，那说明本CPU需要将curlist链表中的请求转移到donelist中，因此pending状态检测返回true。当然，对于本场景，rcp->completed小于rdp->batch，不成立。

（2）如果有新的请求，本CPU的curlist链表为空，那么pending状态检测返回true，因为需要将这些新的请求移动到curlist中并分配batch number。对应这里的场景，curlist非空而且nxtlist没有请求，因此条件不成立

（3）如果donelist中有需要处理的请求，那么也是需要进行处理的

（4）如果本CPU正在进行QS检测的批次（rdp->quiescbatch）不等于当前正在检测GP的批次（rcp->cur ），说明已经启动了新的批次进行GP检测，那么需要本CPU初始化该CPU QS检测的状态，因此pending状态检测返回true。此外，如果rdp->qs_pending等于1表示本CPU的正在进行QS的检测，那么需要调用rcu_check_quiescent_state函数执行实际的QS状态的检测。和上面一样，对于当前的CPU0的状态，这些不成立。

（5）结论：对于CPU0而言，rcu_pending返回0，不会进行任何实际的动作。

3、CPU1上发生了进程切换

发生进程切换的时候，CPU1会执行schedule函数，在该函数中会调用rcu_qsctr_inc(task_cpu(prev))，从而将CPU1的passed_quiesc设置成1。

只是在进程切换的时候修改passed_quiesc是否足够呢？我们举一个极端的场景：加入当前系统中有一个realtime的进程，并且优先级最高，那么实际上在该进程进入阻塞状态之前，系统不会发生进程切换，如果把passed_quiesc的修改任务只交给进程切换，那么这种场景下不就悲剧了吗？因此在周期性tick中，其handler中也有可能修改passed_quiesc，如下：

void rcu_check_callbacks(int cpu, int user) 
{ 
    if (user || －－－－－－－－－－－－－－－－－－－－－－－－－－－－（1） 
        (idle_cpu(cpu) && !in_softirq() && 
                hardirq_count() <= (1 << HARDIRQ_SHIFT))) {－－－－－－－－（2） 
        rcu_qsctr_inc(cpu); 
        rcu_bh_qsctr_inc(cpu); 
    } else if (!in_softirq())－－－－－－－－－－－－－－－－－－－－－－－（3） 
        rcu_bh_qsctr_inc(cpu); 
    tasklet_schedule(&per_cpu(rcu_tasklet, cpu)); 
}

（1）首先，我们知道，rcu_check_callbacks函数是在tick handler中处理，如果timer中断命中了userspace，说明thread已经立刻内核空间，那么必定离开了reader侧的临界区（RCU读侧的临界区不能block，而且我们在rcu_read_lock中disable了preempt，因此RCU读侧的临界区也不会发生抢占）。

（2）OK，我们再思考这样的问题：在进程切换的时候以及在tick handler中命中userspace的场景中修改passed_quiesc是否足够呢？很遗憾，还是不行，考虑这样一个场景，该CPU比较空闲，在idle进程中，而且由于没有其他ready的进程，因此不进行进程切换。此外，在tick handler中，由于always是kernel space，因此也无法标注passed_quiesc的flag。在这样的场景中，由于那个alway idle的CPU上的passed_quiesc的flag无法置位，Grace Period变得非常的长（直到有实际的进程被调度）。

因此，当tick handler命中idle进程的时候，我们还需要进行特别的处理，去掉softirq场景，去掉中断嵌套的场景，那么如果在tick中断中发现CPU实际上是真真正正的执行idle的代码逻辑，那么可以认为本CPU的QS已经通过。

（3）如果tick中断不是打断softirq的执行，那么则说明之前的softirq的handler都执行完毕了，因此bh类型的RCU状态。本文不会对此进行细致描述，读者有兴趣可以自行阅读。

4、CPU1的tick事件到来

CPU1的tick事件到来的时候，会有哪些逻辑动作呢？在rcu_check_quiescent_state函数中，检测到passed_quiesc被置位，这是推动进入下一个状态的原动力。这时候会执行下面的动作：

rdp->qs_pending = 0; 
if (likely(rdp->quiescbatch == rcp->cur)) 
    cpu_quiet(rdp->cpu, rcp, rsp);

一方面由于已经探测到本CPU的QS已经通过，因此clear该CPU的qs_pending，说明当前批次在本CPU上的QS状态探测已经完毕。另外，需要调用cpu_quiet来设置本CPU的Quiescent state，代码如下：

static void cpu_quiet(int cpu, struct rcu_ctrlblk *rcp, struct rcu_state *rsp) 
{ 
    cpu_clear(cpu, rsp->cpumask);－－－－－－－－－－－－－－－－（1） 
    if (cpus_empty(rsp->cpumask)) {－－－－－－－－－－－－－－－（2） 
        rcp->completed = rcp->cur;－－－－－－－－－－－－－－－－（3） 
        rcu_start_batch(rcp, rsp, 0);－－－－－－－－－－－－－－－－（4） 
    } 
}

（1）清除cpumask这个bitmask中代表自己的那个bit

（2）如果cpumask等于0，那么说明整个GP已经过去，一个新的时代，哦，不，一个新的批次就要开始了。

（3）当前批次已经渡过GP，但是这只是一个本CPU才知道的秘密，需要广播出去，让所有CPU都知道，成为尽人皆知的秘密。因此，实际上对于系统中最后一个历经Quiescent state的CPU而言，它的责任重大，它必须肩负起将该批次的Grace period已经过去的消息广播到所有的CPU上去。这个广播操作很简单，就是rcp->completed = rcp->cur，系统中的所有CPU都可以通过查阅rcp->completed了解当前已过Grace Period的批次号是哪个，然后，通过和rdp->batch进行比对，就知道自己的当前批次的请求（注意：这里的当前批次是指该CPU的当前批次，在curlist链表中）是否可以转移到donelist中并进行处理。

（4）我们知道，完成一批就意味着要新启动一批请求的处理，因此调用rcu_start_batch(rcp, rsp, 0)启动下一个批次，但是，由于next_pending等于0，因此实际上没有什么动作

至此，我们可以得到当前的状态如下：

rcu_state	rcu_ctrlblk	CPU0的rcu_data	CPU1的rcu_data
cpumask（0，0）	Cur（5）  Completed（5）  next_pending（0）	Quiescbatch（5）  passed_quiesc（1）  qs_pending（0）  batch（5）  nextlist（）  curlist（＝＝）  donelist（）	Quiescbatch（5）  passed_quiesc（1）  qs_pending（0）  batch（5）  nextlist（）  curlist（＝＝＝）  donelist（）

 

5、CPU0上有新的RCU请求

虽然在上面的tick中，发现了批次5的请求已经渡过GP，但是，真正的处理还需要等待到下一个tick中去完成。在两个tick之间，如果CPU0有新的RCU请求会怎么样呢？实际的动作很简单，这些新的请求都会挂入nextlist链表。

系统中各个CPU上的updater都会在适当的时机调用call_rcu或者synchronize_kernel接口函数，以便注册一个RCU callback函数。这些请求会被挂入各个CPU的struct rcu_data的nextlist链表中，只要用户调用call_rcu或者synchronize_kernel接口函数，callback请求就会挂入，没有任何的条件约束。

这些nextlist链表中RCU callback请求，没有ID（batch number），不和任何批次相关，不会和任何Grace Period的检测相关，各个CPU都会以HZ的频率来检测是否需要启动一次新的批次，以便可以将挂在nextlist链表中的请求转移到curlist中。对于一个特定的CPU而言，这个检测条件很简单：

（1）有新申请的RCU callback请求（nextlist非空）

（2）该CPU当前批次的请求（在curlist链表中）Grace Period已经过去，并且已经转移到donelist进行处理（curlist链表为空）

OK，至此，我们可以得到当前的状态如下：

rcu_state	rcu_ctrlblk	CPU0的rcu_data	CPU1的rcu_data
cpumask（0，0）	Cur（5）  Completed（5）  next_pending（0）	Quiescbatch（5）  passed_quiesc（1）  qs_pending（0）  batch（5）  nextlist（**）  curlist（＝＝）  donelist（）	Quiescbatch（5）  passed_quiesc（1）  qs_pending（0）  batch（5）  nextlist（）  curlist（＝＝＝）  donelist（）

 

6、CPU0的tick到来事件处理

这次tick的到来会处理不少的事情，我们一起来看看__rcu_process_callbacks的代码：

static void __rcu_process_callbacks(struct rcu_ctrlblk *rcp, 
            struct rcu_state *rsp, struct rcu_data *rdp) 
{ 
    if (rdp->curlist && !rcu_batch_before(rcp->completed, rdp->batch)) {－－－－（1） 
        将curlist链表中的请求转移到donelist 
    }

    local_irq_disable(); 
    if (rdp->nxtlist && !rdp->curlist) { 
        将nxtlistlist链表中的请求转移到curlist 
        local_irq_enable();

        rdp->batch = rcp->cur + 1;－－－－－－－－－－－－－－－－－－－－－（2） 
        smp_rmb();

        if (!rcp->next_pending) {－－－－－－－－－－－－－－－－－－－－－－（3） 
            spin_lock(&rsp->lock); 
            rcu_start_batch(rcp, rsp, 1);－－－－－－－－－－－－－－－－－－－（4） 
            spin_unlock(&rsp->lock); 
        } 
    } else { 
        local_irq_enable(); 
    } 
    rcu_check_quiescent_state(rcp, rsp, rdp); 
    if (rdp->donelist) 
        rcu_do_batch(rdp);－－－－－完成callback函数的执行

}

（1）一旦该cpu上的curlist链表有请求，并且completed大于或者等于其批次号（rdp->batch），那么说明该CPU上的curlist链表中的批次已经渡过GP，可以移动到donelist中了。

（2）curlist已经清空，可以把nxtlist中的请求放到curlist中，当然，最重要的是分配一个批次号来标识它。当然，这个ID是6的批次虽然在curlist中，但是并不是系统中的当前批次（当前批次仍然是5）。

（3）由于没有pending的批次（next_pending等于0），因此，调用rcu_start_batch来试图启动一个新的批次。

（4）rcu_start_batch函数中的next_pending参数是用来控制是否设定pending标志的，对于这个场景，我们刚刚分配了一个新的batch number，当然传递1的标志了，这样可以将其设定为pending状态。不过，如果能够顺利的启动一个新的批次，那么pending标志还是会被清除（一个新分配的批次如果立刻启动，那么说明其变成当前的批次，而不是pending了）。具体代码如下：

static void rcu_start_batch(struct rcu_ctrlblk *rcp, struct rcu_state *rsp, int next_pending) 
{ 
    if (next_pending) 
        rcp->next_pending = 1; －－－－－先设置为pending，pending的批次号是6

    if (rcp->next_pending && rcp->completed == rcp->cur) {－－－当前的5号批次刚刚完成 
        启动6号批次的GP探测 
        cpus_andnot(rsp->cpumask, cpu_online_map, nohz_cpu_mask);

        rcp->next_pending = 0;－－－6号批次已经扶正，成为current，因此clear pending flag 
        smp_wmb(); 
        rcp->cur++;－－－－－让cur成员指向6号批次 
    } 
}

如果当前批次刚刚完成（rcp->completed == rcp->cur）并且有pending的批次（批次6刚刚新鲜出炉）要处理，那么还犹豫什么，立刻启动这个pending的批次处理，这时候，pending的6号批次也就变成了current批次了。需要注意的是：当前的批次是6号批次是一个重大事件，需要广播到系统中的所有的CPU上去，怎么广播的呢？通过rcp->cur++这条指令完成。而其他CPU可以通过对比Quiescbatch和rcp->cur就可以知道是否启动了一个新的批次了。

顺便一提的是：虽然CPU1上5号批次尚未完成，CPU0又启动了6号批次GP的处理（整个系统的cpumask（包括CPU1）被重置），不过没有关系，通过对比rcp->completed和该CPU的rdp->batch就可以知道该CPU上的curlist是否完成了GP探测，需要移动到donelist中。

__rcu_process_callbacks中的大部分的代码逻辑已经清晰，除了rcu_check_quiescent_state，代码如下：

static void rcu_check_quiescent_state(struct rcu_ctrlblk *rcp, 
            struct rcu_state *rsp, struct rcu_data *rdp) 
{ 
    if (rdp->quiescbatch != rcp->cur) {－－－－－－－－－－－－－－－－（1） 
        rdp->qs_pending = 1; 
        rdp->passed_quiesc = 0; 
        rdp->quiescbatch = rcp->cur; 
        return; 
    }  
    if (!rdp->qs_pending)－－－－－－－－－－－－－－－－－－－－－－（2） 
        return;  
    if (!rdp->passed_quiesc)－－－－－－－－－－－－－－－－－－－－－（3） 
        return; 
    rdp->qs_pending = 0; －－－－－－－－－－－－－－－－－－－－－－（4）

…… 
}

（1）已经启动了新的6号批次的GP处理，而GP的处理需要各个CPU上的QS的检测，因此，在一个新的批次开始之初，需要reset本CPU上的QS检测的flag。

（2）rcu_check_quiescent_state在各个CPU上都是会反复执行，不过，如果本CPU的QS已经通过，只是在等待其他CPU QS的通过，那么直接return是最好的选择。

（3）如果需要探测本CPU的QS，那么就看看passed_quiesc是否被设置成1

（4）代码执行到这里，说明本CPU的QS已经通过。不过，由于这些代码逻辑和本场景不符，更具体的代码逻辑后续再描述。

至此，我们可以得到当前的状态如下：

rcu_state	rcu_ctrlblk	CPU0的rcu_data	CPU1的rcu_data
cpumask（1，1）	Cur（6）  Completed（5）  next_pending（0）	Quiescbatch（6）  passed_quiesc（0）  qs_pending（1）  batch（6）  nextlist（）  curlist（**）  donelist（＝＝）	Quiescbatch（5）  passed_quiesc（1）  qs_pending（0）  batch（5）  nextlist（）  curlist（＝＝＝）  donelist（）

由上面的状态图可知，当前处理的批次已经变成6，当然5是上次完成的那个批次，在CPU1上还没有处理完毕。此外，CPU0的donelist中的请求也会在本次tick handler中处理。如果请求较少，那么可以一次性处理完毕，这时候donelist是空，否则，会在几次tick handler中（tasklet context）处理完成。

 

7、CPU1的tick到来事件处理

这里留给大家自己看吧，其过程类似上节，我们这里只是给出状态：

rcu_state	rcu_ctrlblk	CPU0的rcu_data	CPU1的rcu_data
cpumask（1，1）	Cur（6）  Completed（5）  next_pending（0）	Quiescbatch（6）  passed_quiesc（0）  qs_pending（1）  batch（6）  nextlist（）  curlist（**）  donelist（）	Quiescbatch（6）  passed_quiesc（0）  qs_pending（1）  batch（5）  nextlist（）  curlist（）  donelist（＝＝＝）

这时候，CPU1上的batch等于5没有任何实际的意义，因为curlist为空了。

 

8、CPU1上有新的RCU请求

状态如下：

rcu_state	rcu_ctrlblk	CPU0的rcu_data	CPU1的rcu_data
cpumask（1，1）	Cur（6）  Completed（5）  next_pending（0）	Quiescbatch（6）  passed_quiesc（0）  qs_pending（1）  batch（6）  nextlist（）  curlist（**）  donelist（）	Quiescbatch（6）  passed_quiesc（0）  qs_pending（1）  batch（5）  nextlist（****）  curlist（）  donelist（）

 

9、CPU0的tick再次到来

只是简单的检测QS状态，不过发现passed_quiesc是0就失望而归了，具体代码大家自己过吧。

 

10、CPU1的tick再次到来

由于curlist为空并且有新的请求，因此nextlist中的请求又需要移到curlist链表中，同时伴随的动作就是分配一个新的batch number 7，__rcu_process_callbacks相关代码如下：

        rdp->batch = rcp->cur + 1;－－－－－－－－－－－－－（1） 
        smp_rmb(); －－－－－－－－－－－－－－－－－－－－（2）

        if (!rcp->next_pending) {－－－－－－－－－－－－－－－（3） 
            spin_lock(&rsp->lock); 
            rcu_start_batch(rcp, rsp, 1);－－－－－－－－－－－－（4） 
            spin_unlock(&rsp->lock); 
        }

（1）新的批次号总是current批次号加一，因此7号批次横空出世，请求挂在本cpu的curlist中。

（2）这里的memory barrier需要结合rcu_start_batch的代码来看，我们整理下面的表格：

CPU A执行分配一个新批次的代码	CPU B执行rcu_start_batch的代码
rdp->batch = rcp->cur + 1;－－－－（X）  smp_rmb(); if (!rcp->next_pending) {－－－－－（Y）  ……启动新的batch    }	rcp->next_pending = 0;  smp_wmb(); rcp->cur++;

在CPU B上，真正启动了一个新的批次，并且smp_wmb的memory barrier保证了先store rcp->next_pending然后store rcp->cur。站在CPU A的角度观察，如果在X处对cur load的值是CPU B中加一之后的值，那么smp_rmp可以保证在Y处，CPU A对next_pending的加载操作必定得到0值。只有这样，才能保证进入rcu_start_batch，从而启动一个新的批次。

（3）如果已经有了pending的批次，那么无论哪一个CPU，只要当前的批次没有渡过GP，那么任何后续的新分配的批次号都是7（只有完成GP之后cur才会加一）。如果已经pending，那么其实是没有必要调用rcu_start_batch函数去启动一个新的批次的。

（4）虽然7号批次横空出世，但是是否能启动GP探测呢？当然不能，目前current的6号批次还没有渡过GP，各个CPU还在不断的检测其passed_quiesc呢。因此，这里的rcu_start_batch过程执行如下：

if (next_pending) 
    rcp->next_pending = 1; －－－－－－－因为有了7号批次，因此设定pending标志

if (rcp->next_pending && 
        rcp->completed == rcp->cur) {－－－－－completed是5，而current是6，因此是false 
    …… 
}

completed是5，而current是6，这就意味着当前批次还没有完成GP探测，不能启动新的批次。至此，我们可以得到当前的状态如下： 

rcu_state	rcu_ctrlblk	CPU0的rcu_data	CPU1的rcu_data
cpumask（1，1）	Cur（6）  Completed（5）  next_pending（1）	Quiescbatch（6）  passed_quiesc（0）  qs_pending（1）  batch（6）  nextlist（）  curlist（**）  donelist（）	Quiescbatch（6）  passed_quiesc（0）  qs_pending（1）  batch（7）  nextlist（）  curlist（****）  donelist（）

这时候，系统中的当前批次是6号批次，正在检测GP，5号批次已经完成，而7号批次是处于pending状态，一旦6号批次完成，7号就可以上位，成为当前批次。

11、6号批次的GP过去之后会怎样？

我们假设CPU0和CPU1的QS都已经通过，也就是说在某个CPU（最后那个通过QS的）cpu_quiet函数中会启动新的批次，这时候的代码如下：

static void cpu_quiet(int cpu, struct rcu_ctrlblk *rcp, struct rcu_state *rsp) 
{ 
    cpu_clear(cpu, rsp->cpumask); 
    if (cpus_empty(rsp->cpumask)) { 
        rcp->completed = rcp->cur; 
        rcu_start_batch(rcp, rsp, 0); 
    } 
}

这时候，  rcu_start_batch函数传递的next_pending参数是0，也就是说不标记pending状态（又不是向curlist中提交请求，当然不会标记pending了）。但是，实际上RCU控制块中的pending已经置位了，这时候rcu_start_batch中的条件成立，因此会立刻启动7号批次。

12、如果你愿意，你可以设定任何的事件，然后走读代码，推动系统的状态持续运行，本文的分析就到此为止了。

 

六、参考文献

1、linux2.6.11 source code

2、开源社区提交的patch说明

 

原创文章，转发请注明出处。蜗窝科技