中斷處理「下半部」機制【轉】

轉自：https://blog.csdn.net/myarrow/article/details/9287169

版權聲明：本文爲博主原創文章，遵循 CC 4.0 BY-SA 版權協議，轉載請附上原文出處連接和本聲明。
本文連接：https://blog.csdn.net/MyArrow/article/details/9287169
1. 中斷處理「下半部」機制
       中斷服務程序通常都是在中斷請求關閉的條件下執行的，以免嵌套而使中斷控制複雜化。可是，中斷是一個隨機事件，它隨時會到來，若是關中斷的時間太長，CPU就不能及時響應其餘的中斷請求，從而形成中斷的丟失。

       所以，Linux內核的目標就是儘量快的處理完中斷請求，盡其所能把更多的處理向後推遲。例如，假設一個數據塊已經達到了網線，當中斷控制器接受到這個中斷請求信號時，Linux內核只是簡單地標誌數據到來了，而後讓處理器恢復到它之前運行的狀態，其他的處理稍後再進行(如把數據移入一個緩衝區，接受數據的進程就能夠在緩衝區找到數據)。

       所以，內核把中斷處理分爲兩部分：上半部(top-half)和下半部(bottom-half)，上半部 (就是中斷服務程序)內核當即執行，而下半部(就是一些內核函數)留着稍後處理。

      首先：一個快速的「上半部」來處理硬件發出的請求，它必須在一個新的中斷產生以前終止。一般，除了在設備和一些內存緩衝區(若是你的設備用到了DMA，就不止這些)之間移動或傳送數據，肯定硬件是否處於健全的狀態以外，這一部分作的工做不多。
      第二：「下半部」運行時是容許中斷請求的，而上半部運行時是關中斷的，這是兩者之間的主要區別。

      內核到底何時執行下半部，以何種方式組織下半部？

      這就是咱們要討論的下半部實現機制，這種機制在內核的演變過程當中不斷獲得改進，在之前的內核中，這個機制叫作bottom-half(如下簡稱BH)。可是，Linux的這種bottom-half機制有兩個缺點：

      1) 在任意一時刻，系統只能有一個CPU能夠執行BH代碼，以防止兩個或多個CPU同時來執行BH函數而相互干擾。所以BH代碼的執行是嚴格「串行化」的。

      2) BH函數不容許嵌套。

 　這兩個缺點在單CPU系統中是可有可無的，但在SMP系統中倒是很是致命的。由於BH機制的嚴格串行化執行顯然沒有充分利用SMP系統的多CPU特色。爲此，在2.4之後的版本中有了新的發展和改進，改進的目標使下半部能夠在多處理機上並行執行，並有助於驅動程序的開發者進行驅動程序的開發。下面主要介紹3種2.6內核中的「下半部」處理機制：

      1) 軟中斷請求(softirq)機制

      2) 小任務(tasklet)機制

      3) 工做隊列機制

      以上三種機制的比較以下圖所示：

 

2. 軟中斷請求(softirq)機制
　　Linux的softirq機制是與SMP緊密不可分的。爲此，整個softirq機制的設計與實現中自始自終都貫徹了一個思想：「誰觸發，誰執行」（Who marks，Who runs），也即觸發軟中斷的那個CPU負責執行它所觸發的軟中斷，並且每一個CPU都有它本身的軟中斷觸發與控制機制。這個設計思想也使得softirq機制充分利用了SMP系統的性能和特色。
2.1 軟中斷描述符
　　Linux在include/linux/interrupt.h頭文件中定義了數據結構softirq_action，來描述一個軟中斷請求，以下所示：
/* PLEASE, avoid to allocate new softirqs, if you need not _really_ high
frequency threaded job scheduling. For almost all the purposes
tasklets are more than enough. F.e. all serial device BHs et
al. should be converted to tasklets, not to softirqs.
*/

enum
{
HI_SOFTIRQ=0, //用於實現高優先級的軟中斷
TIMER_SOFTIRQ,
NET_TX_SOFTIRQ, // 用於網絡數據的發送
NET_RX_SOFTIRQ, // 用於網絡數據的接收
BLOCK_SOFTIRQ,
BLOCK_IOPOLL_SOFTIRQ,
TASKLET_SOFTIRQ, // 用於實現tasklet軟中斷
SCHED_SOFTIRQ,
HRTIMER_SOFTIRQ,
RCU_SOFTIRQ, /* Preferable RCU should always be the last softirq */

NR_SOFTIRQS
};

/* map softirq index to softirq name. update 'softirq_to_name' in
* kernel/softirq.c when adding a new softirq.
*/
extern char *softirq_to_name[NR_SOFTIRQS];

