Linux Kernel - Bottom Halves and Deferring Work

这些东西确实把人搅得很混乱，但它们其实只不过是一些起名的问题，让我们再来梳理一遍。

“下半部（bottom half）”是一个操作系统通用词汇，用于指代中断处理流程中推后执行的那一部分，之所以这样命名，是因为它表示中断处理方案一半的第二部分或者下半部。在 Linux 中，这个词目前确实就是这个含义。所有用于实现将工作推后执行的内核机制都被称为“下半部机制”。一些人错误地把所有的下半部机制都叫做“软中断”，真是在自寻烦恼。

“下半部”这个词也指代 Linux 最早提供的那种将工作推后执行的实现机制。由于该机制也被叫做“BH”，所以，我们就使用它的这个名称，而让“下半部”这个词仍然保持它通常的含义。BH 机制很早以前就被反对使用了，在 2.5 版内核中，它就被完全去除了。

目前，有三种机制可以用来实现将工作推后执行：软中断、tasklet 和工作队列。 tasklet 通过软中断实现，而工作队列与它们完全不同。表 1 揭示了下半部机制的演化历程。

另外一个可以用于将工作推后执行的机制是“内核定时器”。不像本章到目前为止介绍到的所有这些机制，内核定时器把操作推迟到某个确定的时间段之后执行。也就是说，尽管本章讨论的其他机制可以把操作推后到除了现在以外的任何时间进行，但是当你必须保证在一个确定的时间段过去以后再运行时，你应该使用内核定时器。

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

void softirq_handler(struct softirq_action *)

当内核运行一个软中断处理程序的时候，它就会执行这个 action 函数，其唯一的参数为指向相应 softirq_action 结构体的指针。例如，如果 my_softirq 指向 softirq_vec 数组的某项，那么内核会用如下的方式调用软中断处理程序中的函数：

my_softirq->action(my_softirq);

当你看到内核把整个结构体都传递给软中断处理程序而不是仅仅传递数据值的时候，你可能会很吃惊。这个小技巧可以保证将来在结构体中加入新的域时，无须对所有的软中断处理程序都进行变动。如果需要，软中断处理程序可以方便地解析它的参数，从数据成员中提取数值。

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

一个注册的软中断必须在“被标记”后才会执行。这被称作触发软中断（raising the softirq)。通常， 中断处理程序会在返回前标记它的软中断，使其在稍后被执行。 于是，在合适的时刻，该软中断就会运行。在下列地方，待处理的软中断会被检查和执行：

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

不管是用什么办法唤起，软中断都要在 do_softirq() 中执行。该函数很简单。如果有待处理的软中断， do_softirq() 会循环遍历毎一个，调用它们的处理程序。让我们观察一下 do_softirq() 经过简化后的核心部分：

u32 pending;

pending = local_softirq_pending();
if (pending) {
    struct softirq_action *h;

    /* reset the pending bitmask */
    set_softirq_pending(0);
    h = softirq_vec;

    do {
        if (pending & 1)
            h->action(h);
        h++;
        pending >>= 1;
    } while (pending);
 }

在编译期间，通过在<linux/interrupt.h>中定义的一个枚举类型来静态地声明软中断。内核用这些从 0 开始的索引来表示一种相对优先级。索引号小的软中断在索引号大的软中断之前执行。

建立一个新的软中断必须在此枚举类型中加入新的项。而加入时，你不能像在其他地方一样，简单地把新项加到列表的末尾。相反，你必须根据希望赋予它的优先级来决定加入的位置。习惯上， HI_SOFTIRQ 通常作为第一项，而 RCU_SOFTIRQ 作为最后一项。新项可能插在 BLOCK_SOFTIRQ 和 TASKLET_SOFTIRQ 之间。表 2 列举出了已有的 tasklet 类型。

接着，在运行时通过调用 open_softirq() 注册软中断处理程序，该函数有两个参数：软中断的索引号和处理函数。如网络子系统，在 net/coreldev.c 通过以下方式注册自己的软中断：

open_softirq(NET_TX_SOFTIRQ, net_tx_action);
open_softirq(NET_RX_SOFTIRQ, net_rx_action);

