中断处理中的【工作队列】 workqueue 是什么鬼？

2021-12-28 13:45

道哥分享

关注

· 工作队列是什么

· 驱动程序

· 编译、测试

别人的经验，我们的阶梯！

大家好，我是道哥，今天我为大伙儿解说的技术知识点是：【中断处理中的下半部分机制－工作队列】。

在刚开始介绍中断处理的时候，曾经贴出下面这张图：

图中描述了中断处理中的下半部分都有哪些机制，以及如何根据实际的业务场景、限制条件来进行选择。

可以看出：这些不同的实现之间，有些是重复的，或者是相互取代的关系。

也正因为此，它们之间的使用方式几乎是大同小异，至少是在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．＂）；

／／ 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！＂）；

／／初始化工作项

INIT＿WORK（＆mywork， mywork＿handler）；

／／加入到工作队列 system＿wq

schedule＿work（＆mywork）；

｝

return IRQ＿HANDLED；

｝

／／驱动模块初始化函数

static int ＿＿init myirq＿init（void）

｛

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

／／注册中断处理函数

if（request＿irq（irq， myirq＿handler， IRQF＿SHARED， devname，＆mydev）！＝0）
｛
printk（＂register irq［％d］ handler failed．＂， irq）；
return －1；
｝
printk（＂register irq［％d］ handler success．＂， irq）；
return 0；
｝

／／驱动模块退出函数

static void ＿＿exit myirq＿exit（void）

｛
printk（＂myirq＿exit is called．＂）；
／／释放中断处理函数
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