Tasklets 机制浅析
一 、Tasklets 機(jī)制基礎(chǔ)知識(shí)點(diǎn)
1、Taklets 機(jī)制概念
? ? ??Tasklets 機(jī)制是linux中斷處理機(jī)制中的軟中斷延遲機(jī)制。通常用于減少中斷處理的時(shí)間，將本應(yīng)該是在中斷服務(wù)程序中完成的任務(wù)轉(zhuǎn)化成軟中斷完成。
? ? ? 為了最大程度的避免中斷處理時(shí)間過長(zhǎng)而導(dǎo)致中斷丟失，有時(shí)候我們需要把一些在中斷處理中不是非常緊急的任務(wù)放在后面執(zhí)行，而讓中斷處理程序盡快返回。在老版本的 linux 中通常將中斷處理分為 top half handler 、 bottom half handler 。利用 top half handler 處理中斷必須處理的任務(wù)，而 bottom half handler 處理不是太緊急的任務(wù)。
? ? ? 但是 linux2.6 以后的 linux 采取了另外一種機(jī)制，就是軟中斷來代替 bottom half handler 的處理。而 tasklet 機(jī)制正是利用軟中斷來完成對(duì)驅(qū)動(dòng) bottom half 的處理。 Linux2.6 中軟中斷通常只有固定的幾種： HI_SOFTIRQ( 高優(yōu)先級(jí)的 tasklet ，一種特殊的 tasklet) 、 TIMER_SOFTIRQ （定時(shí)器）、 NET_TX_SOFTIRQ （網(wǎng)口發(fā)送）、 NET_RX_SOFTIRQ （網(wǎng)口接收） 、 BLOCK_SOFTIRQ （塊設(shè)備）、 TASKLET_SOFTIRQ （普通 tasklet ）。當(dāng)然也可以通過直接修改內(nèi)核自己加入自己的軟中斷，但是一般來說這是不合理的，軟中斷的優(yōu)先級(jí)比較高，如果不是在內(nèi)核處理頻繁的任務(wù)不建議使用。通常驅(qū)動(dòng)用戶使用 tasklet 足夠了。
? ? ? 機(jī)制流程：當(dāng)linux接收到硬件中斷之后，通過 tasklet 函數(shù)來設(shè)定軟中斷被執(zhí)行的優(yōu)先程度從而導(dǎo)致軟中斷處理函數(shù)被優(yōu)先執(zhí)行的差異性。
? ? ? 特點(diǎn)：tasklet的優(yōu)先級(jí)別較低，而且中斷處理過程中可以被打斷。但被打斷之后，還能進(jìn)行自我恢復(fù)，斷點(diǎn)續(xù)運(yùn)行。
?
2、Tasklets 解決什么問題？
a -- tasklet是I/O驅(qū)動(dòng)程序中實(shí)現(xiàn)可延遲函數(shù)的首選方法；
b -- tasklet和工作隊(duì)列是延期執(zhí)行工作的機(jī)制，其實(shí)現(xiàn)基于軟中斷，但他們更易于使用，因而更適合與設(shè)備驅(qū)動(dòng)程序...tasklet是“小進(jìn)程”，執(zhí)行一些迷你任務(wù)，對(duì)這些人物使用全功能進(jìn)程可能比較浪費(fèi)。
c -- tasklet是并行可執(zhí)行（但是是鎖密集型的）軟件中斷和舊下半?yún)^(qū)的一種混合體，這里既談不上并行性，也談不上性能。引入tasklet是為了替代原來的下半?yún)^(qū)。
? ? ??軟中斷是將操作推遲到未來時(shí)刻執(zhí)行的最有效的方法。但該延期機(jī)制處理起來非常復(fù)雜。因?yàn)槎鄠€(gè)處理器可以同時(shí)且獨(dú)立的處理軟中斷，同一個(gè)軟中斷的處理程序可以在幾個(gè)CPU上同時(shí)運(yùn)行。對(duì)軟中斷的效率來說，這是一個(gè)關(guān)鍵，多處理器系統(tǒng)上的網(wǎng)絡(luò)實(shí)現(xiàn)顯然受惠于此。但處理程序的設(shè)計(jì)必須是完全可重入且線程安全的。另外，臨界區(qū)必須用自旋鎖保護(hù)（或其他IPC機(jī)制），而這需要大量審慎的考慮。
? ? ?我自己的理解，由于軟中斷以ksoftirqd的形式與用戶進(jìn)程共同調(diào)度，這將關(guān)系到OS整體的性能，因此軟中斷在Linux內(nèi)核中也僅僅就幾個(gè)（網(wǎng)絡(luò)、時(shí)鐘、調(diào)度以及Tasklet等），在內(nèi)核編譯時(shí)確定。軟中斷這種方法顯然不是面向硬件驅(qū)動(dòng)的，而是驅(qū)動(dòng)更上一層：不關(guān)心如何從具體的網(wǎng)卡接收數(shù)據(jù)包，但是從所有的網(wǎng)卡接收的數(shù)據(jù)包都要經(jīng)過內(nèi)核協(xié)議棧的處理。而且軟中斷比較“硬”——數(shù)量固定、編譯時(shí)確定、操作函數(shù)必須可重入、需要慎重考慮鎖的問題，不適合驅(qū)動(dòng)直接調(diào)用，因此Linux內(nèi)核為驅(qū)動(dòng)直接提供了一種使用軟中斷的方法，就是tasklet。
軟中斷和 tasklet 的關(guān)系如下圖：
?
? ? ? ?上圖可以看出， ksoftirqd 是一個(gè)后臺(tái)運(yùn)行的內(nèi)核線程，它會(huì)周期的遍歷軟中斷的向量列表，如果發(fā)現(xiàn)哪個(gè)軟中斷向量被掛起了（ pend ），就執(zhí)行對(duì)應(yīng)的處理函數(shù)，對(duì)于 tasklet 來說，此處理函數(shù)就是 tasklet_action ，這個(gè)處理函數(shù)在系統(tǒng)啟動(dòng)時(shí)初始化軟中斷的就掛接了。Tasklet_action 函數(shù)，遍歷一個(gè)全局的 tasklet_vec 鏈表（此鏈表對(duì)于 SMP 系統(tǒng)是每個(gè) CPU 都有一個(gè)），此鏈表中的元素為 tasklet_struct 。下面將介紹各個(gè)函數(shù)
?
二、tasklet數(shù)據(jù)結(jié)構(gòu)
? ? ? ? tasklet通過軟中斷實(shí)現(xiàn)，軟中斷中有兩種類型屬于tasklet，分別是級(jí)別最高的HI_SOFTIRQ和TASKLET_SOFTIRQ。
? ? ? ?Linux內(nèi)核采用兩個(gè)PER_CPU的數(shù)組tasklet_vec[]和tasklet_hi_vec[]維護(hù)系統(tǒng)種的所有tasklet（kernel/softirq.c），分別維護(hù)TASKLET_SOFTIRQ級(jí)別和HI_SOFTIRQ級(jí)別的tasklet：
struct tasklet_head
{
struct tasklet_struct *head;
struct tasklet_struct *tail;
};
static DEFINE_PER_CPU(struct tasklet_head, tasklet_vec);
static DEFINE_PER_CPU(struct tasklet_head, tasklet_hi_vec);
?
tasklet的核心結(jié)構(gòu)體如下（include/linux/interrupt.h）：
struct tasklet_struct
{
struct tasklet_struct *next;
unsigned long state;
atomic_t count;
void (*func)(unsigned long);
unsigned long data;
};
?
各成員的含義如下：
a -- next指針：指向下一個(gè)tasklet的指針。
b -- state：定義了這個(gè)tasklet的當(dāng)前狀態(tài)。這一個(gè)32位的無符號(hào)長(zhǎng)整數(shù)，當(dāng)前只使用了bit［1］和bit［0］兩個(gè)狀態(tài)位。其中，bit［1］＝1表示這個(gè)tasklet當(dāng)前正在某個(gè)CPU上被執(zhí)行，它僅對(duì)SMP系統(tǒng)才有意義，其作用就是為了防止多個(gè)CPU同時(shí)執(zhí)行一個(gè)tasklet的情形出現(xiàn)；bit［0］＝1表示這個(gè)tasklet已經(jīng)被調(diào)度去等待執(zhí)行了。對(duì)這兩個(gè)狀態(tài)位的宏定義如下所示（interrupt.h）
enum
{
TASKLET_STATE_SCHED,
TASKLET_STATE_RUN
};
?
TASKLET_STATE_SCHED置位表示已經(jīng)被調(diào)度（掛起），也意味著tasklet描述符被插入到了tasklet_vec和tasklet_hi_vec數(shù)組的其中一個(gè)鏈表中，可以被執(zhí)行。TASKLET_STATE_RUN置位表示該tasklet正在某個(gè)CPU上執(zhí)行，單個(gè)處理器系統(tǒng)上并不校驗(yàn)該標(biāo)志，因?yàn)闆]必要檢查特定的tasklet是否正在運(yùn)行。
?
c -- 原子計(jì)數(shù)count：對(duì)這個(gè)tasklet的引用計(jì)數(shù)值。NOTE！只有當(dāng)count等于0時(shí)，tasklet代碼段才能執(zhí)行，也即此時(shí)tasklet是被使能的；如果count非零，則這個(gè)tasklet是被禁止的。任何想要執(zhí)行一個(gè)tasklet代碼段的人都首先必須先檢查其count成員是否為0。
d -- 函數(shù)指針func：指向以函數(shù)形式表現(xiàn)的可執(zhí)行tasklet代碼段。
e -- data：函數(shù)func的參數(shù)。這是一個(gè)32位的無符號(hào)整數(shù)，其具體含義可供func函數(shù)自行解釋，比如將其解釋成一個(gè)指向某個(gè)用戶自定義數(shù)據(jù)結(jié)構(gòu)的地址值。
?
?
三、tasklet操作接口
? ? ? ?tasklet對(duì)驅(qū)動(dòng)開放的常用操作包括：
a -- 初始化，tasklet_init()，初始化一個(gè)tasklet描述符。
b -- 調(diào)度，tasklet_schedule()和tasklet_hi_schedule()，將taslet置位TASKLET_STATE_SCHED，并嘗試激活所在的軟中斷。
c -- 禁用/啟動(dòng)，tasklet_disable_nosync()、tasklet_disable()、task_enable()，通過count計(jì)數(shù)器實(shí)現(xiàn)。
d -- 執(zhí)行，tasklet_action()和tasklet_hi_action()，具體的執(zhí)行軟中斷。
e -- 殺死，tasklet_kill()
? ? ? ?即驅(qū)動(dòng)程序在初始化時(shí)，通過函數(shù)task_init建立一個(gè)tasklet，然后調(diào)用函數(shù)tasklet_schedule將這個(gè)tasklet放在 tasklet_vec鏈表的頭部，并喚醒后臺(tái)線程ksoftirqd。當(dāng)后臺(tái)線程ksoftirqd運(yùn)行調(diào)用__do_softirq時(shí)，會(huì)執(zhí)行在中斷向量表softirq_vec里中斷號(hào)TASKLET_SOFTIRQ對(duì)應(yīng)的tasklet_action函數(shù)，然后tasklet_action遍歷 tasklet_vec鏈表，調(diào)用每個(gè)tasklet的函數(shù)完成軟中斷操作。
1、tasklet_int()函數(shù)實(shí)現(xiàn)如下（kernel/softirq.c）
? ? ?用來初始化一個(gè)指定的tasklet描述符
?
void tasklet_init(struct tasklet_struct *t,void (*func)(unsigned long), unsigned long data)
{
t->next = NULL;
t->state = 0;
atomic_set(&t->count, 0);
t->func = func;
t->data = data;
}
?
?
2、tasklet_schedule()函數(shù)
? ? ? 與tasklet_hi_schedule()函數(shù)的實(shí)現(xiàn)很類似，這里只列tasklet_schedule()函數(shù)的實(shí)現(xiàn)（kernel/softirq.c），都挺明白就不描述了：
static inline void tasklet_schedule(struct tasklet_struct *t)
{
if (!test_and_set_bit(TASKLET_STATE_SCHED, &t->state))
__tasklet_schedule(t);
}
void __tasklet_schedule(struct tasklet_struct *t)
{
unsigned long flags;
local_irq_save(flags);
t->next = NULL;
*__this_cpu_read(tasklet_vec.tail) = t;
__this_cpu_write(tasklet_vec.tail, &(t->next));
raise_softirq_irqoff(TASKLET_SOFTIRQ);
local_irq_restore(flags);
}
?
該函數(shù)的參數(shù)t指向要在當(dāng)前CPU上被執(zhí)行的tasklet。對(duì)該函數(shù)的NOTE如下：
a -- 調(diào)用test_and_set_bit()函數(shù)將待調(diào)度的tasklet的state成員變量的bit［0］位（也即TASKLET_STATE_SCHED位）設(shè)置為1，該函數(shù)同時(shí)還返回TASKLET_STATE_SCHED位的原有值。因此如果bit［0］為的原有值已經(jīng)為1，那就說明這個(gè)tasklet已經(jīng)被調(diào)度到另一個(gè)CPU上去等待執(zhí)行了。由于一個(gè)tasklet在某一個(gè)時(shí)刻只能由一個(gè)CPU來執(zhí)行，因此tasklet_schedule()函數(shù)什么也不做就直接返回了。否則，就繼續(xù)下面的調(diào)度操作。
b -- 首先，調(diào)用local_irq_save()函數(shù)來關(guān)閉當(dāng)前CPU的中斷，以保證下面的步驟在當(dāng)前CPU上原子地被執(zhí)行。
c -- 然后，將待調(diào)度的tasklet添加到當(dāng)前CPU對(duì)應(yīng)的tasklet隊(duì)列的首部。
d -- 接著，調(diào)用__cpu_raise_softirq()函數(shù)在當(dāng)前CPU上觸發(fā)軟中斷請(qǐng)求TASKLET_SOFTIRQ。
e -- 最后，調(diào)用local_irq_restore()函數(shù)來開當(dāng)前CPU的中斷。
?
3、tasklet_disable()函數(shù)、task_enable()函數(shù)以及tasklet_disable_nosync()函數(shù)（include/linux/interrupt.h）
? ? ??使能與禁止操作往往總是成對(duì)地被調(diào)用的
static inline void tasklet_disable_nosync(struct tasklet_struct *t)
{
atomic_inc(&t->count);
smp_mb__after_atomic_inc();
}
static inline void tasklet_disable(struct tasklet_struct *t)
{
tasklet_disable_nosync(t);
tasklet_unlock_wait(t);
smp_mb();
}
static inline void tasklet_enable(struct tasklet_struct *t)
{
smp_mb__before_atomic_dec();
atomic_dec(&t->count);
}
?
?
4、tasklet_action()函數(shù)在softirq_init()函數(shù)中被調(diào)用：
void __init softirq_init(void)
{
...
open_softirq(TASKLET_SOFTIRQ, tasklet_action);
open_softirq(HI_SOFTIRQ, tasklet_hi_action);
}
?
tasklet_action()函數(shù)
static void tasklet_action(struct softirq_action *a)
{
struct tasklet_struct *list;
local_irq_disable();
list = __this_cpu_read(tasklet_vec.head);
__this_cpu_write(tasklet_vec.head, NULL);
__this_cpu_write(tasklet_vec.tail, &__get_cpu_var(tasklet_vec).head);
local_irq_enable();
while (list)
{
struct tasklet_struct *t = list;
list = list->next;
if (tasklet_trylock(t))
{
if (!atomic_read(&t->count))
{
if (!test_and_clear_bit(TASKLET_STATE_SCHED, &t->state))
BUG();
t->func(t->data);
tasklet_unlock(t);
continue;
}
tasklet_unlock(t);
}
local_irq_disable();
t->next = NULL;
*__this_cpu_read(tasklet_vec.tail) = t;
__this_cpu_write(tasklet_vec.tail, &(t->next));
__raise_softirq_irqoff(TASKLET_SOFTIRQ);
local_irq_enable();
}
}
注釋如下：
①首先，在當(dāng)前CPU關(guān)中斷的情況下，“原子”地讀取當(dāng)前CPU的tasklet隊(duì)列頭部指針，將其保存到局部變量list指針中，然后將當(dāng)前CPU的tasklet隊(duì)列頭部指針設(shè)置為NULL，以表示理論上當(dāng)前CPU將不再有tasklet需要執(zhí)行（但最后的實(shí)際結(jié)果卻并不一定如此，下面將會(huì)看到）。
②然后，用一個(gè)while{}循環(huán)來遍歷由list所指向的tasklet隊(duì)列，隊(duì)列中的各個(gè)元素就是將在當(dāng)前CPU上執(zhí)行的tasklet。循環(huán)體的執(zhí)行步驟如下：
a -- 用指針t來表示當(dāng)前隊(duì)列元素，即當(dāng)前需要執(zhí)行的tasklet。
b -- 更新list指針為list->next，使它指向下一個(gè)要執(zhí)行的tasklet。
c -- 用tasklet_trylock()宏試圖對(duì)當(dāng)前要執(zhí)行的tasklet（由指針t所指向）進(jìn)行加鎖
? ? ??如果加鎖成功（當(dāng)前沒有任何其他CPU正在執(zhí)行這個(gè)tasklet），則用原子讀函atomic_read()進(jìn)一步判斷count成員的值。如果count為0，說明這個(gè)tasklet是允許執(zhí)行的，于是：
? ?（1）先清除TASKLET_STATE_SCHED位；
? ?（2）然后，調(diào)用這個(gè)tasklet的可執(zhí)行函數(shù)func；
? ?（3）執(zhí)行barrier()操作；
? ?（4）調(diào)用宏tasklet_unlock()來清除TASKLET_STATE_RUN位。
? （5）最后，執(zhí)行continue語句跳過下面的步驟，回到while循環(huán)繼續(xù)遍歷隊(duì)列中的下一個(gè)元素。如果count不為0，說明這個(gè)tasklet是禁止運(yùn)行的，于是調(diào)用tasklet_unlock()清除前面用tasklet_trylock()設(shè)置的TASKLET_STATE_RUN位。
? ??如果tasklet_trylock()加鎖不成功，或者因?yàn)楫?dāng)前tasklet的count值非0而不允許執(zhí)行時(shí)，我們必須將這個(gè)tasklet重新放回到當(dāng)前CPU的tasklet隊(duì)列中，以留待這個(gè)CPU下次服務(wù)軟中斷向量TASKLET_SOFTIRQ時(shí)再執(zhí)行。為此進(jìn)行這樣幾步操作：
? （1）先關(guān)CPU中斷，以保證下面操作的原子性。
? （2）把這個(gè)tasklet重新放回到當(dāng)前CPU的tasklet隊(duì)列的首部；
? （3）調(diào)用__cpu_raise_softirq()函數(shù)在當(dāng)前CPU上再觸發(fā)一次軟中斷請(qǐng)求TASKLET_SOFTIRQ；
? （4）開中斷。
c -- 最后，回到while循環(huán)繼續(xù)遍歷隊(duì)列。
?
5、tasklet_kill()實(shí)現(xiàn)
void tasklet_kill(struct tasklet_struct *t)
{
if (in_interrupt())
printk("Attempt to kill tasklet from interruptn");
while (test_and_set_bit(TASKLET_STATE_SCHED, &t->state))
{
do {
yield();
} while (test_bit(TASKLET_STATE_SCHED, &t->state));
}
tasklet_unlock_wait(t);
clear_bit(TASKLET_STATE_SCHED, &t->state);
}
?
?
四、一個(gè)tasklet調(diào)用例子
? ? ? ?找了一個(gè)tasklet的例子看一下（drivers/usb/atm，usb攝像頭），在其自舉函數(shù)usbatm_usb_probe()中調(diào)用了tasklet_init()初始化了兩個(gè)tasklet描述符用于接收和發(fā)送的“可延遲操作處理”，但此是并沒有將其加入到tasklet_vec[]或tasklet_hi_vec[]中：
tasklet_init(&instance->rx_channel.tasklet,
usbatm_rx_process, (unsigned long)instance);
tasklet_init(&instance->tx_channel.tasklet,
usbatm_tx_process, (unsigned long)instance);
? ? ? 在其發(fā)送接口usbatm_atm_send()函數(shù)調(diào)用tasklet_schedule()函數(shù)將所初始化的tasklet加入到當(dāng)前cpu的tasklet_vec鏈表尾部，并嘗試調(diào)用do_softirq_irqoff()執(zhí)行軟中斷TASKLET_SOFTIRQ：
static int usbatm_atm_send(struct atm_vcc *vcc, struct sk_buff *skb)
{
...
tasklet_schedule(&instance->tx_channel.tasklet);
...
}
? ? ? 在其斷開設(shè)備的接口usbatm_usb_disconnect()中調(diào)用tasklet_disable()函數(shù)和tasklet_enable()函數(shù)重新啟動(dòng)其收發(fā)tasklet（具體原因不詳，這個(gè)地方可能就是由這個(gè)需要，暫時(shí)重啟收發(fā)tasklet）：
void usbatm_usb_disconnect(struct usb_interface *intf)
{
...
tasklet_disable(&instance->rx_channel.tasklet);
tasklet_disable(&instance->tx_channel.tasklet);
...
tasklet_enable(&instance->rx_channel.tasklet);
tasklet_enable(&instance->tx_channel.tasklet);
...
}
? ? ? 在其銷毀接口usbatm_destroy_instance()中調(diào)用tasklet_kill()函數(shù)，強(qiáng)行將該tasklet踢出調(diào)度隊(duì)列。
?
? ? ?從上述過程以及tasklet的設(shè)計(jì)可以看出，tasklet整體是這么運(yùn)行的：驅(qū)動(dòng)應(yīng)該在其硬中斷處理函數(shù)的末尾調(diào)用tasklet_schedule()接口激活該tasklet；內(nèi)核經(jīng)常調(diào)用do_softirq()執(zhí)行軟中斷，通過softirq執(zhí)行tasket，如下圖所示。圖中灰色部分為禁止硬中斷部分，為保護(hù)軟中斷pending位圖和tasklet_vec鏈表數(shù)組，count的改變均為原子操作，count確保SMP架構(gòu)下同時(shí)只有一個(gè)CPU在執(zhí)行該tasklet：
總結(jié)
以上是生活随笔為你收集整理的Tasklets 机制浅析的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： git checkout -b dev
- 下一篇： 编译乱序(Compiler Reorde