软中断处理程序执行的时侯，允许响应中断，但它自己不能休眠。在一个处理程序运行的时侯，当前处理器上的软中断被禁止。但其他的处理器仍可以执行别的软中断。实际上，如果同个软中断在它被执行的同时再次被触发了，那么另外一个处理器可以同时运行其处理程序。这意味着任何共享数据（甚至是仅在软中断处理程序内部使用的全局变量）都需要严格的锁保护。这点很重要，它也是 tasklet 更受青睐的原因。单纯地禁止你的软中断处理程序同时执行不是很理想。如果仅仅通过互斥的加锁方式来防止它自身的并发执行，那么使用软中断就没有任何意义了。因此，大部分软中断处理程序，都通过采取单处理器数据（仅属于某一个处理器的数据，因此根本不需要加锁）或其他一些技巧来避免显式地加锁，从而提供更出色的性能。

引入软中断的主要原因是其可扩展性。如果不需要护展到多个处理器，那么，就使用 tasklet 吧。tasklet 本质上也是软中断，只不过同一个处理程序的多个实例不能在多个处理器上同时运行。

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

raise_softing(NET_TX_SOFTIRQ);

这会触发 NET_TX_SOFTIRQ 软中断。它的处理程序 net_tx_action() 就会在内核下一次执行软中断时投入运行。

在中断处理程序中触发软中断是最常见的形式。在这种情况下， 中断处理程序执行硬件设备的相关操作，然后触发相应的软中断，最后退出。内核在执行完中断处理程序以后，马上就会调用 do_softirq() 函数。于是软中断开始执行中断处理程序留给它去完成的剩余任务。 在这个例子中，“上半部”和“下半部”名字的含义一目了然。

tasklet 由 tasklet_struct 结构表示。毎个结构体单独代表一个 tasklet，它在<linux/interrupt.h>中定义为：

struct tasklet_struct {
    struct tasklet_struct *next; /* next tasklet in the list */
    unsigned long state;         /* state of the tasklet */
    atomic_t count;              /* reference counter */
    void (*func)(unsigned long); /* tasklet handler function */
    unsigned long data;          /* argument to the tasklet function */
};

结构体中的 func 成员是 tasklet 的处理程序（像软中断中的 action 一样），data 是它唯一的参数。

state 成员只能在 0、TASKLET_STATE_SCHED 和 TASKLET_STATE_RUN 之间取值。TASKLET_STATE_SCHIED 表明 tasklet 已被调度，正准备投入运行，TASKLET_STATE_RUN 表明该 tasklet 正在运行。TASKLET_STATE_RUN 只有在多处理器的系统上才会作为一种优化来使用，单处理器系统任何时候都清楚单个 tasklet 是不是正在运行（它要么就是当前正在执行的代码，要么不是）。

count 成员是 tasklet 的引用计数器。如果它不为 0，则 tasklet 被禁止，不允许执行；只有当它为 0 时，tasklet 才被激活，并且在被设置为挂起状态时，该 tasklet 才能够执行。

已调度的 tasklet（等同于被触发的软中断）存放在两个单处理器数据结构： tasklet_vec （普通 tasklet）和 tasklet_hi_vec （高优先级的 tasklet）。这两个数据结构都是由 tasklet_struct 结构体构成的链表。链表中的每个 tasklet_struct 代表一个不同的 tasklet。

tasklet_struct 结构体构成的链表。链表中的每个 ~tasklet_struct! 代表一个不同的 tasklet。

tasklet 由 tasklet_schedule() 和 tasklet_hi_schedule() 函数进行调度，它们接受一个指向 tasklet_struct 结构的指针作为参数。两个函数非常类似（区别在于一个使用 TASKLET_SOFTIRQ 而另个用 HI_SOFTIRQ)。

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

你既可以静态地创建 tasklet，也可以动态地创建它。选择哪种方式取决于你到底是有（或者是想要）一个对 tasklet 的直接引用还是间接引用。如果你准备静态地创建一个 tasklet（也就是有一个它的直接引用），使用下面<linux/interrupt.h>中定义的两个宏中的一个：

DECLARE TASKLET(name, func, data)
DECLAARE TASKLET_DISABLED(name, func, data);

这两个宏都能根据给定的名称静态地创建一个 tasklet_struct 结构。当该 tasklet 被调度以后，给定的函数 func 会被执行，它的参数由 data 给出。这两个宏之间的区别在于引用计数器的初始值设置不同。前面一个宏把创建的 tasklet 的引用计数器设置为 0，该 tasklet 处于激活状态。另一个把引用计数器设置为 1，所以该 tasklet 处于禁止状态。下面是一个例子：