/* 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 *);
};

asmlinkage void do_softirq(void);
asmlinkage void __do_softirq(void);
        其中，函數指針action指向軟中斷請求的服務函數。基於上述軟中斷描述符，Linux在kernel/softirq.c文件中定義了一個全局的softirq_vec數組：
　　static struct softirq_action softirq_vec[NR_SOFTIRQS] __cacheline_aligned_in_smp;

　　在這裏系統一共定義了10個軟中斷請求描述符。軟中斷向量i（0≤i≤9）所對應的軟中斷請求描述符就是softirq_vec[i]。這個數組是個系統全局數組，即它被全部的CPU所共享。這裏須要注意的一點是：每一個CPU雖然都有它本身的觸發和控制機制，而且只執行他本身所觸發的軟中斷請求，可是各個CPU所執行的軟中斷服務例程倒是相同的，也即都是執行softirq_vec[ ]數組中定義的軟中斷服務函數。Linux在kernel/softirq.c中的相關代碼以下：

/*
- No shared variables, all the data are CPU local.
- If a softirq needs serialization, let it serialize itself
by its own spinlocks.
- Even if softirq is serialized, only local cpu is marked for
execution. Hence, we get something sort of weak cpu binding.
Though it is still not clear, will it result in better locality
or will not.

Examples:
- NET RX softirq. It is multithreaded and does not require
any global serialization.
- NET TX softirq. It kicks software netdevice queues, hence
it is logically serialized per device, but this serialization
is invisible to common code.
- Tasklets: serialized wrt itself.
*/

#ifndef __ARCH_IRQ_STAT
irq_cpustat_t irq_stat[NR_CPUS] ____cacheline_aligned;
EXPORT_SYMBOL(irq_stat);
#endif

static struct softirq_action softirq_vec[NR_SOFTIRQS] __cacheline_aligned_in_smp;

DEFINE_PER_CPU(struct task_struct *, ksoftirqd);

char *softirq_to_name[NR_SOFTIRQS] = {
"HI", "TIMER", "NET_TX", "NET_RX", "BLOCK", "BLOCK_IOPOLL",
"TASKLET", "SCHED", "HRTIMER", "RCU"
};

2.2 軟中斷觸發機制
　  要實現「誰觸發，誰執行」的思想，就必須爲每一個CPU都定義它本身的觸發和控制變量。爲此，Linux在include/asm-i386/hardirq.h頭文件中定義了數據結構irq_cpustat_t來描述一個CPU的中斷統計信息，其中就有用於觸發和控制軟中斷的成員變量。數據結構irq_cpustat_t的定義以下：

      IPI: 處理器間的中斷(Inter-Processor Interrupts)

#define NR_IPI 6

typedef struct {
unsigned int __softirq_pending;
#ifdef CONFIG_LOCAL_TIMERS
unsigned int local_timer_irqs;
#endif
#ifdef CONFIG_SMP
unsigned int ipi_irqs[NR_IPI];
#endif
} ____cacheline_aligned irq_cpustat_t;
      中斷處理的相關宏以下：
#define __inc_irq_stat(cpu, member) __IRQ_STAT(cpu, member)++
#define __get_irq_stat(cpu, member) __IRQ_STAT(cpu, member)

#define __IRQ_STAT(cpu, member) (irq_stat[cpu].member)
     irq_cpustat_tirq_stat[NR_CPUS] ____cacheline_aligned;

     1) NR_CPUS：爲系統中CPU個數。
     2) 這樣，每一個CPU都只操做它本身的中斷統計信息結構。假設有一個編號爲id的CPU，那麼它只能操做它本身的中斷統計信息結構irq_stat［id］（0≤id≤NR_CPUS-1），從而使各CPU之間互不影響。

    

    1) 觸發軟中斷函數:

        void raise_softirq(unsigned int nr)；// nr爲中斷號

    2) 設置軟中斷服務函數：

       void open_softirq(int nr, void (*action)(struct softirq_action *)); // nr爲中斷號, action爲中斷處理函數

2.3 初始化軟中斷(softirq_init) 
void __init softirq_init(void)
{
int cpu;

for_each_possible_cpu(cpu) {
int i;

per_cpu(tasklet_vec, cpu).tail =
&per_cpu(tasklet_vec, cpu).head;
per_cpu(tasklet_hi_vec, cpu).tail =
&per_cpu(tasklet_hi_vec, cpu).head;
for (i = 0; i < NR_SOFTIRQS; i++)
INIT_LIST_HEAD(&per_cpu(softirq_work_list[i], cpu));
}

register_hotcpu_notifier(&remote_softirq_cpu_notifier);

open_softirq(TASKLET_SOFTIRQ, tasklet_action); //設置軟中斷服務函數
open_softirq(HI_SOFTIRQ, tasklet_hi_action); //設置軟中斷服務函數

}

