Linux内核设计与实现(6)—下半部和推后执行的工作

6.1下半部

下半部的任务就是执行与中断处理密切相关但中断处理程序本身不执行的工作。对于在上半部和下半部之间划分工作，尽管不存在某种严格的规则，但还是有一些提示可供借鉴：（1）如果一个任务对时间非常敏感，将其放在中断处理程序中执行。（2）如果一个任务和硬件相关，将其放在中断处理程序中执行。（3）如果一个任务要保证不被其它中断打断，将其放在中断处理程序中执行。（4）其它所有任务，考虑放在下半部执行。当我们开始尝试写自己的驱动程序的时候，读一下别人的中断处理程序和相应的下半部会令你受益匪浅。现在的问题是：下半部具体放到以后的什么时候去做呢？下半部并不需要指明一个确切时间，只要把这些任务推迟一点，让他们在系统不太繁忙并且中断恢复后执行就可以了。通常下半部在中断处理程序一返回就会马上执行。下半部执行的关键在于当它们运行的时候，允许响应所有中断。

 

6.1.1为什么要用下半部？

因为在中断处理程序运行的时候，当前的中断线会被屏蔽，如果一个处理程序是SA_INTERRUPT类型，它执行的时候会禁止所有本地中断（而且把本地中断线全局屏蔽掉），再加上中断处理程序要与其它程序——甚至是其它的中断处理程序——异步执行。

具体放到以后什么时候去做呢？

在这里，“以后”仅仅用来强调不是“马上”而已，下半部并不需要指明一个确切时间，只是把这些任务推迟一点，让它们在系统不太繁忙并且中断恢复后执行就可以了，通常下半部在中断处理程序一返回就会马上执行，下半部执行的关键在于当它们运行的时候，允许响应所有的中断。

 

6.2软中断

软中断是用软件方式模拟硬件中断的概念，实现宏观上的异步执行效果，tasklet也是基于软中断实现的。

异步通知所基于的信号也类似于中断。

硬中断是外部设备对CPU的中断

软中断通常是硬中断服务程序对内核的中断。

信号则是由内核（或其它进程）对某个进程的中断。

 

6.2.1软中断的实现

       软中断是在编译期间静态分配的。不像tasklet那样能被动态的注册或去除。软中断由softirq_action结构表示，它定义在中：

struct softirq_action {

             void( *action)(struct softirq_action *);         

/*待执行的函数*/

         Void*date;             /传递给函数的参数*/

                                  } ;

在kernel/softirq.c中定义了一个包含有32个该结构体的数组。     

static strcut softirq_action softirq_vec[32]; 每个注册的软中断都占据该数组中的一项。

（1）       软中断处理程序：

软中断处理程序action的函数原型如下：

void softirq_handler(struct softirq_action *)

当内核运行一个软中断处理程序的时候，它就会执行这个action函数，其唯一的参数为指向相应的softirq_action结构体的指针。

一个软中断不会抢占另外一个软中断，实际上，唯一可以抢占软中断的是中断处理程序，不过，其它的软中断——甚至是相同类型的软中断——可以在其它处理器上同时执行。

（2）       执行软中断：

一个注册的软中断必须在被标记后才会执行。这被称作触发软中断（raising the softirq）。通常，中断处理程序会在返回前标记它的软中断，使其在稍后被执行。软中断被标记后，可以用softirq_pending()检查到这个标记并按照索引号将softirq_pending()的返回值的相应位置1。

在合适的时刻，该软中断就会运行，在下列地方，待处理的软中断会被检查和执行：

在处理完一个硬中断以后

在ksoftirqd内核线程中

在那些显式检查和执行待处理的软中断的代码中，如网络子系统中

不管是用什么办法唤起，软中断都要在do_softirq（）中执行，该函数很简单，如果有待处理的软中断，do_softirq（）会遍历每一个，调用它们的处理程序。