DECLARE TASKLET(my tasklet, my tasklet handler, dev);

这行代码其实等价于：

struct tasklet_struct my_tasklet =(NULL, 0, ATOMIC_INIT(0),
                                   my_tasklet_handler, dev);

这样就创建了一个名为 my_tasklet，处理程序为 my_tasklet_handler 并且是已被激活的 tasklet。当处理程序被调用的时候，dev 就会被传递给它。

此外，通过 tasklet_init 可以动态创建的 tasklet_struct 结构：

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

tasklet 处理程序必须符合规定的函数类型：

void tasklet_handler(unsigned long data)

因为是靠软中断实现，所以 tasklet 不能睡眠。这意味着你不能在 tasklet 中使用信号量或者其他什么阻塞式的函数。由于 tasklet 运行时允许响应中断，所以你必须做好预防工作（如屏蔽中断然后获取一个锁），如果你的 tasklet 和中断处理程序之间共享了某些数据的话。两个相同的 tasklet 决不会同时执行，这点和软中断不同一尽管两个不同的 tasklet 可以在两个处理器上同时执行。如果你的 tasklet 和其他的 tasklet 或者是软中断共享了数据，你必须进行适当地锁保护。

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

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

在 tasklet 被调度以后，只要有机会它就会尽可能早地运行。在它还没有得到运行机会之前，如果有一个相同的 tasklet 又被调度了，那么它仍然只会运行一次。而如果这时它已经开始运行了，比如说在另外一个处理器上，那么这个新的 tasklet 会被重新调度并再次运行。作为一种优化措施，一个 tasklet 总在调度它的处理器上执行——这是希望能更好地利用处理器的高速缓存。

每个处理器都有一组辅助处理软中断（和 tasklet）的内核线程。当内核中出现大量软中断的时候，这些内核进程就会辅助处理它们。因为 tasklet 通过用软件中断实施，下面的讨论同时适用于软中断和 tasklet。简洁起见，我们将主要参考软中断。

我们前面曾经阐述过，对于软中断，内核会选择在几个特殊时机进行处理。而在中断处理程序返回时处理是最常见的。软中断被触发的频率有时可能很高（像在进行大流量的网络通信期间）。更不利的是，处理函数有时还会自行重复触发。也就是说，当一个软中断执行的时候，它可以重新触发自己以便再次得到执行（事实上，网络子系统就会这么做）。如果软中断本身出现的频率就高，再加上它们又有将自己重新设置为可执行状态的能力，那么就会导致用户空间进程无法获得足够的处理器时间，因而处于饥饿状态。而且，单纯的对重新触发的软中断采取不立即处理的策路，也无法让人接受。当软中断最初提出时，就是一个让人进退维谷的问题，亟待解决，而直观的解决方案又都不理想。首先，就让我们看看两种最容易想到的直观的方案。

第一种方案是，只要还有被触发并等待处理的软中断，本次执行就要负责处理，重新触发的软中断也在本次执行返回前被处理。这样做可以保证对内核的软中断采取即时处理的方式，关键在于，对重新触发的软中断也会立即处理。当负载很高的时候这样做就会出问题，此时会有大量被触发的软中断，而它们本身又会重复触发。系统可能会一直处理软中断，根本不能完成其他任务。 用户空间的任务被忽略了——实际上，只有软中断和中断处理程序轮流执行，而系统的用户只能等待。 只有在系统永远处于低负载的情况下，这种方案才会有理想的运行效果；只要系统有哪怕是中等程度的负载量，这种方案就无法让人满意。因为用户空间根本不能容忍有明显的停顿出现。

第二种方案选择不处理重新触发的软中断。在从中断返回的时候，内核和平常一样，也会检查所有挂起的软中断并处理它们。但是，任何自行重新触发的软中断都不会马上处理，它们被放到下一个软中断执行时去处理。而这个时机通常也就是下一次中断返回的时候，这等于是说，一定得等一段时间，新的（或者重新触发的）软中断才能被执行。可是，在比较空闲的系统中，立即处理软中断才是比较好的做法。很不幸，这个方案显然又是一个时好时坏的选择。 尽管它能保证用户空间不处于饥饿状态，但它却让软中断忍受饥饿的痛苦，而根本没有好好利用闲置的系统资源。

