Linux内核设计与实现 第八章 下半部和推后执行的工作

中断处理程序抢占的进程可能正在执行重要的代码，中断处理程序执行完前会禁止中断，硬件就无法与操作系统通信，硬件往往要求中断处理程序迅速处理它的事。所以中断处理程序代码量少，让执行时间变短。
中断处理程序代码量少，而且不能使用会引起阻塞的函数，这极大的限制了中断处理程序能做的事。
由此将中断处理流程分为两部分：a)中断处理程序，即上半部 b)下半部

8.1下半部

下半部的任务就是执行与中断处理密切相关但是中断处理程序本身不执行的工作。
不存在严格明确的规定来说明到底什么任务应该在哪个部分中完成。如何做决定完全取决于驱动开发者自己的判断。尽管在理论上不存在什么错误，但是轻率的实现效果往往不很理想。
什么任务应该在哪个部分中完成的经验：
a)如果一个任务对时间非常敏感，将其放在中断处理程序中执行。
b)如果一个任务和硬件相关，将其放在中断处理程序中执行。
c)如果一个任务要保证不被其他中断打断，将其放在中断处理程序中执行。
d)其他所有任务，考虑放置在下半部执行。
当你开始尝试写自己的驱动程序的时候，读一下别人的中断处理程序和相应的下半部可能会让你受益匪浅。

1）为什么要用下半部

下半部的执行不需要指明一个确切时间，只要把这些任务推迟一点，让它们在系统不太忙时恢复中断，再执行。
下半部的这种设计可以使系统处于中断屏蔽状态的时间尽量可能的短，以此来提高系统的响应能力。

2）下半部的环境

和上半部只能通过中断处理程序实现不同，下半部可以通过多种机制实现
2.6版本的内核提供了三种不同形式的下半部实现机制：软中断、tasklet、工作队列

尽管下半部的执行不需要指明一个确切时间，但是在某些情况需要保证，在一个确定的时间段过去后才可以运行时，就需要使用内核定时器。

8.2软中断

一个软中断不会抢占另一个软中断。唯一可以抢占软中断的是中断处理程序。不过，其他的软中断可以在其他处理器上同时执行。

软中断的代码位于kernel/softirq.c文件中。

Linux2.6，目前只有网络子系统和SCSI子系统直接使用软中断。
内核定时器和tasklet都是建立在软件中断上的。

1）分配索引

在中定义了一个枚举类型来声明软中断

enum
{
    HI_SOFTIRQ=0,
    TIMER_SOFTIRQ,
    NET_TX_SOFTIRQ,
    NET_RX_SOFTIRQ,
    BLOCK_SOFTIRQ,
    IRQ_POLL_SOFTIRQ,
    TASKLET_SOFTIRQ,
    SCHED_SOFTIRQ,
    HRTIMER_SOFTIRQ, /* Unused, but kept as tools rely on the numbering. Sigh! */
    RCU_SOFTIRQ,    /* Preferable RCU should always be the last softirq */

    NR_SOFTIRQS
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

内核用从0开始的索引值来表示一种相对优先级。索引值越小优先级越高。
在枚举中加入新的项时，必须根据希望赋予它的优先级来决定软中断加入的位置。

2）软中断的实现

a)softirq_action
软中断的长相结构体softirq_action

struct softirq_action
{
	void	(*action)(struct softirq_action *);//若一个软中断被标记，do_softirq()会在遍历查看32个软中断时发现它被标记，do_softirq()会调用该软中断的action()函数
};

//软中断处理程序action(struct softirq_action *)
//内核运行一个软中断处理程序时，就是执行函数action()。
//内核将整个软中断的结构体的指针传递给软中断处理程序action()。
//内核调用软中断处理程序中的函数的方法：softirq_vec[x]->action(softirq_vec[x])
//x的取值范围0到31
1
2
3
4
5
6
7
8
9
10

管理32个软中断的注册和激活等操作的数组

static struct softirq_action softirq_vec[32];
1

32个软中断只用了9个:

所有的tasklet都是通过重复运用HI_SOFTIRQ(优先级高的tasklet)和TASKLET_SOFTIRQ(正常优先级的tasklet)这两个软中断实现的

b)触发软中断
触发软中断：将一个已经注册了的软中断标记。
通常，中断处理程序会在返回前标记它的软中断，使其稍后被执行。即在上半部结束之前标记它的软中断，使其稍后被执行。

