深入了解RCU是怎樣實(shí)現(xiàn)的？

深入理解RCU實(shí)現(xiàn)

——基于內(nèi)核2.6.21RCU實(shí)現(xiàn)

RCU(Read-Copy Update)，顧名思義就是讀-拷貝修改，它是基于其原理命名的。對(duì)于被RCU保護(hù)的共享數(shù)據(jù)結(jié)構(gòu)，讀者不需要獲得任何鎖就可以訪問它，但寫者在訪問它時(shí)首先拷貝一個(gè)副本，然后對(duì)副本進(jìn)行修改，最后使用一個(gè)回調(diào)（callback）機(jī)制在適當(dāng)?shù)臅r(shí)機(jī)把指向原來(lái)數(shù)據(jù)的指針重新指向新的被修改的數(shù)據(jù)。那么這個(gè)“適當(dāng)?shù)臅r(shí)機(jī)”是怎么確定的呢？這是由內(nèi)核確定的，也是我們后面討論的重點(diǎn)。

RCU原理

RCU實(shí)際上是一種改進(jìn)的rwlock，讀者幾乎沒有什么同步開銷，它不需要鎖，不使用原子指令，而且在除alpha的所有架構(gòu)上也不需要內(nèi)存柵（Memory Barrier），因此不會(huì)導(dǎo)致鎖競(jìng)爭(zhēng)，內(nèi)存延遲以及流水線停滯。不需要鎖也使得使用更容易，因?yàn)樗梨i問題就不需要考慮了。寫者的同步開銷比較大，它需要延遲數(shù)據(jù)結(jié)構(gòu)的釋放，復(fù)制被修改的數(shù)據(jù)結(jié)構(gòu)，它也必須使用某種鎖機(jī)制同步并行的其它寫者的修改操作。

讀者必須提供一個(gè)信號(hào)給寫者以便寫者能夠確定數(shù)據(jù)可以被安全地釋放或修改的時(shí)機(jī)。有一個(gè)專門的垃圾收集器來(lái)探測(cè)讀者的信號(hào)，一旦所有的讀者都已經(jīng)發(fā)送信號(hào)告知它們都不在使用被RCU保護(hù)的數(shù)據(jù)結(jié)構(gòu)，垃圾收集器就調(diào)用回調(diào)函數(shù)完成最后的數(shù)據(jù)釋放或修改操作。

RCU與rwlock的不同之處是：它既允許多個(gè)讀者同時(shí)訪問被保護(hù)的數(shù)據(jù)，又允許多個(gè)讀者和多個(gè)寫者同時(shí)訪問被保護(hù)的數(shù)據(jù)（注意：是否可以有多個(gè)寫者并行訪問取決于寫者之間使用的同步機(jī)制），讀者沒有任何同步開銷，而寫者的同步開銷則取決于使用的寫者間同步機(jī)制。但RCU不能替代rwlock，因?yàn)槿绻麑懕容^多時(shí)，對(duì)讀者的性能提高不能彌補(bǔ)寫者導(dǎo)致的損失。

讀者在訪問被RCU保護(hù)的共享數(shù)據(jù)期間不能被阻塞，這是RCU機(jī)制得以實(shí)現(xiàn)的一個(gè)基本前提，也就說(shuō)當(dāng)讀者在引用被RCU保護(hù)的共享數(shù)據(jù)期間，讀者所在的CPU不能發(fā)生上下文切換，spinlock和rwlock都需要這樣的前提。寫者在訪問被RCU保護(hù)的共享數(shù)據(jù)時(shí)不需要和讀者競(jìng)爭(zhēng)任何鎖，只有在有多于一個(gè)寫者的情況下需要獲得某種鎖以與其他寫者同步。

寫者修改數(shù)據(jù)前首先拷貝一個(gè)被修改元素的副本，然后在副本上進(jìn)行修改，修改完畢后它向垃圾回收器注冊(cè)一個(gè)回調(diào)函數(shù)以便在適當(dāng)?shù)臅r(shí)機(jī)執(zhí)行真正的修改操作。等待適當(dāng)時(shí)機(jī)的這一時(shí)期稱為grace period，而CPU發(fā)生了上下文切換稱為經(jīng)歷一個(gè)quiescent state，grace period就是所有CPU都經(jīng)歷一次quiescent state所需要的等待的時(shí)間。垃圾收集器就是在grace period之后調(diào)用寫者注冊(cè)的回調(diào)函數(shù)來(lái)完成真正的數(shù)據(jù)修改或數(shù)據(jù)釋放操作的。

要想使用好RCU，就要知道RCU的實(shí)現(xiàn)原理。我們拿linux 2.6.21 kernel的實(shí)現(xiàn)開始分析，為什么選擇這個(gè)版本的實(shí)現(xiàn)呢？因?yàn)檫@個(gè)版本的實(shí)現(xiàn)相對(duì)較為單純，也比較簡(jiǎn)單。當(dāng)然之后內(nèi)核做了不少改進(jìn)，如搶占RCU、可睡眠RCU、分層RCU。但是基本思想都是類似的。所以先從簡(jiǎn)單入手。

首先，上一節(jié)我們提到，寫者在訪問它時(shí)首先拷貝一個(gè)副本，然后對(duì)副本進(jìn)行修改，最后使用一個(gè)回調(diào)（callback）機(jī)制在適當(dāng)?shù)臅r(shí)機(jī)把指向原來(lái)數(shù)據(jù)的指針重新指向新的被修改的數(shù)據(jù)。而這個(gè)“適當(dāng)?shù)臅r(shí)機(jī)”就是所有CPU經(jīng)歷了一次進(jìn)程切換（也就是一個(gè)grace period）。為什么這么設(shè)計(jì)？因?yàn)镽CU讀者的實(shí)現(xiàn)就是關(guān)搶占執(zhí)行讀取，讀完了當(dāng)然就可以進(jìn)程切換了，也就等于是寫者可以操作臨界區(qū)了。那么就自然可以想到，內(nèi)核會(huì)設(shè)計(jì)兩個(gè)元素，來(lái)分別表示寫者被掛起的起始點(diǎn)，以及每cpu變量，來(lái)表示該cpu是否經(jīng)過(guò)了一次進(jìn)程切換(quies state)。就是說(shuō)，當(dāng)寫者被掛起后，

1）重置每cpu變量，值為0。

2）當(dāng)某個(gè)cpu經(jīng)歷一次進(jìn)程切換后，就將自己的變量設(shè)為1。

3）當(dāng)所有的cpu變量都為1后，就可以喚醒寫者了。

下面我們來(lái)分別看linux里是如何完成這三步的。

從一個(gè)例子開始

我們從一個(gè)例子入手，這個(gè)例子來(lái)源于linux kernel文檔中的whatisRCU.txt。這個(gè)例子使用RCU的核心API來(lái)保護(hù)一個(gè)指向動(dòng)態(tài)分配內(nèi)存的全局指針。

C++

struct foo {int a; char b; long c;};DEFINE_SPINLOCK(foo_mutex);struct foo *gbl_foo;void foo_update_a(int new_a){ struct foo *new_fp;struct foo *old_fp;new_fp = kmalloc(sizeof(*new_fp), GFP_KERNEL);spin_lock(&foo_mutex);old_fp = gbl_foo;*new_fp = *old_fp;new_fp->a = new_a;rcu_assign_pointer(gbl_foo, new_fp);spin_unlock(&foo_mutex);synchronize_rcu();kfree(old_fp);}int foo_get_a(void){ int retval;rcu_read_lock();retval = rcu_dereference(gbl_foo)->a;rcu_read_unlock();return retval; }

struct foo {

int a;

char b;

long c;

};

DEFINE_SPINLOCK(foo_mutex);

struct foo *gbl_foo;

void foo_update_a(int new_a)

{

struct foo *new_fp;

struct foo *old_fp;

new_fp = kmalloc(sizeof(*new_fp), GFP_KERNEL);

spin_lock(&foo_mutex);

old_fp = gbl_foo;

*new_fp = *old_fp;

new_fp->a = new_a;

rcu_assign_pointer(gbl_foo, new_fp);

spin_unlock(&foo_mutex);

synchronize_rcu();

kfree(old_fp);

}

int foo_get_a(void)

{

int retval;

rcu_read_lock();

retval = rcu_dereference(gbl_foo)->a;

rcu_read_unlock();

return retval;

}

