目錄

• 工作隊列是什么
•  驅動程序
•  編譯、測試

別人的經驗，我們的階梯！

Linux中斷處理可用下圖總結：

圖中描述了中斷處理中的下半部分都有哪些機制，以及如何根據實際的業務場景、限制條件來進行選擇。

可以看出：這些不同的實現之間，有些是重復的，或者是相互取代的關系。

也正因為此，它們之間的使用方式幾乎是大同小異，至少是在API接口函數的使用方式上，從使用這的角度來看，都是非常類似的。

這篇文章，我們就通過實際的代碼操作，來演示一下工作隊列(workqueue)的使用方式。

工作隊列是什么

工作隊列是Linux操作系統中，進行中斷下半部分處理的重要方式！

從名稱上可以猜到：一個工作隊列就好像業務層常用的消息隊列一樣，里面存放著很多的工作項等待著被處理。

工作隊列中有兩個重要的結構體：工作隊列(workqueue_struct) 和 工作項(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;   // 指向處理函數  
#ifdef CONFIG_LOCKDEP                                                                                   
         struct lockdep_map lockdep_map;  
#endif  
}; 

在內核中，工作隊列中的所有工作項，是通過鏈表串在一起的，并且等待著操作系統中的某個線程挨個取出來處理。

這些線程，可以是由驅動程序通過 kthread_create 創建的線程，也可以是由操作系統預先就創建好的線程。

這里就涉及到一個取舍的問題了。

如果我們的處理函數很簡單，那么就沒有必要創建一個單獨的線程來處理了。

原因有二：

1.   創建一個內核線程是很耗費資源的，如果函數很簡單，很快執行結束之后再關閉線程，太劃不來了，得不償失;
2.   如果每一個驅動程序編寫者都毫無節制地創建內核線程，那么內核中將會存在大量不必要的線程，當然了本質上還是系統資源消耗和執行效率的問題;

為了避免這種情況，于是操作系統就為我們預先創建好一些工作隊列和內核線程。

我們只需要把需要處理的工作項，直接添加到這些預先創建好的工作隊列中就可以了，它們就會被相應的內核線程取出來處理。

例如下面這些工作隊列，就是內核默認創建的（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; 

以上這些默認工作隊列的創建代碼是(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);  
    ... 
 } 

此外，由于工作隊列 system_wq 被使用的頻率很高，于是內核就封裝了一個簡單的函數(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);  
} 

當然了，任何事情有利就有弊！

由于內核默認創建的工作隊列，是被所有的驅動程序共享的。

如果所有的驅動程序都把等待處理的工作項委托給它們來處理，那么就會導致某個工作隊列中過于擁擠。

根據先來后到的原則，工作隊列中后加入的工作項，就可能因為前面工作項的處理函數執行的時間太長，從而導致時效性無法保證。

因此，這里存在一個系統平衡的問題。

關于工作隊列的基本知識點就介紹到這里，下面來實際操作驗證一下。

驅動程序

之前的幾篇文章，在驅動程序中測試中斷處理的操作流程都是一樣的，因此這里就不在操作流程上進行贅述了。

這里直接給出驅動程序的全貌代碼，然后查看 dmesg 的輸出信息。

創建驅動程序源文件和 Makefile： 

$ cd tmp/linux-4.15/drivers  
$ mkdir my_driver_interrupt_wq  
$ touch my_driver_interrupt_wq.c  
$ touch Makefile 

示例代碼全貌

測試場景是：加載驅動模塊之后，如果監測到鍵盤上的ESC鍵被按下，那么就往內核默認的工作隊列system_wq中增加一個工作項，然后觀察該工作項對應的處理函數是否被調用。 

#include <linux/kernel.h>  
#include <linux/module.h>  
#include <linux/interrupt.h>  
static int irq;            
static char * devname;     
static struct work_struct mywork;                
  // 接收驅動模塊加載時傳入的參數  
module_param(irq, int, 0644);  
module_param(devname, charp, 0644);  
// 定義驅動程序的 ID，在中斷處理函數中用來判斷是否需要處理           
 #define MY_DEV_ID           1226  
// 驅動程序數據結構  
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  
// 工作項綁定的處理函數  
static void mywork_handler(struct work_struct *work)  
{  
    printk("mywork_handler is called. \n");  
    // do some other things  
}         
//中斷處理函數  
static irqreturn_t myirq_handler(int irq, void * dev)  
{  
    struct myirq mydev;  
    unsigned char key_code;  
    mydev = *(struct myirq*)dev;       
     // 檢查設備 id，只有當相等的時候才需要處理  
    if (MY_DEV_ID == mydev.devid)  
    {  
        // 讀取鍵盤掃描碼  
        key_code = inb(KBD_DATA_REG);    
        if (key_code == 0x01)  
        { 
             printk("ESC key is pressed! \n");         
             // 初始化工作項 
            INIT_WORK(&mywork, mywork_handler);        
             // 加入到工作隊列 system_wq  
                    schedule_work(&mywork);  
        }  
    }     
    return IRQ_HANDLED;  
}   
// 驅動模塊初始化函數  
static int __init myirq_init(void)  
{  
    printk("myirq_init is called. \n");  
    // 注冊中斷處理函數  
    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;  
}   
// 驅動模塊退出函數  
static void __exit myirq_exit(void)  
{  
    printk("myirq_exit is called. \n");  
    // 釋放中斷處理函數  
    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  

檢查驅動模塊是否加載成功： 

$ 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. 

說明：驅動程序的初始化函數 myirq_init 被調用了，并且成功注冊了 1 號中斷的處理程序。

此時，按一下鍵盤上的 ESC 鍵。

操作系統在捕獲到鍵盤中斷之后，會依次調用此中斷的所有中斷處理程序，其中就包括我們注冊的 myirq_handler 函數。

在這個函數中，當判斷出是ESC按鍵時，就初始化一個工作項(把結構體 work_struct 類型的變量與一個處理函數綁定起來)，然后丟給操作系統預先創建好的工作隊列(system_wq)去處理，如下所示： 

if (key_code == 0x01)  
{  
    printk("ESC key is pressed! \n");  
    INIT_WORK(&mywork, mywork_handler);  
    schedule_work(&mywork);  
} 

因此，當相應的內核線程從這個工作隊列(system_wq)中取出工作項(mywork)來處理的時候，函數 mywork_handler 就會被調用。

現在來看一下 dmesg 的輸出信息： 

[  305.053155] ESC key is pressed!   
[  305.053177] mywork_handler is called. 

可以看到：mywork_handler函數被正確調用了。

完美！