一、workqueue 与 tasklet 的区别
前面已经讲过中断和 tasklet 了，本文将介绍使用 workqueue 来处理中断。首先来看下它和 tasklet 有什么区别。
| 对比项 |
tasklet |
workqueue |
| 上下文 |
softirq |
进程上下文 |
| 能否睡眠 |
不能 |
能 |
| 能否拿 mutex |
不能 |
能 |
| 执行者 |
softirq 机制 |
kworker 内核线程 |
| 延迟 |
通常更低 |
受调度影响 |
| 适合 |
极短、原子性处理 |
耗时、可能睡眠 |
| 新代码推荐 |
少用 |
优先 |
tasklet 运行在软中断上下文中，操作必须是原子的；而 workqueue 运行在内核进程上下文中，操作不必是原子的。因此 workqueue 具有更大的灵活性。特别是，workqueue 可以休眠。
选择规则：
1
2
3
4
5
6
7
8
| 不能睡眠 + 非常短 + 延迟敏感：
tasklet
可能睡眠 + 需要 mutex + 工作耗时：
workqueue
处理逻辑就是中断本身的后续：
threaded IRQ
|
现代内核里，很多过去用 tasklet 的场景会改写为 threaded IRQ 或 workqueue。除非你维护旧驱动或明确需要 softirq 级延迟，否则不应该优先选择 tasklet。
二、workqueue 的使用
2.1 workqueue
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
struct work_struct {
atomic_long_t data;
struct list_head entry;
work_func_t func;
#ifdef CONFIG_LOCKDEP
struct lockdep_map lockdep_map;
#endif
};
|
创建 workqueue：
如果因为某些原因，如需要执行的是个阻塞性质的任务而不愿或不能使用内核提供的共享 workqueue，这时需要自己创建 workqueue。
创建 workqueue 时，有 2 个选择，可选择系统范围内的 ST，也可选择每 CPU 一个内核线程的 MT。
- single threaded (ST):：工作者线程的表现形式之一，在系统范围内，只有一个工作者线程为workqueue服务
- multi threaded (MT)：工作者线程的表现形式之一，在多 CPU 系统上每个 CPU 上都有一个工作者线程为 workqueue 服务
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| #include <linux/workqueue.h>
struct workqueue_struct {
struct cpu_workqueue_struct *cpu_wq;
struct list_head list;
const char *name;
int singlethread;
int freezeable;
int rt;
};
struct workqueue_struct *create_workqueue(const char *name);
struct workqueue_struct *create_singlethread_workqueue(const char *name);
void destroy_workqueue(struct workqueue_struct *queue);
|
相对于 create_singlethread_workqueue，create_workqueue 同样会分配一个 wq 的 workqueue。
不同之处在于，对于多 CPU 系统而言，create_workqueue对每一个 active 的 CPU，都会为之创建一个 per-CPU 的 cwq 结构，对应每一个 cwq，都会生成一个新的 worker_thread。
声明、初始化：
1
2
3
4
5
6
7
8
9
10
11
12
|
typedef void (*work_func_t)(struct work_struct *work);
DECLARE_WORK(work, func);
DECLARE_DELAYED_WORK(work, func);
struct work_struct work;
INIT_WORK(struct work_struct *work, work_func_t func);
PREPARE_WORK(struct work_struct *work, work_func_t func);
INIT_DELAYED_WORK(struct delayed_work *work, work_func_t func);
PREPARE_DELAYED_WORK(struct delayed_work *work, work_func_t func);
|
该系列宏最终都会创建一个以 work命名的工作项，并设置了回调函数 func，例如：
1
2
3
4
5
6
7
| void workqueue_func(void * arg);
DECLARE_WORK(work, workqueue_func);
struct work_struct work;
INIT_WORK(&work, workqueue_func);
|
调度 work：
1
2
3
4
5
|
int schedule_work(struct work_struct *work);
int queue_work(struct workqueue_struct *wq, struct work_struct *work);
|
取消 work：
1
| int cancel_delayed_work(struct work_struct *work);
|
返回值：
- 如果这个工作项在它开始执行前被取消，返回值是非零。内核保证给定工作项的执行不会在调用
cancel_delay_work 成功后被执行。
- 如果
cancel_delay_work 返回 0，则这个工作项可能已经运行在一个不同的处理器，并且仍然可能在调用 cancel_delayed_work 之后被执行。要绝对确保工作函数没有在 cancel_delayed_work 返回 0 后在任何地方运行，你必须跟随这个调用之后接着调用 flush_workqueue。在 flush_workqueue 返回后。任何在改调用之前提交的工作函数都不会在系统任何地方运行。
例子：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
| struct my_work_struct{
int test;
struct work_struct save;
};
struct my_work_struct test_work;
struct workqueue_struct *test_workqueue;
void do_save(struct work_struct *p_work)
{
struct my_work_struct *p_test_work = container_of(p_work, struct my_work_struct, save);
printk("%d\n",p_test_work->test);
}
void test_init()
{
INIT_WORK(&(test_work.save), do_save);
test_work.test = 1;
#if 0
test_workqueue = create_workqueue("test_workqueue");
if (!test_workqueue)
panic("Failed to create test_workqueue\n");
queue_work(test_workqueue, &(test_work.save));
#else
schedule_work(&(test_work.save));
#endif
}
void test_destory(void)
{
if(test_workqueue)
destroy_workqueue(test_workqueue);
}
|
2.2 delayed work
在内核中，除了 work_struct 外还有一个结构体 delayed_work。如果希望“稍后执行”，而不是“马上排队”就可以使用它。该工作队列里拥有一个 timer 定时器结构体，从而实现延时工作。
1
2
3
4
| struct delayed_work {
struct work_struct work;
struct timer_list timer;
};
|
定义：
1
| struct delayed_work my_delaywork;
|
初始化：
绑定 delaywork 对应的函数
1
| INIT_DELAYED_WORK(&my_delaywork , my_delaywork_func);
|
实现对应的回调函数：
1
2
3
4
| static void my_delaywork_func(struct work_struct *work)
{
printk("delaywork running\n");
}
|
调度 delaywork：
1
2
3
4
5
6
7
|
static inline bool schedule_delayed_work(struct delayed_work *dwork,
unsigned long delay);
int queue_delayed_work(struct workqueue_struct *wq,struct delayed_work *dwork, unsigned long delay);
int queue_delayed_work_on(int cpu, struct workqueue_struct *wq,
struct delayed_work *dwork, unsigned long delay);
|
使用delayed workqueue最主要的是调用queue_delayed_work。
**描述：**提交work到某条wq，工作项随后在某个合适时机将被执行。
参数解析：
- wq：workqueue
- work：需要执行的任务。
- dalay：延迟时间，以 ms 为单位。（内部通过注册定时器实现延迟）最少延迟 delay jiffies 之后该工作才会被执行
例子：
1
| schedule_delayed_work(&my_delaywork , msecs_to_jiffies(200));
|
取消 delaywork：
1
| cancel_delayed_work_sync(&my_delaywork);
|
例子：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
| #include <linux/workqueue.h>
struct my_work_struct{
int test;
struct delayed_work my_delay_work;
};
static struct my_work_struct test_work;
static struct workqueue_struct *test_workqueue;
void do_my_delay_work(struct work_struct *p_work)
{
struct my_work_struct *p_test_work = container_of(p_work, struct my_work_struct, my_delay_work.work);
printk("%d\n",p_test_work->test);
schedule_delayed_work(&(test_work.my_delay_work), msecs_to_jiffies(200));
}
void test_init()
{
unsigned long delay_ms;
INIT_DELAYED_WORK(&(test_work.my_delay_work), do_my_delay_work);
test_work.test = 1;
delay_ms = msecs_to_jiffies(200);
#if 1
test_workqueue = create_workqueue("test_workqueue");
if (!test_workqueue)
panic("Failed to create test_workqueue\n");
queue_delayed_work(test_workqueue, &(test_work.my_delay_work), delay_ms);
#else
schedule_delayed_work(&(test_work.my_delay_work), delay_ms);
#endif
}
void test_destory(void)
{
if(test_workqueue)
destroy_workqueue(test_workqueue);
}
|