目錄

· 工作隊(duì)列是什么

· 驅(qū)動(dòng)程序

· 編譯、測(cè)試

別人的經(jīng)驗(yàn)，我們的階梯！

大家好，我是道哥，今天我為大伙兒解說(shuō)的技術(shù)知識(shí)點(diǎn)是：【中斷處理中的下半部分機(jī)制－工作隊(duì)列】。

在剛開(kāi)始介紹中斷處理的時(shí)候，曾經(jīng)貼出下面這張圖：

圖中描述了中斷處理中的下半部分都有哪些機(jī)制，以及如何根據(jù)實(shí)際的業(yè)務(wù)場(chǎng)景、限制條件來(lái)進(jìn)行選擇。

可以看出：這些不同的實(shí)現(xiàn)之間，有些是重復(fù)的，或者是相互取代的關(guān)系。

也正因?yàn)榇耍�它們之間的使用方式幾乎是大同小異，至少是在API接口函數(shù)的使用方式上，從使用這的角度來(lái)看，都是非常類似的。

這篇文章，我們就通過(guò)實(shí)際的代碼操作，來(lái)演示一下工作隊(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＿h(yuǎn)ead        pwqs；            WR： all pwqs of this wq

    struct list＿h(yuǎn)ead        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＿h(yuǎn)ead entry；
           work＿func＿t func；   ／／ 指向處理函數(shù)

＃ifdef CONFIG＿LOCKDEP                                                                                  
           struct lockdep＿map lockdep＿map；

＃endif

｝；

在內(nèi)核中，工作隊(duì)列中的所有工作項(xiàng)，是通過(guò)鏈表串在一起的，并且等待著操作系統(tǒng)中的某個(gè)線程挨個(gè)取出來(lái)處理。

這些線程，可以是由驅(qū)動(dòng)程序通過(guò) kthread＿create 創(chuàng)建的線程，也可以是由操作系統(tǒng)預(yù)先就創(chuàng)建好的線程。

這里就涉及到一個(gè)取舍的問(wèn)題了。

如果我們的處理函數(shù)很簡(jiǎn)單，那么就沒(méi)有必要?jiǎng)?chuàng)建一個(gè)單獨(dú)的線程來(lái)處理了。

原因有二：

1．創(chuàng)建一個(gè)內(nèi)核線程是很耗費(fèi)資源的，如果函數(shù)很簡(jiǎn)單，很快執(zhí)行結(jié)束之后再關(guān)閉線程，太劃不來(lái)了，得不償失；

2．如果每一個(gè)驅(qū)動(dòng)程序編寫(xiě)者都毫無(wú)節(jié)制地創(chuàng)建內(nèi)核線程，那么內(nèi)核中將會(huì)存在大量不必要的線程，當(dāng)然了本質(zhì)上還是系統(tǒng)資源消耗和執(zhí)行效率的問(wèn)題；

為了避免這種情況，于是操作系統(tǒng)就為我們預(yù)先創(chuàng)建好一些工作隊(duì)列和內(nèi)核線程。

我們只需要把需要處理的工作項(xiàng)，直接添加到這些預(yù)先創(chuàng)建好的工作隊(duì)列中就可以了，它們就會(huì)被相應(yīng)的內(nèi)核線程取出來(lá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＿h(yuǎn)ighpri＿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＿h(yuǎn)ighpri＿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＿h(yuǎn)ighpri＿wq ＝ alloc＿workqueue（＂events＿h(yuǎn)ighpri＂， WQ＿HIGHPRI， 0）；                          
           system＿long＿wq ＝ alloc＿workqueue（＂events＿long＂， 0， 0）；
           system＿unbound＿wq ＝ alloc＿workqueue（＂events＿unbound＂， WQ＿UNBOUND，
                                           WQ＿UNBOUND＿M(jìn)AX＿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è)簡(jiǎn)單的函數(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)委托給它們來(lái)處理，那么就會(huì)導(dǎo)致某個(gè)工作隊(duì)列中過(guò)于擁擠。

根據(jù)先來(lái)后到的原則，工作隊(duì)列中后加入的工作項(xiàng)，就可能因?yàn)榍懊婀ぷ黜?xiàng)的處理函數(shù)執(zhí)行的時(shí)間太長(zhǎng)，從而導(dǎo)致時(shí)效性無(wú)法保證。

因此，這里存在一個(gè)系統(tǒng)平衡的問(wèn)題。