如上代碼所示,RCU被用來(lái)保護(hù)全局指針struct foo *gbl_foo。foo_get_a()用來(lái)從RCU保護(hù)的結(jié)構(gòu)中取得gbl_foo的值。而foo_update_a()用來(lái)更新被RCU保護(hù)的gbl_foo的值（更新其a成員）。

首先,我們思考一下,為什么要在foo_update_a()中使用自旋鎖foo_mutex呢?假設(shè)中間沒有使用自旋鎖.那foo_update_a()的代碼如下:

C++

void foo_update_a(int new_a) { struct foo *new_fp; struct foo *old_fp; new_fp = kmalloc(sizeof(*new_fp), GFP_KERNEL); old_fp = gbl_foo; 1:------------------------- *new_fp = *old_fp; new_fp->a = new_a; rcu_assign_pointer(gbl_foo, new_fp); synchronize_rcu(); kfree(old_fp); }

void foo_update_a(int new_a)

{

struct foo *new_fp;

struct foo *old_fp;

new_fp = kmalloc(sizeof(*new_fp), GFP_KERNEL);

old_fp = gbl_foo;

1:-------------------------

*new_fp = *old_fp;

new_fp->a = new_a;

rcu_assign_pointer(gbl_foo, new_fp);

synchronize_rcu();

kfree(old_fp);

}

假設(shè)A進(jìn)程在上圖—-標(biāo)識(shí)處被B進(jìn)程搶點(diǎn).B進(jìn)程也執(zhí)行了goo_ipdate_a().等B執(zhí)行完后，再切換回A進(jìn)程.此時(shí),A進(jìn)程所持的old_fd實(shí)際上已經(jīng)被B進(jìn)程給釋放掉了.此后A進(jìn)程對(duì)old_fd的操作都是非法的。所以在此我們得到一個(gè)重要結(jié)論：RCU允許多個(gè)讀者同時(shí)訪問被保護(hù)的數(shù)據(jù)，也允許多個(gè)讀者在有寫者時(shí)訪問被保護(hù)的數(shù)據(jù)（但是注意：是否可以有多個(gè)寫者并行訪問取決于寫者之間使用的同步機(jī)制）。

說(shuō)明：本文中說(shuō)的進(jìn)程不是用戶態(tài)的進(jìn)程，而是內(nèi)核的調(diào)用路徑，也可能是內(nèi)核線程或軟中斷等。

RCU的核心API

另外,我們?cè)谏厦嬉部吹搅藥讉€(gè)有關(guān)RCU的核心API。它們?yōu)閯e是:

C++

rcu_read_lock() rcu_read_unlock() synchronize_rcu() rcu_assign_pointer() rcu_dereference()

rcu_read_lock()

rcu_read_unlock()

synchronize_rcu()

rcu_assign_pointer()

rcu_dereference()

其中,rcu_read_lock()和rcu_read_unlock()用來(lái)保持一個(gè)讀者的RCU臨界區(qū).在該臨界區(qū)內(nèi)不允許發(fā)生上下文切換。為什么不能發(fā)生切換呢？因?yàn)閮?nèi)核要根據(jù)“是否發(fā)生過(guò)切換”來(lái)判斷讀者是否已結(jié)束讀操作，我們后面再分析。
rcu_dereference()：讀者調(diào)用它來(lái)獲得一個(gè)被RCU保護(hù)的指針。
rcu_assign_pointer()：寫者使用該函數(shù)來(lái)為被RCU保護(hù)的指針分配一個(gè)新的值。

synchronize_rcu()：這是RCU的核心所在，它掛起寫者，等待讀者都退出后釋放老的數(shù)據(jù)。

RCU API實(shí)現(xiàn)分析

lrcu_read_lock()和rcu_read_unlock()

rcu_read_lock()和rcu_read_unlock()的實(shí)現(xiàn)如下:

C++

#define rcu_read_lock() __rcu_read_lock() #define rcu_read_unlock() __rcu_read_unlock() #define __rcu_read_lock() do { preempt_disable(); __acquire(RCU); rcu_read_acquire(); } while (0) #define __rcu_read_unlock() do { rcu_read_release(); __release(RCU); preempt_enable(); } while (0)

#define rcu_read_lock() __rcu_read_lock()

#define rcu_read_unlock() __rcu_read_unlock()

#define __rcu_read_lock()

do {

preempt_disable();

__acquire(RCU);

rcu_read_acquire();

} while (0)

#define __rcu_read_unlock()

do {

rcu_read_release();

__release(RCU);

preempt_enable();

} while (0)

其中__acquire(),rcu_read_acquire(),rcu_read_release(),__release()都是一些選擇編譯函數(shù),可以忽略不可看。因此可以得知：rcu_read_lock(),rcu_read_unlock()只是禁止和啟用搶占.因?yàn)樵谧x者臨界區(qū)，不允許發(fā)生上下文切換。

lrcu_dereference()和rcu_assign_pointer()

rcu_dereference()和rcu_assign_pointer()的實(shí)現(xiàn)如下:

C++

rcu_dereference()和rcu_assign_pointer()的實(shí)現(xiàn)如下: #define rcu_dereference(p) ({ typeof(p) _________p1 = ACCESS_ONCE(p); smp_read_barrier_depends(); (_________p1); }) #define rcu_assign_pointer(p, v) ({ if (!__builtin_constant_p(v) || ((v) != NULL)) smp_wmb(); (p) = (v); })

rcu_dereference()和rcu_assign_pointer()的實(shí)現(xiàn)如下:

#define rcu_dereference(p) ({

typeof(p) _________p1 = ACCESS_ONCE(p);

smp_read_barrier_depends();

(_________p1);

})

#define rcu_assign_pointer(p, v)

({

if (!__builtin_constant_p(v) || ((v) != NULL))

smp_wmb();

(p) = (v);

})

它們的實(shí)現(xiàn)也很簡(jiǎn)單.因?yàn)樗鼈儽旧矶际窃硬僮?。只是為了cache一致性，插上了內(nèi)存屏障。可以讓其它的讀者/寫者可以看到保護(hù)指針的最新值.

lsynchronize_rcu()

synchronize_rcu()在RCU中是一個(gè)最核心的函數(shù),它用來(lái)等待之前的讀者全部退出.我們后面的大部份分析也是圍繞著它而進(jìn)行.實(shí)現(xiàn)如下:

C++

void synchronize_rcu(void) { struct rcu_synchronize rcu; init_completion(&rcu.completion); /* Will wake me after RCU finished */ call_rcu(&rcu.head, wakeme_after_rcu); /* Wait for it */ wait_for_completion(&rcu.completion); }

void synchronize_rcu(void)

{

struct rcu_synchronize rcu;

init_completion(&rcu.completion);

/* Will wake me after RCU finished */

call_rcu(&rcu.head, wakeme_after_rcu);

/* Wait for it */

wait_for_completion(&rcu.completion);

}

我們可以看到,它初始化了一個(gè)本地變量,它的類型為struct rcu_synchronize.調(diào)用call_rcu()之后.一直等待條件變量rcu.competion的滿足。

在這里看到了RCU的另一個(gè)核心API,它就是call_run()。它的定義如下:

C++

void call_rcu(struct rcu_head *head, void (*func)(struct rcu_head *rcu)) { unsigned long flags; struct rcu_data *rdp; head->func = func; head->next = NULL; local_irq_save(flags); rdp = &__get_cpu_var(rcu_data); *rdp->nxttail = head; rdp->nxttail = &head->next; if (unlikely(++rdp->qlen > qhimark)) { rdp->blimit = INT_MAX; force_quiescent_state(rdp, &rcu_ctrlblk); } local_irq_restore(flags); }

void call_rcu(struct rcu_head *head,??void (*func)(struct rcu_head *rcu))

{

unsigned long flags;

struct rcu_data *rdp;

head->func = func;

head->next = NULL;

local_irq_save(flags);

rdp = &__get_cpu_var(rcu_data);

*rdp->nxttail = head;

rdp->nxttail = &head->next;

if (unlikely(++rdp->qlen > qhimark)) {

rdp->blimit = INT_MAX;

force_quiescent_state(rdp, &rcu_ctrlblk);

}

local_irq_restore(flags);

}

該函數(shù)也很簡(jiǎn)單,就是將參數(shù)傳入的回調(diào)函數(shù)fun賦值給一個(gè)struct rcu_head變量，再將這個(gè)struct rcu_head加在了per_cpu變量rcu_data的nxttail 鏈表上。

rcu_data定義如下，是個(gè)每cpu變量:
DEFINE_PER_CPU(struct rcu_data, rcu_data) = { 0L };