2.4 軟中斷服務的執行函數do_softirq
　　函數do_softirq()負責執行數組softirq_vec［i］中設置的軟中斷服務函數。每一個CPU都是經過執行這個函數來執行軟中斷服務的。因爲同一個CPU上的軟中斷服務例程不容許嵌套，所以，do_softirq()函數一開始就檢查當前CPU是否已經正出在中斷服務中，若是是則do_softirq()函數當即返回。舉個例子，假設CPU0正在執行do_softirq()函數，執行過程產生了一個高優先級的硬件中斷，因而CPU0轉去執行這個高優先級中斷所對應的中斷服務程序。衆所周知，全部的中斷服務程序最後都要跳轉到do_IRQ()函數並由它來依次執行中斷服務隊列中的ISR，這裏咱們假定這個高優先級中斷的ISR請求觸發了一次軟中斷，因而do_IRQ()函數在退出以前看到有軟中斷請求，從而調用do_softirq()函數來服務軟中斷請求。所以，CPU0再次進入do_softirq()函數（也即do_softirq()函數在CPU0上被重入了）。可是在這一次進入do_softirq()函數時，它立刻發現CPU0此前已經處在中斷服務狀態中了，所以這一次do_softirq()函數當即返回。因而，CPU0回到該開始時的do_softirq()函數繼續執行，併爲高優先級中斷的ISR所觸發的軟中斷請求補上一次服務。從這裏能夠看出，do_softirq()函數在同一個CPU上的執行是串行的。
asmlinkage void do_softirq(void)
{
__u32 pending;
unsigned long flags;

if (in_interrupt())
return;

local_irq_save(flags);

pending = local_softirq_pending();

if (pending)
__do_softirq();

local_irq_restore(flags);
}

3.小任務( tasklet)機制
　　tasklet機制是一種較爲特殊的軟中斷。

       tasklet一詞的原意是「小片任務」的意思，這裏是指一小段可執行的代碼，且一般以函數的形式出現。軟中斷向量HI_SOFTIRQ和TASKLET_SOFTIRQ均是用tasklet機制來實現的。
　　從某種程度上講，tasklet機制是Linux內核對BH機制的一種擴展。在2.4內核引入了softirq機制後，原有的BH機制正是經過tasklet機制這個橋樑來將softirq機制歸入總體框架中的。正是因爲這種歷史的延伸關係，使得tasklet機制與通常意義上的軟中斷有所不一樣，而呈現出如下兩個顯著的特色：

　　1) 與通常的軟中斷不一樣，某一段tasklet代碼在某個時刻只能在一個CPU上運行，而不像通常的軟中斷服務函數(即softirq_action結構中的action函數指針)那樣——在同一時刻能夠被多個CPU併發地執行。

　　2) 與BH機制不一樣，不一樣的tasklet代碼在同一時刻能夠在多個CPU上併發地執行，而不像BH機制那樣必須嚴格地串行化執行(也即在同一時刻系統中只能有一個CPU執行BH函數)。

3.1 tasklet描述符
　　Linux用數據結構tasklet_struct來描述一個tasklet，每一個結構表明一個獨立的小任務。該數據結構定義在include/linux/interrupt.h頭文件中。以下所示：

/* Tasklets --- multithreaded analogue of BHs.

Main feature differing them of generic softirqs: tasklet
is running only on one CPU simultaneously.

Main feature differing them of BHs: different tasklets
may be run simultaneously on different CPUs.

Properties:
* If tasklet_schedule() is called, then tasklet is guaranteed
to be executed on some cpu at least once after this.
* If the tasklet is already scheduled, but its execution is still not
started, it will be executed only once.
* If this tasklet is already running on another CPU (or schedule is called
from tasklet itself), it is rescheduled for later.
* Tasklet is strictly serialized wrt itself, but not
wrt another tasklets. If client needs some intertask synchronization,
he makes it with spinlocks.
*/

struct tasklet_struct
{
struct tasklet_struct *next;
unsigned long state;
atomic_t count;
void (*func)(unsigned long);
unsigned long data;
};

    • next: 指向下一個tasklet的指針
    • state: 定義了這個tasklet的當前狀態。這一個32位的無符號長整數，當前只使用了bit[1]和bit[0]兩個狀態位。其中，bit[1]＝1 表示這個tasklet當前正在某個CPU上被執行，它僅對SMP系統纔有意義，其做用就是爲了防止多個CPU同時執行一個tasklet的情形出現；bit[0]＝1表示這個tasklet已經被調度去等待執行了。
　　對這兩個狀態位的宏定義以下所示（interrupt.h）：

enum
{
TASKLET_STATE_SCHED, /* Tasklet is scheduled for execution */
TASKLET_STATE_RUN /* Tasklet is running (SMP only) */
};

    • count: 子計數count，對這個tasklet的引用計數值。
      注：只有當count等於0時，tasklet代碼段才能執行，也即此時tasklet是被使能的；若是count非零，則這個tasklet是被禁止的。任何想要執行一個tasklet代碼段的人都首先必須先檢查其count成員是否爲0。
    • func：指向以函數形式表現的可執行tasklet代碼段。
    • data：函數func的參數。這是一個32位的無符號整數，其具體含義可供func函數自行解釋，好比將其解釋成一個指向某個用戶自定義數據結構的地址值。
　　Linux在interrupt.h頭文件中又定義了兩個用來定義tasklet_struct結構變量的輔助宏：