在设计软中断时，开发者就意识到需要一些折中。最终在内核中实现的方案是 不会立即处理重新触发的软中断。而作为改进，当大量软中断出现的时候，内核会唤醒一组内核线程来处理这些负载。 这些线程在最低的优先级上运行（nice 值是 19)，这能避免它们跟其他重要的任务抢夺资源。但它们最终肯定会被执行，所以，这个折中方案能够保证在软中断负担很重的时候，用户程序不会因为得不到处理时间而处于饥饿状态。相应的，也能保证“过量”的软中断终究会得到处理。最后，在空闲系统上，这个方案同样表现良好。软中断处理得非常迅速（因为仅存的内核线程肯定会马上调度）。

每个处理器都有一个这样的线程。所有线程的名字都叫做 ksoftirqd/n，区别在于 n，它对应的是处理器的编号。在一个双 CPU 的机器上就有两个这样的线程，分别叫 ksoftirqd/0 和 ksoftirqd/1。为了保证只要有空闲的处理器，它们就会处理软中断，所以给每个处理器都分配个这样的线程。一旦该线程被初始化，它就会执行类似下面这样的死循环：

for (;;) {
    if (!softirq_pending(cpu))
        schedule();

    set_current_state(TASK_RUNNING);

    while (softirq_pending(cpu)) {
        do_softirq();
        if (need_resched())
            schedule();
    }

    set_current_state(TASK_INTERRUPTIBLE);
}

只要有待处理的软中断（由 softirq_pending() 函数负责发现）, ksoftirq 就会调用 do_softirq() 去处理它们。通过重复执行这样的操作，重新触发的软中断也会被执行。如果有必要的话，每次迭代后都会调用 schedule() 以便让更重要的进程得到处理机会。当所有需要执行的操作都完成以后，该内核线程将自己设置为 TASK_INTERRUPTIBLE 状态，唤起调度程序选择其他可执行进程投入运行。

只要 do_softirq() 函数发现已经执行过的内核线程重新触发了它自己，软中断内核线程就会被唤醒。

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

/*
 * 外部可见的工作队列抽象是
 * 由每个 CPU 的工作队列组成的数组
 */
struct workqueue_struct {
    struct cpu_workqueue_struct cpu_wq[NR_CPUS];
    struct list_head list;
    const char *name;
    int singlethread;
    int freezeable;
    int rt;
};

该结构内是一个由 cpu_workqueue_struct 结构组成的数组，它定义在 kernel/workqueue.c 中，数组的每一项对应系统中的一个处理器。由于系统中毎个处理器对应一个工作者线程，所以对于给定的某台计算机来说，就是每个处理器，每个工作者线程对应一个这样的 cpu_workqueue_struct 结构体。cpu_workqueue_struct 是 kernel/workqueue.c 中的核心数据结构：

struct cpu_workqueue_struct {
    spinlock_t lock;            /* lock protecting this structure */
    struct list_head worklist;  /* list of work */
    wait_queue_head_t more_work;
    struct work_struct *current_struct;
    struct workqueue_struct *wq; /* associated workqueue_struct */
    task_t *thread;              /* associated thread */
};

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

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

工作用<linux/workqueue.h>中定义的 work_struct 结构体表示：

struct work_struct {
    atomic long t data;
    struct list_head entry;
    work_func_t func;
};

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

我们可以看一下 worker_thread() 函数的核心流程，简化如下：

for (;;) {
    prepare_to_wait(&cwq->more_work, &wait, TASK_INTERRUPTIBLE);
    if (list_empty(&cwq->worklist))
        schedule();
    finish_wait(&cwq->more_work, &wait);
    run_workqueue(cwq);
}

该函数在死循环中完成了以下功能：
1、线程将自己设置为休民状态（ state 被设成 TASK_INTERRUPTIBLE)，并把自己加入到等待队列中。
2、如果工作链表是空的，线程调用 schedule() 函数进入睡眠状态。
3、如果链表中有对象，线程不会睡眠。相反，它将自己设置成 TASK_RUNNING，脱离等待队列。
4、如果链表非空，调用 run_workqueue() 函数执行被推后的工作。

下一步，由 run_workqueue() 函数来实际完成推后到此的工作：

while (!list_empty(&cwq->worklist)) {
    struct work_struct *work;
    work_func_t f;
    void *data;
    work = list_entry(cwq->worklist.next, struct work_struct, entry);
    f = work->func;
    list_del_init(cwq->worklist.next);
    work_clear_pending(work);
    f(work);
}