接著我們看下call_rcu注冊(cè)的函數(shù)，我們也可以看到,在synchronize_rcu()中,傳入call_rcu的函數(shù)為wakeme_after_rcu()，其實(shí)現(xiàn)如下:

C++

static void wakeme_after_rcu(struct rcu_head *head) { struct rcu_synchronize *rcu; rcu = container_of(head, struct rcu_synchronize, head); complete(&rcu->completion); }

static void wakeme_after_rcu(struct rcu_head *head)

{

struct rcu_synchronize *rcu;

rcu = container_of(head, struct rcu_synchronize, head);

complete(&rcu->completion);

}

我們可以看到,該函數(shù)將條件變量置真,然后喚醒了在條件變量上等待的進(jìn)程。

由此,我們可以得知,每一個(gè)CPU都有一個(gè)rcu_data.每個(gè)調(diào)用call_rcu()/synchronize_rcu()進(jìn)程的進(jìn)程都會(huì)將一個(gè)rcu_head都會(huì)掛到rcu_data的nxttail鏈表上（這個(gè)rcu_head其實(shí)就相當(dāng)于這個(gè)進(jìn)程在RCU機(jī)制中的體現(xiàn)），然后掛起。當(dāng)讀者都完成讀操作后（經(jīng)過(guò)一個(gè)grace period后）就會(huì)觸發(fā)這個(gè)rcu_head上的回調(diào)函數(shù)來(lái)喚醒寫者。整個(gè)過(guò)程如下圖所示：

看到這里，也就到了問題的關(guān)鍵，內(nèi)核是如何判斷當(dāng)前讀者都已經(jīng)完成讀操作了呢（經(jīng)過(guò)了一個(gè)grace period）？又是由誰(shuí)來(lái)觸發(fā)這個(gè)回調(diào)函數(shù)wakeme_after_rcu呢？下一小節(jié)再來(lái)分析。

從RCU的初始化說(shuō)起

那究竟怎么去判斷當(dāng)前的讀者已經(jīng)操作完了呢？我們?cè)谥翱吹?不是讀者在調(diào)用rcu_read_lock()的時(shí)候要禁止搶占么？因此,我們只需要判斷所有的CPU都進(jìn)過(guò)了一次上下文切換,就說(shuō)明所有讀者已經(jīng)退出了。要徹底弄清楚這個(gè)問題,我們得從RCU的初始化說(shuō)起。

RCU的初始化開始于start_kernel()–>rcu_init()。而其主要是對(duì)每個(gè)cpu調(diào)用了rcu_online_cpu函數(shù)。

lrcu_online_cpu

C++

static void __cpuinit rcu_online_cpu(int cpu) { struct rcu_data *rdp = &per_cpu(rcu_data, cpu); struct rcu_data *bh_rdp = &per_cpu(rcu_bh_data, cpu); rcu_init_percpu_data(cpu, &rcu_ctrlblk, rdp); rcu_init_percpu_data(cpu, &rcu_bh_ctrlblk, bh_rdp); open_softirq(RCU_SOFTIRQ, rcu_process_callbacks, NULL); }

static void __cpuinit rcu_online_cpu(int cpu)

{

struct rcu_data *rdp = &per_cpu(rcu_data, cpu);

struct rcu_data *bh_rdp = &per_cpu(rcu_bh_data, cpu);

rcu_init_percpu_data(cpu, &rcu_ctrlblk, rdp);

rcu_init_percpu_data(cpu, &rcu_bh_ctrlblk, bh_rdp);

open_softirq(RCU_SOFTIRQ, rcu_process_callbacks, NULL);

}

這個(gè)函數(shù)主要完成兩個(gè)操作：初始化兩個(gè)per cpu變量；注冊(cè)RCU_SOFTIRQ軟中斷處理函數(shù)rcu_process_callbacks。我們從這里又看到了另一個(gè)percpu變量：rcu_bh_data.有關(guān)bh的部份之后再來(lái)分析.在這里略過(guò)這些部分。下面看下rcu_init_percpu_data()的實(shí)現(xiàn)。

lrcu_init_percpu_data

C++

static void rcu_init_percpu_data(int cpu, struct rcu_ctrlblk *rcp, struct rcu_data *rdp) { memset(rdp, 0, sizeof(*rdp)); rdp->curtail = &rdp->curlist; rdp->nxttail = &rdp->nxtlist; rdp->donetail = &rdp->donelist; rdp->quiescbatch = rcp->completed; rdp->qs_pending = 0; rdp->cpu = cpu; rdp->blimit = blimit; }

static void rcu_init_percpu_data(int cpu, struct rcu_ctrlblk *rcp, struct rcu_data *rdp)

{

memset(rdp, 0, sizeof(*rdp));

rdp->curtail = &rdp->curlist;

rdp->nxttail = &rdp->nxtlist;

rdp->donetail = &rdp->donelist;

rdp->quiescbatch = rcp->completed;

rdp->qs_pending = 0;

rdp->cpu = cpu;

rdp->blimit = blimit;

}

調(diào)用這個(gè)函數(shù)的第二個(gè)參數(shù)是一個(gè)全局變量rcu_ctlblk。在繼續(xù)向下分析之前我們先看下這些函數(shù)用到的一些重要數(shù)據(jù)結(jié)構(gòu)。

重要數(shù)據(jù)結(jié)構(gòu)

說(shuō)明：下列數(shù)據(jù)結(jié)構(gòu)只列出了主要成員。

C++

struct rcu_ctrlblk { long cur; long completed; cpumask_t cpumask; };

struct rcu_ctrlblk {??

long cur;??

long completed;??

cpumask_t??cpumask;??

};

structrcu_ctrlblk的主要作用就是定義上節(jié)提到的全局變量rcu_ctlblk，這個(gè)變量在系統(tǒng)中是唯一的。另外說(shuō)明一下，為了記錄方便，內(nèi)核將從啟動(dòng)開始的每個(gè)grace period對(duì)應(yīng)一個(gè)數(shù)字表示。

這里的cpumask是為了標(biāo)識(shí)當(dāng)前系統(tǒng)中的所有cpu，以便標(biāo)記哪些cpu發(fā)生過(guò)上下文切換(經(jīng)歷過(guò)一個(gè)quiescent state)。而cur，completed，則用來(lái)同步。我們可以這樣理解，cur和completed是系統(tǒng)級(jí)別的記錄信息，也即系統(tǒng)實(shí)時(shí)經(jīng)歷的grace編號(hào)，一般情況下，新開一個(gè)graceperiod等待周期的話，cur會(huì)加1，當(dāng)graceperiod結(jié)束后，會(huì)將completed置為cur，所以通常情況下，都是completed追著cur跑。

那么我們可能會(huì)猜測(cè)，是不是如果complete= curr -1 的時(shí)候，就表示系統(tǒng)中g(shù)raceperiod還沒有結(jié)束？當(dāng)completed= curr的時(shí)候，就表示系統(tǒng)中不存在graceperiod等待周期了？我們姑且這么理解，實(shí)際上有些許差別，但設(shè)計(jì)思想都是一樣的。

C++

struct rcu_data { long quiescbatch; int passed_quiesc; long batch; struct rcu_head *nxtlist; struct rcu_head **nxttail; struct rcu_head *curlist; struct rcu_head **curtail; struct rcu_head *donelist; struct rcu_head **donetail; };

struct rcu_data {??

long quiescbatch;??????

int passed_quiesc;??

long????????????batch;??????????

struct rcu_head *nxtlist;??

struct rcu_head **nxttail;??

struct rcu_head *curlist;??

struct rcu_head **curtail;??

struct rcu_head *donelist;??

struct rcu_head **donetail;??

};

上面的結(jié)構(gòu)，要達(dá)到的作用是，跟蹤單個(gè)CPU上的RCU事務(wù)。

(1)passed_quiesc：這是一個(gè)flag標(biāo)志，表示當(dāng)前cpu是否已經(jīng)發(fā)生過(guò)搶占了(經(jīng)歷過(guò)一個(gè)quiescent state)，0表示為未發(fā)送過(guò)切換，1表示發(fā)生過(guò)切換；

