linux 驱动程序 tasklet 笔记

Tasklet机制是一种较为特殊的软中断。Tasklet一词的原意是“小片任务”的意思，这里是指一小段可执行的代码，且通常以函数的形式出现。软中断向量HI_SOFTIRQ和TASKLET_SOFTIRQ均是用tasklet机制来实现的。

从 某种程度上讲，tasklet机制是Linux内核对BH机制的一种扩展。在2.4内核引入了softirq机制后，原有的BH机制正是通过 tasklet机制这个桥梁来纳入softirq机制的整体框架中的。正是由于这种历史的延伸关系，使得tasklet机制与一般意义上的软中断有所不 同，而呈现出以下两个显著的特点：

1. 与一般的软中断不同，某一段tasklet代码在某个时刻只能在一个CPU上运行，而不像一般的软中断服务函数（即softirq_action结构中的action函数指针）那样——在同一时刻可以被多个CPU并发地执行。

2. 与BH机制不同，不同的tasklet代码在同一时刻可以在多个CPU上并发地执行，而不像BH机制那样必须严格地串行化执行（也即在同一时刻系统中只能有一个CPU执行BH函数）。

Linux用数据结构tasklet_struct来描述一个tasklet。该数据结构定义在include/linux/interrupt.h头文件中。如下所示：

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

各成员的含义如下：

（1）next指针：指向下一个tasklet的指针。

（2）state：定义了这个 tasklet的当前状态。这一个32位的无符号长整数，当前只使用了bit［1］和bit［0］两个状态位。其中，bit［1］＝1表示这个 tasklet当前正在某个CPU上被执行，它仅对SMP系统才有意义，其作用就是为了防止多个CPU同时执行一个tasklet的情形出 现；bit［0］＝1表示这个tasklet已经被调度去等待执行了。对这两个状态位的宏定义如下所示（interrupt.h）：

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

（3）原子计数count：对这个tasklet的引用计数值。NOTE！只有当count等于0时，tasklet代码段才能执行， 也即此时tasklet是被使能的；如果count非零，则这个tasklet是被禁止的。任何想要执行一个tasklet代码段的人都首先必须先检查其 count成员是否为0。

（4）函数指针func：指向以函数形式表现的可执行tasklet代码段。

（5）data：函数func的参数。这是一个32位的无符号整数，其具体含义可供func函数自行解释，比如将其解释成一个指向某个用户自定义数据结构的地址值。

Linux在interrupt.h头文件中又定义了两个用来定义tasklet_struct结构变量的辅助宏：

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

显然，从上述源代码可以看出，用DECLARE_TASKLET宏定义的tasklet在初始化时是被使能的（enabled），因为 其count成员为0。而用DECLARE_TASKLET_DISABLED宏定义的tasklet在初始时是被禁止的（disabled），因为其 count等于1。

在这里，tasklet状态指两个方面：1. state成员所表示的运行状态；2. count成员决定的使能／禁止状态。

（1） 改变一个tasklet的运行状态state成员中的bit［0］表示一个tasklet是否已被调度去等待执行，bit［1］表示一个tasklet是 否正在某个CPU上执行。对于state变量中某位的改变必须是一个原子操作，因此可以用定义在include/asm/bitops.h头文件中的位操 作来进行。

由于bit［1］这一位（即TASKLET_STATE_RUN）仅仅对于SMP系统才有意义，因此Linux在Interrupt.h头文件中显示地定义了对TASKLET_STATE_RUN位的操作。如下所示：

#ifdef CONFIG_SMP #define tasklet_trylock(t) (!test_and_set_bit(TASKLET_STATE_RUN, &(t)->state)) #define tasklet_unlock_wait(t) while (test_bit(TASKLET_STATE_RUN, &(t)->state)) { /* NOTHING */ } #define tasklet_unlock(t) clear_bit(TASKLET_STATE_RUN, &(t)->state) #else #define tasklet_trylock(t) 1 #define tasklet_unlock_wait(t) do { } while (0) #define tasklet_unlock(t) do { } while (0) #endif

显然，在SMP系统同，tasklet_trylock()宏将把一个tasklet_struct结构变量中的state成员中的 bit［1］位设置成1，同时还返回bit［1］位的非。因此，如果bit［1］位原有值为1（表示另外一个CPU正在执行这个tasklet代码），那 么tasklet_trylock()宏将返回值0，也就表示上锁不成功。如果bit［1］位的原有值为0，那么tasklet_trylock()宏将 返回值1，表示加锁成功。而在单CPU系统中，tasklet_trylock()宏总是返回为1。