软中断在do_softirq()中执行。do_softirq()经过简化后的核心部分：

u32 pending = sofeirq_pending(cpu);

if(pending) {

struct softirq_action *h = softirq_vec;

softirq_pending(cpu) = 0;

do {

       if(pending&1) h->action(h);    //调用action函数

       h++;

      pending>>=1;

       }while(pending);

}

 

6.2.2使用软中断

软中断保留给系统中对时间要求最严格以及最重要的下半部使用。内核定时器和tasklets都是建立在软中断上的，如果你想加入一个新的软中断，首先要想想为什么用tasklet实现不了，tasklet可以动态生成，由于它们对加锁的要求不高，所以使用起来也很方便，当然，对于时间要求养并能自己高效的完成加锁工作的应用，软中断会是正确的选择。

1、  分配索引：在编译期间，可以通过中定义的一个枚举类型来静态的声明软中断。

2、  注册处理程序：接着，在运行时通过调用open_softirq()注册软件中断处理程序，该函数有三个参数：索引号、处理函数和data域存放的数值。例如网络子系统，通过以下方式注册自己的软中断：

open_softirq(NET_TX_SOFTIRQ, net_tx_action,NULL)；

open_softirq(NET_TX_SOFTIRQ, net_rx_action,NULL)；

软中断处理程序的执行的时候，允许响应中断，但自己不能睡眠。

3、  触发你的软中断：

通过在枚举类型的列表中添加新项以及调用open_softirq()进行注册以后，新的软中断处理程序就能够运行。raise_softirq()函数可以将一个软中断设置为挂起状态，让他在下次调用do_softirq()函数时投入运行。一个例子：

raise_softirq(NET_TX_SOFTIRQ);

这会触发NET_TX_SOFTIRQ软中断。它的处理程序net_tx_action()就会在内核下一次执行软中断时投入运行。该函数在触发一个软中断前要禁止中断，触发后再恢复回原来的状态。在中断处理程序中触发软中断是最常见的形式。这样，内核在执行完中断处理程序后，马上就会调用do_softirq。于是软中断开始执行中断处理程序留给它去完成的剩余任务。

 

6.3 Tasklet

tasklet是利用软中断实现的一种下半部机制。它和进程没有任何关系。它和软中断本质上很相似，行为表现也相近，但是，它的接口更简单，锁保护也要求较低。

软中断和tasklet怎样选择呢？

通常你应该用tasklet，软中断一般用的很少，它只在那些执行频率很高和连续性要求很高的情况下才需要，而tasklet却有更广泛的用途。

 

6.3.1 Tasklet的实现

因为tasklet是通过软中断实现的，所以它本身也是软中断。

（1）tasklet结构体：tasklet由tasklet_struct结构表示。每个结构体单独代表一个tasklet，它在中定义：

struct tasklet_struct {

            struct task_struct   *next;      /*指向链表中的下一个tasklet */

            unsigned   long   state;         /* tasklet的状态 */

             atomic_t    count;                 /* 引用计数器 */

             void (*func) (unsigned long);     /* tasklet处理函数 */

            unsigned long data;                  /*给tasklet处理函数的参数 */

}；

结构体中的func成员是tasklet的处理程序，data是它唯一的参数。state成员只能在0、TASKLET_STATE_SCHED和TASKLET_STATE_RUN之间取值。TASKLET_STATE_SCHED表明tasklet已经被调度，正准备投入运行，TASKLET_STATE_RUN表示该tasklet正在运行。只有count为0时，tasklet才被激活，否则被允许，不允许执行。

 

调度tasklet

已调度的tasklet存放在两个单处理器数据结构：tasklet_vec和task_hi_vec中。它们都是由tasklet_struct结构体构成的链表。链表中的每个tasklet_struct代表一个不同的tasklet。

tasklet是由tasklet_schedule()和tasklet_hi_schedule()函数进行调度的，它们接受一个指向tasklet_struct结构的指针作为参数。