#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 }

     顯然，從上述源代碼能夠看出，用DECLARE_TASKLET宏定義的tasklet在初始化時是被使能的（enabled），由於其count成員爲0。而用DECLARE_TASKLET_DISABLED宏定義的tasklet在初始時是被禁止的（disabled），由於其count等於1。
3.2 改變一個tasklet狀態的操做
　　在這裏，tasklet狀態指兩個方面：

       1) state：成員所表示的運行狀態；

       2) count：成員決定的使能／禁止狀態。

3.2.1 改變一個tasklet的運行狀態
　　state成員中的bit[0]表示一個tasklet是否已被調度去等待執行，bit[1]表示一個tasklet是否正在某個CPU上執行。對於state變量中某位的改變必須是一個原子操做，所以能夠用定義在include/asm/bitops.h頭文件中的位操做來進行。
　　因爲bit[1]這一位（即TASKLET_STATE_RUN）僅僅對於SMP系統纔有意義，所以Linux在Interrupt.h頭文件中顯示地定義了對TASKLET_STATE_RUN位的操做。以下所示：
#ifdef CONFIG_SMP
static inline int tasklet_trylock(struct tasklet_struct *t)
{
return !test_and_set_bit(TASKLET_STATE_RUN, &(t)->state);
}

static inline void tasklet_unlock(struct tasklet_struct *t)
{
smp_mb__before_clear_bit();
clear_bit(TASKLET_STATE_RUN, &(t)->state);
}

static inline void tasklet_unlock_wait(struct tasklet_struct *t)
{
while (test_bit(TASKLET_STATE_RUN, &(t)->state)) { barrier(); }
}
#else
#define tasklet_trylock(t) 1
#define tasklet_unlock_wait(t) do { } while (0)
#define tasklet_unlock(t) do { } while (0)
#endif

           顯然，在SMP系統同，tasklet_trylock()宏將把一個tasklet_struct結構變量中的state成員中的bit[1]位設置成1，同時還返回bit[1]位的非。所以，若是bit[1]位原有值爲1（表示另一個CPU正在執行這個tasklet代碼），那麼tasklet_trylock()宏將返回值0，也就表示上鎖不成功。若是bit[1]位的原有值爲0，那麼tasklet_trylock()宏將返回值1，表示加鎖成功。而在單CPU系統中，tasklet_trylock()宏老是返回爲1。
      任何想要執行某個tasklet代碼的程序都必須首先調用宏tasklet_trylock()來試圖對這個tasklet進行上鎖（即設置TASKLET_STATE_RUN位），且只能在上鎖成功的狀況下才能執行這個tasklet。建議！即便你的程序只在CPU系統上運行，你也要在執行tasklet以前調用tasklet_trylock()宏，以便使你的代碼得到良好可移植性。
　　在SMP系統中，tasklet_unlock_wait()宏將一直不停地測試TASKLET_STATE_RUN位的值，直到該位的值變爲0（即一直等待到解鎖），假如：CPU0正在執行tasklet A的代碼，在此期間，CPU1也想執行tasklet A的代碼，但CPU1發現tasklet A的TASKLET_STATE_RUN位爲1，因而它就能夠經過tasklet_unlock_wait()宏等待tasklet A被解鎖（也即TASKLET_STATE_RUN位被清零）。在單CPU系統中，這是一個空操做。
　　宏tasklet_unlock()用來對一個tasklet進行解鎖操做，也即將TASKLET_STATE_RUN位清零。在單CPU系統中，這是一個空操做。

3.2.2 使能／禁止一個tasklet
　　使能與禁止操做每每老是成對地被調用的，tasklet_disable()函數以下（interrupt.h）：

static inline void tasklet_disable(struct tasklet_struct *t)
{
tasklet_disable_nosync(t);
tasklet_unlock_wait(t);
smp_mb();
}

      函數tasklet_disable_nosync()也是一個靜態inline函數，它簡單地經過原子操做將count成員變量的值減1。以下所示（interrupt.h）：
static inline void tasklet_disable_nosync(struct tasklet_struct *t)
{
atomic_inc(&t->count);
smp_mb__after_atomic_inc();
}
      函數tasklet_enable()用於使能一個tasklet，以下所示（interrupt.h）：
static inline void tasklet_enable(struct tasklet_struct *t)
{
smp_mb__before_atomic_dec();
atomic_dec(&t->count);
}

3.3 tasklet描述符的初始化與殺死
　　函數tasklet_init()用來初始化一個指定的tasklet描述符，其源碼以下所示（kernel/softirq.c）：
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;
}

     函數tasklet_kill()用來將一個已經被調度了的tasklet殺死，即將其恢復到未調度的狀態。其源碼以下所示（kernel/softirq.c）：