關(guān)于工作隊(duì)列的基本知識(shí)點(diǎn)就介紹到這里，下面來(lái)實(shí)際操作驗(yàn)證一下。

驅(qū)動(dòng)程序

之前的幾篇文章，在驅(qū)動(dòng)程序中測(cè)試中斷處理的操作流程都是一樣的，因此這里就不在操作流程上進(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

示例代碼全貌

測(cè)試場(chǎng)景是：加載驅(qū)動(dòng)模塊之后，如果監(jiān)測(cè)到鍵盤(pán)上的ESC鍵被按下，那么就往內(nèi)核默認(rèn)的工作隊(duì)列system＿wq中增加一個(gè)工作項(xiàng)，然后觀察該工作項(xiàng)對(duì)應(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ù)中用來(lái)判斷是否需要處理

＃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＿M(jìn)ASK   0x7f

＃define KBD＿STATUS＿M(jìn)ASK     0x80

／／ 工作項(xiàng)綁定的處理函數(shù)

static void mywork＿h(yuǎn)andler（struct work＿struct ＊work）

｛

printk（＂mywork＿h(yuǎn)andler is called． ＂）；

    ／／ do some other things

｝

／／中斷處理函數(shù)

static irqreturn＿t myirq＿h(yuǎn)andler（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）

   ｛

          ／／ 讀取鍵盤(pán)掃描碼

         key＿code ＝ inb（KBD＿DATA＿REG）；

         if （key＿code ＝＝ 0x01）

       ｛

                     printk（＂ESC key is pressed！ ＂）；

                   ／／ 初始化工作項(xiàng)

                    INIT＿WORK（＆mywork， mywork＿h(yuǎn)andler）；

                 ／／ 加入到工作隊(duì)列 system＿wq

                    schedule＿work（＆mywork）；

          ｝

    ｝

    return IRQ＿HANDLED；

｝

／／ 驅(qū)動(dòng)模塊初始化函數(shù)

static int ＿＿init myirq＿init（void）

｛

       printk（＂myirq＿init is called． ＂）；

          ／／ 注冊(cè)中斷處理函數(shù)

       if（request＿irq（irq， myirq＿h(yuǎn)andler， IRQF＿SHARED， devname， ＆mydev）�。�0）
            ｛
                 printk（＂register irq［％d］ handler failed． ＂， irq）；
                 return －1；
              ｝
                 printk（＂register irq［％d］ handler success． ＂， irq）；
                 return 0；
        ｝

／／ 驅(qū)動(dòng)模塊退出函數(shù)

static void ＿＿exit myirq＿exit（void）

｛
             printk（＂myirq＿exit is called． ＂）；
                  ／／ 釋放中斷處理函數(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

編譯、測(cè)試

＄ 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．

說(shuō)明：驅(qū)動(dòng)程序的初始化函數(shù) myirq＿init 被調(diào)用了，并且成功注冊(cè)了 1 號(hào)中斷的處理程序。

此時(shí)，按一下鍵盤(pán)上的 ESC 鍵。

操作系統(tǒng)在捕獲到鍵盤(pán)中斷之后，會(huì)依次調(diào)用此中斷的所有中斷處理程序，其中就包括我們注冊(cè)的 myirq＿h(yuǎn)andler 函數(shù)。

在這個(gè)函數(shù)中，當(dāng)判斷出是ESC按鍵時(shí)，就初始化一個(gè)工作項(xiàng)（把結(jié)構(gòu)體 work＿struct 類型的變量與一個(gè)處理函數(shù)綁定起來(lái)），然后丟給操作系統(tǒng)預(yù)先創(chuàng)建好的工作隊(duì)列（system＿wq）去處理，如下所示：

if （key＿code ＝＝ 0x01）

｛

       printk（＂ESC key is pressed！ ＂）；

       INIT＿WORK（＆mywork， mywork＿h(yuǎn)andler）；

       schedule＿work（＆mywork）；

｝

因此，當(dāng)相應(yīng)的內(nèi)核線程從這個(gè)工作隊(duì)列（system＿wq）中取出工作項(xiàng)（mywork）來(lái)處理的時(shí)候，函數(shù) mywork＿h(yuǎn)andler 就會(huì)被調(diào)用。

現(xiàn)在來(lái)看一下 dmesg 的輸出信息：

［  305．053155］ ESC key is pressed！

［  305．053177］ mywork＿h(yuǎn)andler is called．

可以看到：mywork＿h(yuǎn)andler函數(shù)被正確調(diào)用了。

完美！