Tasklet的实现通过软中断来实现的，tasklet_schedule()调度函数执行一些初始工作，紧接着唤起TASKLET_SOFTIRQ或HI_SOFTIRQ软中断，这样在下一次调用do_softirq（）时就会执行tasklet。

那么do_softirq（）函数什么时候执行呢？

do_softirq（）会尽可能早的在下一个合适的时机执行，由于大部分tasklet和软中断都是在中断处理程序中被设置成待处理状态，所以最近一个中断返回的时候看起来就是执行do_softirq（）的最佳时机。因为TASKLET_SOFTIRQ和HI_SOFTIRQ已经被触发了，所以do_softirq会执行相应的软中断处理程序。

Tasklet_action()和Tasklet_hi_action()两个处理程序就是tasklet处理的核心。

总结：所有的Tasklets都通过重复运用TASKLET_SOFTIRQ或HI_SOFTIRQ这两个软中断实现，当一个tasklet被调度时，内核就会唤起这两个软中断中的一个，随后，该软中断会被特定的函数处理，执行所有已调度的tasklet，这个函数保证同一时间里只有一个给定类别的tasklet会被执行（但其它不同类型的tasklet可以同时执行），所有这些复杂性都被一个简洁的接口隐藏起来了。

 

6.3.2使用tasklet

声明你自己的tasklet

可以静态创建，也可以动态创建，分别对应直接引用和间接引用。选择哪种方式取决于你到底是有（或者是想要）一个对tasklet的直接引用还是间接引用，静态创建一个tasklet（也就是有一个直接引用），使用下面中定义的两个宏中的一个：

 

DECLARE_TASKLET(name, func, data)

实现了定义名称为name的tasklet并将其与func这个函数绑定，而传入这个函数的参数为data。

DECLARE_TASKLET_DISABLED(name, func, data);

DECLARE_TASKLET(my_tasklet, my_tasklet_handler, dev);运行代码实际上等价于：

struct tasklet_struct my_tasklet = { NULL, 0, ATOMIC_INIT(0), my_tasklet_handler, dev };    这样就创建了一个名为my_tasklet,处理程序为tasklet_handler并且已经被激活的tasklet。

还可以通过一个间接引用（一个指针）赋给一个动态创建的tasklet_struct结构的方式来初始化一个tasklet：

Tasklet_init(t,tasklet_handler,dev);/*动态而不是静态创建*/

 

编写你自己的tasklet处理程序

必须符合规定的函数类型：                                               

void tasklet_handler(unsigned long data)

因为是靠软件中断实现，所以tasklet不能睡眠，这意味着你不能在tasklet中使用信号量或者其它什么阻塞式的函数。

 

调度你自己的tasklet

通过调用task_schedule()函数并传递给它相应的tasklet_struct的指针，该tasklet就会被调度以便执行。

tasklet_schedule(&my_tasklet);  /*把my_tasklet标记为挂起*/

 

下面我们看一下软中断和tasklet的异同：

在前期准备工作上，首先要给软中断分配索引，而tasklet则要用宏对处理程序声明。在给软中断分配索引后，还要通过open_softirq()函数来注册处理程序。这样看来，tasklet是一步到位，直接到了处理函数，而软中断需要做更多工作。接下来软中断要等待触发（raise_softirq()或raise_softirq_irqoff）,而tasklet则是等待tasklet_schedule()和tasklet_hi_schedule()对其进行调度。两者虽然在命名上不同，但殊途同归，最终的结果都是等待do_softirq()去执行处理函数，即将下半部设置为待执行状态以便稍后执行。另外，在tasklet的tasklet_schedule()中，需要完成的动作之一便是唤起（触发）TASKLET_SOFTIRQ或HI_SOFTIRQ软中断，说明tasklet仍然是基于软中断的。在进入do_softirq()之后，所做的工作仍然有所不同，不再论述。

