Tasklet

Tasklet

有的时候在驱动程序中需要延迟某些操作的进行，最典型的操作就是在驱动程序的中断处理函数延迟操作。比如在DMA驱动中，当数据传输完毕之后会触发中断的，通常这时候会启动一个tasklet来完成耗时的操作，也就是中断的下半部，让中断尽早的返回。
在Softirq中说过了，Tasklet的实现是基于Softirq的。也就是说Tasklet是Softirq中的一种。根据优先级不同，Linux将Tasklet分为两类如下：

enum
{
HI_SOFTIRQ=0,
TIMER_SOFTIRQ,
NET_TX_SOFTIRQ,
NET_RX_SOFTIRQ,
BLOCK_SOFTIRQ,
BLOCK_IOPOLL_SOFTIRQ,
TASKLET_SOFTIRQ,
SCHED_SOFTIRQ,
HRTIMER_SOFTIRQ,
RCU_SOFTIRQ,    /* Preferable RCU should always be the last softirq */

NR_SOFTIRQS
};

其中HI_SOFTIRQ和TASKLET_SOFTIRQ就是提供给Tasklet使用，HI_SOFTIRQ的优先级比较高，一般驱动程序很少见到使用。

Tasklet定义

linux内核使用tasklet_struct结构体来表示一个Tasklet

struct tasklet_struct
{
struct tasklet_struct *next;
unsigned long state;
atomic_t count;
void (*func)(unsigned long);
unsigned long data;
};

next:    用于指向下一个Tasklet
state:   用于表示当前Tasklet的状态，linux内核定义了两种状态，如下：

enum
{
TASKLET_STATE_SCHED,	/* Tasklet is scheduled for execution */
TASKLET_STATE_RUN	/* Tasklet is running (SMP only) */
};

TASKLET_STATE_SCHED用于表示当前tasklet已经被提交，但是还没有执行，可能还在Tasklet列表中。
TASKLET_STATE_RUN用于表示当前tasklet正在执行，此状态只在SMP系统下有效。
count:  用于表示当前tasklet是否disable/enable。如果count=0表示tasklet是enabled,可以被调度执行，否则不可以被调度。
func:   该tasklet上的延迟回调函数，而date则是该回调函数的参数。

申明Tasklet

如果在驱动程序中需要使用tasklet，就需要先申明一个tasklet，驱动程序可以使用如下的宏同时初始化一个tasklet

#define DECLARE_TASKLET(name, func, data) \
struct tasklet_struct name = { NULL, 0, ATOMIC_INIT(0), func, data }

同样内核也提供了一个相似的宏，用来申明一个disable状态的tasklet

#define DECLARE_TASKLET_DISABLED(name, func, data) \
struct tasklet_struct name = { NULL, 0, ATOMIC_INIT(1), func, data }

除了静态申明以及初始化一个tasklet，内核也提供一个动态初始化tasklet的函数

void tasklet_init(struct tasklet_struct *t,
void (*func)(unsigned long), unsigned long data)
{
t->next = NULL;
t->state = 0;
atomic_set(&t->count, 0);
t->func = func;
t->data = data;
}

管理tasklet

驱动程序在申明了一个tasklet之后，就需要将其加入到系统的tasklet管理链表中来，内核定义了如下的结构用来管理tasklet

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);

head:  用来执行tasklet对象链接的第一个节点。
tail:     tail是个双指针，用来执行tasklet_struct指针的指针，而每个tasklet_struct中有一个next指针，所以可以理解tail总是指向最后一个节点的next。
系统为HI_SOFTIRQ和TASKLET_SOFTIRQ类型定义了各自的tasklet_head管理链表，而且tasklet_vec和tasklet_hi_vec都是per-cpu变量。
也就是说系统使用tasklet_vec用来管理系统中所有softirq类型为TASKLET_SOFTIRQ的tasklet, tasklet_hi_vec同理。

提交tasklet

当在驱动程序中初始化一个tasklet之后，在需要延迟的时候就需要将该tasklet加入到系统的tasklet_vec链表中，提交到系统中。linux系统使用tasklet_schedule函数用来提交一个tasklet

static inline void tasklet_schedule(struct tasklet_struct *t)
{
if (!test_and_set_bit(TASKLET_STATE_SCHED, &t->state))
__tasklet_schedule(t);
}

函数首先要提交的tasklet的state上的TASKLET_STATE_SCHED位是否已经置1，如果置1说明该tasklet已经被提交过，如果没有置1，则返回的是0，同时将state的TASKLET_STATE_SCHED位置1，置1之后表示这个tasklet已经被提交了。test_and_set_bit函数的作用就是返回旧的值，设置新的值。

void __tasklet_schedule(struct tasklet_struct *t)
{
unsigned long flags;

local_irq_save(flags);                                       ---------------A
t->next = NULL;                                              ---------------B
*__this_cpu_read(tasklet_vec.tail) = t;
__this_cpu_write(tasklet_vec.tail, &(t->next));
raise_softirq_irqoff(TASKLET_SOFTIRQ);                       ----------------C
local_irq_restore(flags);                                    ----------------D
}

A： 关闭本地中断，虽然tasklet_vec是per-cpu变量，但是如果不关闭中断仍然会存在并发的情况，需要关闭本地中断。

B：  此处的操作就是将t加入到tasklet_vec链表的末尾，同时需要更改tail和next指针的指向。

C： 这时候就需要触发softirq，在softirq小节有讲到。

D： 恢复中断。

Tasklet机制的初始化

在开机启动的时候，内核使用softirq_init函数来初始化softirq， 用来安装了tasklet的执行函数。

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);
}