說(shuō)明：我們一直強(qiáng)調(diào)發(fā)生過(guò)一次cpu搶占就是經(jīng)歷過(guò)一個(gè)quiescent state，其實(shí)這是不嚴(yán)格的說(shuō)法。為什么呢？因?yàn)樽韵到y(tǒng)啟動(dòng)，各種進(jìn)程頻繁調(diào)度，肯定每個(gè)cpu都會(huì)發(fā)生過(guò)搶占。所以我們這里說(shuō)的“發(fā)生過(guò)搶占”是指從某個(gè)特殊的時(shí)間點(diǎn)開始發(fā)生過(guò)搶占。那么這個(gè)特殊的時(shí)間點(diǎn)是什么時(shí)候呢？就是我們調(diào)用synchronize_rcu將rcu_head掛在每cpu變量上并掛起進(jìn)程時(shí)，我們后面分析就會(huì)證實(shí)這一點(diǎn)。

(2) batch：表示一個(gè)graceperiodid，表示本次被阻塞的寫者，需要在哪個(gè)graceperiod之后被激活；

(3) quiescbatch：表示一個(gè)graceperiodid，用來(lái)比較當(dāng)前cpu是否處于等待進(jìn)程切換完成。

下面我們介紹剩下的三組指針，但首先注意這三組鏈表上的元素都是structrcu_head類型。這三個(gè)鏈表的作用還要結(jié)合整個(gè)RCU寫者從阻塞到喚醒的過(guò)程來(lái)看。

rcu寫者的整體流程，假設(shè)系統(tǒng)中出現(xiàn)rcu寫者阻塞，那么流程如下：

(1) 調(diào)用synchronize_rcu，將代表寫者的rcu_head添加到CPU[n]每cpu變量rcu_data的nxtlist，這個(gè)鏈表代表有需要提交給rcu處理的回調(diào)(但還沒有提交)；

(2) CPU[n]每次時(shí)鐘中斷時(shí)檢測(cè)自己的nxtlist是否為null，若不為空，因此則喚醒rcu軟中斷；

(3) RCU的軟中斷處理函數(shù)rcu_process_callbacks會(huì)挨個(gè)檢查本CPU的三個(gè)鏈表。

下面分析RCU的核心處理函數(shù)，也就是軟中斷處理函數(shù)rcu_process_callbacks。

RCU軟中斷

在“RCU API實(shí)現(xiàn)分析”一節(jié)我們知道調(diào)用synchronize_rcu，將代表寫者的rcu_head添加到了CPU[n]每cpu變量rcu_data的nxtlist。

而另一方面，在每次時(shí)鐘中斷中，都會(huì)調(diào)用update_process_times函數(shù)。

lupdate_process_times

C++

void update_process_times(int user_tick) { //...... if (rcu_pending(cpu)) rcu_check_callbacks(cpu, user_tick); //......}

void update_process_times(int user_tick)

{

//......

if (rcu_pending(cpu))

rcu_check_callbacks(cpu, user_tick);

//......

}

在每一次時(shí)鐘中斷,都會(huì)檢查rcu_pending(cpu)是否為真,如果是,就會(huì)為其調(diào)用rcu_check_callbacks()。

lrcu_pending

rcu_pending()的代碼如下:

C++

int rcu_pending(int cpu) { return __rcu_pending(&rcu_ctrlblk, &per_cpu(rcu_data, cpu)) || __rcu_pending(&rcu_bh_ctrlblk, &per_cpu(rcu_bh_data, cpu)); }

int rcu_pending(int cpu)

{

return __rcu_pending(&rcu_ctrlblk, &per_cpu(rcu_data, cpu)) ||

__rcu_pending(&rcu_bh_ctrlblk, &per_cpu(rcu_bh_data, cpu));

}

同上面一樣,忽略bh的部份，我們看__rcu_pending。

l__rcu_pending

C++

static int __rcu_pending(struct rcu_ctrlblk *rcp, struct rcu_data *rdp) { /* This cpu has pending rcu entries and the grace period * for them has completed. */ if (rdp->curlist && !rcu_batch_before(rcp->completed, rdp->batch)) return 1; /* This cpu has no pending entries, but there are new entries */ if (!rdp->curlist && rdp->nxtlist) return 1; /* This cpu has finished callbacks to invoke */ if (rdp->donelist) return 1; /* The rcu core waits for a quiescent state from the cpu */ if (rdp->quiescbatch != rcp->cur || rdp->qs_pending) return 1; /* nothing to do */ return 0; }

static int __rcu_pending(struct rcu_ctrlblk *rcp, struct rcu_data *rdp)

{

/* This cpu has pending rcu entries and the grace period

* for them has completed.

if (rdp->curlist && !rcu_batch_before(rcp->completed, rdp->batch))

return 1;

/* This cpu has no pending entries, but there are new entries */

if (!rdp->curlist && rdp->nxtlist)

return 1;

/* This cpu has finished callbacks to invoke */

if (rdp->donelist)

return 1;

/* The rcu core waits for a quiescent state from the cpu */

if (rdp->quiescbatch != rcp->cur || rdp->qs_pending)

return 1;

/* nothing to do */

return 0;

}

可以上面有四種情況會(huì)返回1，分別對(duì)應(yīng):

A.該CPU上有等待處理的回調(diào)函數(shù),且已經(jīng)經(jīng)過(guò)了一個(gè)batch(grace period).rdp->datch表示rdp在等待的batch序號(hào);

B.上一個(gè)等待已經(jīng)處理完了,又有了新注冊(cè)的回調(diào)函數(shù);

C.等待已經(jīng)完成,但尚末調(diào)用該次等待的回調(diào)函數(shù);

D.在等待quiescent state.

如果rcu_pending返回1,就會(huì)進(jìn)入到rcu_check_callbacks()，rcu_check_callbacks()中主要工作就是調(diào)用raise_softirq(RCU_SOFTIRQ)，觸發(fā)RCU軟中斷。而RCU軟中斷的處理函數(shù)為rcu_process_callbacks，其中分別針對(duì)每cpu變量rcu_bh_data和rcu_data調(diào)用__rcu_process_callbacks。我們主要分析針對(duì)rcu_data的調(diào)用。

l__rcu_process_callbacks

1) 先看nxtlist里有沒有待處理的回調(diào)(rcu_head)，如果有的話，說(shuō)明有寫者待處理，那么還要分兩種情況：

1.1)如果系統(tǒng)是第一次出現(xiàn)寫者阻塞，也即之前的寫者都已經(jīng)處理完畢，那么此時(shí)curlist鏈表一定為空（curlist專門存放已被rcu檢測(cè)到的寫者請(qǐng)求），于是就把nxtlist里的所有成員都移動(dòng)到curlist指向，并把當(dāng)前CPU需要等待的graceperiodid：rdp->batch設(shè)置為當(dāng)前系統(tǒng)處理的graceperiod的下一個(gè)grace周期，即rcp->cur+ 1。由于這算是一個(gè)新的graceperiod，即start_rcu_batch，所以還接著需要增加系統(tǒng)的graceperiod計(jì)數(shù)，即rcp->cur++，同時(shí)，將全局的cpusmask設(shè)置為全f，代表新的graceperiod開始，需要檢測(cè)所有的cpu是否都經(jīng)過(guò)了一次進(jìn)程切換。代碼如下：

C++

void __rcu_process_callbacks(struct rcu_ctrlblk *rcp, struct rcu_data *rdp) { if (rdp->nxtlist && !rdp->curlist) { move_local_cpu_nxtlist_to_curlist(); rdp->batch = rcp->cur + 1; if (!rcp->next_pending) { rcp->next_pending = 1; rcp->cur++; cpus_andnot(rcp->cpumask, cpu_online_map, nohz_cpu_mask); } } }

void __rcu_process_callbacks(struct rcu_ctrlblk *rcp,??

struct rcu_data *rdp)??

{??

if (rdp->nxtlist && !rdp->curlist) {??

move_local_cpu_nxtlist_to_curlist();??

rdp->batch = rcp->cur + 1;??

if (!rcp->next_pending) {??

rcp->next_pending = 1;??

rcp->cur++;??

cpus_andnot(rcp->cpumask, cpu_online_map, nohz_cpu_mask);??

}??

}

接著跳轉(zhuǎn)至1.2。

1.2) 如果系統(tǒng)之前已經(jīng)有寫者在被rcu監(jiān)控著，但還沒來(lái)得及經(jīng)過(guò)一個(gè)graceperiod，這個(gè)時(shí)候curlist不為空，nxtlist也不為空，寫者會(huì)被加入nxtlist中。由于curlist不為空，說(shuō)明上個(gè)rcu周期的寫者還沒有處理完，于是不會(huì)將本次阻塞的寫者加入curlist，一直到上次的curlist里的rcu_head被處理完（都移動(dòng)到了donelist），才會(huì)將后來(lái)的寫者納入RCU考慮（移動(dòng)到curlist）（假如這個(gè)期間又來(lái)了多個(gè)寫者，則多個(gè)寫者的rcu_head共享下一個(gè)graceperiod，也就是下一個(gè)graceperiod結(jié)束后這多個(gè)寫者都會(huì)被喚醒）。進(jìn)入下一步。