软中断和工作队列都是异步发生的（就是说，在中断处理返回的时候）

 

6.4工作队列

 

工作队列（work queue）是另外一种将工作推后执行的形式，他和我们前面讨论过的其他形式完全不同。工作队列可以把工作推后，交由一个内核线程去执行——这个下半部总是会在进程上下文执行。这样，通过工作队列执行的代码能占尽进程上下文的所有优势，最重要的是工作队列允许重新调度甚至是睡眠。

如果你需要用一个可以重新调度的实体来执行你的下半部处理，你应该使用工作队列，它是唯一能在进程上下文运行的下半部实现的机制，也只有它才可以睡眠，这意味着你在你需要获得大量的内存时，在你需要获取信号量时，在你需要执行阻塞式的IO操作时，它都会非常有用，如果你不需要用一个内核线程来推后执行工作，那么就考虑使用tasklet吧！

 

6.4.1工作队列的实现

工作队列子系统是一个用于创建内核线程的接口，通过它创建的进程负责执行由内核其他部分排到队列里的其他任务。它创建的这些内核线程被称作工作者线程。工作队列可以让驱动程序创建一个专门的工作者线程来处理需要推后的工作。不过，工作队列子系统提供了一个默认的工作者线程来处理这些工作。因此，工作队列最基本的表现形式就转变成了一个把需要推后执行的任务交给特定的通用线程这样一个接口。

 

表示线程的数据结构

工作者线程用workqueue_struct结构表示：

struct workqueue_struct {

            struct cpu_workqueue_struct cpu_wq[NR_CPUS];

            const char *name;

           struct   list_head   list;

          };

该结构内是一个由cpu_workqueue_struct结构组成的数组，定义在kernel/workqueue.c中，数组的每一项对应一个系统中的处理器。每个工作者线程都对应这样的cpu_workqueue_struct结构体。cpu_workqueue_struct是kernel/workqueue.c中的核心数据结构：   

struct cpu_workqueue_struct     {

                 spinlock_t    lock;             /* 锁定以便保护该结构体 */

                 long    romove_sequeue; /* 最近一个被加上的（下一个要运行的） */

                 long   insert_sequeue;     /*下一个要加上的   */

                wait_queue_head_t    more_work;

                wait_queue_head_t     work_done;

                struct   workqueue_struct   *wq;           /* 有关联的workqueue_struct结构 */

                task_t    *thread;                                 /* 有关联的线程 */

               int   run_depth;                                     /* run_workqueue()循环深度   */

               }；

由此可以看出，每个工作者线程类型关联一个自己的workqueue_struct。在该结构体里面，给每个线程分配一个cpu_workqueue_struct，因而也就是给每个处理器分配一个，因为每个处理器都有一个该类型的线程。

表示工作的数据结构

        所有工作者线程都是用普通的内核线程实现的，它们都要执行worker_thread()函数。在它初始化完以后，这个函数（worker_thread）开始休眠。当有操作被插入到队列的时候，线程就会被唤醒，以便执行这些操作。当没有剩余的操作时，它又会继续睡眠。

工作用中定义的work_struct结构体表示：  

struct    work_struct {

                      unsigned   long   pending;        /* 这个工作是否正在等待处理 */

                     struct   list_head entry;             /* l连接所有工作的链表 */

                    void   (* func) (void *);               /* 处理函数 */

                   void   *wq_data;                         /* 内部使用 */

                    struct timer_list timer;               /* 延迟的工作队列所用到的定时器 */

}；

这些结构体被连接成链表，在每个处理器的每种类型的队列都对应这样一个链表。当一个工作者线程被唤醒时，它会执行它的链表上的所有工作。当工作完毕时，他会将相应的work_struct对象从链表中移去,当链表上不再有对象的时候，它就会继续睡眠。

 

6.4.2使用工作队列

（1）创建推后的工作

首先要做的是实际创建一些需要推后执行的工作。可以通过DECLARE_WORK在编译时静态的创建该结构体：