可以看到为每个cpu都初始化了tasklet_vec和tasklet_hi_vec， 同时为TASKLET_SOFTIRQ和HI_SOFTIRQ安装了各自对于的处理函数。

执行Tasklet

当一个tasklet提交到系统之后，系统会在合适的时机处理该tasklet，最终调用到tasklet_action函数中。

static void tasklet_action(struct softirq_action *a)
{
struct tasklet_struct *list;

local_irq_disable();                                                      ------------------------A
list = __this_cpu_read(tasklet_vec.head);                                 ------------------------B
__this_cpu_write(tasklet_vec.head, NULL);                                  -----------------------C
__this_cpu_write(tasklet_vec.tail, this_cpu_ptr(&tasklet_vec.head));       ------------------------D
local_irq_enable();

while (list) {
struct tasklet_struct *t = list;

list = list->next;

if (tasklet_trylock(t)) {                                           ------------------------E
if (!atomic_read(&t->count)) {                              -----------------------F
if (!test_and_clear_bit(TASKLET_STATE_SCHED,        ------------------------H
&t->state))
BUG();
t->func(t->data);                                    -----------------------P
tasklet_unlock(t);                                   -----------------------G
continue;
}
tasklet_unlock(t);
}

local_irq_disable();                                              -----------------------Q
t->next = NULL;                                                   -----------------------X
*__this_cpu_read(tasklet_vec.tail) = t;
__this_cpu_write(tasklet_vec.tail, &(t->next));
__raise_softirq_irqoff(TASKLET_SOFTIRQ);                          ---------------------Y
local_irq_enable();
}
}

A:    因为接下来需要操作到tasklet_vec，需要关闭本地中断，防止并发操作影响到tasklet_vec的数据。
B:    将整个tasklet_vec链表的数据保存到临时变量list中。
C:    将整个tasklet_vec的head指向NULL。
D:    同时初始化tail执行为NULL
E:    这时候就从list链表取出一个tasklet，首先通过tasklet_trylock函数判断

static inline int tasklet_trylock(struct tasklet_struct *t)
{
return !test_and_set_bit(TASKLET_STATE_RUN, &(t)->state);
}

该函数就是判断state的TASKLET_START_RUN是否设置为1， 如果已经设置为1，则返回0，否则返回1。同时会设置TASKLET_START_RUN的位为1。
那这里为什么需要判断这个TASKLET_START_RUN标志呢？  根据前面的知识此标志只有在SMP系统有效。所以在SMP系统下有如下的情况：
假如有来两个处理的系统A处理器和B处理器。当前有个tasklet任务T已经提交到处理器A，并且已经调度执行了，此时TASKLET_STATE_SCHED状态已经清0。此时假设外部发生了一次中断，系统将此次中断处理权交给了B处理器，而在B的中断处理函数中调用了tasklet_schedule把tasklet T提交到B处理器，因为此taksklet T的状态TASKLET_STATE_SCHED的状态已经清0，从而可以提交成功。这样一下同一个Tasklet就有可能出现在两个处理器上执行，所以引入了TASKLET_START_RUN标志。还是前面的情况，因为在A处理器已经设置了TASKLET_STATE_RUN标志，所以在B处理器执行的时候就会失败，从而防止了统一个tasklet执行在多个处理器上。当然了这种情况只出现在SMP系统中。

F:    读取count的值，判断是否是enable的tasklet。disable的tasklet是不允许执行的。
H:    将state的TASKLET_STATE_SCHED的位清0。
P:    执行该tasklet的回调处理函数
G:    将state的TASKLET_START_RUN的位清0
Q:    因为接下来需要操作到tasklet_vec变量，为了防止中断引入并发问题，关闭中断
X:    将上述不符合条件的tasklet重新加入到tasklet_vec链表中进行管理。因为在上述的执行的过程中也有新的tasklet加入到tasklet_vec中
Y:    再次触发softirq。

其他Tasklet操作

enable一个tasklet

static inline void tasklet_enable(struct tasklet_struct *t)
{
smp_mb__before_atomic();
atomic_dec(&t->count);
}

将指定的tasklet对象t上的coun减去1。

disable一个tasklet

static inline void tasklet_unlock_wait(struct tasklet_struct *t)
{
while (test_bit(TASKLET_STATE_RUN, &(t)->state)) { barrier(); }
}

static inline void tasklet_disable_nosync(struct tasklet_struct *t)
{
atomic_inc(&t->count);
smp_mb__after_atomic();
}

static inline void tasklet_disable(struct tasklet_struct *t)
{
tasklet_disable_nosync(t);
tasklet_unlock_wait(t);
smp_mb();
}

内核提供了两个版本的disable函数，相比tasklet_disable_nosync函数tasklet_disable函数在disable的时候会调用tasklet_unlock_wait等待，主要是等待TASKLET_START_RUN的状态被清除，也就是等待tasklet被处理完毕后在disable。

kill掉一个tasklet

void tasklet_kill(struct tasklet_struct *t)
{
if (in_interrupt())
pr_notice("Attempt to kill tasklet from interrupt\n");

while (test_and_set_bit(TASKLET_STATE_SCHED, &t->state)) {
do {
yield();
} while (test_bit(TASKLET_STATE_SCHED, &t->state));
}
tasklet_unlock_wait(t);
clear_bit(TASKLET_STATE_SCHED, &t->state);

该函数最终会通过清除TAKLET_STATE_SCHED状态位，使器不再被调用。而如果当前tasklet已经提交但是没有执行，tasklet_kill将会睡眠直到该tasklet从tasklet_vec链表中删除。如果当前tasklet已经提交并且正在运行，这时候就需要等待tasklet运行完毕。该函数一般会在驱动的卸载函数中被调用到。