2) rcu_process_callbacks調(diào)用每CPU函數(shù)rcu_check_quiescent_state開始監(jiān)控，檢測(cè)所有的CPU是否會(huì)經(jīng)歷一個(gè)進(jìn)程切換。這個(gè)函數(shù)是如何得知需要開始監(jiān)控的？答案在于quiescbatch與全局的rcp->cur比較。初始化時(shí)rdp->quiescbatch =rcp->completed = rcp->cur。由于1.1有新graceperiod開啟，所以rcp->cur已經(jīng)加1了，于是rdp->quiescbatch和rcp->curr不等，進(jìn)而將此cpu的rdp->passed_quiesc設(shè)為0，表示這個(gè)周期開始，我要等待這個(gè)cpu經(jīng)歷一個(gè)進(jìn)程切換，等待該CPU將passed_quiesc置為1。即與前面講到的passed_quiesc標(biāo)志置0的時(shí)機(jī)吻合。最后將rdp->quiescbatch置為 rcp->cur，以避免下次再進(jìn)入軟中斷里將passed_quiesc重復(fù)置0。

C++

void rcu_check_quiescent_state(struct rcu_ctrlblk *rcp,??

struct rcu_data *rdp)??

{??

if (rdp->quiescbatch != rcp->cur) {??

/* start new grace period: */??

rdp->qs_pending = 1;??

rdp->passed_quiesc = 0;??

rdp->quiescbatch = rcp->cur;??

return;??

}??

}

3) 本次軟中斷結(jié)束，下次軟中斷到來(lái)，再次進(jìn)入rcu_check_quiescent_state進(jìn)行檢測(cè)，如果本CPU的rdp->passed_quiesc已經(jīng)置1，則需要cpu_quiet將本CPU標(biāo)志位從全局的rcp->cpumask中清除，如果cpumask為0了，則說(shuō)明自上次RCU寫者被掛起以來(lái)，所有CPU都已經(jīng)歷了一次進(jìn)程切換，于是本次rcu等待周期結(jié)束，將rcp->completed置為rcp->cur，重置cpumask為全f，并嘗試重新開啟一個(gè)新的grace period。我們可以看到RCU用了如此多的同步標(biāo)志，卻少用spinlock鎖，是多么巧妙的設(shè)計(jì)，不過(guò)這也提高了理解的難度。

C++

void rcu_check_quiescent_state(struct rcu_ctrlblk *rcp, struct rcu_data *rdp) { if (rdp->quiescbatch != rcp->cur) { /* start new grace period: */ rdp->qs_pending = 1; rdp->passed_quiesc = 0; rdp->quiescbatch = rcp->cur; return; } if (!rdp->passed_quiesc) return; /*this cpu has passed a quies state*/ if (likely(rdp->quiescbatch == rcp->cur)) { cpu_clear(cpu, rcp->cpumask); if (cpus_empty(rcp->cpumask)) rcp->completed = rcp->cur; } }

void rcu_check_quiescent_state(struct rcu_ctrlblk *rcp,??

struct rcu_data *rdp)??

{??

if (rdp->quiescbatch != rcp->cur) {??

/* start new grace period: */??

rdp->qs_pending = 1;??

rdp->passed_quiesc = 0;??

rdp->quiescbatch = rcp->cur;??

return;??

}??

if (!rdp->passed_quiesc)??

return;??

/*this cpu has passed a quies state*/??

if (likely(rdp->quiescbatch == rcp->cur)) {??

cpu_clear(cpu, rcp->cpumask);??

if (cpus_empty(rcp->cpumask))??

rcp->completed = rcp->cur;??

}??

}

4)下次再進(jìn)入rcu軟中斷__rcu_process_callbacks，發(fā)現(xiàn)rdp->batch已經(jīng)比rcp->completed小了（因?yàn)樯弦徊襟E中，后者增大了），則將rdp->curlist上的回調(diào)移動(dòng)到rdp->donelist里，接著還會(huì)再次進(jìn)入rcu_check_quiescent_state，但是由于當(dāng)前CPU的rdp->qs_pending已經(jīng)為1了，所以不再往下清除cpu掩碼。__rcu_process_callbacks

代碼變成了：

C++

void __rcu_process_callbacks(struct rcu_ctrlblk *rcp, struct rcu_data *rdp) { if (rdp->curlist && !rcu_batch_before(rcp->completed, rdp->batch)) { *rdp->donetail = rdp->curlist; rdp->donetail = rdp->curtail; rdp->curlist = NULL; rdp->curtail = &rdp->curlist; } if (rdp->nxtlist && !rdp->curlist) { move_local_cpu_nxtlist_to_curlist(); rdp->batch = rcp->cur + 1; if (!rcp->next_pending) { rcp->next_pending = 1; rcp->cur++; cpus_andnot(rcp->cpumask, cpu_online_map, nohz_cpu_mask); } } if (rdp->donelist) rcu_do_batch(rdp); }

void __rcu_process_callbacks(struct rcu_ctrlblk *rcp,??

struct rcu_data *rdp)??

{??

if (rdp->curlist && !rcu_batch_before(rcp->completed, rdp->batch)) {??

*rdp->donetail = rdp->curlist;??

rdp->donetail = rdp->curtail;??

rdp->curlist = NULL;??

rdp->curtail = &rdp->curlist;??

}??

if (rdp->nxtlist && !rdp->curlist) {??

move_local_cpu_nxtlist_to_curlist();??

rdp->batch = rcp->cur + 1;??

if (!rcp->next_pending) {??

rcp->next_pending = 1;??

rcp->cur++;??

cpus_andnot(rcp->cpumask, cpu_online_map, nohz_cpu_mask);??

}??

if (rdp->donelist)??

rcu_do_batch(rdp);??

}

5)經(jīng)過(guò)千山萬(wàn)水終于來(lái)到rcu_do_batch（如果rdp->donelist有的話）在此函數(shù)里，執(zhí)行RCU寫者掛載的回調(diào)，即wakeme_after_rcu。

lrcu_do_batch

C++

static void rcu_do_batch(struct rcu_data *rdp) { struct rcu_head *next, *list; int count = 0; list = rdp->donelist; while (list) { next = list->next; prefetch(next); list->func(list); list = next; if (++count >= rdp->blimit) break; } rdp->donelist = list; local_irq_disable(); rdp->qlen -= count; local_irq_enable(); if (rdp->blimit == INT_MAX && rdp->qlen) blimit = blimit; if (!rdp->donelist) rdp->donetail = &rdp->donelist; else raise_rcu_softirq(); }

static void rcu_do_batch(struct rcu_data *rdp)

{

struct rcu_head *next, *list;

int count = 0;

list = rdp->donelist;

while (list) {

next = list->next;

prefetch(next);

list->func(list);

list = next;

if (++count >= rdp->blimit)

break;

}

rdp->donelist = list;

local_irq_disable();

rdp->qlen -= count;

local_irq_enable();

if (rdp->blimit == INT_MAX && rdp->qlen)

blimit = blimit;

if (!rdp->donelist)

rdp->donetail = &rdp->donelist;

else

raise_rcu_softirq();

}

它遍歷處理掛在鏈表上的回調(diào)函數(shù).在這里,注意每次調(diào)用的回調(diào)函數(shù)有最大值限制.這樣做主要是防止一次調(diào)用過(guò)多的回調(diào)函數(shù)而產(chǎn)生不必要系統(tǒng)負(fù)載.如果donelist中還有沒處理完的數(shù)據(jù),打開RCU軟中斷,在下次軟中斷到來(lái)的時(shí)候接著處理.

注意：

僅當(dāng)系統(tǒng)檢測(cè)到一個(gè)grace period的所有CPU都經(jīng)歷了進(jìn)程切換后，才會(huì)給系統(tǒng)一個(gè)信息要求啟動(dòng)新batch，在此期間的所有寫者請(qǐng)求，都暫存在本地CPU的nxtlist鏈表里。

進(jìn)程切換

在每一次進(jìn)程切換的時(shí)候,都會(huì)調(diào)用rcu_qsctr_inc().如下代碼片段如示:

C++

