linux內(nèi)核分析筆記----上半部與下半部(上)嗨，嗨，如果您記性好的話，我在上一篇博客中提到過這樣一件事：中斷處理是分為兩個部分：中斷處理程序是上半部，它接收到一個中斷，就立即執(zhí)行，但只做有嚴(yán)格時限的工作；而另外被叫做下半部的另外一個部分主要做被允許能稍后完成的工作。這個下半部正是今天的重點(diǎn)。
下半部的任務(wù)就是執(zhí)行與中斷處理密切相關(guān)但中斷處理程序本生身不執(zhí)行的任務(wù)。最好情況當(dāng)然是中斷處理程序把所有的工作都交給下半部執(zhí)行，而自己啥都不做。因?yàn)槲覀兛偸窍Ｍ袛嗵幚沓绦虮M可能快的返回。但是，中斷處理程序注定要完成一部分工作。遺憾的是，并沒有誰嚴(yán)格規(guī)定說什么任務(wù)應(yīng)該在哪個部分中完成，換句話說，這個決定完全由想我們這樣的驅(qū)動工程師來做。記住，中斷處理程序會異步執(zhí)行，并且在最好的情況下它也會鎖定當(dāng)前的中斷線，因此將中斷處理程序縮短的越小就越好。當(dāng)然啦，沒有規(guī)則并不是沒有經(jīng)驗(yàn)和教訓(xùn)：
1.如果一個任務(wù)對時間非常敏感，感覺告訴我還是將其放在中斷處理程序中執(zhí)行是個好的選擇。
2.如果一個任務(wù)和硬件相關(guān)，還是將其放在中斷處理程序中執(zhí)行吧。
3.如果一個任務(wù)要保證不被其他中斷(特別是相同的中斷)打斷，那就將其放在中斷處理程序中吧。
4.其他所有任務(wù)，除非你有更好的理由，否則全部丟到下半部執(zhí)行。
總之，一句話：中斷處理程序要執(zhí)行的越快越好。
我們前邊老是說下半部會在以后執(zhí)行，那么這個以后是個什么概念呢？遺憾的說，這個只是相對于馬上而言的。下半部并需要指定一個明確的時間，只要把這個任務(wù)推遲一點(diǎn)，讓他們在系統(tǒng)不太繁忙并且中斷恢復(fù)后執(zhí)行就可以了。通常下半部在中斷處理程序已返回就會馬上執(zhí)行，關(guān)鍵在于當(dāng)它們運(yùn)行時，允許相應(yīng)所有的中斷。
上半部只能通過中斷處理程序來完成，下半部的實(shí)現(xiàn)卻是有很多種方式。這些用來實(shí)現(xiàn)下半部的機(jī)制分別由不同的接口和子系統(tǒng)組成。最早的是“bottom half”，這種機(jī)制也被稱為“BH”,它提供的接口很簡單，提供了一個靜態(tài)創(chuàng)建，由32個bottom half組成的鏈表，上半部通過一個32位整數(shù)中的一位來標(biāo)識出哪個bottom half可執(zhí)行。每個BH都在全局范圍內(nèi)進(jìn)行同步，即使分屬于不同的處理器，也不允許任何兩個bottom half同時執(zhí)行。這種方式使用方便但不夠靈活，簡單卻有性能瓶頸。以需要更好的方法了。第二種方法，任務(wù)隊列(task queues).內(nèi)核定義了一組隊列。其中每個隊列都包含一個由等待調(diào)用的函數(shù)組成鏈表。根據(jù)其所處隊列的位置，這些函數(shù)會在某個時刻被執(zhí)行，驅(qū)動程序可根據(jù)需要把它們自己的下半部注冊到合適的隊列上去。這種方法已經(jīng)不錯，但仍然不夠靈活，它沒辦法代替整個BH接口。對于一些性能要求較高的子系統(tǒng)，像網(wǎng)絡(luò)部分，它也不能勝任。在2.3開發(fā)版中，又引入了軟中斷(softirqs)和tasklet，這里的軟中斷和實(shí)現(xiàn)系統(tǒng)調(diào)用所提到的軟中斷(軟件中斷)不是同一個概念。如果無須考慮和過去開發(fā)的驅(qū)動程序相兼容的話， 軟中斷和tasklet可以完全代替BH接口。軟中斷是一組靜態(tài)定義的下半部接口，有32個，可以在所有的處理器上同時執(zhí)行----即使兩個類型完全相同。task是一種基于軟中斷實(shí)現(xiàn)的靈活性強(qiáng)，動態(tài)創(chuàng)建的下半部實(shí)現(xiàn)機(jī)制。兩個不同類型的tasklet可以在不同的處理器上同時執(zhí)行，但類型相同的tasklet不能同時執(zhí)行。tasklet其實(shí)是一種在性能和易用性之間尋求平衡的產(chǎn)物。軟中斷必須在編譯期間就進(jìn)行靜態(tài)注冊，而tasklet可以通過代碼進(jìn)行動態(tài)注冊?，F(xiàn)在都是2.6內(nèi)核了，我們說點(diǎn)新鮮的吧，linux2.6內(nèi)核提供了三種不同形式的下半部實(shí)現(xiàn)機(jī)制：軟中斷，tasklets和工作對列，這些會依次介紹到。這時，可能有人會想到定時器的概念，定時器也確實(shí)是這樣，但定時器提供了精確的推遲時間，我們這里還不至于這樣，所以先放下，我們后面再說定時器。好，下面我開始說說詳細(xì)的各種機(jī)制：
1.軟中斷 實(shí)際上軟中斷使用的并不多，反而是后面的tasklet比較多，但tasklet是通過軟中斷實(shí)現(xiàn)的，軟中斷的代碼位于/kernel/softirq.c中。軟中斷是在編譯期間靜態(tài)分配的，由softirq_action結(jié)構(gòu)表示，它定義在linux/interrupt.h中：
1
2
3
4
struct softirq_action{
void (*action)(struct softirq_action *); //待執(zhí)行的函數(shù)
void *data; //傳給函數(shù)的參數(shù)
};
kernel/softirq.c中定義了一個包含有32個結(jié)構(gòu)體的數(shù)組：
1
static struct softirq_action softirq_vec[32]
每個被注冊的軟中斷都占據(jù)該數(shù)組的一項，因此最多可能有32個軟中斷，這是沒法動態(tài)改變的。由于大部分驅(qū)動程序都使用tasklet來實(shí)現(xiàn)它們的下半部，所以現(xiàn)在的內(nèi)核中，只用到了6個。上面的軟中斷結(jié)構(gòu)中，第一項是軟中斷處理程序，原型如下：
1
void softirq_handler(struct softirq_action *)
當(dāng)內(nèi)核運(yùn)行一個軟中斷處理程序時，它會執(zhí)行這個action函數(shù)，其唯一的參數(shù)為指向相應(yīng)softirq_action結(jié)構(gòu)體的指針。例如，如果my_softirq指向softirq_vec數(shù)組的實(shí)現(xiàn)，那么內(nèi)核會用如下的方式調(diào)用軟中斷處理程序中的函數(shù)：
1
my_softirq->action(my_softirq)
一個軟中斷不會搶占另外一個軟中斷，實(shí)際上，唯一可以搶占軟中斷的是中斷處理程序。不過，其他的軟中斷----甚至是相同類型的軟中斷-----可以在其他類型的機(jī)器上同時執(zhí)行。一個注冊的軟中斷必須在被標(biāo)記后才能執(zhí)行----觸發(fā)軟中斷(rasing the softirq).通常，中斷處理程序會在返回前標(biāo)記它的軟中斷，使其在稍后執(zhí)行。在下列地方，待處理的軟中斷會被檢查和執(zhí)行：
1.處理完一個硬件中斷以后 2.在ksoftirqd內(nèi)核線程中 3.在那些顯示檢查和執(zhí)行待處理的軟中斷的代碼中，如網(wǎng)絡(luò)子系統(tǒng)中。
軟中斷會在do_softirq()中執(zhí)行，如果有待處理的軟中斷，do_softirq會循環(huán)遍歷每一個，調(diào)用他們的處理程序。核心部分如下：
1
2
3
4
5
6
7
8
9
10
11
u32 pending = softirq_pending(cpu);
if(pending){
struct softirq_action *h = softirq_vec;
softirq_pending(cpu) = 0;
do{
if(pending &1)
h->action(h);
h++;
pending >>=1;
}while(pending);
}
上述代碼會檢查并執(zhí)行所有待處理的軟中斷，softirq_pending()，用它來獲得待處理的軟中斷的32位位圖-----如果第n位被設(shè)置為1，那么第n位對應(yīng)類型的軟中斷等待處理，一旦待處理的軟中斷位圖被保存，就可以將實(shí)際的軟中斷位圖清零了。pending &1是判斷pending的第一位是否被置為1.一旦pending為0，就表示沒有待處理的中斷了，因?yàn)閜ending最多可能設(shè)置32位，循環(huán)最多也只能執(zhí)行32位。軟中斷保留給系統(tǒng)中對時間要求最嚴(yán)格以及最重要的下半部使用。所以使用之前還是要想清楚的。下面簡要的說明如何使用軟中斷：
1.分配索引：在編譯期間，要通過<linux/interrupt.h>中建立的一個枚舉類型來靜態(tài)地聲明軟中斷。內(nèi)核用這些從0開始的索引來表示一種相對優(yōu)先級，
索引號越小就越先執(zhí)行。所以，可以根據(jù)自己的需要將自己的索引號放在合適的位置。
2.注冊處理程序：接著，在運(yùn)行時通過調(diào)用open_softirq()注冊中斷處理程序，例如網(wǎng)絡(luò)子系統(tǒng)，如下：
1
2
open_softirq(NET_TX_SOFTIRQ,net_tx_action,NULL);
open_softirq(NET_RX_SOFTIRQ,net_rx_action,NULL);
函數(shù)有三個參數(shù)，軟中斷索引號，處理函數(shù)和data域存放的數(shù)組。軟中斷處理程序執(zhí)行的時候，允許響應(yīng)中斷，但它自己不能休眠。在一個處理程序運(yùn)
行的時候，當(dāng)前處理器的軟中斷被禁止，但其他的處理器仍可以執(zhí)行別的軟中斷。實(shí)際上，如果一個軟中斷在它被執(zhí)行的時候同時再次被觸發(fā)了，那么
另外一個處理器可以同時運(yùn)行其處理程序。這意味著對共享數(shù)據(jù)都需要嚴(yán)格的鎖保護(hù)。大部分軟中斷處理程序都通過采取單處理數(shù)據(jù)(僅屬于某一個處
理器的數(shù)據(jù))或者其他一些技巧來避免顯示的加鎖，從而提供更出色的性能。
3.觸發(fā)軟中斷：經(jīng)過上面兩項，新的軟中斷處理程序就能夠運(yùn)行，raist_softirq(中斷索引號)函數(shù)可以將一個軟中斷設(shè)置為掛起狀態(tài)，從而讓它在下次調(diào)
用do_softirq()函數(shù)時投入運(yùn)行。該函數(shù)在觸發(fā)一個軟中斷之前先要禁止中斷，觸發(fā)后再恢復(fù)回原來的狀態(tài)，如果中斷本來就已經(jīng)被禁止了，那么可以調(diào)用另一函數(shù)raise_softirq_irqoff(),這會帶來一些優(yōu)化效果。
在中斷處理程序中觸發(fā)軟中斷是最常見的形式，中斷處理程序執(zhí)行硬件設(shè)備的相關(guān)操作，然后觸發(fā)相應(yīng)的軟中斷，最后退出。內(nèi)核在執(zhí)行完中斷處理程序后，馬上就會調(diào)用do_softirq()函數(shù)。于是，軟中斷就開始執(zhí)行中斷處理程序留給它去完成的剩下任務(wù)。
2.Tasklets：tasklet是通過軟中斷實(shí)現(xiàn)的，所以它們本身也是軟中斷。它由兩類軟中斷代表：HI_SOFTIRQ和TASKLET_SOFTIRQ.區(qū)別在于前者會先于后者執(zhí)行。Tasklets由tasklet_struct結(jié)構(gòu)表示，每個結(jié)構(gòu)體單獨(dú)代表一個tasklet,在linux/interrupt.h中定義：
1
2
3
4
5
6
7
struct tasklet_struct{
struct tasklet_struct *next;
unsigned long state;
atomic_t count;
void (*func)(unsigned long);
unsigned long data;
};
結(jié)構(gòu)體中func成員是tasklet的處理程序，data是它唯一的參數(shù)。state的取值只能是0，TASKLET_STATE_SCHED(表明tasklet已經(jīng)被調(diào)度，正準(zhǔn)備投入運(yùn)行)和TASKLET_STATE_RUN(表明該tasklet正在運(yùn)行，它只有在多處理器的系統(tǒng)上才會作為一種優(yōu)化來使用，單處理器系統(tǒng)任何時候都清楚單個tasklet是不是正在運(yùn)行)之間取值。count是tasklet的引用計數(shù)器，如果不為0，則tasklet被禁止，不允許執(zhí)行；只有當(dāng)它為0時，tasklet才被激活，并且被設(shè)置為掛起狀態(tài)時，該tasklet才能夠執(zhí)行。已調(diào)度的tasklet(等同于被觸發(fā)的軟中斷)存放在兩個單處理器數(shù)據(jù)結(jié)構(gòu)：tasklet_vec(普通tasklet)和task_hi_vec高優(yōu)先級的tasklet)中，這兩個數(shù)據(jù)結(jié)構(gòu)都是由tasklet_struct結(jié)構(gòu)體構(gòu)成的鏈表，鏈表中的每個tasklet_struct代表一個不同的tasklet。Tasklets由tasklet_schedule()和tasklet_hi_schedule()進(jìn)行調(diào)度，它們接收一個指向tasklet_struct結(jié)構(gòu)的指針作為參數(shù)。兩個函數(shù)非常相似(區(qū)別在于一個使用TASKLET_SOFTIRQ而另外一個使用HI_SOFTIRQ).有關(guān)tasklet_schedule()的操作細(xì)節(jié)：
1.檢查tasklet的狀態(tài)是否為TASKLET_STATE_SCHED.如果是，說明tasklet已經(jīng)被調(diào)度過了，函數(shù)返回。
2.保存中斷狀態(tài)，然后禁止本地中斷。在執(zhí)行tasklet代碼時，這么做能夠保證處理器上的數(shù)據(jù)不會弄亂。
3.把需要調(diào)度的tasklet加到每個處理器一個的tasklet_vec鏈表或task_hi_vec鏈表的表頭上去。
4.喚起TASKLET_SOFTIRQ或HI_SOFTIRQ軟中斷，這樣在下一次調(diào)用do_softirq()時就會執(zhí)行該tasklet。
5.恢復(fù)中斷到原狀態(tài)并返回。
那么作為tasklet處理的核心tasklet_action()和tasklet_hi_action(),具體做了些什么呢：
1.禁止中斷，并為當(dāng)前處理器檢索tasklet_vec或tasklet_hi_vec鏈表。
2.將當(dāng)前處理器上的該鏈表設(shè)置為NULL,達(dá)到清空的效果。
3.運(yùn)行相應(yīng)中斷。
4.循環(huán)遍歷獲得鏈表上的每一個待處理的tasklet。
5.如果是多處理器系統(tǒng)，通過檢查TASKLET_STATE_RUN來判斷這個tasklet是否正在其他處理器上運(yùn)行。如果它正在運(yùn)行，那么現(xiàn)在就不要執(zhí)行，跳
到下一個待處理的tasklet去。
6.如果當(dāng)前這個tasklet沒有執(zhí)行，將其狀態(tài)設(shè)置為TASKLETLET_STATE_RUN,這樣別的處理器就不會再去執(zhí)行它了。
7.檢查count值是否為0，確保tasklet沒有被禁止。如果tasklet被禁止，則跳到下一個掛起的tasklet去。
8.現(xiàn)在可以確定這個tasklet沒有在其他地方執(zhí)行，并且被我們設(shè)置為執(zhí)行狀態(tài)，這樣它在其他部分就不會被執(zhí)行，并且引用計數(shù)器為0，現(xiàn)在可以執(zhí)行
tasklet的處理程序了。
9.重復(fù)執(zhí)行下一個tasklet，直至沒有剩余的等待處理的tasklets。
說了這么多，我們該怎樣使用這個tasklet呢，這個我在linux設(shè)備驅(qū)動理論帖講的太多了。但別急，下邊為了博客完整，我仍然會大致講講：
1.聲明自己的tasklet：投其所好，既可以靜態(tài)也可以動態(tài)創(chuàng)建，這取決于選擇是想有一個對tasklet的直接引用還是間接引用。靜態(tài)創(chuàng)建方法(直接引用)，可以使用下列兩個宏的一個(在linux/interrupt.h中定義)：
1
2
DECLARE_TASKLET(name,func,data)
DECLARE_TASKLET_DISABLED(name,func,data)
這兩個宏之間的區(qū)別在于引用計數(shù)器的初始值不同，前面一個把創(chuàng)建的tasklet的引用計數(shù)器設(shè)置為0，使其處于激活狀態(tài)，另外一個將其設(shè)置為1，處于禁止?fàn)顟B(tài)。而動態(tài)創(chuàng)建(間接引用)的方式如下：
1
tasklet_init(t,tasklet_handler,dev);
2.編寫tasklet處理程序：函數(shù)類型是void tasklet_handler(unsigned long data).因?yàn)槭强寇浿袛鄬?shí)現(xiàn)，所以tasklet不能休眠，也就是說不能在tasklet中使用信號量或者其他什么阻塞式的函數(shù)。由于tasklet運(yùn)行時允許響應(yīng)中斷，所以必須做好預(yù)防工作，如果新加入的tasklet和中斷處理程序之間共享了某些數(shù)據(jù)額的話。兩個相同的tasklet絕不能同時執(zhí)行，如果新加入的tasklet和其他的tasklet或者軟中斷共享了數(shù)據(jù)，就必須要進(jìn)行適當(dāng)?shù)劓i保護(hù)。
3.調(diào)度自己的tasklet：前邊的工作一做完，接下來就剩下調(diào)度了。通過一下方法實(shí)現(xiàn)：tasklet_schedule(&my_tasklet).在tasklet被調(diào)度以后，只要有合適的機(jī)會就會得到運(yùn)行。如果在還沒有得到運(yùn)行機(jī)會之前，如果有一個相同的tasklet又被調(diào)度了，那么它仍然只會運(yùn)行一次。如果這時已經(jīng)開始運(yùn)行，那么這個新的tasklet會被重新調(diào)度并再次運(yùn)行。一種優(yōu)化策略是一個tasklet總在調(diào)度它的處理器上執(zhí)行。調(diào)用tasklet_disable()來禁止某個指定的tasklet，如果該tasklet當(dāng)前正在執(zhí)行，這個函數(shù)會等到它執(zhí)行完畢再返回。調(diào)用tasklet_disable_nosync()也是來禁止的，只是不用在返回前等待tasklet執(zhí)行完畢，這么做安全性就不咋嘀了(因?yàn)闆]法估計該tasklet是否仍在執(zhí)行)。tasklet_enable()激活一個tasklet?？梢允褂胻asklet_kill()函數(shù)從掛起的對列中去掉一個tasklet。這個函數(shù)會首先等待該tasklet執(zhí)行完畢，然后再將其移去。當(dāng)然，沒有什么可以阻止其他地方的代碼重新調(diào)度該tasklet。由于該函數(shù)可能會引起休眠，所以禁止在中斷上下文中使用它。
接下來的問題，我在前邊說過，對于軟中斷，內(nèi)核會選擇幾個特殊的實(shí)際進(jìn)行處理(常見的是中斷處理程序返回時)。軟中斷被觸發(fā)的頻率有時會很好，而且還可能會自行重復(fù)觸發(fā)，這帶來的結(jié)果就是用戶空間的進(jìn)程無法獲得足夠的處理器時間，因?yàn)樘幱陴囸I狀態(tài)。同時，如果單純的對重復(fù)觸發(fā)的軟中斷采取不立即處理的策略也是無法接受的。兩種極端但完美的情況是什么樣的呢：
1.只要還有被觸發(fā)并等待處理的軟中斷，本次執(zhí)行就要負(fù)責(zé)處理，重新觸發(fā)的軟中斷也在本次執(zhí)行返回前被處理。問題在于，用戶進(jìn)程可能被忽略而使其
處于饑餓狀態(tài)。
2.選擇不處理重新觸發(fā)的軟中斷。在從中斷返回的時候，內(nèi)核和平常一樣，也會檢查所有掛起的軟中斷并處理它們，但是，任何自行重新觸發(fā)的軟中斷都
不會馬上處理，它們被放到下一個軟中斷執(zhí)行時機(jī)去處理。問題在于新的或重新觸發(fā)的軟中斷必要要等一定的時間才能被執(zhí)行。
我現(xiàn)在想的是來個折衷方案吧，那多好，內(nèi)核開發(fā)者門還真是想到了。內(nèi)核選中的方案是不會立即處理重新觸發(fā)的軟中斷，作為改進(jìn)，當(dāng)大量軟中斷出現(xiàn)的時候，內(nèi)核會喚醒一組內(nèi)核線程來處理這些負(fù)載。這些線程在最低優(yōu)先級上運(yùn)行(nice值為19)。這種這種方案能夠保證在軟中斷負(fù)擔(dān)很重的時候用戶程序不會因?yàn)榈貌坏教幚頃r間而處理饑餓狀態(tài)。相應(yīng)的，也能保證“過量”的軟中斷終究會得到處理。最后，在空閑系統(tǒng)上，這個方案同樣表現(xiàn)良好，軟中斷處理得非常迅速(因?yàn)閮H存的內(nèi)存線程肯定會馬上調(diào)度)。為了保證只要有空閑的處理器，它們就會處理軟中斷，所以給每個處理器都分配一個這樣的線程。所有線程的名字都叫做ksoftirad/n，區(qū)別在于n，它對應(yīng)的是處理器的編號。一旦該線程被初始化，它就會執(zhí)行類似下面這樣的死循環(huán)：
1
2
3
4
5
6
7
8
9
10
11
for(;;){
if(!softirq_pending(cpu))
schedule();
set_current_state(TASK_RUNNING);
while(softirq_pending(cpu)){
do_softirq();
if(need_resched())
schedule();
}
set_current_state(TASK_INTERRUPTIBLE);
}
softirq_pending()負(fù)責(zé)發(fā)現(xiàn)是否有待處理的軟中斷。當(dāng)所有需要執(zhí)行的操作都完成以后，該內(nèi)核線程將自己設(shè)置為TASK_INTERRUPTIBLE狀態(tài)，喚起調(diào)度程序選擇其他可執(zhí)行進(jìn)程投入運(yùn)行。最后，只要do_softirq()函數(shù)發(fā)現(xiàn)已經(jīng)執(zhí)行過的內(nèi)核線程重新觸發(fā)了自己，軟中斷內(nèi)核線程就會被喚醒。
linux內(nèi)核分析筆記----上半部與下半部(下)接著上節(jié)的來，我們在上節(jié)說了軟中斷和tasklet，那這最后就是工作隊列了哦..
工作隊列和前面討論的其他形式都不相同，它可以把工作推后，交由一個內(nèi)核線程去執(zhí)行----該工作總是會在進(jìn)程上下文執(zhí)行。這樣，通過工作隊列執(zhí)行代碼能占盡進(jìn)程上下文的所有優(yōu)勢，最重要的就是工作隊列允許重新調(diào)度甚至是睡眠。相比較前邊兩個，這個選擇起來就很容易了。我說過，前邊兩個是不允許休眠的，這個是允許休眠的，這就很明白了是不？這意味著在你需要獲得大量內(nèi)存的時候，在你需要獲取信號量時，在你需要執(zhí)行阻塞式的I/O操作時，它都會非常有用(先說話， 這個不是我說的，是書上這么說的哦)。
工作隊列子系統(tǒng)是一個用于創(chuàng)建內(nèi)核線程的接口，通過它創(chuàng)建的進(jìn)程負(fù)責(zé)執(zhí)行由內(nèi)核其他部分排到隊列里的任務(wù)。它創(chuàng)建的這些內(nèi)核線程被稱作工作者線程(worker threads).工作隊列可以讓你的驅(qū)動程序創(chuàng)建一個專門的工作者線程來處理需要推后的工作。不過，工作隊列子系統(tǒng)提供了一個缺省的工作者線程來處理這些工作。因此，工作隊列最基本的表現(xiàn)形式就轉(zhuǎn)變成一個把需要推后執(zhí)行的任務(wù)交給特定的通用線程這樣一種接口。缺省的工作線程叫做event/n.每個處理器對應(yīng)一個線程，這里的n代表了處理器編號。除非一個驅(qū)動程序或者子系統(tǒng)必須建立一個屬于自己的內(nèi)核線程，否則最好還是使用缺省線程。
1.工作這線程結(jié)構(gòu)用下面的結(jié)構(gòu)表示：
1
2
3
struct workqueue_struct{
struct cpu_workqueue_struct cpu_wq[NR_CPUS];
}
結(jié)構(gòu)中數(shù)組的每一項對應(yīng)系統(tǒng)的一個CPU.接下來，在看看在kernel/workqueue.c中的核心數(shù)據(jù)結(jié)構(gòu)cpu_workqueue_struct:
1
2
3
4
5
6
7
8
9
10
struct cpu_workqueue_struct{
spinlock_t lock;
atomic_t nr_queued;
struct list_head worklist;
wait_queue_head_t more_work;
wait_queue_head_t work_done;
struct workqueue_struct *wq;
task_t *thread;
struct completion exti;
}
2.表示工作的數(shù)據(jù)結(jié)構(gòu)：所有的工作者線程都是用普通的內(nèi)核線程來實(shí)現(xiàn)的，它們都要執(zhí)行worker_thread()函數(shù)。在它初始化完以后，這個函數(shù)執(zhí)行一個死循環(huán)執(zhí)行一個循環(huán)并開始休眠，當(dāng)有操作被插入到隊列的時候，線程就會被喚醒，以便執(zhí)行這些操作。當(dāng)沒有剩余的時候，它又會繼續(xù)休眠。工作有work_struct(linux/workqueue)結(jié)構(gòu)表示:
1
2
3
4
5
6
7
8
struct work_struct{
unsigned long pending;
struct list_head entry; //連接所有工作的鏈表
void (*func)(void *); //處理函數(shù)
void *data; //傳遞給處理函數(shù)的參數(shù)
void *wq_data;
struct timer_list timer; //y延遲工作隊列所用到的定時器
}
當(dāng)一個工作線程被喚醒時，它會執(zhí)行它的鏈表上的所有工作。工作一旦執(zhí)行完畢，它就將相應(yīng)的work_struct對象從鏈表上移去，當(dāng)鏈表不再有對象時，它就繼續(xù)休眠。woker_thread函數(shù)的核心流程如下：
1
2
3
4
5
6
7
8
9
10
11
for(;;){
set_task_state(current,TASK_INTERRUPTIBLE);
add_wait_queue(&cwq->more_work,&wait);
if(list_empty(&cwq->worklist))
schedule();
else
set_task_state(current,TASK_RUNNING);
remove_wait_queue(&cwq->more_work,&wait);
if(!list_empty(&cwq->worklist))
run_workqueue(cwq);
}
分析一下上面的代碼。首先線程將自己設(shè)置為休眠狀態(tài)并把自己加入等待隊列。如果工作對列是空的，線程調(diào)用schedule()函數(shù)進(jìn)入睡眠狀態(tài)。如果鏈表有對象，線程就將自己設(shè)為運(yùn)行態(tài)，脫離等待隊列。然后，再次調(diào)用run_workqueue()執(zhí)行推后的工作。好了，接下來，問題就糾結(jié)在run_workqueue(),它完成實(shí)際推后到此的工作：
1
2
3
4
5
6
7
8
while(!list_empty(&cwq->worklist)){
struct work_struct *work = list_entry(cwq->worklist.next,struct work_struct,entry);
void (*f)(void *) = work->func;
void *data = work->data;
list_del_init(cwq->worklist.next);
clear_bit(0,&work->pending);
f(data);
}
該函數(shù)循環(huán)遍歷鏈表上每個待處理的工作，執(zhí)行鏈表上每個結(jié)點(diǎn)上的work_struct的func成員函數(shù)：
1.當(dāng)鏈表不為空時，選取下一個節(jié)點(diǎn)對象。
2.獲取我們希望執(zhí)行的函數(shù)func及其參數(shù)data。
3.把該結(jié)點(diǎn)從鏈表上接下來，將待處理標(biāo)志位pending清0。
4.調(diào)用函數(shù)。
5.重復(fù)執(zhí)行。
老師說的好：光說不練，不是好漢。現(xiàn)在我們繼續(xù)來看看怎么用吧：
1.首先，實(shí)際創(chuàng)建一些需要推后完成的工作，可以在編譯時靜態(tài)地創(chuàng)建該數(shù)據(jù)結(jié)構(gòu)：
1
DECLARE_WORK(name,void (*func)(void *),void *data);
當(dāng)然了，如果愿意，我們當(dāng)然可以在運(yùn)行時通過指針動態(tài)創(chuàng)建一個工作：
1
INIT_WORK(struct work_struct *work, void (*func)(void *),void *data);
2.工作隊列處理函數(shù)，會由一個工作者線程執(zhí)行，因此，函數(shù)會運(yùn)行在進(jìn)程上下文中，默認(rèn)情況下，允許相應(yīng)中斷，并且不持有鎖。如果需要，函數(shù)可以睡眠。需要注意的是，盡管處理函數(shù)運(yùn)行在進(jìn)程上下文中，但它不能訪問用戶空間，因?yàn)閮?nèi)核線程在用戶空間沒有相應(yīng)的內(nèi)存映射。函數(shù)原型如下：
1
void work_hander(void *data);
3.對工作進(jìn)行調(diào)度。前面的準(zhǔn)備工作做完以后，下面就可以開始調(diào)度了，只需調(diào)用schedule_work(&work).這樣work馬上就會被調(diào)度，一旦其所在的處理器上的工作者線程被喚醒，它就會被執(zhí)行。當(dāng)然如果不想快速執(zhí)行，而是想延遲一段時間執(zhí)行，按就用schedule_delay_work(&work,delay);delay是要延遲的時間節(jié)拍，后面講.
4.刷新操作。插入隊列的工作會在工作者線程下一次被喚醒的時候執(zhí)行。有時，在繼續(xù)下一步工作之前，你必須保證一些操作已經(jīng)執(zhí)行完畢等等。由于這些原因，內(nèi)核提供了一個用于刷新指定工作隊列的函數(shù)：void flush_scheduled_work(void); 這個函數(shù)會一直等待，直到隊列中所有的對象都被執(zhí)行后才返回。在等待所有待處理的工作執(zhí)行的時候，該函數(shù)會進(jìn)入休眠狀態(tài)，所以只能在進(jìn)程上下文中使用它。需要說明的是，該函數(shù)并不取消任何延遲執(zhí)行的工作。取消延遲執(zhí)行的工作應(yīng)該調(diào)用：int cancel_delayed_work(struct work_struct *work);這個函數(shù)可以取消任何與work_struct 相關(guān)掛起的工作。
5.創(chuàng)建新的工作隊列。前邊說過最好使用缺省線程，可如果你堅持要使用自己創(chuàng)建的線程，咋辦？這時你就應(yīng)該創(chuàng)建一個新的工作隊列和與之相應(yīng)的工作者線程，方法很簡單，用下面的函數(shù)：struct workqueue_struct *create_workqueue(const char *name);name是新內(nèi)核線程的名字。這樣就會創(chuàng)建所有的工作者線程(系統(tǒng)中的每個處理器都有一個)并且做好所有開始處理工作之前的準(zhǔn)備工作。在創(chuàng)建之后，就調(diào)用下面的函數(shù)吧：
1
2
int queue_work(struct workqueue_struct *wq, struct work_struct *work);
int queue_delayed_work(struct workqueue_struct *wq,struct work_struct *work,unsigned long delay);
這兩個函數(shù)和schedule_work()和schedule_delayed_work()相近，唯一的區(qū)別在于它們可以針對特定的工作隊列而不是缺省的event隊列進(jìn)行操作。
好了，工作隊列也說完了，我還是結(jié)合前邊一篇，把這三個地板不實(shí)現(xiàn)的策略比較一下，方便以后選擇.
首先，tasklet是基于軟中斷實(shí)現(xiàn)的，兩者相近，工作隊列機(jī)制與它們完全不同，靠內(nèi)核線程來實(shí)現(xiàn)。軟中斷提供的序列化的保障最少，這就要求中斷處理函數(shù)必須格外小心地采取一些步驟確保共享數(shù)據(jù)的安全，兩個甚至更多相同類別的軟中斷有可能在不同的處理器上同時執(zhí)行。如果被考察的代碼本身多線索化的工作做得非常好，它完全使用單處理器變量，那么軟中斷就是非常好的選擇。對于時間要求嚴(yán)格和執(zhí)行效率很高的應(yīng)用來說，它執(zhí)行的也最快。否則選擇tasklets意義更大。tasklet接口簡單，而且兩種同種類型的tasklet不能同時執(zhí)行，所以實(shí)現(xiàn)起來也會簡單一些。如果需要把任務(wù)推遲到進(jìn)程上下文中完成，那你只能選擇工作隊列了。如果不需要休眠，那軟中斷和tasklet可能更合適。另外就是工作隊列造成的開銷最大，當(dāng)然這是相對的，針對大部分情況，工作隊列都能提供足夠的支持。從方便度上考慮就是:工作隊列，tasklets，最后才是軟中斷。我們在做驅(qū)動的時候，關(guān)于這三個下半部實(shí)現(xiàn)，需要考慮兩點(diǎn)：首先，是不是需要一個可調(diào)度的實(shí)體來執(zhí)行需要推后完成的工作(即休眠的需要)，如果有，工作隊列就是唯一的選擇，否則最好用tasklet。性能如果是最重要的，那還是軟中斷吧。
最后，就是一些禁止下半部的相關(guān)部分了，給一個表：
函數(shù)
描述
void local_bh_disable()禁止本地處理器的軟中斷和tasklet的處理
void local_bh_enable()激活本地處理器的軟中斷和tasklet的處理
這些函數(shù)有可能被嵌套使用----最后被調(diào)用的local_bh_enable()最終激活下半部。函數(shù)通過preempt_count為每個進(jìn)程維護(hù)一個計數(shù)器。當(dāng)計數(shù)器變?yōu)?時，下半部才能夠被處理。因?yàn)橄掳氩康奶幚硪呀?jīng)被禁止了，所以local_bh_enable()還需要檢查所有現(xiàn)存的待處理的下半部并執(zhí)行它們。
好了，這一次講完了，畫了兩次，我們在這兩次中提到了一些同時發(fā)生的問題，這時可能存在數(shù)據(jù)共享互斥訪問的問題，這個就是內(nèi)核同步方面的事情了，我們后面再慢慢說這個事。