open_softirq(软件中断的索引号，软件中断处理函数)//注册你的软件中断处理程序。
raise_softirq(软件中断的索引号)//将一个已经注册了的软中断标记。
1
2

c)被标记的软中断如何被do_softirq()执行
待处理的软中断被检查和执行的地方，即调用函数do_softirq()的地方：
①从一个硬件中断代码处返回时
②在ksoftirqd内核线程中
③在哪些显式检查和执行待处理的软中断的代码中，如网络子系统中。

8.3tasklet

tasklet是利用软中断实现的一种下半部机制。
下半部机制的选择：通常应该使用tasklet,只在哪些执行频率很高和连续性要求高的情况下才需要使用软中断。

第一类：重复运用软中断HI_SOFTIRQ(优先级高的tasklet)实现的tasklet。
第二类：重复运用软中断TASKLET_SOFTIRQ(正常优先级的tasklet)实现的tasklet。
两类仅是优先级不同，其他没有区别。

大多数情况下，为了控制一个寻常的硬件，tasklet机制都是实现自己的下半部的最佳选择。tasklet可以动态创建，使用方便，执行起来也还算快。
相同优先级的一个tasklet不会抢占另一个tasklet。tasklet也是软中断，唯一可以抢占软中断的是中断处理程序。不过，另一个优先级的tasklet可以在其他处理器上同时执行。
如果一个tasklet和其他tasklet或者软中断共享了数据，就必须进行适当地锁保护。
中断上半部和下半部软中断机制和下半部tasklet机制都不能使用信号量或者其他什么会导致阻塞的函数。

1）tasklet的实现

a)tasklet的长相

struct tasklet_struct
{
     struct tasklet_struct *next;//链表中的下一个tasklet。
     unsigned long state;        //tasklet的状态。三种状态是：a)0  b)TASKLET_STATE_SCHED:tasklet已经被调度，正准备投入运行。 c)TASKLET_STATE_RUN：tasklet正在被某处理器运行       
     atomic_t count;             //引用计数器。count≠0时tasklet被禁止执行，count=0时tasklet被允许执行，tasklet就可以被激活和设置成挂起状态，最终执行。
     void (*func)(unsigned long);//tasklet处理函数。
     unsigned long data;         //给tasklet处理函数的参数。
};
1
2
3
4
5
6
7
8

b)调度tasklet
tasklet_schedule()和tasklet_hi_schedule()调度tasklet，内核就会唤起软中断HI_SOFTIRQ和软中断TASKLET_SOFTIRQ中的一个，对应的软中断处理函数就会执行，最后软中断处理函数执行所有已调度的tasklet。
当然，软中断处理函数还要保证同一时间里只有一个给定类别的tasklet会被执行（但是其他不同类型的tasklet可以同时执行）

tasklet_schedule()和tasklet_hi_schedule()非常类似，区别是一个使用软中断HI_SOFTIRQ，另一个使用软中断TASKLET_SOFTIRQ。

do_softirq()执行tasklet_action()和tasklet_hi_action()。下图是描述tasklet_action()和tasklet_hi_action()干什么：

2）使用tasklet

大多数情况下，为了控制一个寻常的硬件，tasklet机制都是实现自己的下半部的最佳选择。tasklet可以动态创建，使用方便，执行起来也还算快。
相同优先级的一个tasklet不会抢占另一个tasklet。tasklet也是软中断，唯一可以抢占软中断的是中断处理程序。不过，另一个优先级的tasklet可以在其他处理器上同时执行。
如果一个tasklet和其他tasklet或者软中断共享了数据，就必须进行适当地锁保护。
中断上半部和下半部软中断机制和下半部tasklet机制都不能使用信号量或者其他什么会导致阻塞的函数。

a)声明自己的tasklet

DECLARE_TASKLET(tasklet长相结构体的指针,软中断处理函数的函数指针,软中断处理函数的参数)//静态创建一个tasklet_struct结构，特别的设置引用计数为0，当该tasklet被调用以后，给定函数func会被执行。
DECLARE_TASKLET_DSABLED(tasklet长相结构体的指针,软中断处理函数的函数指针,软中断处理函数的参数)//静态创建一个tasklet_struct结构，特别的设置引用计数为1，当该tasklet被调用以后，给定函数func会被执行。
tasklet_init(tasklet长相结构体的指针,软中断处理函数的函数指针,软中断处理函数的参数)//通过指针赋给一个动态创建的tasklet_struct结构体的方式来初始化一个tasklet结构体的成员。
1
2
3