DECLARE_WORK(name, void (*func) (void *), void *data);

这样就会静态的创建一个名为name，处理函数为func，参数为data的work_struct结构体。也可以在运行时通过指针创建一个工作：

INIT_WORK(struct work_struct *work, void (*func)(void *), void   *data);

这样就动态的初始化了一个由work指向的工作。

（2）工作队列的处理函数

原型是：void   work_handler(void   *data)

这个函数会由一个工作者线程执行，因此，函数会运行在进程上下文中，默认情况下，允许响应中断，并且不持有任何锁，如果需要，函数可以睡眠，注意的是，尽管操作处理函数运行在进程上下文中，但它不能访问用户空间，因为内核线程在用户空间没有相关的内存映射，通常在系统调用发生时，内核会代表用户空间的进程运行，此时它才能访问用户空间，也只有在此时它才会映射用户空间的内存。

（3）对工作进行调度

现在工作已经创建，我们可以调度它了，要把给定工作的处理函数提交给默认的events工作线程，只需调用： schedule_work(&work); work马上就会被调度，一旦其所在的处理器上的工作者线程被唤醒，它就会被执行。

（4）刷新操作

刷新工作队列的函数就是确保在卸载模块之前，要确保一些操作已经执行完毕了，该函数如下：

Void flush_scheduled_work(void);

该函数会一直等待，直到队列中所有对象都被执行以后才返回，在等待所以待处理的工作执行的时候，该函数会进入休眠状态，所以只能在进程上下文中使用它。

 

（5）创建新的工作队列

当缺省的队列不能满足你的需要时，你应该创建一个新的工作队列和与之对应的工作者线程。

 

 

 

6.5下半部之间的选择

1 从设计的角度考虑

软中断提供的执行序列化的保障最少，这就要求软中断必须采取一些步骤确保共享数据的安全。如果被考察的代码本身多线索化的工作就做得非常好，软中断就很好，对于时间要求严格和执行频率很高的话，它执行的也快。如果代码本身多线索化的工作就做得不充分，就选择tasklet比较好，由于两个同种类型的tasklet不能同时执行，实现起来也很简单一些。

2 如果你需要把任务推到进程上下文中完成，只能选择工作队列。

如果不需要睡眠，那么软中断和工作队列就更合适。工作队列造成的开销最大，因为他要牵扯到内核线程甚至是上下文切换。

3 说到易用性，工作队列最好，使用缺省的events队列简直不费吹灰之力。接下来就tasklet。他的的接口很简单，最后才是软中断，它必须静态创建。

 

6.6在下半部之间加锁

使用tasklet的一个好处是在于它自己负责执行的序列化保障，两个相同类型的tasklet不允许同时执行，即使在不同的处理器上也不行，意味着你无须考虑相同类型的tasklet内部的同步问题。当然，tasklet之间的同步（两个不同类型的tasklet共享同一数据时）需要正确使用锁机制。

因为软中断根本不保障执行序列化，（即使相同类型的软中断也有可能有两个实例在同时执行）所以所有的共享数据都需要合适的锁。

如果进程上下文和一个下半部共享数据，在访问这些数据之前，你需要禁止下半部的处理并得到锁的使用权，所做的这些是为了本地和SMP的保护并且防止死锁的出现。

如果中断上下文和一个下半部共享数据，在访问数据之前，你需要禁止中断并得到锁的使用权，所做的这些是为了本地和SMP的保护并且防止死锁的出现。

任何在工作队列中被共享的数据也需要使用锁机制，其中有关锁的要点和在一般内核代码中没什么区别，因为工作队列本来就是在进程上下文中执行的。

 

禁止下半部

一般单纯禁止下半部的处理是不够的，为了保证共享数据的安全，更常见的做法是先得到一个锁然后在禁止下半部的处理，驱动程序中通常使用的都是这种方法。