第8章 下半部和推后执行的工作

操作系统必须有一个快速、异步、简单的机制负责对硬件做出迅速响应并完成那些时间要求很严格的操作，中断处理程序很适合用于实现这些功能；然而对于其他对时间要求相对宽松的任务，就应该推后到中断被激活以后再去运行。

一、下半部

下半部的任务就是执行与中断处理密切相关但中断处理程序本身不执行的工作，这对系统的响应能力和性能都至关重要。并没有规定如何在上半部和下半部划分工作，但有以下经验：

• 若任务对时间非常敏感，将其放在中断处理程序中执行
• 若任务和硬件相关，将其放在中断处理程序中执行
• 若任务要保证不被其他中断（特别是相同的中断）中断，将其放在中断处理程序中执行
• 其他所有任务，考虑放在下半部执行

通常下半部在中断处理程序一返回就会马上执行，下半部执行的关键在于当它们运行时允许响应所有的中断。

当前，有三种机制可以用来实现将工作推后执行：

• 软中断
• tasklet
• 工作队列

二、软中断

2.1 软中断的实现

软中断是在编译期间静态分配的，最多有32个软中断，其结构：

struct softirq_action {
  void (*action) (struct softirq_action *);
};

static struct softirq_action softirq_vec[NR_SOFTIRQS]; //NR_SOFTIRQS=32

---

软中断处理程序：

void softirq_handler(struct softirq_action *);

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

---

执行软中断：一个注册的软中断必须在被标记后才会执行。这被称为触发软中断。通常中断处理程序在返回前标记它的软中断，使其在稍后被执行。于是在合适的时刻，该软中断就会被运行：

• 从一个硬件中断代码返回时
• 在ksoftirqd内核线程中
• 在那些显示检查和执行待处理的软中断的代码中，如网络子系统中

软中断最后都在do_softirq()中执行，其会循环遍历每一个待处理的软中断，调用它们的处理程序。

2.2 使用软中断

软中断保留给系统中对时间要求最严格以及最重要的下半部使用。目前只有两个子系统（网络和SCSI）直接使用软中断。此外，内核定时器和tasklet都是建立在软中断上。

在编译期间，通过一个枚举类型来静态声明软中断，并通过索引值来表示相对优先级，索引号小的软中断在索引号大的软中断之前执行。

image.png

---

注册软中断处理程序：

open_softirq(软中断索引号，处理函数)

例如：
open_softirq(NET_TX_SOFTIRQ, next_tx_action);
open_softirq(NET_RX_SOFTIRQ, next_rx_action);

软中断处理程序执行的时候，允许响应中断，但它自己不能休眠。在一个处理程序运行的时候，当前处理器上的软中断被终止。但其他的处理器仍可以执行别的软中断。实际上，如果同一个软中断在被执行的同时再被触发了，那么另一个处理器可以同时处理其处理程序，这意味着任何共享数据都需要严格的锁保护。

---

触发软中断：将一个软中断设置为挂起状态，让它在下次调用do_softirq()函数时投入运行。

raise_softirq(NET_TX_SOFTIRQ);

若中断被禁止了：
raise_softirq_irqoff(NET_TX_SOFTIRQ);

在中断处理程序中触发软中断是最常见的形式。在这种情况下，中断处理程序执行硬件设备的相关操作，然后触发相应的软中断，最后退出。内核在执行完中断处理程序后，马上会调用do_softirq函数。于是软中断开始执行中断处理程序留给它去完成的剩余任务。

三、tasklet

3.1 tasklet实现

tasklet是通过软中断实现的，由两类软中断代表：HI_SOFTIRQ和TASKLET_SOFTIRQ。两者的区别在于优先级。

tasklet的结构体如下：

struct tasklet_struct {
  struct tasklet_struct *next;   // 链表中的下一个tasklet，每个tasklet_struct代表一个不同的tasklet
  unsigned long state;           // tasklet的状态，0、TASKLET_STATE_SCHED、TASKLET_STATE_RUN
  atomic_t count;                // 引用计数器，若不为0，则tasklet被禁止，不允许运行；为0时，tasklet才被激活，并设置为挂起状态，才能被执行
  void (*func)(unsigned long);   // tasklet处理函数
  unsigned long data;            // 给tasklet处理函数的参数 
};

---

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

tasklet_schedule()会检查tasklet的状态，若可执行则调用_tasklet_schedule(),把需要调度的tasklet放到tasklet_vec链表表头。触发TASKLET_SOFTIRQ软中断，这样在下一次调用do_softirq时就会执行待处理的tasklet。

do_softirq执行对应的tasklet_action()。tasklet_action()会遍历tasklet_vec链表中待处理的tasklet，判断并标志tasklet的状态以及count值，若tasklet需要执行则运行其处理函数tasklet_struct#func。

3.2 使用tasklet