void tasklet_kill(struct tasklet_struct *t)
{
if (in_interrupt())
printk("Attempt to kill tasklet from interrupt\n");

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);
}
3.4 tasklet對列
　　多個tasklet能夠經過tasklet描述符中的next成員指針連接成一個單向對列。爲此，Linux專門在頭文件include/linux/interrupt.h中定義了數據結構tasklet_head來描述一個tasklet對列的頭部指針。以下所示：
/*
* Tasklets
*/
struct tasklet_head
{
struct tasklet_struct *head;
struct tasklet_struct **tail;
};

       儘管tasklet機制是特定於軟中斷向量HI_SOFTIRQ和TASKLET_SOFTIRQ的一種實現，可是tasklet機制仍然屬於softirq機制的總體框架範圍內的，所以，它的設計與實現仍然必須堅持「誰觸發，誰執行」的思想。爲此，Linux爲系統中的每個CPU都定義了一個tasklet對列頭部，來表示應該有各個CPU負責執行的tasklet對列。以下所示（kernel/softirq.c）：
#define DEFINE_PER_CPU_SECTION(type, name, sec) \
__PCPU_ATTRS(sec) PER_CPU_DEF_ATTRIBUTES \
__typeof__(type) name

#define DEFINE_PER_CPU(type, name) \
DEFINE_PER_CPU_SECTION(type, name, "")

static DEFINE_PER_CPU(struct tasklet_head, tasklet_vec);
static DEFINE_PER_CPU(struct tasklet_head, tasklet_hi_vec);
        即：struct tasklet_head tasklet_vec[NR_CPUS] __cacheline_aligned;
　　       struct tasklet_head tasklet_hi_vec[NR_CPUS] __cacheline_aligned;
       其中，tasklet_vec［］數組用於軟中斷向量TASKLET_SOFTIRQ，而tasklet_hi_vec［］數組則用於軟中斷向量HI_SOFTIRQ。也即，若是CPUi（0≤i≤NR_CPUS-1）觸發了軟中斷向量TASKLET_SOFTIRQ，那麼對列tasklet_vec［i］中的每個tasklet都將在CPUi服務於軟中斷向量TASKLET_SOFTIRQ時被CPUi所執行。一樣地，若是CPUi（0≤i≤NR_CPUS-1）觸發了軟中斷向量HI_SOFTIRQ，那麼隊列tasklet_hi_vec［i］中的每個tasklet都將CPUi在對軟中斷向量HI_SOFTIRQ進行服務時被CPUi所執行。
　　隊列tasklet_vec［I］和tasklet_hi_vec［I］中的各個tasklet是怎樣被所CPUi所執行的呢？其關鍵就是軟中斷向量TASKLET_SOFTIRQ和HI_SOFTIRQ的軟中斷服務程序——tasklet_action()函數和tasklet_hi_action()函數。下面咱們就來分析這兩個函數。

3.5 軟中斷向量TASKLET_SOFTIRQ和HI_SOFTIRQ
　　Linux爲軟中斷向量TASKLET_SOFTIRQ和HI_SOFTIRQ實現了專用的觸發函數和軟中斷服務函數。
      • 專用的觸發函數

       tasklet_schedule()函數和tasklet_hi_schedule()函數分別用來在當前CPU上觸發軟中斷向量TASKLET_SOFTIRQ和HI_SOFTIRQ，並把指定的tasklet加入當前CPU所對應的tasklet隊列中去等待執行。

      • 專用的軟中斷服務函數

         tasklet_action()函數和tasklet_hi_action()函數則分別是軟中斷向量TASKLET_SOFTIRQ和HI_SOFTIRQ的軟中斷服務函數。在初始化函數softirq_init()中，這兩個軟中斷向量對應的描述符softirq_vec[0]和softirq_vec[6]中的action函數指針就被分別初始化成指向函數tasklet_hi_action()和函數tasklet_action()。

3.5.1 軟中斷向量TASKLET_SOFTIRQ的觸發函數tasklet_schedule
　　該函數實如今include/linux/interrupt.h頭文件中，是一個inline函數。其源碼以下所示：
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); // 觸發軟中斷TASKLET_SOFTIRQ
local_irq_restore(flags);
}

static inline void tasklet_schedule(struct tasklet_struct *t)
{
if (!test_and_set_bit(TASKLET_STATE_SCHED, &t->state))
__tasklet_schedule(t);
}

         • 調用test_and_set_bit()函數將待調度的tasklet的state成員變量的bit［0］位（也即TASKLET_STATE_SCHED位）設置爲1，該函數同時還返回TASKLET_STATE_SCHED位的原有值。所以若是bit［0］爲的原有值已經爲1，那就說明這個tasklet已經被調度到另外一個CPU上去等待執行了。因爲一個tasklet在某一個時刻只能由一個CPU來執行，所以tasklet_schedule()函數什麼也不作就直接返回了。不然，就繼續下面的調度操做。