b)编写自己的tasklet处理程序
tasklet处理函数必须符合如下规定的函数类型：

void tasklet_handler(unsigned long data)
1

c)调度tasklet

tasklet_schedule(&mou_tasklet)//把某tasklet标记为挂起
tasklet_disable(&mou_tasklet)//禁止某指定的tasklet，如果此tasklet当前正在执行，等到此tasklet执行完毕，tasklet_disable函数才返回。
tasklet_enable(&mou_tasklet)//激活某指定的tasklet
tasklet_kill(&mou_tasklet)//从挂起队列中去掉一个指定的tasklet，此函数可能会引起休眠，所以禁止在中断上下文中使用它。
1
2
3
4

d)ksoftirqd（原文讲的虽略长但很明了）

3）老的BH机制

Linux2.5彻底放弃了BH机制

8.4工作队列

工作队列是一种将工作推后执行的形式，它和我们前面讨论的所有其他形式都不相同。工作队列可以把工作推后，交由一个内核线程去执行。
如果推后执行的任务需要睡眠，那么就选择工作队列。如果推后执行的任务不需要睡眠，那么就选择软中断或tasklet。
唯一能在进程上下文中运行的下半部实现机制，也只有它才可以睡眠。这意味着在你需要获得大量的内存时，在你需要获取信号量时，在你需要执行阻塞式的I/O操作时，它都会非常有用。

工作队列（work queue）是另外一种将中断的部分工作推后的一种方式，它可以实现一些tasklet不能实现的工作，比如工作队列机制可以睡眠。这种差异的本质原因是，在工作队列机制中，将推后的工作交给一个称之为工作者线程（worker thread）的内核线程去完成（单核下一般会交给默认的线程events/0）。因此，在该机制中，当内核在执行中断的剩余工作时就处在进程上下文（process context）中。也就是说由工作队列所执行的中断代码会表现出进程的一些特性，最典型的就是可以重新调度甚至睡眠。

1）工作队列的实现

工作队列子系统是一个用于创建内核线程的接口，工作队列子系统创建的内核线程负责执行由内核其他部分排到队列里的任务。
工作队列子系统创建的内核线程也叫工作者线程。
每个处理器与之对应的工作者线程叫默认的工作者线程，即缺省的工作者线程。
缺省的工作者线程会在多个地方得到被推后的工作。
如果下半部是需要执行大量的处理操作的任务A，将此任务A交给工作者线程，可能会导致排在任务A后面的需要完成的工作处于饥饿状态。此时就可以选择建立一个专属的内核线程来处理任务A。

a)表示线程的数据结构

//描述工作队列的工作队列结构体，描述特定的某类工作队列
struct workqueue_struct {
    struct cpu_workqueue_struct cpu_wq[NR_CPUS];// cpu_wq是所有处理器的工作队列集合成的工作队列数组，cpu_wq[NR_CPUS]代表某一个工作队列
    struct list_head list;//队列描述符的指针
    const char *name;  //
    int singlethread;    //
    int freezeable;        //
    int rt;                       //
};

//描述CPU工作队列的CPU工作队列结构体，描述使用特定的某类工作队列的cpu集合
struct cpu_workqueue_struct {

    spinlock_t lock;                               //锁保护

    struct list_head worklist;               //工作链表头
    wait_queue_head_t more_work;   //
    struct work_struct *current_work;//

    struct workqueue_struct *wq;      //队列描述符指针，每个处理器都有工作队列，工作队列都有一个它的队列
    struct task_struct *thread;            //线程描述符指针，每个处理器都有工作队列，工作队列都有一个它的线程
};

b)表示工作的数据结构
//描述工作的工作结构体
struct work_struct {
    atomic_long_t data;   
    struct list_head entry;//用于把work挂到其他队列上。
    work_func_t func;       //工作任务的处理函数
};
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

当一个工作者线程被唤醒时，它会执行它的链表上的所有工作。工作完毕，它就将相应的work_struct对象从链表上移去。当链表上不再有对象的时候，它就会继续休眠。

c)工作队列实现机制的总结

2）使用工作队列

