原图

为什么有 tasklet

linux kernel 已经把中断处理分成了 top half 和 bottom half,看起来已经不错了,那为何还要提供 softirq、tasklet 和 workqueue 这些 bottom half 机制。

workqueue 和 softirq、tasklet 有本质的区别:

  • workqueue 运行在 process context,而 softirq 和 tasklet 运行在 interrupt context。因此,出现 workqueue 是不奇怪的,在有 sleep 需求的场景中需要。
  • softirq 更倾向于性能。软中断的种类是编译时静态定义的,在运行时不能添加或删除,同一种软中断的处理函数可以在多个处理器上同时执行,处理函数必须是可以重入的,需要使用锁保护临界区。
  • tasklet 更倾向于易用性。Tasklet 可以在运行时添加或删除,一个 Tasklet 同一时刻只能在一个处理器上执行,不要求处理函数是可以重入的。

数据结构

Tasklet 的数据结构如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
<include/linux/interrupt.h>
struct tasklet_struct
{
struct tasklet_struct *next;
unsigned long state;
atomic_t count;
bool use_callback;
union {
void (*func)(unsigned long data);
void (*callback)(struct tasklet_struct *t);
};
unsigned long data;
};

成员 next 用来把 Tasklet 添加到单向链表中。 成员 state 是 Tasklet 的状态,取值如下:

  • 0: Tasklet 没有被调度
  • (1 << TASKLET_STATE_SCHED): Tasklet 被调度,即将被执行
  • (1<< TASKLET_STATE_RUN):只在多处理器系统中使用,表示 Tasklet 正在执行

成员 count 是计数,0 表示允许 Tasklet 被执行,非零值表示禁止 Tasklet 被执行成员 func 是处理函数,成员 data 是传给处理函数的参数。 每个处理器有两条单向链表:低优先级 Tasklet 链表和高优先级 Tasklet 链表。

1
2
3
4
5
6
7
<kernel/softirq.c>
struct tasklet_head {
struct tasklet_struct *head;
struct tasklet_struct **tail;
};
static DEFINE_PER_CPU(struct tasklet_head, tasklet_vec);
static DEFINE_PER_CPU(struct tasklet_head, tasklet_hi_vec);

实现(扩展🙈)

Tasklet 是基于软中断实现的,根据优先级分为两种,低优先级 Tasklet 和高优先级 Tasklet。软中断 HI_SOFTIRO 执行高优先级 Tasklet,软中断 TASKLET_SOFTIRQ 执行低优先级 Tasklet。

调度 Tasklet

函数 tasklet_schedule() 用来调度低优先级 Tasklet,函数 tasklet_hi_schedule() 用来调度高优先级 Tasklet。以函数 tasklet_schedule() 为例说明,其代码如下

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
<include/linux/interrupt.h、kernel/softirq.c>
static inline void tasklet_schedule(struct tasklet_struct *t)
{
if (!test_and_set_bit(TASKLET_STATE_SCHED, &t->state))
__tasklet_schedule(t);
}
...
void __tasklet_schedule(struct tasklet_struct *t)
{
__tasklet_schedule_common(t, &tasklet_vec,
TASKLET_SOFTIRQ);
}
...
static void __tasklet_schedule_common(struct tasklet_struct *t,
struct tasklet_head __percpu *headp,
unsigned int softirq_nr)
{
struct tasklet_head *head;
unsigned long flags;

local_irq_save(flags);
head = this_cpu_ptr(headp);
t->next = NULL;
*head->tail = t;
head->tail = &(t->next);
raise_softirq_irqoff(softirq_nr);
local_irq_restore(flags);
}

如果 Tasklet 没有被调度过,那么首先设置调度标志位,然后把 Tasklet 添加到当前处理器的低优先级 Tasklet 链表的尾部,最后触发软中断 TASKLET_SOFTIRQ。

执行 Tasklet

初始化的时候,把软中断 TASKLET_SOFTIRQ 的处理函数注册为函数 tasklet_action,把软中断 HI_SOFTIRQ 的处理函数注册为函数 tasklet_hi_action()。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
<kernel/softirq.c>
void __init softirq_init(void)
{
int cpu;

for_each_possible_cpu(cpu) {
per_cpu(tasklet_vec, cpu).tail =
&per_cpu(tasklet_vec, cpu).head;
per_cpu(tasklet_hi_vec, cpu).tail =
&per_cpu(tasklet_hi_vec, cpu).head;
}

open_softirq(TASKLET_SOFTIRQ, tasklet_action);
open_softirq(HI_SOFTIRQ, tasklet_hi_action);
}

以函数 tasklet_action() 为例说明,其代码如下

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
47
48
49
50
51
52
53
54
55
56
57
<kernel/softirq.c>
static __latent_entropy void tasklet_action(struct softirq_action *a)
{
tasklet_action_common(a, this_cpu_ptr(&tasklet_vec), TASKLET_SOFTIRQ);
}
static void tasklet_action_common(struct softirq_action *a,
struct tasklet_head *tl_head,
unsigned int softirq_nr)
{
struct tasklet_struct *list;

local_irq_disable();
list = tl_head->head;
tl_head->head = NULL;

/* 1. 把当前处理器的低优先级 Tasklet 链表中的所有 Tasklet 移到临时链表 list 中 */
tl_head->tail = &tl_head->head;
local_irq_enable();

/* 2. 遍历临时链表 list,依次处理每个 Tasklet */
while (list) {
struct tasklet_struct *t = list;

list = list->next;

/* 3. ,尝试锁住 Tasklet,确保一个 Tasklet 同一时刻只在一个处理器上执行 */
if (tasklet_trylock(t)) {

/* 4. 如果 Tasklet 的计数为 0,表示允许 Tasklet 被执行 */
if (!atomic_read(&t->count)) {

/* 5. 清除 Tasklet 的调度标志位,其他处理器可以调度这个 Tasklet,但是不能执行这个 Tasklet */
if (tasklet_clear_sched(t)) {
if (t->use_callback)
/* 6. 执行 Tasklet 的处理函数 */
t->callback(t);
else
/* 6. 执行 Tasklet 的处理函数 */
t->func(t->data);
}
tasklet_unlock(t);
continue;
}

/* 7. 释放 Tasklet 的锁,其他处理器就可以执行这个 Tasklet 了 */
tasklet_unlock(t);
}

/* 8. 如果尝试锁住 Tasklet 失败(表示 Tasklet 正在其他处理器上执行),或者禁止 Tasklet 被执行,那么把 Tasklet 重新添加到当前处理器的低优先级 Tasklet 链表的尾部,然后触发软中断 TASKLET_SOFTIRQ */
local_irq_disable();
t->next = NULL;
*tl_head->tail = t;
tl_head->tail = &t->next;
__raise_softirq_irqoff(softirq_nr);
local_irq_enable();
}
}

在上面的注释 4 如果 Tasklet 的计数为 0,表示允许 Tasklet 被执行 ,tasklet 就是通过这个来保证只在一个 cpu 上执行。如果该 tasklet 已经在别的 cpu 上执行了,那么我们将其挂入该 cpu 的 tasklet 链表的尾部,这样,在下一个 tasklet 执行时机到来的时候,kernel 会再次尝试执行该 tasklet,在这个时间点,也许其他 cpu 上的该 tasklet 已经执行完毕了。通过这样代码逻辑,保证了特定的 tasklet 只会在一个 cpu 上执行,不会在多个 cpu 上并发。

参考文献

http://www.wowotech.net/irq_subsystem/soft-irq.html

http://www.wowotech.net/irq_subsystem/tasklet.html

《Linux 内核深度解析》