　　  • 首先，調用local_irq_save()函數來關閉當前CPU的中斷，以保證下面的步驟在當前CPU上原子地被執行。
　      • 而後，將待調度的tasklet添加到當前CPU對應的tasklet隊列的尾部。
　　  • 接着，調用raise_softirq_irqoff函數在當前CPU上觸發軟中斷請求TASKLET_SOFTIRQ。
　　  • 最後，調用local_irq_restore()函數來開當前CPU的中斷。

3.5.2 軟中斷向量TASKLET_SOFTIRQ的服務程序tasklet_action
　　函數tasklet_action()是tasklet機制與軟中斷向量TASKLET_SOFTIRQ的聯繫紐帶。正是該函數將當前CPU的tasklet隊列中的各個tasklet放到當前CPU上來執行的。該函數實如今kernel/softirq.c文件中，其源代碼以下：

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();
}
}

       • 首先，在當前CPU關中斷的狀況下，「原子」地讀取當前CPU的tasklet隊列頭部指針，將其保存到局部變量list指針中，而後將當前CPU的tasklet隊列頭部指針設置爲NULL，以表示理論上當前CPU將再也不有tasklet須要執行（但最後的實際結果卻並不必定如此，下面將會看到）。
　　• 而後，用一個while{}循環來遍歷由list所指向的tasklet隊列，隊列中的各個元素就是將在當前CPU上執行的tasklet。循環體的執行步驟以下：
　　• 用指針t來表示當前隊列元素，即當前須要執行的tasklet。
　　• 更新list指針爲list->next，使它指向下一個要執行的tasklet。
　　• 用tasklet_trylock()宏試圖對當前要執行的tasklet（由指針t所指向）進行加鎖，若是加鎖成功（當前沒有任何其餘CPU正在執行這個tasklet），則用原子讀函數atomic_read()進一步判斷count成員的值。若是count爲0，說明這個tasklet是容許執行的，因而：
         (1) 先清除TASKLET_STATE_SCHED位；

         (2) 而後，調用這個tasklet的可執行函數func；

         (3) 調用宏tasklet_unlock()來清除TASKLET_STATE_RUN位 

         (4) 最後，執行continue語句跳過下面的步驟，回到while循環繼續遍歷隊列中的下一個元素。若是count不爲0，說明這個tasklet是禁止運行的，因而調用tasklet_unlock()清除前面用tasklet_trylock()設置的TASKLET_STATE_RUN位。

3.6 tasklet使用總結
     1) 聲明和使用小任務大多數狀況下，爲了控制一個經常使用的硬件設備，小任務機制是實現下半部的最佳選擇。小任務能夠動態建立，使用方便，執行起來也比較快。咱們既能夠靜態地建立小任務，也能夠動態地建立它。選擇那種方式取決於究竟是想要對小任務進行直接引用仍是一個間接引用。若是準備靜態地建立一個小任務（也就是對它直接引用），使用下面兩個宏中的一個：
       DECLARE_TASKLET(name,func, data)
       DECLARE_TASKLET_DISABLED(name,func, data)
      這兩個宏都能根據給定的名字靜態地建立一個tasklet_struct結構。當該小任務被調度之後，給定的函數func會被執行，它的參數由data給出。這兩個宏之間的區別在於引用計數器的初始值設置不一樣。第一個宏把建立的小任務的引用計數器設置爲0，所以，該小任務處於激活狀態。另外一個把引用計數器設置爲1，因此該小任務處於禁止狀態。例如：
        DECLARE_TASKLET(my_tasklet,my_tasklet_handler, dev);
        這行代碼其實等價於
        struct tasklet_struct my_tasklet = { NULL, 0, ATOMIC_INIT(0),
                                                        tasklet_handler,dev};
        這樣就建立了一個名爲my_tasklet的小任務，其處理程序爲tasklet_handler，而且已被激活。當處理程序被調用的時候，dev就會被傳遞給它。

 
       2)  編寫本身的小任務處理程序小任務處理程序必須符合以下的函數類型：

            void  tasklet_handler(unsigned long data)
           因爲小任務不能睡眠，所以不能在小任務中使用信號量或者其它產生阻塞的函數。可是小任務運行時能夠響應中斷。
       3) 調度本身的小任務經過調用tasklet_schedule()函數並傳遞給它相應的tasklt_struct指針，該小任務就會被調度以便適當的時候執行：
            tasklet_schedule(&my_tasklet);        /*把my_tasklet標記爲掛起 */
            在小任務被調度之後，只要有機會它就會盡量早的運行。在它尚未獲得運行機會以前，若是一個相同的小任務又被調度了，那麼它仍然只會運行一次。

           能夠調用tasklet_disable()函數來禁止某個指定的小任務。若是該小任務當前正在執行，這個函數會等到它執行完畢再返回。調用tasklet_enable()函數能夠激活一個小任務，若是但願把以DECLARE_TASKLET_DISABLED（）建立的小任務激活，也得調用這個函數，如：
           tasklet_disable(&my_tasklet);        /*小任務如今被禁止,這個小任務不能運行*/
          tasklet_enable(&my_tasklet);        /*  小任務如今被激活*/
          也能夠調用tasklet_kill()函數從掛起的隊列中去掉一個小任務。該函數的參數是一個指向某個小任務的tasklet_struct的長指針。在小任務從新調度它自身的時候，從掛起的隊列中移去已調度的小任務會頗有用。這個函數首先等待該小任務執行完畢，而後再將它移去。
