中财网学术前沿报告范文:LINUX系统中断服务下半部之tasklet详解
来源：百度文库 编辑：中财网 时间：2024/05/05 22:47:24
【摘要】本文详解了中断服务下半部之tasklet实现机制。介绍了tasklet链表的组织形式tasklet_vec，在此基础之上分析了tasklet执行流程。最后介绍了tasklet相关的API，如何编写自己的tasklet处理程序及定义一个tasklet对象并向内核提交等待调度运行。
【关键字】中断下半部，tasklet，tasklet_vec，tasklet_schedule，TASKLET_SOFTIRQ
1 tasklet概述
tasklet是利用软中断实现的一种下半部机制。tasklet和软中断在本质上很相似，行为表现也相近，选择到底是用软中断还是tasklet其实很简单：
tasklet内部对软中断进行了封装，外部接口更简单，锁保护也要求较低。
下半部和推后执行的工作，软中断的使用者屈指可数。它只在那些执行频率很高和连续性要求很高的情况下才需要。
因为tasklet是通过软中断实现的，所以它们本身也是软中断。Tasklet有两类软中断代表：HI_SOFTIRQ和TASKLET_SOFTIRQ。这两者之间惟一的实际区别在于HI_SOFTIRQ类型的软中断先于TASKLET_SOFTIRQ类型的软中断执行。
2 Tasklet的组织形式
2.1 tasklet_struct定义
tasklet由tasklet_struct结构表示。每个结构体单独代表一个tasklet，它在< include/linux/interrupt.h>中定义:
260/* Tasklets --- multithreaded analogue of BHs.
262 Main feature differing them of generic softirqs: tasklet is running only on one CPU simultaneously.
265 Main feature differing them of BHs: different tasklets may be run simultaneously on different CPUs.
268 Properties:
269 * If tasklet_schedule() is called, then tasklet is guaranteed to be executed on some cpu at least once after this.
271 * If the tasklet is already scheduled, but its excecution is still not
272 started, it will be executed only once.
273 * If this tasklet is already running on another CPU (or schedule is called
274 from tasklet itself), it is rescheduled for later.
275 * Tasklet is strictly serialized wrt itself, but not wrt another tasklets. If client needs some intertask synchronization, he makes it with spinlocks.
278 */
280struct tasklet_struct
281{
282 struct tasklet_struct *next;
283 unsigned long state;
284 atomic_t count;
285 void (*func)(unsigned long);
286 unsigned long data;
287};
next：链表中的下一个tasklet；
state：tasklet的状态；
count：引用计数器；
func：tasklet处理函数；
data：给tasklet处理函数的参数
state成员只能在0、TASKLET_STATE_SCHED和TASKLET_STATE_RUN之间取值。
TASKLET_STATE_SCHED表明tasklet已被调度，正准备投入运行，TASKLET_STATE_RUN表明该tasklet正在某CPU上运行。TASKLET_STATE_RUN只有在多处理器的系统上才会作为一种优化来使用，单处理器系统任何时候都清楚单个tasklet是不是正在运行。
count成员是tasklet的引用计数器。如果它不为0，则tasklet被禁止，不允许执行；只有当它为0时，tasklet才被激活，并且在被设置为TASKLET_STATE_SCHED状态时，该tasklet才能够执行。
2.2 Tasklet队列tasklet_vec
已调度的tasklet(等同于被触发的软中断))a存放在两个单处理器数据结构：tasklet_vec (普通tasklet)和tasklet_hi_vec(高优先级的tasklet)中。这两个数据结构都是由tasklet_struct结构体构成的链表。链表中的每个tasklet_struct代表一个不同的tasklet。
333struct tasklet_head
334{
335 struct tasklet_struct *list;
336};
337
340static DEFINE_PER_CPU(struct tasklet_head, tasklet_vec) = { NULL };
341static DEFINE_PER_CPU(struct tasklet_head, tasklet_hi_vec) = { NULL };
tasklet由tasklet_schedule()和tasklet_hi_schedule()函数进行调度，它们接受一个指向tasklet_struct结构的指针作为参数。tasklet_schedule()将此tasklet添加到当前CPU的tasklet_vec链表的头部，并置上TASKLET_STATE_SCHED标识，然后置软中断触发表示，等待调度软中断时再执行tasklet_vec中所有注册的tasklet。
3 检查当前tasklet的运行状态
在SMP机器上，tasklet机制可以保证同一个tasklet不会同时在多个CPU上同时运行，其是通过state域实现的。当其为TASKLET_STATE_RUN时表示其他CPU正在运行当前tasklet，则本CPU上的相关工作推后进行。
这些机制已经封装好了，用户程序不用管。这样相当于为用户提供了更简单可靠的接口。
302#ifdef CONFIG_SMP
303static inline int tasklet_trylock(struct tasklet_struct *t)
304{
305 return !test_and_set_bit(TASKLET_STATE_RUN, &(t)->state);
306}
若不是TASKLET_STATE_RUN状态，则设置为TASKLET_STATE_RUN，防止其他CPU调度，返回成功；否则返回失败。
307
308static inline void tasklet_unlock(struct tasklet_struct *t)
309{
310 smp_mb__before_clear_bit();
311 clear_bit(TASKLET_STATE_RUN, &(t)->state);
312}
当前CPU上的处理工作完成后，清除TASKLET_STATE_RUN。
313
314static inline void tasklet_unlock_wait(struct tasklet_struct *t)
315{
316 while (test_bit(TASKLET_STATE_RUN, &(t)->state)) { barrier(); }
317}
等待其他处理器上处理完毕，否则barrier。
318#else 单CPU上，一切为空。
319#define tasklet_trylock(t) 1
320#define tasklet_unlock_wait(t) do { } while (0)
321#define tasklet_unlock(t) do { } while (0)
322#endif
4 执行tasklet
当tasklet挂起等待运行后，do_softirq()会尽可能早地在下一个合适的时机执行。由于大部分tasklet和软中断都是在中断处理程序中被设置成待处理状态，所以最近一个中断返回的时候看起来就是执行do_softirq()的最佳时机。因为TASKLET_SOFTIRQ和HI_SOFTIRQ已经被触发，所以do_softirq()会执行相应的软中断处理程序。
tasklet_action为注册的对应tasklet软中断执行函数，传递的参数为softirq_action，符合softirq的API接口。
369static void tasklet_action(struct softirq_action *a)
370{
371 struct tasklet_struct *list;
372
373 local_irq_disable();
374 list = __get_cpu_var(tasklet_vec).list;
375 __get_cpu_var(tasklet_vec).list = NULL;
376 local_irq_enable();
377
378 while (list) {
379 struct tasklet_struct *t = list;
380
381 list = list->next;
382
383 if (tasklet_trylock(t)) {
384 if (!atomic_read(&t->count)) {
385 if (!test_and_clear_bit(TASKLET_STATE_SCHED, &t->state))
386 BUG();
387 t->func(t->data);
388 tasklet_unlock(t);
389 continue;
390 }
391 tasklet_unlock(t);
392 }
393
394 local_irq_disable();
395 t->next = __get_cpu_var(tasklet_vec).list;
396 __get_cpu_var(tasklet_vec).list = t;
397 __raise_softirq_irqoff(TASKLET_SOFTIRQ);
398 local_irq_enable();
399 }
400}
tasklet_action和tasklet_hi_action是tasklet处理的核心。其流程如下：
1) 373行，禁止中断。没有必要首先保存其状态，因为这里的代码总是作为软中断被调用，而且中断总是被激活的。
2) 374行，得到注册在当前处理器上的tasklet链表tasklet_vec或tasklet_hi_vec。
3) 375行，将当前处理器上的该链表设置为NULL，达到清空的效果。
4) 376行，允许响应中断。
5) 378行，循环遍历获得链表上的每一个待处理的tasklet。
6) 379行，得到当前的链表头。
7) 381行，保存后续链表。
8) 383行，如果是多处理器系统，通过检查TASKLET_STATE_RUN状态标志来判断这个tasklet是否注册到其他CPU上并且目前正在其他处理器上运行。如果没有运行，将其状态标志设置为TASKLET_STATE_RUN，这样别的处理器就不会再去执行它了，转384，否则转394行。这就保证了同一时间里，相同类型的tasklet只能有一个执行。
9) 384行，检查count值是否为0，确保tasklet没有被禁止。如果没有被禁止，则执行其注册的函数，tasklet运行完毕，清除tasklet的state域的TASKLET_STATE_RUN状态标志。然后回378行。如果tasklet被禁止了，则跳到391行。
10) 394行，将当前禁止的或者其他CPU正在处理的tasklet保存在tasklet_vec链表头部，重设TASKLET_SOFTIRQ标识，等待下次调度。回378行，重复执行下一个tasklet，直至没有剩余的等待处理的tasklet。
Tasklet的实现很简单，但非常巧妙。我们可以看到，所有的tasklet都通过重复运用TASKLET_SOFTIRQ和HI_SOFTIRQ这两个软中断来实现。当一个tasklet被调度时，内核就会唤起这两个软中断中的一个。随后，该软中断会被特定的函数处理，执行所有已调度的tasklet。这个函数保证同一时间里某tasklet只能在一个CPU上运行，但其他不同类型的tasklet可以同时执行。
5 Tasklet的API
5.1 自定义tasklet处理程序
tasklet处理程序必须符合规定的函数类型:
void tasklet_handler(unsigned long data)
因为是靠软中断实现，这意味着tasklet处理程序应注意如下几点：
2 tasklet不能睡眠。这意味着你不能在tasklet中使用信号量或者其他什么阻塞式的函数。
2 由于tasklet运行时允许响应中断，如果你的tasklet和中断处理程序之间共享了某些数据的话，所以你必须做好预防工作(比如屏蔽中断然后获取一个锁)。
2 两个相同的tasklet决不会在不同CPU上同时执行，这点和软中断不同，tasklet自身无需实现SMP的互斥。
2 但两个不同的tasklet可以在两个处理器上同时执行。如果你的tasklet和其他的tasklet或者是软中断共享了数据，你必须进行适当地锁保护。
5.2 Tasklet的初始化
大多数情况下，为了控制一个寻常的硬件设备，tasklet机制都是实现你自己的下半部的最佳选择。tasklet可以静态地创建或者动态创建，使用方便，执行起来也还算快。你既可以全局静态地创建tasklet，也可以动态地创建它。选择哪种方式取决于你到底是有一个对tasklet的直接引用还是间接引用。最好使用系统提供的初始化API，避免直接操作tasklet的成员，这样可以提高可移植性。
静态创建一个tasklet可使用下面include/linux/interrupt.h中定义的两个宏中的一个:
289#define DECLARE_TASKLET(name, func, data) \
290struct tasklet_struct name = { NULL, 0, ATOMIC_INIT(0), func, data }
291
292#define DECLARE_TASKLET_DISABLED(name, func, data) \
293struct tasklet_struct name = { NULL, 0, ATOMIC_INIT(1), func, data }
这两个宏都能根据给定的名称静态地创建一个tasklet_struct结构。当该tasklet被调度以后，给定的函数func会被执行，它的参数由data给出。这两个宏之间的区别在于引用计数器的初始值设置不同。前面一个宏把创建的tasklet的引用计数器设置为0，该tasklet处于激活状态。另一个把引用计数器设置为1，所以该tasklet处于禁止状态。
下面是一个例子:
DECLARE_TASKLET(my_tasklet, my_tasklet_handler, dev);
这样就创建了一个名为my_tasklet，处理程序为tasklet_handler并且已被激活的tasklet。当处理程序被调用的时候，dev就会被传递给它。
还可以通过将一个间接引用(一个指针)赋给一个动态创建的tasklet_struct结构的方式来初始化一个tasklet。该函数可以在运行过程中动态初始化未初始化或者已经使用过的tasklet。
436void tasklet_init(struct tasklet_struct *t,
437 void (*func)(unsigned long), unsigned long data)
438{
439 t->next = NULL;
440 t->state = 0;
441 atomic_set(&t->count, 0);
442 t->func = func;
443 t->data = data;
444}
5.3 调度tasklet
通过调用tasklet_schedule()函数并传递给它相应的tasklet_struct的指针，该 tasklet就会被调度以便执行:
tasklet_schedule(&my_tasklet);/*把my_tasklet标记为挂起*/
在tasklet被调度以后，只要有机会它就会尽可能早地运行。在它还没有得到运行机会之前，
如果有一个相同的tasklet又被调度了，那么它仍然只会运行一次。而如果这时它已经开始运行了，比如说在另外一个处理器上，那么这个新的tasklet会被重新调度并再次运行。作为一种优化措施，一个tasklet总在调度它的处理器上执行—这是希望能更好地利用处理器的高速缓存。
include/linux/interrupt.h
326static inline void tasklet_schedule(struct tasklet_struct *t)
327{
328 if (!test_and_set_bit(TASKLET_STATE_SCHED, &t->state))
329 __tasklet_schedule(t);
/////////////////////
kernel/softirq.c
343void fastcall __tasklet_schedule(struct tasklet_struct *t)
344{
345 unsigned long flags;
346
347 local_irq_save(flags);
348 t->next = __get_cpu_var(tasklet_vec).list;
349 __get_cpu_var(tasklet_vec).list = t;
350 raise_softirq_irqoff(TASKLET_SOFTIRQ);
351 local_irq_restore(flags);
352}
/////////////////////
330}
tasklet_schedule()流程如下:
1) 检查tasklet的状态是否为TASKLET_STATE_SCHED。如果是，说明tasklet已经被调度过(有可能是一个tasklet已经被调度过但还没来得及执行，而该tasklet又被唤起了一次)，函数立即返回。
2) 保存中断状态，然后禁止本地中断。
3) 把需要调度的tasklet加到每个处理器一个的tasklet_vec链表或tasklet_hi_vec链表的表头上去。
4) 唤起TASKLET_SOFTIRQ或HI_SOFTIRQ软中断，这样在下一次调用do_softirq()时就会执行该tasklet。
5) 恢复中断到原状态并返回。
5.4 禁止或者使能tasklet
341static inline void tasklet_disable_nosync(struct tasklet_struct *t)
342{
343 atomic_inc(&t->count);
344 smp_mb__after_atomic_inc();
345}
可用来禁止指定的tasklet，不过它无须在返回前等待tasklet执行完毕。这么做往往不太安全，因为你无法估计该tasklet是否仍在执行。
347static inline void tasklet_disable(struct tasklet_struct *t)
348{
349 tasklet_disable_nosync(t);
350 tasklet_unlock_wait(t);
351 smp_mb();
352}
你可以调用tasklet_disable()函数来禁止某个指定的tasklet。如果该tasklet当前正在执行，这个函数会等到它执行完毕再返回。
354static inline void tasklet_enable(struct tasklet_struct *t)
355{
356 smp_mb__before_atomic_dec();
357 atomic_dec(&t->count);
358}
调用tasklet_enable()函数可以激活一个tasklet，如果希望激活DECLARE_TASKLET_DISABLED ()创建的tasklet，你也得调用这个函数，如:
tasklet_enable(&my_tasklet):/*tasklet现在被激活*/
5.5 删除tasklet
你可以通过调用tasklet_kill()函数从挂起的队列中去掉一个tasklet。该函数的参数是一个指向某个tasklet的tasklet_struct的长指针。在处理一个经常重新调度它自身的tasklet的时候，从挂起的队列中移去已调度的tasklet会很有用。这个函数首先等待该tasklet执行完毕，然后再将它移去。由于该函数可能会引起休眠，所以禁止在中断上下文中使用它。
448void tasklet_kill(struct tasklet_struct *t)
449{
450 if (in_interrupt())
451 printk("Attempt to kill tasklet from interrupt\n");
452
453 while (test_and_set_bit(TASKLET_STATE_SCHED, &t->state)) {
454 do
455 yield();
456 while (test_bit(TASKLET_STATE_SCHED, &t->state));
457 }
458 tasklet_unlock_wait(t);
459 clear_bit(TASKLET_STATE_SCHED, &t->state);
460}