任何想要执行某个tasklet代码的 程序都必须首先调用宏tasklet_trylock()来试图对这个tasklet进行上锁（即设置TASKLET_STATE_RUN位），且只能在 上锁成功的情况下才能执行这个tasklet。建议！即使你的程序只在CPU系统上运行，你也要在执行tasklet之前调用 tasklet_trylock()宏，以便使你的代码获得良好可移植性。

在SMP系统中，tasklet_unlock_wait()宏 将一直不停地测试TASKLET_STATE_RUN位的值，直到该位的值变为0（即一直等待到解锁），假如：CPU0正在执行tasklet A的代码，在此期间，CPU1也想执行tasklet A的代码，但CPU1发现tasklet A的TASKLET_STATE_RUN位为1，于是它就可以通过tasklet_unlock_wait()宏等待tasklet A被解锁（也即TASKLET_STATE_RUN位被清零）。在单CPU系统中，这是一个空操作。

宏tasklet_unlock()用来对一个tasklet进行解锁操作，也即将TASKLET_STATE_RUN位清零。在单CPU系统中，这是一个空操作。

（2）使能／禁止一个tasklet

使能与禁止操作往往总是成对地被调用的，tasklet_disable()函数如下

（interrupt.h）：

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

函数tasklet_disable_nosync()也是一个静态inline函数，它简单地通过原子操作将count成员变量的值减1。如下所示（interrupt.h）：

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

函数tasklet_enable()用于使能一个tasklet，如下所示（interrupt.h）：

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

函数tasklet_init()用来初始化一个指定的tasklet描述符，其源码如下所示（kernel/softirq.c）：

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

函数tasklet_kill()用来将一个已经被调度了的tasklet杀死，即将其恢复到未调度的状态。其源码如下所示（kernel/softirq.c）：

void tasklet_kill(struct tasklet_struct *t) { if (in_interrupt()) printk("Attempt to kill tasklet from interruptn"); while (test_and_set_bit(TASKLET_STATE_SCHED, &t->state)) { current->state = TASK_RUNNING; do { current->policy |= SCHED_YIELD; schedule(); } while (test_bit(TASKLET_STATE_SCHED, &t->state)); } tasklet_unlock_wait(t); clear_bit(TASKLET_STATE_SCHED, &t->state); }

多个tasklet可以通过tasklet描述符中的next成员指针链接成一个单向对列。为此，Linux专门在头文件include/linux/interrupt.h中定义了数据结构tasklet_head来描述一个tasklet对列的头部指针。如下所示：

struct tasklet_head { struct tasklet_struct *list; } __attribute__ ((__aligned__(SMP_CACHE_BYTES)));

尽管tasklet机制是特定于软中断向量HI_SOFTIRQ和TASKLET_SOFTIRQ的一种实现，但是tasklet机制 仍然属于softirq机制的整体框架范围内的，因此，它的设计与实现仍然必须坚持“谁触发，谁执行”的思想。为此，Linux为系统中的每一个CPU都 定义了一个tasklet对列头部，来表示应该有各个CPU负责执行的tasklet对列。如下所示（kernel/softirq.c）：

struct tasklet_head tasklet_vec[NR_CPUS] __cacheline_aligned; struct tasklet_head tasklet_hi_vec[NR_CPUS] __cacheline_aligned;

其中，tasklet_vec［］数组用于软中断向量TASKLET_SOFTIRQ，而tasklet_hi_vec［］数组则用于 软中断向量HI_SOFTIRQ。也即，如果CPUi（0≤i≤NR_CPUS-1）触发了软中断向量TASKLET_SOFTIRQ，那么对列 tasklet_vec［i］中的每一个tasklet都将在CPUi服务于软中断向量TASKLET_SOFTIRQ时被CPUi所执行。同样地，如果 CPUi（0≤i≤NR_CPUS-1）触发了软中断向量HI_SOFTIRQ，那么队列tasklet_vec［i］中的每一个tasklet都将 CPUi在对软中断向量HI_SOFTIRQ进行服务时被CPUi所执行。

