在前一章也提到過，之所以中斷會分成上下兩部分，是由於中斷對時限的要求非常高，需要盡快的響應硬件。
主要內容：
- 中斷下半部處理
- 實現中斷下半部的機制
- 總結中斷下半部的實現
- 中斷實現示例
1. 中斷下半部處理
那么對於一個中斷，如何划分上下兩部分呢？哪些處理放在上半部，哪些處理放在下半部？
這里有一些經驗可供借鑒：
- 如果一個任務對時間十分敏感，將其放在上半部
- 如果一個任務和硬件有關，將其放在上半部
- 如果一個任務要保證不被其他中斷打斷，將其放在上半部
- 其他所有任務，考慮放在下半部
2. 實現中斷下半部的機制
實現下半部的方法很多，隨着內核的發展，產生了一些新的方法，也淘汰了一些舊方法。
目前使用最多的是以下3中方法
- 2.1 軟中斷
- 2.2 tasklet
- 2.3 工作隊列
2.1 軟中斷
軟中斷的代碼在：kernel/softirq.c
軟中斷的流程如下：
流程圖中幾個步驟的說明：
① 注冊軟中斷的函數 open_softirq參見 kernel/softirq.c文件)
/* * 將軟中斷類型和軟中斷處理函數加入到軟中斷序列中 * @nr - 軟中斷類型 * @(*action)(struct softirq_action *) - 軟中斷處理的函數指針 */ void open_softirq(int nr, void (*action)(struct softirq_action *)) { /* softirq_vec是個struct softirq_action類型的數組 */ softirq_vec[nr].action = action; }
軟中斷類型目前有10個，其定義在 include/linux/interrupt.h 文件中：
enum { HI_SOFTIRQ=0, TIMER_SOFTIRQ, NET_TX_SOFTIRQ, NET_RX_SOFTIRQ, BLOCK_SOFTIRQ, BLOCK_IOPOLL_SOFTIRQ, TASKLET_SOFTIRQ, SCHED_SOFTIRQ, HRTIMER_SOFTIRQ, RCU_SOFTIRQ, /* Preferable RCU should always be the last softirq */ NR_SOFTIRQS };
struct softirq_action 的定義也在 include/linux/interrupt.h 文件中
/* * 這個結構體的字段是個函數指針，字段名稱是action * 函數指針的返回指是void型 * 函數指針的參數是 struct softirq_action 的地址，其實就是指向 softirq_vec 中的某一項 * 如果 open_softirq 是這樣調用的： open_softirq(NET_TX_SOFTIRQ, my_tx_action); * 那么 my_tx_action 的參數就是 softirq_vec[NET_TX_SOFTIRQ]的地址 */ struct softirq_action { void (*action)(struct softirq_action *); };
② 觸發軟中斷的函數 raise_softirq 參見 kernel/softirq.c文件
/* * 觸發某個中斷類型的軟中斷 * @nr - 被觸發的中斷類型 * 從函數中可以看出，在處理軟中斷前后有保存和恢復寄存器的操作 */ void raise_softirq(unsigned int nr) { unsigned long flags; local_irq_save(flags); raise_softirq_irqoff(nr); local_irq_restore(flags); }
③ 執行軟中斷 do_softirq 參見 kernel/softirq.c文件
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); }
④ 執行相應的軟中斷 - 執行自己寫的中斷處理
linux中，執行軟中斷有專門的內核線程，每個處理器對應一個線程，名稱ksoftirqd/n (n對應處理器號)
通過top命令查看我的單核虛擬機，CentOS系統中的ksoftirqd線程如下：
[root@vbox ~]# top | grep ksoftirq 4 root 20 0 0 0 0 S 0.0 0.0 0:00.02 ksoftirqd/0
2.2 tasklet
tasklet也是利用軟中斷來實現的，但是它提供了比軟中斷更好用的接口(其實就是基於軟中斷又封裝了一下)，
所以除了對性能要求特別高的情況，一般建議使用tasklet來實現自己的中斷。
tasklet對應的結構體在 <linux/interrupt.h> 中
struct tasklet_struct { struct tasklet_struct *next; /* 鏈表中的下一個tasklet */ unsigned long state; /* tasklet狀態 */ atomic_t count; /* 引用計數器 */ void (*func)(unsigned long); /* tasklet處理函數 */ unsigned long data; /* tasklet處理函數的參數 */ };
tasklet狀態只有3種值：
- 值 0 表示該tasklet沒有被調度
- 值 TASKLET_STATE_SCHED 表示該tasklet已經被調度
- 值 TASKLET_STATE_RUN 表示該tasklet已經運行
引用計數器count 的值不為0，表示該tasklet被禁止。
tasklet使用流程如下：
1. 聲明tasklet (參見<linux/interrupt.h>)
/* 靜態聲明一個tasklet */ #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 傳遞一個tasklet_struct指針給初始化函數 */ extern void tasklet_init(struct tasklet_struct *t, void (*func)(unsigned long), unsigned long data);
2. 編寫處理程序
參照tasklet處理函數的原型來寫自己的處理邏輯
void tasklet_handler(unsigned long date)
3. 調度tasklet
中斷的上半部處理完后調度tasklet，在適當時候進行下半部的處理
tasklet_schedule(&my_tasklet) /* my_tasklet就是之前聲明的tasklet_struct */
2.3 工作隊列
工作隊列子系統是一個用於創建內核線程的接口，通過它可以創建一個工作者線程來專門處理中斷的下半部工作。
工作隊列和tasklet不一樣，不是基於軟中斷來實現的。
缺省的工作者線程名稱是 events/n (n對應處理器號)。
通過top命令查看我的單核虛擬機，CentOS系統中的events線程如下：
[root@vbox ~]# top | grep event 7 root 20 0 0 0 0 S 0.0 0.0 0:03.71 events/0
工作隊列主要用到下面3個結構體，弄懂了這3個結構體的關系，也就知道工作隊列的處理流程了。
/* 在 include/linux/workqueue.h 文件中定義 */ struct work_struct { atomic_long_t data; /* 這個並不是處理函數的參數，而是表示此work是否pending等狀態的flag */ #define WORK_STRUCT_PENDING 0 /* T if work item pending execution */ #define WORK_STRUCT_FLAG_MASK (3UL) #define WORK_STRUCT_WQ_DATA_MASK (~WORK_STRUCT_FLAG_MASK) struct list_head entry; /* 中斷下半部處理函數的鏈表 */ work_func_t func; /* 處理中斷下半部工作的函數 */ #ifdef CONFIG_LOCKDEP struct lockdep_map lockdep_map; #endif }; /* 在 kernel/workqueue.c文件中定義 * 每個工作者線程對應一個 cpu_workqueue_struct ，其中包含要處理的工作的鏈表 * (即 work_struct 的鏈表，當此鏈表不空時，喚醒工作者線程來進行處理) */ /* * The per-CPU workqueue (if single thread, we always use the first * possible cpu). */ struct cpu_workqueue_struct { spinlock_t lock; /* 鎖保護這種結構 */ struct list_head worklist; /* 工作隊列頭節點 */ wait_queue_head_t more_work; struct work_struct *current_work; struct workqueue_struct *wq; /* 關聯工作隊列結構 */ struct task_struct *thread; /* 關聯線程 */ } ____cacheline_aligned; /* 也是在 kernel/workqueue.c 文件中定義的 * 每個 workqueue_struct 表示一種工作者類型，系統默認的就是 events 工作者類型 * 每個工作者類型一般對應n個工作者線程，n就是處理器的個數 */ /* * The externally visible workqueue abstraction is an array of * per-CPU workqueues: */ struct workqueue_struct { struct cpu_workqueue_struct *cpu_wq; /* 工作者線程 */ struct list_head list; const char *name; int singlethread; int freezeable; /* Freeze threads during suspend */ int rt; #ifdef CONFIG_LOCKDEP struct lockdep_map lockdep_map; #endif };
使用工作者隊列的方法見下圖：
① 創建推后執行的工作 - 有靜態創建和動態創建2種方法
/* 靜態創建一個work_struct * @n - work_struct結構體，不用事先定義 * @f - 下半部處理函數 */ #define DECLARE_WORK(n, f) \ struct work_struct n = __WORK_INITIALIZER(n, f) /* 動態創建一個 work_struct * @_work - 已經定義好的一個 work_struct * @_func - 下半部處理函數 */ #ifdef CONFIG_LOCKDEP #define INIT_WORK(_work, _func) \ do { \ static struct lock_class_key __key; \ \ (_work)->data = (atomic_long_t) WORK_DATA_INIT(); \ lockdep_init_map(&(_work)->lockdep_map, #_work, &__key, 0);\ INIT_LIST_HEAD(&(_work)->entry); \ PREPARE_WORK((_work), (_func)); \ } while (0) #else #define INIT_WORK(_work, _func) \ do { \ (_work)->data = (atomic_long_t) WORK_DATA_INIT(); \ INIT_LIST_HEAD(&(_work)->entry); \ PREPARE_WORK((_work), (_func)); \ } while (0) #endif
工作隊列處理函數的原型：
typedef void (*work_func_t)(struct work_struct *work);
② 刷新現有的工作，這個步驟不是必須的，可以直接從第①步直接進入第③步
刷新現有工作的意思就是在追加新的工作之前，保證隊列中的已有工作已經執行完了。
/* 刷新系統默認的隊列，即 events 隊列 */ void flush_scheduled_work(void); /* 刷新用戶自定義的隊列 * @wq - 用戶自定義的隊列 */ void flush_workqueue(struct workqueue_struct *wq);
③ 調度工作 - 調度新定義的工作，使之處於等待處理器執行的狀態
/* 調度第一步中新定義的工作，在系統默認的工作者線程中執行此工作 * @work - 第一步中定義的工作 */ schedule_work(struct work_struct *work); /* 調度第一步中新定義的工作，在系統默認的工作者線程中執行此工作 * @work - 第一步中定義的工作 * @delay - 延遲的時鍾節拍 */ int schedule_delayed_work(struct delayed_work *work, unsigned long delay); /* 調度第一步中新定義的工作，在用戶自定義的工作者線程中執行此工作 * @wq - 用戶自定義的工作隊列類型 * @work - 第一步中定義的工作 */ int queue_work(struct workqueue_struct *wq, struct work_struct *work); /* 調度第一步中新定義的工作，在用戶自定義的工作者線程中執行此工作 * @wq - 用戶自定義的工作隊列類型 * @work - 第一步中定義的工作 * @delay - 延遲的時鍾節拍 */ int queue_delayed_work(struct workqueue_struct *wq, struct delayed_work *work, unsigned long delay);
3. 總結中斷下半部的實現
下面對實現中斷下半部工作的3種機制進行總結，便於在實際使用中決定使用哪種機制
| 下半部機制 |
上下文 |
復雜度 |
執行性能 |
順序執行保障 |
| 軟中斷 | 中斷 | 高 (需要自己確保軟中斷的執行順序及鎖機制) |
好 (全部自己實現，便於調優) |
沒有 |
| tasklet | 中斷 | 中 (提供了簡單的接口來使用軟中斷) |
中 | 同類型不能同時執行 |
| 工作隊列 | 進程 | 低 (在進程上下文中運行，與寫用戶程序差不多) |
差 | 沒有 (和進程上下文一樣被調度) |
4. 中斷實現示例
4.1 軟中斷的實現
本來想用內核模塊的方法來測試一下軟中斷的流程，但是編譯時發現軟中斷注冊函數(open_softirq)和觸發函數(raise_softirq)
並沒有用EXPORT_SYMBOL導出，所以自定義的內核模塊中無法使用。
測試的代碼如下：
#include <linux/interrupt.h> #include "kn_common.h" MODULE_LICENSE("Dual BSD/GPL"); static void my_softirq_func(struct softirq_action*); static int testsoftirq_init(void) { // 注冊softirq，這里注冊的是定時器的下半部 open_softirq(TIMER_SOFTIRQ, my_softirq_func); // 觸發softirq raise_softirq(TIMER_SOFTIRQ); return 0; } static void testsoftirq_exit(void) { printk(KERN_ALERT "*************************\n"); print_current_time(0); printk(KERN_ALERT "testrbtree is exited!\n"); printk(KERN_ALERT "*************************\n"); } static void my_softirq_func(struct softirq_action* act) { printk(KERN_ALERT "=========================\n"); print_current_time(0); printk(KERN_ALERT "my softirq function is been called!....\n"); printk(KERN_ALERT "=========================\n"); } module_init(testsoftirq_init); module_exit(testsoftirq_exit);
其中頭文件 kn_common.h 的相關內容參見之前的博客《Linux內核設計與實現》讀書筆記（六）- 內核數據結構
由於內核沒有用EXPORT_SYMBOL導出open_softirq和raise_softirq函數，所以編譯時有如下警告：
WARNING: "open_softirq" [/root/chap08/mysoftirq.ko] undefined! WARNING: "raise_softirq" [/root/chap08/mysoftirq.ko] undefined!
注：編譯用的系統時centos6.3 (uname -r結果 - 2.6.32-279.el6.x86_64)
沒辦法，只能嘗試修改內核代碼(將open_softirq和raise_softirq用EXPORT_SYMBOL導出)，再重新編譯內核，然后再嘗試能否測試軟中斷。
主要修改2個文件，(既然要修改代碼，干脆加了一種軟中斷類型)：
/* 修改 kernel/softirq.c */ // ... 略 ... char *softirq_to_name[NR_SOFTIRQS] = { "HI", "TIMER", "NET_TX", "NET_RX", "BLOCK", "BLOCK_IOPOLL", "TASKLET", "SCHED", "HRTIMER", "RCU", "WYB" }; /* 追加了一種新的softirq，即 "WYB"，我名字的縮寫 ^_^ */ // ... 略 ... void raise_softirq(unsigned int nr) { unsigned long flags; local_irq_save(flags); raise_softirq_irqoff(nr); local_irq_restore(flags); } EXPORT_SYMBOL(raise_softirq); /* 追加的代碼 */ void open_softirq(int nr, void (*action)(struct softirq_action *)) { softirq_vec[nr].action = action; } EXPORT_SYMBOL(open_softirq); /* 追加的代碼 */ // ... 略 ... /* 還修改了 include/linux/interrupt.h */ enum { HI_SOFTIRQ=0, TIMER_SOFTIRQ, NET_TX_SOFTIRQ, NET_RX_SOFTIRQ, BLOCK_SOFTIRQ, BLOCK_IOPOLL_SOFTIRQ, TASKLET_SOFTIRQ, SCHED_SOFTIRQ, HRTIMER_SOFTIRQ, RCU_SOFTIRQ, /* Preferable RCU should always be the last softirq */ WYB_SOFTIRQS, /* 追加的一種中斷類型 */ NR_SOFTIRQS };
重新編譯內核后，在新的內核上再次實驗軟中斷代碼：
(編譯內核方法參見：《Linux內核設計與實現》讀書筆記（五）- 系統調用 3.3節)
測試軟中斷的代碼：testsoftirq.c
#include <linux/interrupt.h> #include "kn_common.h" MODULE_LICENSE("Dual BSD/GPL"); static void my_softirq_func(struct softirq_action*); static int testsoftirq_init(void) { printk(KERN_ALERT "interrupt's top half!\n"); // 注冊softirq，這里注冊的是自定義的軟中斷類型 open_softirq(WYB_SOFTIRQS, my_softirq_func); // 觸發softirq raise_softirq(WYB_SOFTIRQS); return 0; } static void testsoftirq_exit(void) { printk(KERN_ALERT "*************************\n"); print_current_time(0); printk(KERN_ALERT "testsoftirq is exited!\n"); printk(KERN_ALERT "*************************\n"); } static void my_softirq_func(struct softirq_action* act) { printk(KERN_ALERT "=========================\n"); print_current_time(0); printk(KERN_ALERT "my softirq function is been called!....\n"); printk(KERN_ALERT "=========================\n"); } module_init(testsoftirq_init); module_exit(testsoftirq_exit);
Makefile：
obj-m += mysoftirq.o mysoftirq-objs := testsoftirq.o kn_common.o #generate the path CURRENT_PATH:=$(shell pwd) #the current kernel version number LINUX_KERNEL:=$(shell uname -r) #the absolute path LINUX_KERNEL_PATH:=/usr/src/kernels/$(LINUX_KERNEL) #complie object all: make -C $(LINUX_KERNEL_PATH) M=$(CURRENT_PATH) modules rm -rf modules.order Module.symvers .*.cmd *.o *.mod.c .tmp_versions *.unsigned #clean clean: rm -rf modules.order Module.symvers .*.cmd *.o *.mod.c *.ko .tmp_versions *.unsigned
測試軟中斷的方法如下：
make insmod mysoftirq.ko rmmod mysoftirq dmesg | tail -9 # 運行結果 interrupt's top half! ========================= 2013-4-22 14:4:57 my softirq function is been called!.... ========================= ************************* 2013-4-22 14:5:2 testsoftirq is exited! *************************
4.2 tasklet的實現
tasklet的實驗用默認的內核即可，我們切換到centos6.3的默認內核(uname -r: 2.6.32-279.el6.x86_64)
從中我們也可以看出，內核之所以沒有導出open_softirq和raise_softirq函數，可能還是因為提倡我們盡量用tasklet來實現中斷的下半部工作。
tasklet測試代碼：testtasklet.c
#include <linux/interrupt.h> #include "kn_common.h" MODULE_LICENSE("Dual BSD/GPL"); static void my_tasklet_func(unsigned long); /* mytasklet 必須定義在testtasklet_init函數的外面，否則會出錯 */ DECLARE_TASKLET(mytasklet, my_tasklet_func, 1000); static int testtasklet_init(void) { printk(KERN_ALERT "interrupt's top half!\n"); // 如果在這里定義的話，那么 mytasklet是函數的局部變量， // 后面調度的時候會找不到 mytasklet // DECLARE_TASKLET(mytasklet, my_tasklet_func, 1000); // 調度tasklet， 處理器會在適當時候執行這個tasklet tasklet_schedule(&mytasklet); return 0; } static void testtasklet_exit(void) { printk(KERN_ALERT "*************************\n"); print_current_time(0); printk(KERN_ALERT "testtasklet is exited!\n"); printk(KERN_ALERT "*************************\n"); } static void my_tasklet_func(unsigned long data) { printk(KERN_ALERT "=========================\n"); print_current_time(0); printk(KERN_ALERT "my tasklet function is been called!....\n"); printk(KERN_ALERT "parameter data is %ld\n", data); printk(KERN_ALERT "=========================\n"); } module_init(testtasklet_init); module_exit(testtasklet_exit);
Makefile：
obj-m += mytasklet.o mytasklet-objs := testtasklet.o kn_common.o #generate the path CURRENT_PATH:=$(shell pwd) #the current kernel version number LINUX_KERNEL:=$(shell uname -r) #the absolute path LINUX_KERNEL_PATH:=/usr/src/kernels/$(LINUX_KERNEL) #complie object all: make -C $(LINUX_KERNEL_PATH) M=$(CURRENT_PATH) modules rm -rf modules.order Module.symvers .*.cmd *.o *.mod.c .tmp_versions *.unsigned #clean clean: rm -rf modules.order Module.symvers .*.cmd *.o *.mod.c *.ko .tmp_versions *.unsigned
測試tasklet的方法如下：
make insmod mytasklet.ko rmmod mytasklet dmesg | tail -10 # 運行結果 interrupt's top half! ========================= 2013-4-22 14:53:14 my tasklet function is been called!.... parameter data is 1000 ========================= ************************* 2013-4-22 14:53:20 testtasklet is exited! *************************
4.3 工作隊列的實現
workqueue的例子的中靜態定義了一個工作，動態定義了一個工作。
靜態定義的工作由系統工作隊列（events/n）調度，
動態定義的工作由自定義的工作隊列（myworkqueue）調度。
測試工作隊列的代碼：testworkqueue.c
#include <linux/workqueue.h> #include "kn_common.h" MODULE_LICENSE("Dual BSD/GPL"); static void my_work_func(struct work_struct *); static void my_custom_workqueue_func(struct work_struct *); /* 靜態創建一個工作，使用系統默認的工作者線程，即 events/n */ DECLARE_WORK(mywork, my_work_func); static int testworkqueue_init(void) { /*自定義的workqueue */ struct workqueue_struct *myworkqueue = create_workqueue("myworkqueue"); /* 動態創建一個工作 */ struct work_struct *mywork2; mywork2 = kmalloc(sizeof(struct work_struct), GFP_KERNEL); INIT_WORK(mywork2, my_custom_workqueue_func); printk(KERN_ALERT "interrupt's top half!\n"); /* 刷新系統默認的隊列 */ flush_scheduled_work(); /* 調度工作 */ schedule_work(&mywork); /* 刷新自定義的工作隊列 */ flush_workqueue(myworkqueue); /* 調度自定義工作隊列上的工作 */ queue_work(myworkqueue, mywork2); return 0; } static void testworkqueue_exit(void) { printk(KERN_ALERT "*************************\n"); print_current_time(0); printk(KERN_ALERT "my workqueue test is exited!\n"); printk(KERN_ALERT "*************************\n"); } static void my_work_func(struct work_struct *work) { printk(KERN_ALERT "=========================\n"); print_current_time(0); printk(KERN_ALERT "my workqueue function is been called!....\n"); printk(KERN_ALERT "=========================\n"); } static void my_custom_workqueue_func(struct work_struct *work) { printk(KERN_ALERT "=========================\n"); print_current_time(0); printk(KERN_ALERT "my cutomize workqueue function is been called!....\n"); printk(KERN_ALERT "=========================\n"); kfree(work); } module_init(testworkqueue_init); module_exit(testworkqueue_exit);
Makefile:
obj-m += myworkqueue.o myworkqueue-objs := testworkqueue.o kn_common.o #generate the path CURRENT_PATH:=$(shell pwd) #the current kernel version number LINUX_KERNEL:=$(shell uname -r) #the absolute path LINUX_KERNEL_PATH:=/usr/src/kernels/$(LINUX_KERNEL) #complie object all: make -C $(LINUX_KERNEL_PATH) M=$(CURRENT_PATH) modules rm -rf modules.order Module.symvers .*.cmd *.o *.mod.c .tmp_versions *.unsigned #clean clean: rm -rf modules.order Module.symvers .*.cmd *.o *.mod.c *.ko .tmp_versions *.unsigned
測試workqueue的方法如下：
make insmod myworkqueue.ko rmmod myworkqueue dmesg | tail -13 # 運行結果 interrupt's top half! ========================= 2013-4-23 9:55:29 my workqueue function is been called!.... ========================= ========================= 2013-4-23 9:55:29 my cutomize workqueue function is been called!.... ========================= ************************* 2013-4-23 9:55:29 my workqueue is exited! *************************