asmlinkage void __sched schedule(void) { //......rcu_qsctr_inc(cpu); //...... }

asmlinkage void __sched schedule(void)

{

//......

rcu_qsctr_inc(cpu);

//......

}

rcu_qsctr_inc()代碼如下:

C++

static inline void rcu_qsctr_inc(int cpu) { struct rcu_data *rdp = &per_cpu(rcu_data, cpu); rdp->passed_quiesc = 1; }

static inline void rcu_qsctr_inc(int cpu)

{

struct rcu_data *rdp = &per_cpu(rcu_data, cpu);

rdp->passed_quiesc = 1;

}

該函數(shù)將對(duì)應(yīng)CPU上的rcu_data的passed_quiesc成員設(shè)為了1?；蛟S你已經(jīng)發(fā)現(xiàn)了,這個(gè)過(guò)程就標(biāo)識(shí)該CPU經(jīng)過(guò)了一次quiescent state，和之前在軟中斷的初始化為0相呼應(yīng)。

幾種RCU情況分析

1.如果CPU 1上有進(jìn)程調(diào)用rcu_read_lock進(jìn)入臨界區(qū),之后退出來(lái),發(fā)生了進(jìn)程切換,新進(jìn)程又通過(guò)rcu_read_lock進(jìn)入臨界區(qū).由于RCU軟中斷中只判斷一次上下文切換,因此,在調(diào)用回調(diào)函數(shù)的時(shí)候,仍然有進(jìn)程處于RCU的讀臨界區(qū),這樣會(huì)不會(huì)有問題呢?
這樣是不會(huì)有問題的.還是上面的例子:

C++

spin_lock(&foo_mutex); old_fp = gbl_foo; *new_fp = *old_fp; new_fp->a = new_a; rcu_assign_pointer(gbl_foo, new_fp); spin_unlock(&foo_mutex); synchronize_rcu(); kfree(old_fp);

spin_lock(&foo_mutex);

old_fp = gbl_foo;

*new_fp = *old_fp;

new_fp->a = new_a;

rcu_assign_pointer(gbl_foo, new_fp);

spin_unlock(&foo_mutex);

synchronize_rcu();

kfree(old_fp);

使用synchronize_rcu ()只是為了等待持有old_fd(也就是調(diào)用rcu_assign_pointer ()更新之前的gbl_foo)的進(jìn)程退出.而不需要等待所有的讀者全部退出.這是因?yàn)?在rcu_assign_pointer ()之后的讀取取得的保護(hù)指針,已經(jīng)是更新好的新值了.

2. 如果一個(gè)CPU連續(xù)調(diào)用synchronize_rcu()或者call_rcu()它們會(huì)有什么影響呢?
如果當(dāng)前有請(qǐng)求在等待,就會(huì)新請(qǐng)?zhí)峤坏幕卣{(diào)函數(shù)掛到taillist上,一直到前一個(gè)等待完成,再將taillist的數(shù)據(jù)移到curlist,并開啟一個(gè)新的等待,因此,也就是說(shuō),在前一個(gè)等待期間提交的請(qǐng)求,都會(huì)放到一起處理.也就是說(shuō),他們會(huì)共同等待所有CPU切換完成.
舉例說(shuō)明如下:

帶bh的RCU API

在上面的代碼分析的時(shí)候,經(jīng)?？吹綆в衎h的RCU代碼.現(xiàn)在來(lái)看一下這些帶bh的RCU是什么樣的。

C++

#define rcu_read_lock_bh() __rcu_read_lock_bh() #define rcu_read_unlock_bh() __rcu_read_unlock_bh() #define __rcu_read_lock_bh() do { local_bh_disable(); __acquire(RCU_BH); rcu_read_acquire(); }while (0) #define __rcu_read_unlock_bh() do { rcu_read_release(); __release(RCU_BH); local_bh_enable(); } while (0)

#define rcu_read_lock_bh() __rcu_read_lock_bh()

#define rcu_read_unlock_bh() __rcu_read_unlock_bh()

#define __rcu_read_lock_bh()

do {

local_bh_disable();

__acquire(RCU_BH);

rcu_read_acquire();

}while (0)

#define __rcu_read_unlock_bh()

do {

rcu_read_release();

__release(RCU_BH);

local_bh_enable();

} while (0)

根據(jù)上面的分析：bh RCU跟普通的RCU相比不同的是,普通RCU是禁止內(nèi)核搶占,而bh RCU是禁止下半部.
其實(shí),帶bh的RCU一般在軟中斷使用,不過(guò)計(jì)算quiescent state并不是發(fā)生一次上下文切換。而是發(fā)生一次softirq.我們?cè)诤竺娴姆治鲋锌傻玫接∽C.

lcall_rcu_bh()

C++

void call_rcu_bh(struct rcu_head *head, void (*func)(struct rcu_head *rcu)) { unsigned long flags; struct rcu_data *rdp; head->func = func; head->next = NULL; local_irq_save(flags); rdp = &__get_cpu_var(rcu_bh_data); *rdp->nxttail = head; rdp->nxttail = &head->next; if (unlikely(++rdp->qlen > qhimark)) { rdp->blimit = INT_MAX; force_quiescent_state(rdp, &rcu_bh_ctrlblk); } local_irq_restore(flags); }

void call_rcu_bh(struct rcu_head *head, void (*func)(struct rcu_head *rcu))

{

unsigned long flags;

struct rcu_data *rdp;

head->func = func;

head->next = NULL;

local_irq_save(flags);

rdp = &__get_cpu_var(rcu_bh_data);

*rdp->nxttail = head;

rdp->nxttail = &head->next;

if (unlikely(++rdp->qlen > qhimark)) {

rdp->blimit = INT_MAX;

force_quiescent_state(rdp, &rcu_bh_ctrlblk);

}

local_irq_restore(flags);

}

它跟call_rcu()不相同的是,rcu是取per_cpu變量rcu__data和全局變量rcu_ctrlblk.而bh RCU是取rcu_bh_data,rcu_bh_ctrlblk.他們的類型都是一樣的,這樣做只是為了區(qū)分BH和普通RCU的等待.

對(duì)于rcu_bh_qsctr_inc

C++

static inline void rcu_bh_qsctr_inc(int cpu) { struct rcu_data *rdp = &per_cpu(rcu_bh_data, cpu); rdp->passed_quiesc = 1; }

static inline void rcu_bh_qsctr_inc(int cpu)

{

struct rcu_data *rdp = &per_cpu(rcu_bh_data, cpu);

rdp->passed_quiesc = 1;

}

它跟rcu_qsctr_inc()機(jī)同,也是更改對(duì)應(yīng)成員.所不同的是,調(diào)用rcu_bh_qsctr_inc()的地方發(fā)生了變化.

C++