a)创建推后的工作

DECLARE_WORK(n, f)         //n是工作名，f处理函数
INIT_WORK(_work, _func)  //_work是工作名，_func处理函数
1
2

b)工作队列处理函数格式

void work_handler(void *data)
1

c)对工作进行调度
想要把给定工作的处理函数提交给缺省的events工作线程，只需调用：

schedule_work(&work)//调度work，即一旦此work所在的处理器上的工作者线程被唤醒，此work的处理函数就会被执行。
1

d)刷新操作
确保不再有待处理的工作：

void flush_scheduled_work(void)//函数会一直等到，直到队列中所有对象都被执行以后才返回。
1

取消延迟执行的工作调用：

int cancel_delayed_work(struct work_struct *work)//取消执行某指定的工作
1

d)创建新的工作队列
如果缺省的队列不能满足需要，应该创建一个新的工作队列和与之相应的工作者线程。由于这么做会在每个处理器上都创建一个工作者线程，所以只有在明确了必须要靠自己的一套线程来提高性能的情况下，再创建自己的工作队列。

创建一个新的任务队列和与之相关的工作者线程，只需调用一个函数：

struct workqueue_struct *create_workqueue(const char *name);
1

name参数用于该内核线程的名字。比如，缺省的events队列的创建就调用的是：

static struct workqueue_struct *keventd_wq；
keventd_wq = create_workqueue("events");
1
2

这样就会创建所有的工作者线程，并且做好所有开始处理工作之前的准备工作。

创建一个工作时无须考虑工作队列的类型。在创建之后，可以调用下面列举的函数。这些函数与schedule_work()以及schedule_delayed_work()相近，唯一的区别在于它们针对给定的工作队列而不是缺省的events队列进行操作。

int queue_work(struct workqueue_struct *wq, struct work_struct *work)；
int queue_delayed_work(struct workqueue_struct *wq,struct delayed_work *dwork, unsigned long delay)；
1
2

最后，可以调用下面的函数刷新指定的工作队列：

void flush_workqueue(struct workqueue_struct *wq)；
1

这个函数只是它在返回前等待清空的是给定的队列。

3）老的任务队列机制

任务队列被工作队列取代了

8.5下半部机制的选择

上述简单来说，是否需要一个可调度的实体（是否有休眠需要）来执行需要推后完成的工作–有，工作队列就是唯一选择。
否则最好使用tasklet，若是必须专注于性能的提高，那么就考虑软中断。

8.6在下半部之间加锁

在使用下半部机制时，即使是在单处理器的系统上，避免共享数据被同时访问也是至关重要的。
记住，一个下半部实际上可能在任何时候执行。

使用tasklet的一个好处在于，它自己负责执行的序列化保障：两个相同类型的tasklet不允许同时执行，即使在不同的处理器上也不行。

Q：通常在哪些地方枷锁？
A：
如果进程上下文和一个下半部共享数据，在进程上下文中访问这些数据之前，需要禁止下半部的处理并得到锁的使用权。是为了本地和SMP的保护并且防止死锁的出现；
如果中断上下文和一个下半部共享数据，在中断上下文中访问数据之前，需要禁止中断并得到锁的使用权。为了本地和SMP的保护并防止死锁的出现；
任何在工作队列中被共享的数据也需要使用锁机制，其中有关锁的要点和一般内核代码中没什么区别，工作队列本来就是在进程上下文中执行的。

8.7禁止下半部

一般如果需要禁止下半部，那么单纯禁止下半部的处理是不够的的。
为了保证共享数据的安全，更常见的做法是，先得到一个锁然后在禁止下半部的处理。

下半部机制控制函数如下所示：

实际上在linux kernel中，task_struct、thread_info都用来保存进程相关信息，即进程PCB信息。
结构体thread_info中的成员preempt_count（内核抢占的时候也是它）是进程维护的一个计数器。
当计数器变为0时，下半部才能够被处理。因为此时会取消下半部处理的禁止，所以local_bh_enable()还需要检查所有现存的待处理的下半部并执行它们。

local_bh_enable()和local_bh_disable()函数并不能禁止工作队列的执行（工作队列是在进程上下文中运行的，不会涉及异步执行的问题，所以也就不需要禁止工作队列执行。而软中断和tasklet是异步发生的–在中断处理返回的时候，所以内核要禁止它们）