下半部和推后執行的工作
- 1 下半部
- 為什么要用下半部
- 下半部的環境
- 內核定時器
- 2 軟中斷
- 軟中斷的實現
- 軟中斷處理程序
- 執行軟中斷
- 使用軟中斷
- 3 tasklet
- tasklet的實現
- 使用tasklet
- ksoftirqd
- 4 工作隊列
- 工作隊列的實現
- 工作、工作隊列和工作者線程之間的關系
- 使用工作隊列
- 5 下半部機制的選擇
- 6 在下半部之間加鎖
中斷處理程序本身存在一些局限,所以它只能完成整個中斷處理流程的上半部分。這些局限包括:
- 中斷處理程序以異步方式執行并且它有可能會打斷其他重要代碼(甚至包括其他處理程序)的執行。因此,為了避免被打斷的代碼停止時間過長,中斷處理程序應該執行地越快越好。
- 如果當前有一個中斷處理程序正在執行,在最好的情況下(如果設置了SA_INTERRUPT),與該中斷同級的其他中斷會被屏蔽,在最壞的情況下,當前處理器上所有其他中斷都會被屏蔽。因此,仍應該讓它們執行的越快越好。
- 由于中斷處理程序往往需要對硬件進行操作,所以它們通常有很高的時限要求。
- 中斷處理程序不在進程上下文中運行,所以它們不能阻塞,這限制了它們所做的事情。
現在,為什么中斷處理程序只能作為整個硬件中斷處理流程一部分的原因就很明顯了。我們必須有一個快速、異步、簡單的處理程序負責對硬件做出迅速響應并完成那些時間要求很嚴格的額操作。中斷處理程序很適合于實現這些功能,可是,對于那些對時間要求相對寬松的任務,就應該推后到中斷被激活以后再去運行。
這樣,整個中斷流程就被分為了兩個部分,第一個部分是中斷處理程序(上半部),還有我們接下來所討論的下半部。
1 下半部
下半部的任務是執行與中斷處理密切相關但中斷處理程序(上半部)本身不執行的工作。
為什么要用下半部
我們希望盡量減少中斷處理程序中需要完成的工作量,因為在它運行的時候當前的中斷線在所有處理器上都會被屏蔽。更糟糕的是如果一個處理程序是SA_INTERRUPT類型,它執行的時候會禁止所有的本地中斷。而縮短中斷被屏蔽的時間對系統的響應能力和性能都至關重要。再加上中斷處理程序要與其他程序異步執行,我們必須盡力縮短中斷處理程序的執行。解決的方法就是把一些工作放到以后去做。
下半部并不需要指明一個確切時間,只要把這些任務推遲一點,讓它們在系統不太繁忙并且中斷恢復后執行就可以了。下半部執行的關鍵在于當它們運行的時候,允許響應所有的中斷。
下半部的環境
和上半部只能通過中斷處理程序實現不同,下半部可以通過多種機制實現,這些用來實現下半部的機制分別由不同的接口和子系統組成。
最早的Linux只提供bottom half這種機制用于實現下半部,這種機制也被稱為BH,我們現在也這么叫它,以避免和下半部這個通用詞匯混淆。BH接口也非常簡單,它提供了一個靜態創建、由32個bottom half組成的鏈表。上半部通過一個32位整數中的一位來標識出哪個bottom half可以執行。每個BH都在全局范圍內進行同步,即使分屬于不同的處理器,也不允許任何;兩個bottom half同時執行。
不久,內核開發者引入了任務隊列機制來實現工作的推后執行,并用它來替代BH機制。內核為此定義了一組隊列。其中每個隊列都包含一個由等待調用的函數組成鏈表。根據其所處隊列的位置,這些函數會在某個時刻被執行。驅動程序可以把它們自己的下半部注冊到合適的隊列上去。這種機制表現得還不錯,但不夠靈活,對于一些性能要求較高的子系統,像網絡部分,它不能勝任。
在2.3版本中,內核開發者引入了軟中斷(softirqs)和tasklet。如果無需考慮過去的驅動程序兼容的話,軟中斷和tasklet可以完全替代BH接口。軟中斷是一組靜態定義的下半部接口,有32個,可以在所有處理器上同時執行,即使兩個類型相同也可以。tasklet是一種基于軟中斷實現的靈活性強、動態創建的下半部實現機制。兩個不同類型的tasklet可以在不同的處理器上同時執行,但類型相同的tasklet不能同時執行。使用軟中斷需要特別小心,因為兩個相同的軟中斷有可能同時被執行。此外,軟中斷必須在編譯期間就進行靜態注冊,與此相反,tasklet可以通過代碼進行動態注冊。
在2.5版本,BH接口最終被棄置了。所有的BH使用者必須轉而使用其他下半部接口。此外,任務隊列也被工作隊列接口取代了。工作隊列是一種簡單但很有用的方法,它們先對要推后執行的工作排隊,稍后在進程上下文中執行它們。
在2.6版本中,內核提供了三種不同形式的下半部實現機制:軟中斷、tasklet和工作隊列。
內核定時器
另外一個可以用于將工作推后執行的機制是內核定時器。內核定時器把操作推遲到某個確定的時間段之后執行。內核定時器是以軟中斷為基礎建立的。
2 軟中斷
軟中斷使用得比較少,而tasklet是下半部更常用的一種形式。軟中斷的代碼位于kernel/softirq.c文件中。
軟中斷的實現
軟中斷是在編譯期間靜態分配的。它不像tasklet那樣能被動態地注冊或去除。軟中斷由softirq_action結構表示,定義在include/linux/interrupt.h中
/* softirq mask and active fields moved to irq_cpustat_t in* asm/hardirq.h to get better cache usage. KAO*/struct softirq_action
{void (*action)(struct softirq_action *);void *data;
};
kernel/softirq.c中定義了一個包含有32個該結構體的數組
static struct softirq_action softirq_vec[32] __cacheline_aligned_in_smp;
每個被注冊的軟中斷都占據該數組的一項。因此最多可能有32個軟中斷。注意這是個定制,注冊的軟中斷數目的最大值沒法動態改變。
軟中斷處理程序
軟中斷處理程序action的函數原型如下:
void softirq_handler(struct softirq_action *);
當內核運行一個軟中斷處理程序的時候,它就會執行這個函數,其唯一的參數為指向相應softirq_action結構體的指針。例如,如果my_softirq指向softirq_vec數組的某項,那么內核會用如下的方式調用軟中斷處理程序中的函數:
my_softirq->action(my_softirq);
內核把整個結構體都傳遞給軟中斷處理程序。
一個軟中斷不會搶占另一個軟中斷。唯一可以搶占軟中斷的是中斷處理程序。不過,其他軟中斷(可以是相同類型的軟中斷),可以在其他處理器上同時執行
執行軟中斷
一個注冊的軟中斷必須在被標記后才會執行。這被稱作觸發軟中斷。通常,中斷處理程序會在返回前標記它的軟中斷,使其在稍后執行。在下列地方,待處理的軟中斷會被檢查和執行:
- 從一個硬件中斷代碼處返回時
- 在ksoftirqd內核線程中
- 在那些顯式檢查和執行待處理的軟中斷的代碼中,如網絡子系統。
不管用什么方法喚起,軟中斷都要在do_softirq()中執行。如果有待處理的軟中斷,do_softirq()會循環遍歷每一個,調用它們的處理程序。它是先去獲取軟中斷32位位圖,如果第n位被設置為1,那么第n為對應類型的軟中斷等待處理,do_softirq()會去調用softirq_vec數組的第n個軟中斷處理程序。
使用軟中斷
軟中斷保留給系統中對時間要求最嚴格以及最重要的下半部使用,在2.6.10版本中,只有兩個子系統:網絡和SCSI,直接使用軟中斷。此外,內核定時器和tasklet都是建立在軟中斷上的。對于時間要求嚴格并能自己高效完成加鎖工作的應用,軟中斷是正確的選擇。當你想加入一個新的軟中斷時,首先應該問問自己為什么tasklet實現不了?
-
分配索引
在編譯期間,可以通過include/linux/interrupt.h中定義的一個枚舉類型來靜態地聲明軟中斷。內核用這些從0開始的索引來表示一種相對優先級。索引號小的軟中斷在索引號大的軟中斷之前執行。
建立一個新的軟中斷必須在此枚舉類型中加入新的項。而加入時,不能簡單地加入到末尾。應該根據希望賦予它的優先級來決定加入的位置。 -
注冊你的處理程序
接著,在運行時通過調用open_softirq()注冊軟中斷處理程序,該函數有三個參數:軟中斷的索引號、處理函數和data域存放的數值。例如網絡子系統,通過以下方式注冊自己的軟中斷:
open_softirq(NET_TX_SOFTIRQ,net_tx_action,NULL);
軟中斷處理程序執行的時候,允許響應中斷,但它自己不能休眠。在一個處理程序運行的時候,當前處理器上的軟中斷被禁止。但其他的處理器仍可以執行別的軟中斷。實際上,如果同一個軟中斷在它被執行的同時再次被觸發了,那么另外一個處理器可以同時運行其處理程序。這意味著任何共享數據,都需要嚴格的鎖保護。
- 觸發你的軟中斷
通過在枚舉類型的列表中添加新項以及調用open_softirq()進行注冊以后,新的軟中斷處理程序就能夠運行。raise_softirq()函數可以將一個軟中斷設置為掛起狀態,讓它在下次調用do_softirq()函數時投入運行。比如,網絡子系統可能會調用:
raise_softirq(NET_TX_SOFTIRQ);
這會觸發NET_TX_SOFTIRQ軟中斷,它的處理程序net_tx_action()就會在內核下一次執行軟中斷時投入運行。該函數在觸發一個軟中斷之前先要禁止中斷,觸發后再 恢復回原來的狀態。如果中斷本來就已經被禁止了,那么可以調用另一個函數raise_softirq_irqoff()。
3 tasklet
tasklet是利用軟中斷實現的一種下半部機制。它和進程沒有任何關系。
tasklet的實現
因為tasklet是通過軟中斷實現的,所以它們本身也是軟中斷。tasklet由兩類軟中斷代表:HI_SOFTIRQ和TASKLET_SOFTIRQ。這兩者唯一的區別就是HI_SOFTIRQ類型的軟中斷優先與TASKLET_SOFTIRQ類型的軟中斷執行。
- tasklet結構體
tasklet是由tasklet_struct結構表示。每個結構體單獨代表一個tasklet,它在include/linux/interrupt.h中定義。
/* Tasklets --- multithreaded analogue of BHs.Main feature differing them of generic softirqs: taskletis running only on one CPU simultaneously.Main feature differing them of BHs: different taskletsmay be run simultaneously on different CPUs.Properties:* If tasklet_schedule() is called, then tasklet is guaranteedto be executed on some cpu at least once after this.* If the tasklet is already scheduled, but its excecution is still notstarted, it will be executed only once.* If this tasklet is already running on another CPU (or schedule is calledfrom tasklet itself), it is rescheduled for later.* Tasklet is strictly serialized wrt itself, but notwrt another tasklets. If client needs some intertask synchronization,he makes it with spinlocks.*/struct tasklet_struct
{struct tasklet_struct *next; /* 鏈表中的下一個tasklet */unsigned long state; /* tasklet的狀態 */atomic_t count; /* 引用計數器 */void (*func)(unsigned long); /* tasklet處理函數 */unsigned long data; /* 給tasklet處理函數的參數 */
};
結構體中的func是tasklet的處理函數,data是它唯一的參數。state只能在0、TASKLET_STATE_SCHED和TASKLET_STATE_RUN之間取值。TASKLET_STATE_SCHED表明tasklet已被調度,正準備投入運行,TASKLET_STATE_RUN表明該tasklet正在運行。TASKLET_STATE_RUN只有在多處理器的系統上才會作為一種優化來使用,單處理器系統任何時候都清楚單個tasklet是不是正在運行。
count成員是tasklet的引用計數器。如果它不為0,則tasklet被禁止,不允許執行;只有當它為0時,tasklet才被激活,并且在被設備為掛起狀態,該tasklet才能夠執行。
- 調度tasklet
已調度的tasklet存放在兩個單處理器數據結構:task_vec(普通tasklet)和tasklet_hi_vec(高優先級的tasklet)中。這兩個數據結構都是由tasklet_struct結構體構成的鏈表。鏈表中的每個tasklet_struct代表一個不同的tasklet。
tasklet由tasklet_schedule()和tasklet_hi_schedule()函數進行調度,它們接受一個指向tasklet_struct結構的指針作為參數。兩個函數非常類似(區別在于一個使用TASKLET_SOFTIRQ而另一個用HI_SOFTIRQ)
所有的tasklet都通過重復運用HI_SOFTIRQ和TASKLET_SOFTIRQ這兩個軟中斷實現的。當一個tasklet被調度時,內核就會喚起這兩個軟中斷的一個。隨后,該軟中斷會被特定的函數處理,執行所有已調度的tasklet。同一時間只有一個給定類型的tasklet會被執行(但其他不同類型的tasklet可以被同時執行)。
使用tasklet
- 聲明你自己的tasklet
既可以靜態創建tasklet,也可以動態創建它。選擇那種方式取決于你到時是對tasklet的直接引用還是間接引用。如果你真準備靜態創建一個tasklet(直接引用),使用include/linux/interrupt.h中定義的兩個宏中一個:
#define DECLARE_TASKLET(name, func, data) \
struct tasklet_struct name = { NULL, 0, ATOMIC_INIT(0), func, data }#define DECLARE_TASKLET_DISABLED(name, func, data) \
struct tasklet_struct name = { NULL, 0, ATOMIC_INIT(1), func, data }
這兩個宏都能根據給定的名稱靜態創建一個tasklet_struct結構。當該tasklet被調度以后,給定的函數func會執行,它的參數由data給出。這兩個宏之間的區別在于引用計數器的初始值設置不同,DECLARE_TASKLET設置為0,該tasklet處于激活狀態;DECLARE_TASKLET_DISABLED的引用計數器為1,該tasklet處于禁止狀態。
還可以通過將一個間接引用(一個指針)賦給一個動態創建的tasklet_struct結構的方式來初始化一個tasklet。
void tasklet_init(struct tasklet_struct *t,void (*func)(unsigned long), unsigned long data);
- 編寫你自己的tasklet處理程序
tasklet的處理程序必須符合規定的函數類型:
void tasklet_handler(unsigned long data);
因為是靠軟中斷實現,所以tasklet不能睡眠。這意味著你不能在tasklet中使用信號量或者其他什么阻塞式的函數。
- 調度你自己的tasklet
通過tasklet_schedule()函數并傳遞給它相應的tasklet_struct的指針,該tasklet就會被調度以便執行。
tasklet_schedule(&my_tasklet);
tasklet被調度以后,只要有機會它就會盡可能早地運行。
可以調用task_disable()函數來禁止某個指定的tasklet。如果該tasklet正在運行,這個函數會等到它執行完畢再返回。也可以調用tasklet_diable_nosync()函數,它也可用來禁止指定的tasklet,不過無須在返回前等待tasklet執行完畢。這么做往往不太安全,因為無法估計該tasklet是否仍在執行。調用tasklet_enable()函數可以激活一個tasklet。
你可以通過調用tasklet_kill()函數從掛起的隊列中去掉一個tasklet。
void tasklet_kill(struct tasklet_struct *t);
該函數的參數是一個指向某個tasklet的tasklet_struct的長指針。這個函數首先等待該tasklet執行完畢,然后再將它移去。
ksoftirqd
每個處理器都有一組輔助軟中斷的內核線程。當內核中出現大量軟中斷的時候,這些內核進程就會輔助處理它們。
軟中斷被觸發的頻率有時會很高,處理函數還會自行重復觸發。當一個軟中斷執行的時候,它可以重新觸發自己以便再次得到執行。如果軟中斷本身出現的頻率就高,再加上它們又有將自己重新設置為可執行狀態的能力,那么就會導致用戶空間進程無法獲得足夠的處理器時間,因而處于饑餓狀態。
在設計軟中斷時,開發者需要一些折中。最終在內核中實現的方案是不會立即處理重新觸發的軟中斷。而作為改進,當大量軟中斷出現的時候,內核會喚醒一組內核線程來處理這些負責,這些線程在最低的優先級上運行(nice值為19),這能避免它們跟其他重要的任務搶奪資源。但它們最終肯定會被執行,所以,這個折中方案能夠保證軟中斷負擔很重的時候用戶程序不會因為得不到處理時間而處于饑餓狀態。相應的,也能保證“過量”的軟中斷 終究會得到處理。
每個處理器都有一個這樣的線程。所有線程的名字都叫做ksoftirqd/n,區別在于n,它是對于的處理器的編號。在一個雙CPU的機器上就有兩個這樣的線程,分別叫做ksoftirqd/0和ksoftirqd/1。為了保證只要有空閑的處理器,它們就會處理軟中斷,所以給每個處理器都分配一個這樣的線程。一旦線程被初始化,就會執行類似下面的死循環。
只要有待處理的軟中斷,ksoftirqd就會調用do_softirq()去處理它們。通過重復執行這樣的操作,重新觸發的軟中斷也會被執行。每次迭代后都會調用schedule()以便讓更重要的進程得到處理機會。當所有需要執行的操作都完成以后,該內核線程將自己設置為TASK_INTERRUPTIBLE狀態,喚起調度程序選擇其他可執行進程投入運行。
只要do_softirq()函數發現已經執行過的內核線程重新觸發了它自己,軟中斷內核線程就會被喚醒。
4 工作隊列
工作隊列是另外一種將工作推后執行的形式,工作隊列把工作推后,交由一個內核線程去執行,然后在進程上下文執行。這樣,通過工作隊列執行的代碼能占盡進程上下文的所有優勢。最重要的就是工作隊列允許重新調度甚至是睡眠。
工作隊列的實現
工作隊列子系統是一個用于創建內核線程的接口,通過它創建的進程負責執行由內核其他部分排到隊列里的任務。它創建的這些內核線程被稱作工作者線程。工作隊列可以讓你的驅動程序創建一個專門的工作者線程來處理需要推后的工作。工作隊列子系統提供了一個默認的工作者線程來處理這些工作。
默認的工作者線程叫做events/n,這里的n是處理器編號。每個處理器對應一個線程。比如,單處理器的系統只有events/0這樣一個線程。而雙處理器的系統就會多一個events/1線程。
- 表示線程的數據結構
工作者線程用workqueue_struct結構表示,定義在kernel/workqueue.c:
/** The externally visible workqueue abstraction is an array of* per-CPU workqueues:*/
struct workqueue_struct {struct cpu_workqueue_struct cpu_wq[NR_CPUS];const char *name;struct list_head list; /* Empty if single thread */
};
該結構內的cpu_workqueue_struct ,定義在kernel/workqueue.c中,cpu_workqueue_struct的每一項對應系統中的一個處理器。由于系統中每個處理器對應一個工作者線程,對于計算機來說,就是每個處理器。每個工作者線程對應一個這樣的cpu_workqueue_struct結構體。cpu_workqueue_struct是kernel/workqueue.c中的數據結構:
/** The per-CPU workqueue (if single thread, we always use cpu 0's).** The sequence counters are for flush_scheduled_work(). It wants to wait* until until all currently-scheduled works are completed, but it doesn't* want to be livelocked by new, incoming ones. So it waits until* remove_sequence is >= the insert_sequence which pertained when* flush_scheduled_work() was called.*/
struct cpu_workqueue_struct {spinlock_t lock; /* 鎖,來保護該結構體 */long remove_sequence; /* 最近一個 */long insert_sequence; /* Next to add */struct list_head worklist; /* 工作列表 */wait_queue_head_t more_work;wait_queue_head_t work_done;struct workqueue_struct *wq;task_t *thread;int run_depth; /* Detect run_workqueue() recursion depth */
} ____cacheline_aligned;
每個工作者線程關聯一個自己的workqueue_struct。在該結構體里面,給每個線程分配一個cpu_workqueue_struct,因而也就是給每個處理器分配一個。
- 表示工作的數據結構
所有的工作者線程都是用普通的內核線程實現的,它們都要執行worker_thread()函數。在它初始化以后,這個函數執行一個死循環并開始休眠。當有操作被插入到隊列里的時候,線程就會被喚醒,當沒有剩余的操作時,它又會繼續休眠。
工作用include/linux/workqueue.h中定義的work_struct結構體表示:
struct work_struct {unsigned long pending; /* 這個工作是否正在等待處理 */struct list_head entry; /* 連接所有工作的鏈表 */void (*func)(void *); /* 處理函數 */void *data; /* 處理函數的參數 */void *wq_data; /* 內部使用 */struct timer_list timer; /* 延遲的工作隊列所用到的處理器 */
};
這樣結構體被連接成鏈表,在每個處理器上的每種類型的隊列都對應這樣一個鏈表。當一個工作者線程被喚醒時,它會執行它的鏈表上的所有工作。工作被執行完畢,它就將相應的work_struct對象從鏈表中刪除。當鏈表不再有對象的時候,它就會繼續休眠。
- run_workqueue()
下一步,由run_workqueue()函數來實際完成推后到此的工作:
static inline void run_workqueue(struct cpu_workqueue_struct *cwq)
{unsigned long flags;/** Keep taking off work from the queue until* done.*/spin_lock_irqsave(&cwq->lock, flags);cwq->run_depth++;if (cwq->run_depth > 3) {/* morton gets to eat his hat */printk("%s: recursion depth exceeded: %d\n",__FUNCTION__, cwq->run_depth);dump_stack();}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);spin_unlock_irqrestore(&cwq->lock, flags);BUG_ON(work->wq_data != cwq);clear_bit(0, &work->pending);f(data);spin_lock_irqsave(&cwq->lock, flags);cwq->remove_sequence++;wake_up(&cwq->work_done);}cwq->run_depth--;spin_unlock_irqrestore(&cwq->lock, flags);
}
該函數循環遍歷鏈表上每個待處理的工作,執行鏈表每個節點上的work_struct中的func成員函數:
- 當鏈表不為空時,選取下一個節點對象
- 獲取我們希望執行的函數func以及參數data
- 把該節點從鏈表上解下來,將待處理標志位pending清0
- 調用函數
- 重復執行
工作、工作隊列和工作者線程之間的關系
我們先看一張圖,該圖闡述了這幾個數據結構的邏輯關系
位于最高一層的是工作者線程。系統允許有多種類型的工作者線程存在。對于指定的一個類型,系統的每個CPU上都有一個該類的工作者線程。在默認情況下內核只有events這一種類型的工作者線程。每個工作者線程都由一個cpu_workqueue_struct結構體表示。而workqueue_struct結構體則表示給定類型的所有工作者線程。比如,在系統默認的通用events工作者類型之外,我們自己加入了一種falcon工作者類型,并且使用的是一個擁有四個處理器的計算久,那么,系統中現在有四個events類型的線程(也就有四個cpu_workqueue_struct結構體)和四個falcon類型的線程(會有另外四個cpu_workqueue_struct結構體)。同時,又一個對應events類型的workqueue_struct和一個對應falcon類型的workqueue_struct。
工作處于最低一層,你的驅動程序創建這些需要推后執行的工作。它們用work_struct結構表示。這個結構體中最重要的部分是一個指針,它指向一個函數,而正是該函數負責處理需要推后執行的具體任務。工作會被提交給某個具體的工作者線程,然后工作者線程會被喚醒并執行這些排好的工作。
使用工作隊列
- 創建推后的工作
首先要做的是實際創建一些需要推后完成的工作。可以通過DECLARE_WORK在編譯時靜態地創建該結構體:
DECLARE_WORK(name,void (*func)(void *),void *data);
這樣就會靜態創建一個名為name,處理函數為func,參數為data的work_struct結構體。
同樣,也可以在運行時通過指針創建一個工作:
INIT_WORK(struct work_struct *work,void (*func) (void *),void *data);
- 工作隊列處理函數
工作隊列處理函數的原型是:
void work_handler(void *data);
這個函數會由一個工作者線程執行,因此,函數會運行在進程上下文中。
- 對工作進行調度
現在工作已經被創建,我們可以調度它了。想要把給定工作的處理函數提交給默認的events工作線程,只需調用:
schedule_work(&work);
work馬上就會被調度,一旦其所在的處理器上的工作者線程被喚醒,它就會被執行。有時候我們不希望工作立馬執行,而是希望它經過一段延遲后在執行。這種情況下,我們可以調用schedule_delayed_work函數將工作在指定的時間執行:
schedule_delayed_work(&work,delay);
work指向的work_struct知道delay指定的時鐘節拍用完后才會執行。
- 刷新操作
插入隊列的工作會在工作者線程下一次被喚醒的時候執行。有時,在繼續下一步工作之前,你必須保證一些操作已經執行完畢了。處于以上目的,內核準備了一個用于刷新指定工作隊列的函數:
void flush_scheduled_work(void);
函數會一直等待,知道隊列中所有對象都被執行以后才返回,該函數并不取消任何延遲執行的工作。取消延遲執行的工作應該調用:
int cancel_delayed_work(struct work_struct *work);
這個函數可以取消任何與work_struct相關的掛起工作。
- 創建新的工作隊列
如果默認的隊列不能滿足你的需求,你應該創建一個新的工作隊列和與之相應的工作者線程。由于這么做會在每個處理器上都創建一個工作者線程,所以只有在你明確了必須要靠自己的一套線程來提高性能的情況下,再創建自己的工作隊列。
創建一個新的任務隊列和與之相關的工作者線程,只需調用一個簡單的函數:
struct workqueue_struct *create_workqueue(const char *name);
name用于該內核線程的命名。比如,默認events隊列的創建就調用的是:
struct workqueue_struct *keventd_wq;
keventd_wq=create_workqueue("events");
這個函數會創建所有的工作者線程(系統中每個處理器都有一個)并且做好所有開始處理工作之前的準備工作。
創建一個工作的時候無需考慮工作隊列的類型。在創建之后,我們可以調用下面的函數,將工作加到指定的工作隊列上。
你可以調用下面的函數刷新指定的工作隊列,確保所有指定工作隊列上的任務都已經完成:
flush_workqueue(struct workqueue_struct *wq);
5 下半部機制的選擇
在內核2.6版本中,有三種選擇:軟中斷、tasklet和工作隊列。tasklet基于軟中斷實現,所以二者很相近。工作隊列機制與它們完全不同,它靠內核線程實現。
從設計角度考慮,軟中斷提供的執行序列化的保障最少。這就要求軟中斷處理函數必須采取一些步驟確保共享數據的安全,因為兩個甚至更多相同類別的軟中斷有可能在不同的處理器上同時運行。
如果你需要把任務推后到進程上下文中完成,那么在這三者就只能選擇工作隊列了。如果明確的說,并不需要睡眠,那么軟中斷和tasklet可能更合適。
如果說到易于使用,工作隊列當仁不讓。使用默認的events隊列非常簡單。接下來是tasklet,它的接口也很簡單。最后才是軟中斷,它必須靜態創建,并且需要慎重考慮其實現。
下表是對三種下半部接口的比較:
簡單地說,一般地驅動程序的編寫者需要做出兩個選擇。首先,你是不是需要一個休眠的功能?要是有,工作隊列就是唯一選擇。否則最好用tasklet。要是必須專注于性能的提高,那么就考慮軟中斷吧。
6 在下半部之間加鎖
如果需要禁止所有的下半部處理(就是所有的軟中斷和所有的tasklet),可以調用local_bh_disable()函數。允許下半部進行處理,可以調用local_bh_enbale()函數。
這些函數有可能被嵌套使用,最后被調用的local_bh_enable()最終激活下半部。比如,第一次調用local_bh_disable函數,則本地軟中斷處理被禁止。如果local_bh_disable被調用三次,則本地處理仍然被禁止,只有當第四次調用local_bh_enbale函數時,軟中斷才被最新激活。
這兩個函數并不能禁止工作隊列的執行,因為工作隊列是在進程上下文運行。