队列tasklet_vec［I］和 tasklet_hi_vec［I］中的各个tasklet是怎样被所CPUi所执行的呢？其关键就是软中断向量TASKLET_SOFTIRQ和 HI_SOFTIRQ的软中断服务程序——tasklet_action()函数和tasklet_hi_action()函数。下面我们就来分析这两个 函数。

Linux为软中断向量TASKLET_SOFTIRQ和HI_SOFTIRQ实现了专用的触发函数和软中断服务函数。其 中，tasklet_schedule()函数和tasklet_hi_schedule()函数分别用来在当前CPU上触发软中断向量 TASKLET_SOFTIRQ和HI_SOFTIRQ，并把指定的tasklet加入当前CPU所对应的tasklet队列中去等待执行。而 tasklet_action()函数和tasklet_hi_action()函数则分别是软中断向量TASKLET_SOFTIRQ和 HI_SOFTIRQ的软中断服务函数。在初始化函数softirq_init()中，这两个软中断向量对应的描述符softirq_vec［0］和 softirq_vec［3］中的action函数指针就被分别初始化成指向函数tasklet_hi_action()和函数 tasklet_action（）。

（1）软中断向量TASKLET_SOFTIRQ的触发函数tasklet_schedule（）

该函数实现在include/linux/interrupt.h头文件中，是一个inline函数。其源码如下所示：

static inline void tasklet_schedule(struct tasklet_struct *t) { if (!test_and_set_bit(TASKLET_STATE_SCHED, &t->state)) { int cpu = smp_processor_id(); unsigned long flags; local_irq_save(flags); t->next = tasklet_vec[cpu].list; tasklet_vec[cpu].list = t; __cpu_raise_softirq(cpu, TASKLET_SOFTIRQ); local_irq_restore(flags); } }

该函数的参数t指向要在当前CPU上被执行的tasklet。对该函数的NOTE如下：

①调用 test_and_set_bit()函数将待调度的tasklet的state成员变量的bit［0］位（也即TASKLET_STATE_SCHED 位）设置为1，该函数同时还返回TASKLET_STATE_SCHED位的原有值。因此如果bit［0］为的原有值已经为1，那就说明这个 tasklet已经被调度到另一个CPU上去等待执行了。由于一个tasklet在某一个时刻只能由一个CPU来执行，因此 tasklet_schedule()函数什么也不做就直接返回了。否则，就继续下面的调度操作。

②首先，调用local_irq_save()函数来关闭当前CPU的中断，以保证下面的步骤在当前CPU上原子地被执行。

③然后，将待调度的tasklet添加到当前CPU对应的tasklet队列的首部。

④接着，调用__cpu_raise_softirq()函数在当前CPU上触发软中断请求TASKLET_SOFTIRQ。

⑤最后，调用local_irq_restore()函数来开当前CPU的中断。

（2）软中断向量TASKLET_SOFTIRQ的服务程序tasklet_action（）

函 数tasklet_action()是tasklet机制与软中断向量TASKLET_SOFTIRQ的联系纽带。正是该函数将当前CPU的 tasklet队列中的各个tasklet放到当前CPU上来执行的。该函数实现在kernel/softirq.c文件中，其源代码如下：

