Linux RCU機(jī)制詳解.doc

一:前言RCU機(jī)制出現(xiàn)的比較早,只是在linux kernel中一直到2.5版本的時(shí)候才被采用.關(guān)于RCU機(jī)制,這里就不做過多的介紹了,網(wǎng)上有很多有關(guān)RCU介紹和使用的文檔.請自行查閱.本文主要是從linux kernel源代碼的角度.來分析RCU的實(shí)現(xiàn).在討論RCU的實(shí)現(xiàn)之前.有必要重申以下幾點(diǎn):1:RCU使用在讀者多而寫者少的情況.RCU和讀寫鎖相似.但RCU的讀者占鎖沒有任何的系統(tǒng)開銷.寫者與寫寫者之間必須要保持同步,且寫者必須要等它之前的讀者全部都退出之后才能釋放之前的資源.2:RCU保護(hù)的是指針.這一點(diǎn)尤其重要.因?yàn)橹羔樫x值是一條單指令.也就是說是一個(gè)原子操作.因它更改指針指向沒必要考慮它的同步.只需要考慮cache的影響.3:讀者是可以嵌套的.也就是說rcu_read_lock()可以嵌套調(diào)用.4:讀者在持有rcu_read_lock()的時(shí)候,不能發(fā)生進(jìn)程上下文切換.否則,因?yàn)閷懻咝枰却x者完成,寫者進(jìn)程也會(huì)一直被阻塞.以下的代碼是基于linux kernel 2.6.26二:使用RCU的實(shí)例Linux kernel中自己附帶有詳細(xì)的文檔來介紹RCU,這些文檔位于linux-/Documentation/RCU. 這些文檔值得多花點(diǎn)時(shí)間去仔細(xì)研讀一下.下面以whatisRCU.txt中的例子作為今天分析的起點(diǎn):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被用來保護(hù)全局指針struct foo *gbl_foo. foo_get_a()用來從RCU保護(hù)的結(jié)構(gòu)中取得gbl_foo的值.而foo_update_a()用來更新被RCU保護(hù)的gbl_foo的值.另外,我們思考一下,為什么要在foo_update_a()中使用自旋鎖foo_mutex呢?假設(shè)中間沒有使用自旋鎖.那foo_update_a()的代碼如下: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)程對old_fd的操作都是非法的.另外,我們在上面也看到了幾個(gè)有關(guān)RCU的核心API.它們?yōu)閯e是:rcu_read_lock()rcu_read_unlock()synchronize_rcu()rcu_assign_pointer()rcu_dereference()其中,rcu_read_lock()和rcu_read_unlock()用來保持一個(gè)讀者的RCU臨界區(qū).在該臨界區(qū)內(nèi)不允許發(fā)生上下文切換.rcu_dereference():讀者調(diào)用它來獲得一個(gè)被RCU保護(hù)的指針.Rcu_assign_pointer():寫者使用該函數(shù)來為被RCU保護(hù)的指針分配一個(gè)新的值.這樣是為了安全從寫者到讀者更改其值.這個(gè)函數(shù)會(huì)返回一個(gè)新值三:RCU API實(shí)現(xiàn)分析Rcu_read_lock()和rcu_read_unlock()的實(shí)現(xiàn)如下:#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_read_acquire(),rcu_read_release(),rcu_read_release()都是一些選擇編譯函數(shù),可以忽略不可看.因此可以得知.rcu_read_lock(),rcu_read_unlock()只是禁止和啟用搶占.因?yàn)樵谧x者臨界區(qū),不允許發(fā)生上下文切換.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)也很簡單.因?yàn)樗鼈儽旧矶际窃硬僮?因?yàn)橹皇菫榱薱ache一致性,插上了內(nèi)存屏障.可以讓其它的讀者/寫者可以看到保護(hù)指針的最新值.synchronize_rcu()在RCU中是一個(gè)最核心的函數(shù),它用來等待之前的讀者全部退出.我們后面的大部份分析也是圍繞著它而進(jìn)行.實(shí)現(xiàn)如下:void synchronize_rcu(void) struct rcu_synchronize rcu; init_completion(&pletion); /* Will wake me after RCU finished */ call_rcu(&rcu.head, wakeme_after_rcu); /* Wait for it */ wait_for_completion(&pletion);我們可以看到,它初始化了一個(gè)本地變量,它的類型為struct rcu_synchronize.調(diào)用call_rcu()之后.一直等待條件變量petion的滿足.在這里看到了RCU的另一個(gè)核心API,它就是call_run().它的定義如下:void call_rcu(struct rcu_head *head, void (*func)(struct rcu_head *rcu)它用來等待之前的讀者操作完成之后,就會(huì)調(diào)用函數(shù)func.我們也可以看到,在synchronize_rcu()中,讀者操作完了要調(diào)用的函數(shù)就是wakeme_after_rcu().另外,call_rcu()用在不可睡眠的條件中,如果中斷環(huán)境,禁止搶占環(huán)境等.而synchronize_rcu()用在可睡眠的環(huán)境下.先跟蹤看一下wakeme_after_rcu():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)程.看下call_rcu():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ù)也很簡單,就是將head加在了per_cpu變量rcu_data的tail鏈表上.Rcu_data定義如下:DEFINE_PER_CPU(struct rcu_data, rcu_data) = 0L ;由此,我們可以得知,每一個(gè)CPU都有一個(gè)rcu_data.每個(gè)調(diào)用call_rcu()/synchronize_rcu()進(jìn)程所代表的head都會(huì)掛到rcu_data的tail鏈表上.那究竟怎么去判斷當(dāng)前的寫者已經(jīng)操作完了呢?我們在之前看到,不是讀者在調(diào)用rcu_read_lock()的時(shí)候要禁止搶占么?因此,我們只需要判斷如有的CPU都進(jìn)過了一次上下文切換,就說明所有讀者已經(jīng)退出了.引用( (/developerworks/cn/linux/l-rcu/)中有關(guān)這個(gè)過程的描述:“等待適當(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)用寫者注冊的回調(diào)函數(shù)來完成真正的數(shù)據(jù)修改或數(shù)據(jù)釋放操作的”要徹底弄清楚這個(gè)問題,我們得從RCU的初始化說起.四:從RCU的初始化說起RCU的初始化位于start_kernel()rcu_init().代碼如下:void _init rcu_init(void) _rcu_init();void _init _rcu_init(void) rcu_cpu_notify(&rcu_nb, CPU_UP_PREPARE, (void *)(long)smp_processor_id(); /* Register notifier for non-boot CPUs */ register_cpu_notifier(&rcu_nb);Reqister_cpu_notifier()是關(guān)于通知鏈表的操作,可以忽略不看.跟進(jìn)rcu_cpu_notify():static int _cpuinit rcu_cpu_notify(struct notifier_block *self, unsigned long action, void *hcpu) long cpu = (long)hcpu; switch (action) case CPU_UP_PREPARE: case CPU_UP_PREPARE_FROZEN: rcu_online_cpu(cpu); break; case CPU_DEAD: case CPU_DEAD_FROZEN: rcu_offline_cpu(cpu); break; default: break; return NOTIFY_OK;注意到,在_rcu_init()中是以CPU_UP_PREPARE為參數(shù)調(diào)用此函數(shù),對應(yīng)流程轉(zhuǎn)入rcu_online_cpu中: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è)per_cpu變量,rcu_bh_data.有關(guān)bh的部份之后再來分析.在這里略過這些部份.Rcu_init_percpu_data()如下: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.定義如下:static struct rcu_ctrlblk rcu_ctrlblk = .cur = -300, .completed = -300, .lock = _SPIN_LOCK_UNLOCKED(&rcu_ctrlblk.lock), .cpumask = CPU_MASK_NONE,;static struct rcu_ctrlblk rcu_bh_ctrlblk = .cur = -300, .completed = -300, .lock = _SPIN_LOCK_UNLOCKED(&rcu_bh_ctrlblk.lock), .cpumask = CPU_MASK_NONE,;在rcu_init_percpu_data中,初始化了三個(gè)鏈表,分別是taillist,curlist和donelist.另外, 將rdp-quiescbatch 賦值為 rcp-completed.這個(gè)是一個(gè)很重要的操作.Rdp- quiescbatch表示rcu_data已經(jīng)完成的grace period序號(在代碼中也被稱為了batch),rcp-completed表示全部變量rcu_ctrlblk計(jì)數(shù)已經(jīng)完成的grace period序號.將rdp-quiescbatch = rcp-completed;,表示不需要等待grace period.回到rcu_online_cpu()中:open_softirq(RCU_SOFTIRQ, rcu_process_callbacks, NULL);初始化了RCU_SOFTIRQ類型的軟中斷.但這個(gè)軟中斷什么時(shí)候被打開,還需要之后來分析.之后,每個(gè)CPU的初始化都會(huì)經(jīng)過start_kernel()-rcu_init().相應(yīng)的,也為每個(gè)CPU初始化了RCU的相關(guān)結(jié)構(gòu).五:等待RCU讀者操作完成之前,我們看完了RCU的初始化,現(xiàn)在可以來看一下RCU如何來判斷當(dāng)前的RCU讀者已經(jīng)退出了.在每一次進(jìn)程切換的時(shí)候,都會(huì)調(diào)用rcu_qsctr_inc().如下代碼片段如示:asmlinkage void _sched schedule(void) rcu_qsctr_inc(cpu); .Rcu_qsctr_inc()代碼如下:static inline void rcu_qsctr_inc(int cpu) struct rcu_data *rdp = &per_cpu(rcu_data, cpu); rdp-passed_quiesc = 1;該函數(shù)將對應(yīng)CPU上的rcu_data的passed_quiesc成員設(shè)為了1.或許你已經(jīng)發(fā)現(xiàn)了,這個(gè)過程就標(biāo)識(shí)該CPU經(jīng)過了一次quiescent state.沒錯(cuò):-)另外,在時(shí)鐘中斷中,會(huì)進(jìn)行以下操作:void update_process_times(int user_tick) if (rcu_pending(cpu) rcu_check_callbacks(cpu, user_tick); 在每一次時(shí)鐘中斷,都會(huì)檢查是否有需要更新的RCU需要處理,如果有,就會(huì)為其調(diào)用rcu_check_callbacks().Rcu_pending()的代碼如下: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的部份.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,分別對應(yīng):1:該CPU上有等待處理的回調(diào)函數(shù),且已經(jīng)經(jīng)過了一個(gè)batch(grace period).rdp-datch表示rdp在等待的batch序號2:上一個(gè)等待已經(jīng)處理完了,又有了新注冊的回調(diào)函數(shù).3:等待已經(jīng)完成,但尚末調(diào)用該次等待的回調(diào)函數(shù).4:在等待quiescent state.關(guān)于rcp和rdp結(jié)構(gòu)中成員的含義,我們等用到的時(shí)候再來分析.如果rcu_pending返回1,就會(huì)進(jìn)入到rcu_check_callbacks().代碼如下:void rcu_check_callbacks(int cpu, int user) if (user | (idle_cpu(cpu) & !in_softirq() & hardirq_count() rcu_qsctr_inc(cpu); rcu_bh_qsctr_inc(cpu); else if (!in_softirq() rcu_bh_qsctr_inc(cpu); raise_rcu_softirq();如果已經(jīng)CPU中運(yùn)行的進(jìn)程是用戶空間進(jìn)程或者是CPU空閑且不處于中斷環(huán)境,那么,它也已經(jīng)進(jìn)過了一次切換.注意,RCU只能在內(nèi)核空間使用.最后調(diào)用raise_rcu_softirq()打開了軟中斷處理.相應(yīng)的,也就調(diào)用RCU的軟中斷處理函數(shù).結(jié)合上面分析的初始化流程,軟中斷的處理函數(shù)為rcu_process_callbacks().代碼如下:static void rcu_process_callbacks(struct softirq_action *unused) _rcu_process_callbacks(&rcu_ctrlblk, &_get_cpu_var(rcu_data); _rcu_process_callbacks(&rcu_bh_ctrlblk, &_get_cpu_var(rcu_bh_data);在閱讀_rcu_process_callbacks()之前,先來了解一下rdp中幾個(gè)鏈表的含義:每次新注冊的回調(diào)函數(shù),都會(huì)鏈入到rdp-taillist.當(dāng)前等待grace period完成的函數(shù)都會(huì)鏈入到rdp-curlist上.到等待的grace period已經(jīng)到來,就會(huì)將curlist上的鏈表移到donelist上.當(dāng)一個(gè)grace period過了之后,就會(huì)將taillist上的數(shù)據(jù)移到rdp-curlist上.之后加冊的回調(diào)函數(shù)又會(huì)將其加到rdp-taillist上._rcu_process_callbacks()代碼分段分析如下:static 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; 如果有需要處理的回調(diào)函數(shù),且已經(jīng)經(jīng)過了一次grace period.就將curlist上的數(shù)據(jù)移到donetlist上.其中,crp-completed表示已經(jīng)完成的grace period.rdp-batch表示該CPU正在等待的grace period序號. if (rdp-nxtlist & !rdp-curlist) local_irq_disable(); rdp-curlist = rdp-nxtlist; rdp-curtail = rdp-nxttail; rdp-nxtlist = NULL; rdp-nxttail = &rdp-nxtlist; local_irq_enable(); /* * start the next batch of callbacks */ /* determine batch number */ rdp-batch = rcp-cur + 1; /* see the comment and corresponding wmb() in * the rcu_start_batch() */ smp_rmb(); if (!rcp-next_pending) /* and start it/schedule start if its a new batch */ spin_lock(&rcp-lock); rcp-next_pending = 1; rcu_start_batch(rcp); spin_unlock(&rcp-lock); 如果上一個(gè)等待的回調(diào)函數(shù)處理完了,而且又有了新注冊的回調(diào)函數(shù).就將taillist上的數(shù)據(jù)移動(dòng)到curlist上.并開啟新的grace period等待.注意里面幾個(gè)變量的賦值: rdp-batch = rcp-cur + 1表示該CPU等待的grace period置為當(dāng)前已發(fā)生grace period序號的下一個(gè).每次啟動(dòng)一個(gè)新的grace period等待之后,就會(huì)將rcp-next_pending.在啟動(dòng)的過程中,也就是rcu_start_batch()的過程中,會(huì)將rcp-next_pending置為1.設(shè)置這個(gè)變量主要是防止多個(gè)寫者競爭的情況 /更新相關(guān)信息 rcu_check_quiescent_state(rcp, rdp); /處理等待完成的回調(diào)函數(shù) if (rdp-donelist) rcu_do_batch(rdp);接著,更新相關(guān)的信息,例如,判斷當(dāng)前CPU是否進(jìn)行了quiescent state.或者grace period是否已經(jīng)完成.最后再處理掛在rdp-donelist上的鏈表.這里面有幾個(gè)子函數(shù)值得好好分析,分別分析如下:第一個(gè)要分析的是rcu_start_batch():static void rcu_start_batch(struct rcu_ctrlblk *rcp) if (rcp-next_pending & rcp-completed = rcp-cur) rcp-next_pending = 0; smp_wmb(); rcp-cur+; smp_mb(); cpus_andnot(rcp-cpumask, cpu_online_map, nohz_cpu_mask); rcp-signaled = 0; 這個(gè)函數(shù)的代碼雖然很簡單,但隱藏了很多玄機(jī).每次啟動(dòng)一個(gè)新的grace period等待的時(shí)候就將rcp-cur加1,將rcp-cpumask中,將存在的CPU的位置1.其中,if判斷必須要滿足二個(gè)條件:第一:rcp-next_pending必須為1.我們把這個(gè)函數(shù)放到_rcu_process_callbacks()這個(gè)大環(huán)境中看一下:static void _rcu_process_callbacks(struct rcu_ctrlblk *rcp, struct rcu_data *rdp) if (rdp-nxtlist & !rdp-curlist) if (!rcp-next_pending) /* and start it/schedule start if its a new batch */ spin_lock(&rcp-lock); rcp-next_pending = 1; rcu_start_batch(rcp); spin_unlock(&rcp-lock); 首先，rcp-next_pending為0才會(huì)調(diào)用rcu_start_batch()啟動(dòng)一個(gè)新的進(jìn)程然后，將rcp-next_pending置為1,再調(diào)用rcu_start_batch().在這里要注意中間的自旋鎖.然后在rcu_start_batch()中,再次判斷rcp-next_pending為1后,再進(jìn)行后續(xù)操作,并將rcp-next_pending置為0.為什么這里需要這樣的判斷呢? 如果其它CPU正在開啟一個(gè)新的grace period等待,那就用不著再次開啟一個(gè)新的等待了,直接返回即可.第二: rcu_start_batch()中if要滿足的第二個(gè)條件為rcp-completed = rcp-cur.也就是說前面的grace period全部都完成了.每次開啟新等待的時(shí)候都會(huì)將rcp-cur加1.每一個(gè)等待完成之后,都會(huì)將rc- completed等于rcp-cur.第二個(gè)要分析的函數(shù)是rcu_check_quiescent_state().代碼如下:static 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; /* Grace period already completed for this cpu? * qs_pending is checked instead of the actual bitmap to avoid * cacheline trashing. */ if (!rdp-qs_pending) return; /* * Was there a quiescent state since the beginning of the grace * period? If no, then exit and wait for the next call. */ if (!rdp-passed_quiesc) return; rdp-qs_pending = 0; spin_lock(&rcp-lock); /* * rdp-quiescbatch/rcp-cur and the cpu bitmap can come out of sync * during cpu startup. Ignore the quiescent state. */ if (likely(rdp-quiescbatch = rcp-cur) cpu_quiet(rdp-cpu, rcp); spin_unlock(&rcp-lock);首先,如果rdp-quiescbatch != rcp-cur.則說明又開啟了一個(gè)新的等待,因此需要重新處理這個(gè)等待,首先將rdp-quiescbatch 更新為rcp-cur.然后,使rdp-qs_pending為1.表示有等待需要處理. passed_quiesc也被清成了0.然后,再判斷rdp-passed_quiesc是否為真,記得我們在之前分析過,在每次進(jìn)程切換或者進(jìn)程切換的時(shí)候,都會(huì)調(diào)用rcu_qsctr_inc().該函數(shù)會(huì)將rdp-passed_quiesc置為1.因此,在這里判斷這個(gè)值是為了檢測該CPU上是否發(fā)生了上下文切換.之后,就是一段被rcp-lock保護(hù)的一段區(qū)域.如果還是等待沒有發(fā)生改變,就會(huì)調(diào)用cpu_quiet(rdp-cpu, rcp)將該CPU位清零.如果是一個(gè)新的等待了,就用不著清了,因?yàn)樾枰匦屡袛嘣揅PU上是否發(fā)生了上下文切換.cpu_quiet()函數(shù)代碼如下:static void cpu_quiet(int cpu, struct rcu_ctrlblk *rcp) cpu_clear(cpu, rcp-cpumask); if (cpus_empty(rcp-cpumask) /* batch completed ! */ rcp-completed = rcp-cur; rcu_start_batch(rcp); 它清除當(dāng)前CPU對應(yīng)的位,如果CPMMASK為空,對應(yīng)所有的CPU都發(fā)生了進(jìn)程切換,就會(huì)將rcp-completed = rcp-cur.并且根據(jù)需要是否開始一個(gè)grace period等待.最后一個(gè)要分析的函數(shù)是rcu_do_batch().它進(jìn)行的是清尾的工作.如果等待完成了,那就必須要處理donelist鏈表上掛載的數(shù)據(jù)了.代碼如下: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 rdp-blimit = blimit; if (!rdp-donelist) rdp-donetail = &rdp-donelist; else raise_rcu_softirq();它遍歷處理掛在鏈表上的回調(diào)函數(shù).在這里,注意每次調(diào)用的回調(diào)函數(shù)有最大值限制.這樣做主要是防止一次調(diào)用過多的回調(diào)函數(shù)而產(chǎn)生不必要系統(tǒng)負(fù)載.如果donelist中還有沒處理完的數(shù)據(jù),打開RCU軟中斷,在下次軟中斷到來的時(shí)候接著處理.五:幾種RCU情況分析1:如果CPU 1上有進(jìn)程調(diào)用rcu_read_lock進(jìn)入臨界區(qū),之后退出來,發(fā)生了進(jìn)程切換,新進(jìn)程又通過rcu_read­_lock進(jìn)入臨界區(qū).由于RCU軟中斷中只判斷一次上下文切換,因此,在調(diào)用回調(diào)函數(shù)的時(shí)候,仍然有進(jìn)程處于RCU的讀臨界區(qū),這樣會(huì)不會(huì)有問題呢?這樣是不會(huì)有問題的.還是上面的例子: 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:上面分析的似乎是針對有掛載鏈表的CPU而言的,那對于只調(diào)用rcu_read_lock()的CPU,它們是怎么處理的呢?首先,每次啟動(dòng)一次等待,肯定是會(huì)更新rcp-cur的.因此,在rcu_pending()的判斷中,下面語句會(huì)被滿足: if (rdp-quiescbatch != rcp-cur | rdp-qs_pending) return 1;因此會(huì)進(jìn)入到RCU的軟中斷.在軟中斷處理中:rcu_process_callbacks() _rcu_process_callbacks() rcu_check_quiescent_state()中,如果該CPU上有進(jìn)程切換,就會(huì)各新rcp中的CPU 掩碼數(shù)組.3:如果一個(gè)CPU連續(xù)調(diào)用synchronize_rcu()或者call_rcu()它們會(huì)有什么影響呢?如果當(dāng)前有請求在等待,就會(huì)新請?zhí)峤坏幕卣{(diào)函數(shù)掛到