4.tasklet的簡單用法
    下面是tasklet的一個簡單應用,以模塊的造成加載。

#include <linux/module.h>
#include <linux/init.h>
#include <linux/fs.h>
#include <linux/kdev_t.h>
#include <linux/cdev.h>
#include <linux/kernel.h>
#include <linux/interrupt.h>

static struct t asklet_struct my_tasklet;

static void tasklet_handler (unsigned long d ata)
{
printk(KERN_ALERT,"tasklet_handler is running./n");
}

static int __init test_init(void)
{
tasklet_init(&my_tasklet,tasklet_handler,0);
tasklet_schedule(&my_tasklet);
return0;
}

static void __exit test_exit(void)
{
tasklet_kill(&tasklet);
printk(KERN_ALERT,"test_exit is running./n");
}
MODULE_LICENSE("GPL");

module_init(test_init);
module_exit(test_exit);

       從這個例子能夠看出，所謂的小任務機制是爲下半部函數的執行提供了一種執行機制，也就是說，推遲處理的事情是由tasklet_handler實現，什麼時候執行，經由小任務機制封裝後交給內核去處理。

4. 中斷處理的工做隊列機制
     工做隊列（work queue）是另一種將工做推後執行的形式，它和前面討論的tasklet有所不一樣。工做隊列能夠把工做推後，交由一個內核線程去執行，也就是說，這個下半部分能夠在進程上下文中執行。這樣，經過工做隊列執行的代碼能佔盡進程上下文的全部優點。最重要的就是工做隊列容許被從新調度甚至是睡眠。
      那麼，什麼狀況下使用工做隊列，什麼狀況下使用tasklet。若是推後執行的任務須要睡眠，那麼就選擇工做隊列；若是推後執行的任務不須要睡眠，那麼就選擇tasklet。另外，若是須要用一個能夠從新調度的實體來執行你的下半部處理，也應該使用工做隊列。它是惟一能在進程上下文運行的下半部實現的機制，也只有它才能夠睡眠。這意味着在須要得到大量的內存時、在須要獲取信號量時，在須要執行阻塞式的I/O操做時，它都會很是有用。若是不須要用一個內核線程來推後執行工做，那麼就考慮使用tasklet。
    

4.1 工做、工做隊列
      如前所述，咱們把推後執行的任務叫作工做（work），描述它的數據結構爲work_struct，這些工做以隊列結構組織成工做隊列（workqueue），其數據結構爲workqueue_struct，而工做線程就是負責執行工做隊列中的工做。系統默認的工做者線程爲events,本身也能夠建立本身的工做者線程。表示工做的數據結構用<linux/workqueue.h>中定義的work_struct結構表示：

typedef void (*work_func_t)(struct work_struct *work);

struct work_struct {
atomic_long_t data;
struct list_head entry; // 鏈接全部工做的鏈表
work_func_t func; // 要執行的函數
#ifdef CONFIG_LOCKDEP
struct lockdep_map lockdep_map;
#endif
};
       這些結構被鏈接成鏈表。當一個工做者線程被喚醒時，它會執行它的鏈表上的全部工做。工做被執行完畢，它就將相應的work_struct對象從鏈表上移去。當鏈表上再也不有對象的時候，它就會繼續休眠。
表示工做隊列的數據結構用<kernel/workqueue.c>中定義的workqueue_struct:

/*
* The externally visible workqueue abstraction is an array of
* per-CPU workqueues:
*/
struct workqueue_struct {
unsigned int flags; /* I: WQ_* flags */
union {
struct cpu_workqueue_struct __percpu *pcpu;
struct cpu_workqueue_struct *single;
unsigned long v;
} cpu_wq; /* I: cwq's */
struct list_head list; /* W: list of all workqueues */

struct mutex flush_mutex; /* protects wq flushing */
int work_color; /* F: current work color */
int flush_color; /* F: current flush color */
atomic_t nr_cwqs_to_flush; /* flush in progress */
struct wq_flusher *first_flusher; /* F: first flusher */
struct list_head flusher_queue; /* F: flush waiters */
struct list_head flusher_overflow; /* F: flush overflow list */

mayday_mask_t mayday_mask; /* cpus requesting rescue */
struct worker *rescuer; /* I: rescue worker */

int saved_max_active; /* W: saved cwq max_active */
const char *name; /* I: workqueue name */
#ifdef CONFIG_LOCKDEP
struct lockdep_map lockdep_map;
#endif
};

4.2 建立推後的工做
4.2.1 靜態地建立工做(work_struct) 
     要使用工做隊列，首先要作的是建立一些須要推後完成的工做。能夠經過DECLARE_WORK在編譯時靜態地建該結構：
     DECLARE_WORK(name, func); 

     其定義以下：