该函数循环遍历链表上每个待处理的工作，执行链表每个节点上的 workqueue_struct 中的 func 成员函数：
1）当链表不为空时，选取下一个节点对象。
2）获取我们希望执行的函数 func 及其参数 data 。
3）把该节点从链表上解下来，将待处理标志位 pending 清零。
4）调用函数。
5）重复执行。

这些数据结构之间的关系确实让人觉得混乱，难以理清头绪。图 1 给出了示意图，把所有这些关系放在一起进行解释。

位于最高一层的是工作者线程。系统允许有多种类型的工作者线程存在。对于指定的一个类型，系统的每个 CPU 上都有一个该类的工作者线程。内核中有些部分可以根据需要来创建工作者线程，而在默认情况下内核只有 event 这一种类型的工作者线程。毎个工作者线程都由一个 cpu_workequeue_struct 结构体表示。而 workqueue_struct 结构体则表示给定类型的所有工作者线程。

例如，除系统默认的通用 events 工作者类型之外，我自己还加入了一种 falcon 工作者类型。并且使用的是一个拥有四个处理器的计算机。那么，系统中现在有四个 event 类型的线程（因而也就有四个 cpu_workqueue_struct 结构体）和四个 falcon 类型的线程（因而会有另外四个 cpu_workqueue_struct 结构体）。同时，有一个对应 event 类型的 workqueue_struct 和一个对应 faicon 类型的 workqueue_struct 。

工作处于最底层，让我们从这里开始。你的驱动程序创建这些需要推后执行的工作。它们用 work_struct 结构来表示。这个结构体中最重要的部分是一个指针，它指向一个函数，而正是该函数负责处理需要推后执行的具体任务。工作会被提交给某个具体的工作者线程——在这种情况下，就是特殊的 falcon 线程。然后这个工作者线程会被唤醒并执行这些排好的工作。

大部分驱动程序都使用的是现存的默认工作者线程。它们使用起来简单、方便。可是，在有些要求更严格的情况下，驱动程序需要自己的工作者线程。比如说 XFS 文件系统就为自己创建了两种新的工作者线程。

首先要做的是实际创建一些需要推后完成的工作。可以通过 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 指向的工作，处理函数为 func，参数为 data。

工作队列处理函数的原型是：

void work_handler(void *data)

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

在工作队列和内核其他部分之间使用锁机制就像在其他的进程上下文中使用锁机制一样方便。这使編写处理函数变得相对容易。

现在工作已经被创建，我们可以调度它了。想要把给定工作的处理函数提交给缺省的 events 工作线程，只需调用

schedule_work(&work);

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

有时候你并不希望工作马上就被执行，而是希望它经过一段延迟以后再执行。在这种情况下，你可以调度它在指定的时间执行：

schedule_delayed_work(&work, delay);

这时， &work 指向的 work_struct 直到 delay 指定的时钟节拍用完以后オ会执行。

排入队列的工作会在工作者线程下一次被唤醒的时候执行。有时，在继续下一步工作之前，你必须保证一些操作已经执行完毕了。这一点对模块来说就很重要，在卸载之前，它就有可能需要调用下面的函数。而在内核的其他部分，为了防止竞争条件的出现，也可能需要确保不再有待处理的工作。

出于以上目的，内核准备了一个用于刷新指定工作队列的函数：

void flush_scheduled_work(void);

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

注意，该函数并不取消任何延迟执行的工作。就是说，任何通过 schedule_delayed_work() 调度的工作，如果其延迟时间未结東，它并不会因为调用 flush_scheduled_work() 而被刷新掉。取消延迟执行的工作应该调用：

int cancel_delayed_work(struct work struct *work);

这个函数可以取消任何与 work_struct 相关的挂起工作。

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

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

struct workqueue_struct *create_workqueue(const char *name);

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

struct workqueue_struct *keventd_wq;
keventd_wg= create workqueue("events");

这样就会创建所有的工作者线程（系统中的每个处理器都有一个），并且做好所有开始处理工作之前的准备工作。

创建一个工作的时候无须考虑工作队列的类型。在创建之后，可以调用下面列举的函数。这些函数与 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 work_struct *work,
                       unsigned long delay)

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

flush_workqueue(struct workqueue_struct *wq);

该函数和前面讨论过的 flush_scheduled_work() 作用相同，只是它在返回前等待清空的是给定的队列。