asmlinkage void __do_softirq(void) { //...... do { if (pending & 1) { h->action(h); rcu_bh_qsctr_inc(cpu); } h++; pending >>= 1; } while (pending); //...... }

asmlinkage void __do_softirq(void)

{

//......

do {

if (pending & 1) {

h->action(h);

rcu_bh_qsctr_inc(cpu);

}

h++;

pending >>= 1;

} while (pending);

//......

}

也就是說(shuō)，在發(fā)生軟中斷的時(shí)候，才會(huì)認(rèn)為是經(jīng)過(guò)了一次quiescent state.

RCU鏈表操作

為了操作鏈表，在include/linux/rculist.h有一套專門的RCU API。如：list_entry_rcu、list_add_rcu、list_del_rcu、list_for_each_entry_rcu等。即對(duì)所有kernel 的list的操作都有一個(gè)對(duì)應(yīng)的RCU操作。那么這些操作和原始的list操作有哪些不同呢？我們先對(duì)比幾個(gè)看下。

llist_entry_rcu

C++

#define list_entry_rcu(ptr, type, member) \container_of(rcu_dereference(ptr), type, member)#define list_entry(ptr, type, member) \container_of(ptr, type, member)

#define list_entry_rcu(ptr, type, member) \

container_of(rcu_dereference(ptr), type, member)

#define list_entry(ptr, type, member) \

container_of(ptr, type, member)

l__list_for_each_rcu

C++

#define __list_for_each_rcu(pos, head) \for (pos = rcu_dereference((head)->next); \pos != (head); \pos = rcu_dereference(pos->next))#define __list_for_each(pos, head) \for (pos = (head)->next; pos != (head); pos = pos->next)

#define __list_for_each_rcu(pos, head) \

for (pos = rcu_dereference((head)->next); \

pos != (head); \

pos = rcu_dereference(pos->next))

#define __list_for_each(pos, head) \

for (pos = (head)->next; pos != (head); pos = pos->next)

從__list_for_each_rcu和list_entry_rcu的實(shí)現(xiàn)可以看出，其將指針的獲取替換為使用rcu_dereference。

llist_replace_rcu

C++

static inline void list_replace_rcu(struct list_head *old,struct list_head *new){new->next = old->next;new->prev = old->prev;rcu_assign_pointer(new->prev->next, new);new->next->prev = new;old->prev = LIST_POISON2;}static inline void list_replace(struct list_head *old,struct list_head *new){new->next = old->next;new->next->prev = new;new->prev = old->prev;new->prev->next = new;}

static inline void list_replace_rcu(struct list_head *old,

struct list_head *new)

{

new->next = old->next;

new->prev = old->prev;

rcu_assign_pointer(new->prev->next, new);

new->next->prev = new;

old->prev = LIST_POISON2;

}

static inline void list_replace(struct list_head *old,

struct list_head *new)

{

new->next = old->next;

new->next->prev = new;

new->prev = old->prev;

new->prev->next = new;

}

從list_replace_rcu的實(shí)現(xiàn)可以看出，RCU API的實(shí)現(xiàn)將指針的賦值替換為rcu_assign_pointer。

llist_del_rcu

C++

static inline void list_del_rcu(struct list_head *entry){__list_del(entry->prev, entry->next);entry->prev = LIST_POISON2;}static inline void list_del(struct list_head *entry){__list_del(entry->prev, entry->next);entry->next = LIST_POISON1;entry->prev = LIST_POISON2;}

static inline void list_del_rcu(struct list_head *entry)

{

__list_del(entry->prev, entry->next);

entry->prev = LIST_POISON2;

}

static inline void list_del(struct list_head *entry)

{

__list_del(entry->prev, entry->next);

entry->next = LIST_POISON1;

entry->prev = LIST_POISON2;

}

從list_del_rcu的實(shí)現(xiàn)，可以看出RCU API的實(shí)現(xiàn)沒有將刪除項(xiàng)的next指針置為無(wú)效。這樣實(shí)現(xiàn)是為了防止刪除節(jié)點(diǎn)時(shí)，讀者還在遍歷該節(jié)點(diǎn)。

RCU鏈表API使用

下面看下RCU list API的幾個(gè)應(yīng)用示例。

只有增加和刪除的鏈表操作

在這種應(yīng)用情況下，絕大部分是對(duì)鏈表的遍歷，即讀操作，而很少出現(xiàn)的寫操作只有增加或刪除鏈表項(xiàng)，并沒有對(duì)鏈表項(xiàng)的修改操作，這種情況使用RCU非常容易，從rwlock轉(zhuǎn)換成RCU非常自然。路由表的維護(hù)就是這種情況的典型應(yīng)用，對(duì)路由表的操作，絕大部分是路由表查詢，而對(duì)路由表的寫操作也僅僅是增加或刪除，因此使用RCU替換原來(lái)的rwlock順理成章。系統(tǒng)調(diào)用審計(jì)也是這樣的情況。

這是一段使用rwlock的系統(tǒng)調(diào)用審計(jì)部分的讀端代碼：

C++

static enum audit_state audit_filter_task(struct task_struct *tsk) { struct audit_entry *e; enum audit_state state; read_lock(&auditsc_lock); /* Note: audit_netlink_sem held by caller. */ list_for_each_entry(e, &audit_tsklist, list) { if (audit_filter_rules(tsk, &e->rule, NULL, &state)) { read_unlock(&auditsc_lock); return state; } } read_unlock(&auditsc_lock); return AUDIT_BUILD_CONTEXT; }

static enum audit_state audit_filter_task(struct task_struct *tsk)

{

struct audit_entry *e;

enum audit_state?? state;

read_lock(&auditsc_lock);

/* Note: audit_netlink_sem held by caller. */

list_for_each_entry(e, &audit_tsklist, list) {

if (audit_filter_rules(tsk, &e->rule, NULL, &state)) {

read_unlock(&auditsc_lock);

return state;

}

read_unlock(&auditsc_lock);

return AUDIT_BUILD_CONTEXT;

}

使用RCU后將變成：

C++

static enum audit_state audit_filter_task(struct task_struct *tsk) { struct audit_entry *e; enum audit_state state; rcu_read_lock(); /* Note: audit_netlink_sem held by caller. */ list_for_each_entry_rcu(e, &audit_tsklist, list) { if (audit_filter_rules(tsk, &e->rule, NULL, &state)) { rcu_read_unlock(); return state; } } rcu_read_unlock(); return AUDIT_BUILD_CONTEXT; }

static enum audit_state audit_filter_task(struct task_struct *tsk)

{

struct audit_entry *e;

enum audit_state?? state;

rcu_read_lock();

/* Note: audit_netlink_sem held by caller. */

list_for_each_entry_rcu(e, &audit_tsklist, list) {

if (audit_filter_rules(tsk, &e->rule, NULL, &state)) {

rcu_read_unlock();

return state;

}

rcu_read_unlock();

return AUDIT_BUILD_CONTEXT;

}

這種轉(zhuǎn)換非常直接，使用rcu_read_lock和rcu_read_unlock分別替換read_lock和read_unlock，鏈表遍歷函數(shù)使用_rcu版本替換就可以了。

使用rwlock的寫端代碼：

C++

static inline int audit_del_rule(struct audit_rule *rule, struct list_head *list) { struct audit_entry *e; write_lock(&auditsc_lock); list_for_each_entry(e, list, list) { if (!audit_compare_rule(rule, &e->rule)) { list_del(&e->list); write_unlock(&auditsc_lock); return 0; } } write_unlock(&auditsc_lock); return -EFAULT; /* No matching rule */ } static inline int audit_add_rule(struct audit_entry *entry, struct list_head *list) { write_lock(&auditsc_lock); if (entry->rule.flags & AUDIT_PREPEND) { entry->rule.flags &= ~AUDIT_PREPEND; list_add(&entry->list, list); } else { list_add_tail(&entry->list, list); } write_unlock(&auditsc_lock); return 0; }

static inline int audit_del_rule(struct audit_rule *rule,

struct list_head *list)

{

struct audit_entry??*e;

write_lock(&auditsc_lock);

list_for_each_entry(e, list, list) {

if (!audit_compare_rule(rule, &e->rule)) {

list_del(&e->list);

write_unlock(&auditsc_lock);

return 0;

}

write_unlock(&auditsc_lock);

return -EFAULT;???????? /* No matching rule */

}

static inline int audit_add_rule(struct audit_entry *entry,

struct list_head *list)

{

write_lock(&auditsc_lock);

if (entry->rule.flags & AUDIT_PREPEND) {

entry->rule.flags &= ~AUDIT_PREPEND;

list_add(&entry->list, list);

} else {

list_add_tail(&entry->list, list);

}

write_unlock(&auditsc_lock);

return 0;

}

使用RCU后寫端代碼變成為：

C++

static inline int audit_del_rule(struct audit_rule *rule, struct list_head *list) { struct audit_entry *e; /* Do not use the _rcu iterator here, since this is the only * deletion routine. */ list_for_each_entry(e, list, list) { if (!audit_compare_rule(rule, &e->rule)) { list_del_rcu(&e->list); call_rcu(&e->rcu, audit_free_rule, e); return 0; } } return -EFAULT; /* No matching rule */ } static inline int audit_add_rule(struct audit_entry *entry, struct list_head *list) { if (entry->rule.flags & AUDIT_PREPEND) { entry->rule.flags &= ~AUDIT_PREPEND; list_add_rcu(&entry->list, list); } else { list_add_tail_rcu(&entry->list, list); } return 0; }

static inline int audit_del_rule(struct audit_rule *rule,

struct list_head *list)

{

struct audit_entry??*e;

/* Do not use the _rcu iterator here, since this is the only

* deletion routine. */

list_for_each_entry(e, list, list) {

if (!audit_compare_rule(rule, &e->rule)) {

list_del_rcu(&e->list);

call_rcu(&e->rcu, audit_free_rule, e);

return 0;

}

return -EFAULT;???????? /* No matching rule */

}

static inline int audit_add_rule(struct audit_entry *entry,

struct list_head *list)

{

if (entry->rule.flags & AUDIT_PREPEND) {

entry->rule.flags &= ~AUDIT_PREPEND;

list_add_rcu(&entry->list, list);

} else {

list_add_tail_rcu(&entry->list, list);

}

return 0;

}

對(duì)于鏈表刪除操作，list_del替換為list_del_rcu和call_rcu，這是因?yàn)楸粍h除的鏈表項(xiàng)可能還在被別的讀者引用，所以不能立即刪除，必須等到所有讀者經(jīng)歷一個(gè)quiescent state才可以刪除。另外，list_for_each_entry并沒有被替換為list_for_each_entry_rcu，這是因?yàn)?，只有一個(gè)寫者在做鏈表刪除操作，因此沒有必要使用_rcu版本。

通常情況下，write_lock和write_unlock應(yīng)當(dāng)分別替換成spin_lock和spin_unlock，但是對(duì)于只是對(duì)鏈表進(jìn)行增加和刪除操作而且只有一個(gè)寫者的寫端，在使用了_rcu版本的鏈表操作API后，rwlock可以完全消除，不需要spinlock來(lái)同步讀者的訪問。對(duì)于上面的例子，由于已經(jīng)有audit_netlink_sem被調(diào)用者保持，所以spinlock就沒有必要了。

這種情況允許修改結(jié)果延后一定時(shí)間才可見，而且寫者對(duì)鏈表僅僅做增加和刪除操作，所以轉(zhuǎn)換成使用RCU非常容易。

寫端需要對(duì)鏈表?xiàng)l目進(jìn)行修改操作

如果寫者需要對(duì)鏈表?xiàng)l目進(jìn)行修改，那么就需要首先拷貝要修改的條目，然后修改條目的拷貝，等修改完畢后，再使用條目拷貝取代要修改的條目，要修改條目將被在經(jīng)歷一個(gè)grace period后安全刪除。

對(duì)于系統(tǒng)調(diào)用審計(jì)代碼，并沒有這種情況。這里假設(shè)有修改的情況，那么使用rwlock的修改代碼應(yīng)當(dāng)如下：

C++

static inline int audit_upd_rule(struct audit_rule *rule, struct list_head *list, __u32 newaction, __u32 newfield_count) { struct audit_entry *e; struct audit_newentry *ne; write_lock(&auditsc_lock); /* Note: audit_netlink_sem held by caller. */ list_for_each_entry(e, list, list) { if (!audit_compare_rule(rule, &e->rule)) { e->rule.action = newaction; e->rule.file_count = newfield_count; write_unlock(&auditsc_lock); return 0; } } write_unlock(&auditsc_lock); return -EFAULT; /* No matching rule */ }

static inline int audit_upd_rule(struct audit_rule *rule,

struct list_head *list,

__u32 newaction,

__u32 newfield_count)

{

struct audit_entry??*e;

struct audit_newentry *ne;

write_lock(&auditsc_lock);

/* Note: audit_netlink_sem held by caller. */

list_for_each_entry(e, list, list) {

if (!audit_compare_rule(rule, &e->rule)) {

e->rule.action = newaction;

e->rule.file_count = newfield_count;

write_unlock(&auditsc_lock);

return 0;

}

write_unlock(&auditsc_lock);

return -EFAULT;???????? /* No matching rule */

}

如果使用RCU，修改代碼應(yīng)當(dāng)為；

C++

static inline int audit_upd_rule(struct audit_rule *rule, struct list_head *list, __u32 newaction, __u32 newfield_count) { struct audit_entry *e; struct audit_newentry *ne; list_for_each_entry(e, list, list) { if (!audit_compare_rule(rule, &e->rule)) { ne = kmalloc(sizeof(*entry), GFP_ATOMIC); if (ne == NULL) return -ENOMEM; audit_copy_rule(&ne->rule, &e->rule); ne->rule.action = newaction; ne->rule.file_count = newfield_count; list_replace_rcu(e, ne); call_rcu(&e->rcu, audit_free_rule, e); return 0; } } return -EFAULT; /* No matching rule */ }

static inline int audit_upd_rule(struct audit_rule *rule,

struct list_head *list,

__u32 newaction,

__u32 newfield_count)

{

struct audit_entry??*e;

struct audit_newentry *ne;

list_for_each_entry(e, list, list) {

if (!audit_compare_rule(rule, &e->rule)) {

ne = kmalloc(sizeof(*entry), GFP_ATOMIC);

if (ne == NULL)

return -ENOMEM;

audit_copy_rule(&ne->rule, &e->rule);

ne->rule.action = newaction;

ne->rule.file_count = newfield_count;

list_replace_rcu(e, ne);

call_rcu(&e->rcu, audit_free_rule, e);

return 0;

}

return -EFAULT;???????? /* No matching rule */

}

修改操作立即可見

前面兩種情況，讀者能夠容忍修改可以在一段時(shí)間后看到，也就說(shuō)讀者在修改后某一時(shí)間段內(nèi)，仍然看到的是原來(lái)的數(shù)據(jù)。在很多情況下，讀者不能容忍看到舊的數(shù)據(jù)，這種情況下，需要使用一些新措施，如System V IPC，它在每一個(gè)鏈表?xiàng)l目中增加了一個(gè)deleted字段，標(biāo)記該字段是否刪除，如果刪除了，就設(shè)置為真，否則設(shè)置為假，當(dāng)代碼在遍歷鏈表時(shí)，核對(duì)每一個(gè)條目的deleted字段，如果為真，就認(rèn)為它是不存在的。

還是以系統(tǒng)調(diào)用審計(jì)代碼為例，如果它不能容忍舊數(shù)據(jù)，那么，讀端代碼應(yīng)該修改為：

C++

static enum audit_state audit_filter_task(struct task_struct *tsk) { struct audit_entry *e; enum audit_state state; rcu_read_lock(); list_for_each_entry_rcu(e, &audit_tsklist, list) { if (audit_filter_rules(tsk, &e->rule, NULL, &state)) { spin_lock(&e->lock); if (e->deleted) { spin_unlock(&e->lock); rcu_read_unlock(); return AUDIT_BUILD_CONTEXT; } rcu_read_unlock(); return state; } } rcu_read_unlock(); return AUDIT_BUILD_CONTEXT; }

static enum audit_state audit_filter_task(struct task_struct *tsk)

{

struct audit_entry *e;

enum audit_state?? state;

rcu_read_lock();

list_for_each_entry_rcu(e, &audit_tsklist, list) {

if (audit_filter_rules(tsk, &e->rule, NULL, &state)) {

spin_lock(&e->lock);

if (e->deleted) {

spin_unlock(&e->lock);

rcu_read_unlock();

return AUDIT_BUILD_CONTEXT;

}

rcu_read_unlock();

return state;

}

rcu_read_unlock();

return AUDIT_BUILD_CONTEXT;

}

注意，對(duì)于這種情況，每一個(gè)鏈表?xiàng)l目都需要一個(gè)spinlock保護(hù)，因?yàn)閯h除操作將修改條目的deleted標(biāo)志。此外，該函數(shù)如果搜索到條目，返回時(shí)應(yīng)當(dāng)保持該條目的鎖。

寫端的刪除操作將變成：

C++

static inline int audit_del_rule(struct audit_rule *rule,

struct list_head *list)

{

struct audit_entry??*e;

/* Do not use the _rcu iterator here, since this is the only

* deletion routine. */

list_for_each_entry(e, list, list) {

if (!audit_compare_rule(rule, &e->rule)) {

spin_lock(&e->lock);

list_del_rcu(&e->list);

e->deleted = 1;

spin_unlock(&e->lock);

call_rcu(&e->rcu, audit_free_rule, e);

return 0;

}

return -EFAULT;???????? /* No matching rule */

}

刪除條目時(shí)，需要標(biāo)記該條目為已刪除。這樣讀者就可以通過(guò)該標(biāo)志立即得知條目是否已經(jīng)刪除。

小結(jié)

RCU是2.6內(nèi)核引入的新的鎖機(jī)制，在絕大部分為讀而只有極少部分為寫的情況下，它是非常高效的，因此在路由表維護(hù)、系統(tǒng)調(diào)用審計(jì)、SELinux的AVC、dcache和IPC等代碼部分中，使用它來(lái)取代rwlock來(lái)獲得更高的性能。但是，它也有缺點(diǎn)，延后的刪除或釋放將占用一些內(nèi)存，尤其是對(duì)嵌入式系統(tǒng)，這可能是非常昂貴的內(nèi)存開銷。此外，寫者的開銷比較大，尤其是對(duì)于那些無(wú)法容忍舊數(shù)據(jù)的情況以及不只一個(gè)寫者的情況，寫者需要spinlock或其他的鎖機(jī)制來(lái)與其他寫者同步。

閱讀全文