static void tasklet_action(struct softirq_action *a) { int cpu = smp_processor_id(); struct tasklet_struct *list; local_irq_disable(); list = tasklet_vec[cpu].list; tasklet_vec[cpu].list = NULL; local_irq_enable(); while (list != NULL) { struct tasklet_struct *t = list; list = list->next; if (tasklet_trylock(t)) { if (atomic_read(&t->count) == 0) { clear_bit(TASKLET_STATE_SCHED, &t->state); t->func(t->data); /* * talklet_trylock() uses test_and_set_bit that imply * an mb when it returns zero, thus we need the explicit * mb only here: while closing the critical section. */ #ifdef CONFIG_SMP smp_mb__before_clear_bit(); #endif tasklet_unlock(t); continue; } tasklet_unlock(t); } local_irq_disable(); t->next = tasklet_vec[cpu].list; tasklet_vec[cpu].list = t; __cpu_raise_softirq(cpu, TASKLET_SOFTIRQ); local_irq_enable(); } }

注释如下：

①首先，在当前CPU关中断的情况下，“原子”地读取当前CPU的tasklet队列头部指针，将其保存到局 部变量list指针中，然后将当前CPU的tasklet队列头部指针设置为NULL，以表示理论上当前CPU将不再有tasklet需要执行（但最后的 实际结果却并不一定如此，下面将会看到）。

②然后，用一个while{}循环来遍历由list所指向的tasklet队列，队列中的各个元素就是将在当前CPU上执行的tasklet。循环体的执行步骤如下：

• 用指针t来表示当前队列元素，即当前需要执行的tasklet。
• 更新list指针为list->next，使它指向下一个要执行的tasklet。
• 用 tasklet_trylock()宏试图对当前要执行的tasklet（由指针t所指向）进行加锁，如果加锁成功（当前没有任何其他CPU正在执行这个 tasklet），则用原子读函数atomic_read()进一步判断count成员的值。如果count为0，说明这个tasklet是允许执行的， 于是：（1）先清除TASKLET_STATE_SCHED位；（2）然后，调用这个tasklet的可执行函数func；（3）执行barrier() 操作；（4）调用宏tasklet_unlock()来清除TASKLET_STATE_RUN位。（5）最后，执行continue语句跳过下面的步 骤，回到while循环继续遍历队列中的下一个元素。如果count不为0，说明这个tasklet是禁止运行的，于是调用 tasklet_unlock()清除前面用tasklet_trylock()设置的TASKLET_STATE_RUN位。
• 如果 tasklet_trylock()加锁不成功，或者因为当前tasklet的count值非0而不允许执行时，我们必须将这个tasklet重新放回到 当前CPU的tasklet队列中，以留待这个CPU下次服务软中断向量TASKLET_SOFTIRQ时再执行。为此进行这样几步操作：（1）先关 CPU中断，以保证下面操作的原子性。（2）把这个tasklet重新放回到当前CPU的tasklet队列的首部；（3）调用 __cpu_raise_softirq()函数在当前CPU上再触发一次软中断请求TASKLET_SOFTIRQ；（4）开中断。
• 最后，回到while循环继续遍历队列。

（3）软中断向量HI_SOFTIRQ的触发函数tasklet_hi_schedule()

该函数与tasklet_schedule()几乎相同，其源码如下（include/linux/interrupt.h）：

static inline void tasklet_hi_schedule(struct tasklet_struct *t) { if (!test_and_set_bit(TASKLET_STATE_SCHED, &t->state)) { int cpu = smp_processor_id(); unsigned long flags; local_irq_save(flags); t->next = tasklet_hi_vec[cpu].list; tasklet_hi_vec[cpu].list = t; __cpu_raise_softirq(cpu, HI_SOFTIRQ); local_irq_restore(flags); } }

（4）软中断向量HI_SOFTIRQ的服务函数tasklet_hi_action（）

该函数与tasklet_action()函数几乎相同，其源码如下（kernel/softirq.c）：

static void tasklet_hi_action(struct softirq_action *a) { int cpu = smp_processor_id(); struct tasklet_struct *list; local_irq_disable(); list = tasklet_hi_vec[cpu].list; tasklet_hi_vec[cpu].list = NULL; local_irq_enable(); while (list != NULL) { struct tasklet_struct *t = list; list = list->next; if (tasklet_trylock(t)) { if (atomic_read(&t->count) == 0) { clear_bit(TASKLET_STATE_SCHED, &t->state); t->func(t->data); tasklet_unlock(t); continue; } tasklet_unlock(t); } local_irq_disable(); t->next = tasklet_hi_vec[cpu].list; tasklet_hi_vec[cpu].list = t; __cpu_raise_softirq(cpu, HI_SOFTIRQ); local_irq_enable(); } }

Bottom Half机制在新的softirq机制中被保留下来，并作为softirq框架的一部分。其实现也似乎更为复杂些，因为它是通过tasklet机制这个中 介桥梁来纳入softirq框架中的。实际上，软中断向量HI_SOFTIRQ是内核专用于执行BH函数的。原有的32个BH函数指针被保留，定义在 kernel/softirq.c文件中：

static void (*bh_base[32])(void);

但是，每个BH函数都对应有一个tasklet，并由tasklet的可执行函数func来负责调用相应的bh函数（func函数的参数指定调用哪一个BH函数）。与32个BH函数指针相对应的tasklet的定义如下所示（kernel/softirq.c）：

struct tasklet_struct bh_task_vec[32];

上述tasklet数组使系统全局的，它对所有的CPU均可见。由于在某一个时刻只能有一个CPU在执行BH函数，因此定义一个全局的自旋锁来保护BH函数，如下所示（kernel/softirq.c）：

spinlock_t global_bh_lock = SPIN_LOCK_UNLOCKED;

在softirq机制的初始化函数softirq_init()中将bh_task_vec［32］数组中的每一个tasklet中的 func函数指针都设置为指向同一个函数bh_action，而data成员（也即func函数的调用参数）则被设置成该tasklet在数组中的索引 值，如下所示：

void __init softirq_init() { …… for (i=0; i<32; i++) tasklet_init(bh_task_vec+i, bh_action, i); …… }

因此，bh_action()函数将负责相应地调用参数所指定的bh函数。该函数是连接tasklet机制与Bottom Half机制的关键所在。

该函数的源码如下（kernel/softirq.c）：

static void bh_action(unsigned long nr) { int cpu = smp_processor_id(); if (!spin_trylock(&global_bh_lock)) goto resched; if (!hardirq_trylock(cpu)) goto resched_unlock; if (bh_base[nr]) bh_base[nr](); hardirq_endlock(cpu); spin_unlock(&global_bh_lock); return; resched_unlock: spin_unlock(&global_bh_lock); resched: mark_bh(nr); }

对该函数的注释如下：

①首先，调用spin_trylock()函数试图对自旋锁global_bh_lock进行加 锁，同时该函数还将返回自旋锁global_bh_lock的原有值的非。因此，如果global_bh_lock已被某个CPU上锁而为非0值（那个 CPU肯定在执行某个BH函数），那么spin_trylock()将返回为0表示上锁失败，在这种情况下，当前CPU是不能执行BH函数的，因为另一个 CPU正在执行BH函数，于是执行goto语句跳转到resched程序段，以便在当前CPU上再一次调度该BH函数。

②调用hardirq_trylock()函数锁定当前CPU，确保当前CPU不是处于硬件中断请求服务中，如果锁定失败，跳转到resched_unlock程序段，以便先对global_bh_lock解锁，在重新调度一次该BH函数。

③此时，我们已经可以放心地在当前CPU上执行BH函数了。当然，对应的BH函数指针bh_base［nr］必须有效才行。

④从BH函数返回后，先调用hardirq_endlock()函数（实际上它什么也不干，调用它只是为了保此加、解锁的成对关系），然后解除自旋锁global_bh_lock，最后函数就可以返回了。

⑤resched_unlock程序段：先解除自旋锁global_bh_lock，然后执行reched程序段。

⑥resched程序段：当某个CPU正在执行BH函数时，当前CPU就不能通过bh_action（）函数来调用执行任何BH函数，所以就通过调用mark_bh()函数在当前CPU上再重新调度一次，以便将这个BH函数留待下次软中断服务时执行。

（1）init_bh()函数

该函数用来在bh_base［］数组登记一个指定的bh函数，如下所示（kernel/softirq.c）：

void init_bh(int nr, void (*routine)(void)) { bh_base[nr] = routine; mb(); }

（2）remove_bh()函数

该函数用来在bh_base［］数组中注销指定的函数指针，同时将相对应的tasklet杀掉。

如下所示（kernel/softirq.c）：

void remove_bh(int nr) { tasklet_kill(bh_task_vec+nr); bh_base[nr] = NULL; }

（3）mark_bh()函数

该函数用来向当前CPU标记由一个BH函数等待去执行。它实际上通过调用 tasklet_hi_schedule()函数将相应的tasklet加入到当前CPU的tasklet队列tasklet_hi_vec［cpu］ 中，然后触发软中断请求HI_SOFTIRQ，如下所示（include/linux/interrupt.h）：

static inline void mark_bh(int nr)

 

{

tasklet_hi_schedule(bh_task_vec+nr);

}

在32个BH函数指针中，大多数已经固定用于一些常见的外设，比如：第0个BH函数就固定地用于时钟中断。Linux在头文件include/linux/interrupt.h中定义了这些已经被使用的BH函数所引，如下所示：

enum { TIMER_BH = 0, TQUEUE_BH, DIGI_BH, SERIAL_BH, RISCOM8_BH, SPECIALIX_BH, AURORA_BH, ESP_BH, SCSI_BH, IMMEDIATE_BH, CYCLADES_BH, CM206_BH, JS_BH, MACSERIAL_BH, ISICOM_BH };