声明tasklet：

静态创建tasklet：
DECLARE_TASKLET(name, func, data);//count初始值为0，tasklet处于激活状态
DECLARE_TASKLET_DISABLE(name, func, data);//count初始值为1，tasklet处于禁止状态

//创建一个名为my_tasklet，处理程序为my_tasklet_handler并且被激活的tasklet
//当my_tasklet_handler被调用时dev会被传递给它
DECLARE_TASKLET(my_tasklet, my_tasklet_handler, dev);

动态创建tasklet并初始化：
tasklet_init(t, tasklet_handler, dev)

---

tasklet处理程序原型如下：

void tasklet_handler(unsigned long data);

tasklet不能休眠。两个相同的tasklet不会同时运行，但tasklet运行时允许响应中断，所以tasklet与中断处理程序之间共享数据需要做好并发保护；两个不通的tasklet可能在两个处理器同时执行，因此它们之间的共享数据也需要做好并发保护。

---

调度tasklet：

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

tasklet被调度后，只要有机会就会尽可能早地运行：

• 如果它没得到运行机会之前，有个相同的tasklet被调度了，那么它只会运行一次；
• 如果它以及开始运行了，在另一个处理器上又被调度了，那么会再运行一次。

tasklet_disable(&my_tasklet);  //禁止tasklet
tasklet_enable(&my_tasklet);  //激活tasklet

tasklet_kill(&my_tasklet);  //从挂起的队列中去掉my_tasklet，可能会休眠，禁止在中断上下文中使用

---

当大量软中断出现时，内核会唤醒一组内核线程来处理这些负载。这些线程以最低优先级运行（nice值19），避免跟其他重要任务抢夺资源。但它们最后肯定会被执行，因此能够保证在软中断负载重时用户程序不会因为得不到处理器时间而处于饥饿状态，也能保证空闲系统上软中断迅速得到处理。

每个处理器都有一个这样的线程，称为ksoftirqd/n，n表示处理器编号。该线程一旦被初始化，就会陷入循环：若没有待处理软中断，则调用schedule让出时间；若有则调用do_softirq来处理。

四、工作队列

工作队列把工作推后，交由一个内核线程执行，这个下半部总是会在进程上下文中执行，因此工作队列执行的代码拥有进程上下文的所有优势。最重要的是工作队列运行重新调度甚至休眠。

4.1 工作队列的实现

工作队列子系统是一个用于创建内核线程的接口，通过它创建的进程负责执行由内核其他部分放入队列里的任务。这些内核线程称为工作线程events/n，n表示处理器编号，每个处理器对应一个线程。

//工作线程结构：
struct workqueue_struct {
  struct cpu_workqueue_struct cpu_wq[NR_CPUS]; //每个元素对应一个处理器
  struct list_head list;
  const char *name;
  int singlethread;
  int freezeable;
  int rt;
};

//处理器的结构：
struct cpu_workqueue_struct{
  spinlock_t lock; //锁保护这种结构
  struct list_head worklist; //工作列表
  wait_queue_head_t more_work;
  struct work_struct *current_struct; //工作链表
  struct workqueue_struct *wq; // 关联工作队列结构
  task_t *thread;   //关联线程
};

// 工作结构：
struct work_struct {
  atomic_long_t data;
  struct list_head entry;
  work_fun_t func;
};

工作线程会执行worker_thread()函数，在其初始化后会执行一个死循环并开始休眠，当有操作被插入到队列current_struct时，线程就会被唤醒，以便执行这些操作work->func；当没有剩余操作时，又继续休眠。

image.png

4.2 使用工作队列

创建工作：

// 静态创建名称为name，处理函数为func，参数为data的work_struct结构体：
DECLARE_WORK(name, void (*func)(void *), void *data);

//运行时创建work_struct结构体：
INIT_WORK(struct work_struct *work, void (*func)(void *), void *data);

---

工作队列处理函数运行在进程上下文，原型如下：

void work_handle(void *data);

该函数默认允许响应中断，并且不持有任何锁。如有需要，可以休眠。但不能访问用户空间。

---

工作调度，一旦工作所在处理器上的工作者线程被唤醒，工作就会被执行：

schedule_work(&work);

schedule_delayed_work(&work, delay); //延迟执行工作
int cancel_delayed_work(struct work_struct *work); //取消延迟执行的工作

---

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

void flush_scheduled_work(void);

---

创建新的工作队列和与之相关的工作者线程：

struct workqueue_struct *create_workqueue(const char *name);

调度工作到指定的工作队列：

int queue_work(struct workqueue_struct *wq,  
               struct work_struct *work);

int queue_delayed_work(struct workqueue_struct *wq,  
               struct work_struct *work, 
               unsigned long delay);

刷新指定工作队列：

flush_workqueue(struct workqueue_struct *wq);

五、下半部机制的选择

image.png