#define DECLARE_WORK(n, f) \
struct work_struct n = __WORK_INITIALIZER(n, f)
       舉例以下：
static void do_poweroff(struct work_struct *dummy)
{
kernel_power_off();
}

static DECLARE_WORK(poweroff_work, do_poweroff);
        即建立了一個全局靜態變量：static work_struct poweroff_work，且被初始化了，其執行函數爲do_poweroff。

4.2.2 動態初始化工做(work_struct)
     先定義一具struct work_struct 變量，在須要使用時調用INIT_WORK進行初始化，而後即可以使用。

#define INIT_WORK(_work, _func) \
do { \
__INIT_WORK((_work), (_func), 0); \
} while (0)
       舉例以下：

void __cfg80211_scan_done(struct work_struct *wk)
{
struct cfg80211_registered_device *rdev;

rdev = container_of(wk, struct cfg80211_registered_device,
scan_done_wk);

cfg80211_lock_rdev(rdev);
___cfg80211_scan_done(rdev, false);
cfg80211_unlock_rdev(rdev);
}

struct cfg80211_registered_device {

struct work_struct scan_done_wk;
struct work_struct sched_scan_results_wk;
struct work_struct conn_work;
struct work_struct event_work;
struct cfg80211_wowlan *wowlan;
}
struct cfg80211_registered_device *rdev;
rdev = kzalloc(alloc_size, GFP_KERNEL);

INIT_WORK(&rdev->scan_done_wk, __cfg80211_scan_done); // 其執行函數爲： __cfg80211_scan_done
INIT_WORK(&rdev->sched_scan_results_wk, __cfg80211_sched_scan_results);

4.3 對工做進行調度
      如今工做已經被建立，咱們能夠調度它了。想要把給定工做的待處理函數提交給缺省的events工做線程，只需調用:      int schedule_work(struct work_struct *work);  
      它把work放入全局工做隊列：system_wq，其定義以下：
struct workqueue_struct *system_wq __read_mostly;
int schedule_work(struct work_struct *work)
{
return queue_work(system_wq, work);
}

queue_work：把一個工做放入工做隊列:

int queue_work(struct workqueue_struct *wq, struct work_struct *work)
{
int ret;

ret = queue_work_on(get_cpu(), wq, work);
put_cpu();

return ret;
}

        把work放入工做隊列，work立刻就會被調度，一旦其所在的處理器上的工做者線程被喚醒，它就會被執行。有時候並不但願工做立刻就被執行，而是但願它通過一段延遲之後再執行。在這種狀況下，能夠調度它在指定的時間執行：
struct delayed_work {
struct work_struct work;
struct timer_list timer;
};

int schedule_delayed_work(struct delayed_work *dwork, unsigned long delay/*jiffies*/)
{
return queue_delayed_work(system_wq, dwork, delay);
}

#define DECLARE_DELAYED_WORK(n, f) \
struct delayed_work n = __DELAYED_WORK_INITIALIZER(n, f)

#define INIT_DELAYED_WORK(_work, _func) \
do { \
INIT_WORK(&(_work)->work, (_func)); \
init_timer(&(_work)->timer); \
} while (0)

4.4 建立工做者線程
        工做放入工做隊列以後，由管理此工做隊列的工做者來執行這些work，經過alloc_workqueue或create_singlethread_workqueue來建立工做者線程，它最後調用kthread_create建立線程，其線程名爲alloc_workqueue中指定的name，其舉例以下：

static int __init init_workqueues(void)
{
unsigned int cpu;
int i;
...
system_wq = alloc_workqueue("events", 0, 0);
system_long_wq = alloc_workqueue("events_long", 0, 0);
system_nrt_wq = alloc_workqueue("events_nrt", WQ_NON_REENTRANT, 0);
system_unbound_wq = alloc_workqueue("events_unbound", WQ_UNBOUND,
WQ_UNBOUND_MAX_ACTIVE);
system_freezable_wq = alloc_workqueue("events_freezable",
WQ_FREEZABLE, 0);
BUG_ON(!system_wq || !system_long_wq || !system_nrt_wq ||
!system_unbound_wq || !system_freezable_wq);
return 0;
}

#define create_singlethread_workqueue(name) \
alloc_workqueue((name), WQ_UNBOUND | WQ_MEM_RECLAIM, 1)


#define alloc_workqueue(name, flags, max_active) \
__alloc_workqueue_key((name), (flags), (max_active), NULL, NULL)

如：cfg80211_wq = create_singlethread_workqueue("cfg80211");建立了一個名爲cfg80211的kernel線程。

部分摘自：http://blog.csdn.net/lmac21121/article/details/7741229————————————————版權聲明：本文爲CSDN博主「Arrow」的原創文章，遵循 CC 4.0 BY-SA 版權協議，轉載請附上原文出處連接及本聲明。原文連接：https://blog.csdn.net/myarrow/article/details/9287169