Linux 工作隊(duì)列 workqueue 是什么鬼？
目錄
- 工作隊(duì)列是什么
- 驅(qū)動(dòng)程序
- 編譯、測試
別人的經(jīng)驗(yàn)，我們的階梯！
Linux中斷處理可用下圖總結(jié)：
圖中描述了中斷處理中的下半部分都有哪些機(jī)制，以及如何根據(jù)實(shí)際的業(yè)務(wù)場景、限制條件來進(jìn)行選擇。
可以看出：這些不同的實(shí)現(xiàn)之間，有些是重復(fù)的，或者是相互取代的關(guān)系。
也正因?yàn)榇?，它們之間的使用方式幾乎是大同小異，至少是在API接口函數(shù)的使用方式上，從使用這的角度來看，都是非常類似的。
這篇文章，我們就通過實(shí)際的代碼操作，來演示一下工作隊(duì)列(workqueue)的使用方式。
工作隊(duì)列是什么
工作隊(duì)列是Linux操作系統(tǒng)中，進(jìn)行中斷下半部分處理的重要方式！
從名稱上可以猜到：一個(gè)工作隊(duì)列就好像業(yè)務(wù)層常用的消息隊(duì)列一樣，里面存放著很多的工作項(xiàng)等待著被處理。
工作隊(duì)列中有兩個(gè)重要的結(jié)構(gòu)體：工作隊(duì)列(workqueue_struct) 和 工作項(xiàng)(work_struct)：
- struct workqueue_struct {
- struct list_head pwqs; /* WR: all pwqs of this wq */
- struct list_head list; /* PR: list of all workqueues */
- ...
- char name[WQ_NAME_LEN]; /* I: workqueue name */
- ...
- /* hot fields used during command issue, aligned to cacheline */
- unsigned int flags ____cacheline_aligned; /* WQ: WQ_* flags */
- struct pool_workqueue __percpu *cpu_pwqs; /* I: per-cpu pwqs */
- struct pool_workqueue __rcu *numa_pwq_tbl[]; /* PWR: unbound pwqs indexed by node */
- };
- struct work_struct {
- atomic_long_t data;
- struct list_head entry;
- work_func_t func; // 指向處理函數(shù)
- #ifdef CONFIG_LOCKDEP
- struct lockdep_map lockdep_map;
- #endif
- };
在內(nèi)核中，工作隊(duì)列中的所有工作項(xiàng)，是通過鏈表串在一起的，并且等待著操作系統(tǒng)中的某個(gè)線程挨個(gè)取出來處理。
這些線程，可以是由驅(qū)動(dòng)程序通過 kthread_create 創(chuàng)建的線程，也可以是由操作系統(tǒng)預(yù)先就創(chuàng)建好的線程。
這里就涉及到一個(gè)取舍的問題了。
如果我們的處理函數(shù)很簡單，那么就沒有必要?jiǎng)?chuàng)建一個(gè)單獨(dú)的線程來處理了。
原因有二：
- 創(chuàng)建一個(gè)內(nèi)核線程是很耗費(fèi)資源的，如果函數(shù)很簡單，很快執(zhí)行結(jié)束之后再關(guān)閉線程，太劃不來了，得不償失;
- 如果每一個(gè)驅(qū)動(dòng)程序編寫者都毫無節(jié)制地創(chuàng)建內(nèi)核線程，那么內(nèi)核中將會存在大量不必要的線程，當(dāng)然了本質(zhì)上還是系統(tǒng)資源消耗和執(zhí)行效率的問題;
為了避免這種情況，于是操作系統(tǒng)就為我們預(yù)先創(chuàng)建好一些工作隊(duì)列和內(nèi)核線程。
我們只需要把需要處理的工作項(xiàng)，直接添加到這些預(yù)先創(chuàng)建好的工作隊(duì)列中就可以了，它們就會被相應(yīng)的內(nèi)核線程取出來處理。
例如下面這些工作隊(duì)列，就是內(nèi)核默認(rèn)創(chuàng)建的（include/linux/workqueue.h）：
- /*
- * System-wide workqueues which are always present.
- *
- * system_wq is the one used by schedule[_delayed]_work[_on]().
- * Multi-CPU multi-threaded. There are users which expect relatively
- * short queue flush time. Don't queue works which can run for too
- * long.
- *
- * system_highpri_wq is similar to system_wq but for work items which
- * require WQ_HIGHPRI.
- *
- * system_long_wq is similar to system_wq but may host long running
- * works. Queue flushing might take relatively long.
- *
- * system_unbound_wq is unbound workqueue. Workers are not bound to
- * any specific CPU, not concurrency managed, and all queued works are
- * executed immediately as long as max_active limit is not reached and
- * resources are available.
- *
- * system_freezable_wq is equivalent to system_wq except that it's
- * freezable.
- *
- * *_power_efficient_wq are inclined towards saving power and converted
- * into WQ_UNBOUND variants if 'wq_power_efficient' is enabled; otherwise,
- * they are same as their non-power-efficient counterparts - e.g.
- * system_power_efficient_wq is identical to system_wq if
- * 'wq_power_efficient' is disabled. See WQ_POWER_EFFICIENT for more info.
- */
- extern struct workqueue_struct *system_wq;
- extern struct workqueue_struct *system_highpri_wq;
- extern struct workqueue_struct *system_long_wq;
- extern struct workqueue_struct *system_unbound_wq;
- extern struct workqueue_struct *system_freezable_wq;
- extern struct workqueue_struct *system_power_efficient_wq;
- extern struct workqueue_struct *system_freezable_power_efficient_wq;
以上這些默認(rèn)工作隊(duì)列的創(chuàng)建代碼是(kernel/workqueue.c)：
- int __init workqueue_init_early(void)
- {
- ...
- system_wq = alloc_workqueue("events", 0, 0);
- system_highpri_wq = alloc_workqueue("events_highpri", WQ_HIGHPRI, 0);
- system_long_wq = alloc_workqueue("events_long", 0, 0);
- system_unbound_wq = alloc_workqueue("events_unbound", WQ_UNBOUND,
- WQ_UNBOUND_MAX_ACTIVE);
- system_freezable_wq = alloc_workqueue("events_freezable",
- WQ_FREEZABLE, 0);
- system_power_efficient_wq = alloc_workqueue("events_power_efficient",
- WQ_POWER_EFFICIENT, 0);
- system_freezable_power_efficient_wq = alloc_workqueue("events_freezable_power_efficient",
- WQ_FREEZABLE | WQ_POWER_EFFICIENT,
- 0);
- ...
- }
此外，由于工作隊(duì)列 system_wq 被使用的頻率很高，于是內(nèi)核就封裝了一個(gè)簡單的函數(shù)(schedule_work)給我們使用：
- /**
- * schedule_work - put work task in global workqueue
- * @work: job to be done
- *
- * Returns %false if @work was already on the kernel-global workqueue and
- * %true otherwise.
- *
- * This puts a job in the kernel-global workqueue if it was not already
- * queued and leaves it in the same position on the kernel-global
- * workqueue otherwise.
- */
- static inline bool schedule_work(struct work_struct *work){
- return queue_work(system_wq, work);
- }
當(dāng)然了，任何事情有利就有弊！
由于內(nèi)核默認(rèn)創(chuàng)建的工作隊(duì)列，是被所有的驅(qū)動(dòng)程序共享的。
如果所有的驅(qū)動(dòng)程序都把等待處理的工作項(xiàng)委托給它們來處理，那么就會導(dǎo)致某個(gè)工作隊(duì)列中過于擁擠。
根據(jù)先來后到的原則，工作隊(duì)列中后加入的工作項(xiàng)，就可能因?yàn)榍懊婀ぷ黜?xiàng)的處理函數(shù)執(zhí)行的時(shí)間太長，從而導(dǎo)致時(shí)效性無法保證。
因此，這里存在一個(gè)系統(tǒng)平衡的問題。
關(guān)于工作隊(duì)列的基本知識點(diǎn)就介紹到這里，下面來實(shí)際操作驗(yàn)證一下。
驅(qū)動(dòng)程序
之前的幾篇文章，在驅(qū)動(dòng)程序中測試中斷處理的操作流程都是一樣的，因此這里就不在操作流程上進(jìn)行贅述了。
這里直接給出驅(qū)動(dòng)程序的全貌代碼，然后查看 dmesg 的輸出信息。
創(chuàng)建驅(qū)動(dòng)程序源文件和 Makefile：
- $ cd tmp/linux-4.15/drivers
- $ mkdir my_driver_interrupt_wq
- $ touch my_driver_interrupt_wq.c
- $ touch Makefile
示例代碼全貌
測試場景是：加載驅(qū)動(dòng)模塊之后，如果監(jiān)測到鍵盤上的ESC鍵被按下，那么就往內(nèi)核默認(rèn)的工作隊(duì)列system_wq中增加一個(gè)工作項(xiàng)，然后觀察該工作項(xiàng)對應(yīng)的處理函數(shù)是否被調(diào)用。
- #include <linux/kernel.h>
- #include <linux/module.h>
- #include <linux/interrupt.h>
- static int irq;
- static char * devname;
- static struct work_struct mywork;
- // 接收驅(qū)動(dòng)模塊加載時(shí)傳入的參數(shù)
- module_param(irq, int, 0644);
- module_param(devname, charp, 0644);
- // 定義驅(qū)動(dòng)程序的 ID，在中斷處理函數(shù)中用來判斷是否需要處理
- #define MY_DEV_ID 1226
- // 驅(qū)動(dòng)程序數(shù)據(jù)結(jié)構(gòu)
- struct myirq
- {
- int devid;
- };
- struct myirq mydev ={ MY_DEV_ID };
- #define KBD_DATA_REG 0x60
- #define KBD_STATUS_REG 0x64
- #define KBD_SCANCODE_MASK 0x7f
- #define KBD_STATUS_MASK 0x80
- // 工作項(xiàng)綁定的處理函數(shù)
- static void mywork_handler(struct work_struct *work)
- {
- printk("mywork_handler is called. \n");
- // do some other things
- }
- //中斷處理函數(shù)
- static irqreturn_t myirq_handler(int irq, void * dev)
- {
- struct myirq mydev;
- unsigned char key_code;
- mydev = *(struct myirq*)dev;
- // 檢查設(shè)備 id，只有當(dāng)相等的時(shí)候才需要處理
- if (MY_DEV_ID == mydev.devid)
- {
- // 讀取鍵盤掃描碼
- key_code = inb(KBD_DATA_REG);
- if (key_code == 0x01)
- {
- printk("ESC key is pressed! \n");
- // 初始化工作項(xiàng)
- INIT_WORK(&mywork, mywork_handler);
- // 加入到工作隊(duì)列 system_wq
- schedule_work(&mywork);
- }
- }
- return IRQ_HANDLED;
- }
- // 驅(qū)動(dòng)模塊初始化函數(shù)
- static int __init myirq_init(void)
- {
- printk("myirq_init is called. \n");
- // 注冊中斷處理函數(shù)
- if(request_irq(irq, myirq_handler, IRQF_SHARED, devname, &mydev)!=0)
- {
- printk("register irq[%d] handler failed. \n", irq);
- return -1;
- }
- printk("register irq[%d] handler success. \n", irq);
- return 0;
- }
- // 驅(qū)動(dòng)模塊退出函數(shù)
- static void __exit myirq_exit(void)
- {
- printk("myirq_exit is called. \n");
- // 釋放中斷處理函數(shù)
- free_irq(irq, &mydev);
- }
- MODULE_LICENSE("GPL");
- module_init(myirq_init);
- module_exit(myirq_exit);
Makefile 文件
- ifneq ($(KERNELRELEASE),)
- obj-m := my_driver_interrupt_wq.o
- else
- KERNELDIR ?= /lib/modules/$(shell uname -r)/build
- PWD := $(shell pwd)
- default:
- $(MAKE) -C $(KERNELDIR) M=$(PWD) modules
- clean:
- $(MAKE) -C $(KERNEL_PATH) M=$(PWD) clean
- endif
編譯、測試
- $ make
- $ sudo insmod my_driver_interrupt_wq.ko irq=1 devname=mydev
檢查驅(qū)動(dòng)模塊是否加載成功：
- $ lsmod | grep my_driver_interrupt_wq
- my_driver_interrupt_wq 16384 0
再看一下 dmesg 的輸出信息：
- $ dmesg
- ...
- [ 188.247636] myirq_init is called.
- [ 188.247642] register irq[1] handler success.
說明：驅(qū)動(dòng)程序的初始化函數(shù) myirq_init 被調(diào)用了，并且成功注冊了 1 號中斷的處理程序。
此時(shí)，按一下鍵盤上的 ESC 鍵。
操作系統(tǒng)在捕獲到鍵盤中斷之后，會依次調(diào)用此中斷的所有中斷處理程序，其中就包括我們注冊的 myirq_handler 函數(shù)。
在這個(gè)函數(shù)中，當(dāng)判斷出是ESC按鍵時(shí)，就初始化一個(gè)工作項(xiàng)(把結(jié)構(gòu)體 work_struct 類型的變量與一個(gè)處理函數(shù)綁定起來)，然后丟給操作系統(tǒng)預(yù)先創(chuàng)建好的工作隊(duì)列(system_wq)去處理，如下所示：
- if (key_code == 0x01)
- {
- printk("ESC key is pressed! \n");
- INIT_WORK(&mywork, mywork_handler);
- schedule_work(&mywork);
- }
因此，當(dāng)相應(yīng)的內(nèi)核線程從這個(gè)工作隊(duì)列(system_wq)中取出工作項(xiàng)(mywork)來處理的時(shí)候，函數(shù) mywork_handler 就會被調(diào)用。
現(xiàn)在來看一下 dmesg 的輸出信息：
- [ 305.053155] ESC key is pressed!
- [ 305.053177] mywork_handler is called.
可以看到：mywork_handler函數(shù)被正確調(diào)用了。
完美！
