工作队列workqueue应用

Posted 2020-10-26 yuxi_o

工作队列是另一种将工作推后执行的形式，它可以把工作交给一个内核线程去执行，这个下半部是在进程上下文中执行的，因此，它可以重新调度还有睡眠。

区分使用软中断/tasklet还是工作队列比较简单，如果推后的工作不需要睡眠，那么就选择软中断或tasklet，但如果需要一个可以重新调度，可以睡眠，可以获取内存，可以获取信号量，可以执行阻塞式I/O操作时，那么，请选择工作队列吧！

在老的内核当中（2.6.36之前）工作队列是内核创建一个专门的工作者线程，它有一条任务链表，当设备或者内核某进程有部分任务需要推后处理的时候就把任务挂载在工作者线程的任务链表上，然后会在未来的某个时刻，工作者线程被调度，它就会去处理挂载在任务链表上的任务了。内核有一个缺省的工作者线程叫events/n，n是处理器的编号：每个处理器对应一个线程。如单处理器只有一个events/0，而双处理器会多出一个events/1。当然，我们也可以创建我们自己的工作者线程。假如有某个任务会被频繁的触发，那么便可以为它创建一个专门的工作者线程，比如触摸屏CTP。

然而在2.6.36之后的内核当中对工作队列子系统作了改变，采用的机制改变为并发管理工作队列机制（Concurrency Managed Workqueue (cmwq)）。在原来的机制当中，当kernel需要创建一个workqueue（create_workqueue()方式）的时候，它会在每一个cpu上创建一个work_thread，为每一个cpu分配一个struct cpu_workqueue_struct，随着Kernel创建越来越多的workqueue，这将占用大量的的内存资源，并且加重了进程调度的任务量。而在新的工作队列机制中它不再在每次create_workqueue时都为workqueue创建一个work thread，而是在系统启动的时候给每个cpu创建一个work thread，当有任务项work_struct需要处理时，系统会将任务项work_struct交给某个处理器的work thread上去处理。

本文基于linux 3.14.77描述，与2.6内核在work_struct定义上不同，但原理相同。

1. 工作\\工作队列\\工作者线程关系

推后执行的任务叫工作work_struct，一个work_struct代表一个工作队列节点。

工作以队列结构组织成工作队列（workqueue)，workqueue_struct。

工作线程就是负责执行工作队列中的工作。

2.定义

work_struct定义在linux/workqueue.h中，实现在kernel/workqueue.c中。

typedef void (*work_func_t)(struct work_struct *work);

这个函数会有一个工作者线程执行，因此函数运行在进程上下文。默认情况下，允许先响应中断，并且不持有锁。
尽管该函数运行在进程上下文，但它不能访问用户空间，因为内核线程在用户空间没有相关的内存映射。（通常在系统调用发生时，内核会代表用户空间的进程运行，此时它才能访问用户空间，也只有此时它才会映射用户空间的内存)。

struct work_struct {
    atomic_long_t data;
    struct list_head entry; /*连接所有工作的链表*/
    work_func_t func;
#ifdef CONFIG_LOCKDEP
    struct lockdep_map lockdep_map;
#endif
};

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

初始化并绑定处理函数（宏定义）

DECLARE_WORK(name,work_func_t*); // 静态
INIT_WORK(struct work_struct *work, work_func_t *);  // 静态

工作队列调度

static inline bool schedule_work(struct work_struct *work)
{
    return queue_work(system_wq, work);
}

如果当前work_struct已经被调度了，schedule_work返回0，否则返回非0。当使用schedule_work方法将工作节点添加到系统工作队列，实际上就是添加到当前CPU的工作队列中。

/*
 * System-wide workqueues which are always present.
 *
 * system_wq is the one used by schedule[_delayed]_work[_on]().
 * Multi-CPU multi-threaded.  There are users which expect relatively
 * short queue flush time.  Don\'t queue works which can run for too
 * long.
 *
 * system_long_wq is similar to system_wq but may host long running
 * works.  Queue flushing might take relatively long.
 *
 * system_unbound_wq is unbound workqueue.  Workers are not bound to
 * any specific CPU, not concurrency managed, and all queued works are
 * executed immediately as long as max_active limit is not reached and
 * resources are available.
 *
 * system_freezable_wq is equivalent to system_wq except that it\'s
 * freezable.
 *
 * *_power_efficient_wq are inclined towards saving power and converted
 * into WQ_UNBOUND variants if \'wq_power_efficient\' is enabled; otherwise,
 * they are same as their non-power-efficient counterparts - e.g.
 * system_power_efficient_wq is identical to system_wq if
 * \'wq_power_efficient\' is disabled.  See WQ_POWER_EFFICIENT for more info.
 */
extern struct workqueue_struct *system_wq;
extern struct workqueue_struct *system_long_wq;
extern struct workqueue_struct *system_unbound_wq;
extern struct workqueue_struct *system_freezable_wq;
extern struct workqueue_struct *system_power_efficient_wq;
extern struct workqueue_struct *system_freezable_power_efficient_wq;

Linux内核会为每个处理器创建一个线程，用来处理工作队列中的工作。这类线程实际上就是普通的内核线程（调用worker_thread函数创建）。除了创建工作线程外，Linux内核还会创建一个全局的工作队列（system_wq），这些工作在Linux内核启动时就已经做完了。因此，如果工作不紧急的话，可以直接使用system_wq作为工作队列，也就是将工作节点（work_struct）直接添加到system_wq中。schedule_work函数的任务实际上就是将work参数指定的工作节点添加到系统的工作队列中（system_wq）。

/*通过传递不同的参数来创建用途不同的工作队列，实现在kernel/workqueue.c

events：普通工作队列，大部分的延迟处理函数都在这个工作队列上运行，要求任务执行时间短，避免互相影响。

events_long：需要长时间运行的工作项可在这个工作队列上运行。

events_unbound：该工作队列上的任务不会绑定在特定CPU上，只要某处理器空闲，就可以处理该工作队列上的任务。

events_freezable：类似于events，但工作项都是被挂起的 */

system_wq = alloc_workqueue("events", 0, 0);

system_long_wq = alloc_workqueue("events_long", 0, 0);

system_unbound_wq = alloc_workqueue("events_unbound",WQ_UNBOUND,WQ_UNBOUND_MAX_ACTIVE);

这些由Linux内核创建的工作线程的命名规则如下：kworker/cpu_id:thread_id 。对于未绑定CPU的线程池中的线程，则显示为kworker/u:thread_id。

大部分格式都是  kworker /u2:0 或者  kworker /0:0H, 查看资料得知：

内核中有很多kworker，有绑定cpu的和不绑定cpu的，它支持cpu的hotplug时work的迁移。

u：是unbound的缩写，代表没有绑定特定的CPU，kworker /u2:0中的 2 是 work_pool 的ID。

不带u的就是绑定特定cpu的workerq，它在init_workqueues中初始化，给每个cpu分配worker，如果该worker的nice小于0，说明它的优先级很高，所以就加了H属性。

参考：http://www.cnblogs.com/rohens-hbg/p/6129069.html

3. 应用

struct work_struct mywork;
void work_handler(struct work_struct *work);  // 底半部处理函数

INIT_WORK(&mywork, work_handler);
schedule_work(&mywork); //调度工作

 

4. 工作队列扩展

默认情况下可直接采用系统工作队列处理工作，当然也可以创建新的工作线程和工作队列来处理工作。

struct workqueue_struct;
create_workqueue(name); // 宏定义，创建一个新的工作队列
create_singlethread_workqueue("helloworld"); //宏定义， 创建一个单线程的工作队列不与任何CPU绑定
int queue_work(struct workqueue_struct *wq, struct work_struct *work);   //当前CPU
int queue_work_on(int cpu, struct workqueue_struct *wq, struct work_struct *work);    // 指定CPU
void flush_workqueue(struct workqueue_struct *wq);  //  确保工作队列中的工作都完成
void destroy_workqueue(struct workqueue_struct *wq);  //  销毁工作队列  

应用

struct workqueue_struct *queue = NULL;
queue = create_workqueue("newProcess");//创建多个CPU上的工作者进程
struct work_struct mywork;
void work_handler(struct work_struct *work);  // 底半部处理函数
INIT_WORK(&mywork, work_handler); //初始化work
queue_work(queue, &mywork); // 把工作添加到当前cpu的工作队列中
flush_workqueue(queue); //刷新
destroy_workqueue(queue);//销毁新创建的工作者进程

 

5. delayed_work

对于周期性的任务，内核提供了一套封装好的快捷机制，本质上利用了工作队列和定时器实现，这套机制就是delayed_work。

void delayed_work_timer_fn(unsigned long __data);
struct delayed_work {
    struct work_struct work;
struct timer_list timer;

    /* target workqueue and CPU ->timer uses to queue ->work */
    struct workqueue_struct *wq;
    int cpu;
};

DECLARE_DELAYED_WORK(n, f); //宏定义
INIT_DELAYED_WORK(_work, _func); //宏定义
static inline bool schedule_delayed_work(struct delayed_work *dwork, unsigned long delay); // 指定延时后执行调度

其中delay的单位是jiffies，因此一种常见的用法如下：

schedule_delayed_work(&work, msecs_to_jiffies(poll_interval));

如果要周期性的执行任务，通常会在delayed_work的工作函数中再次调用schedule_delayed_work()，周而复始。

bool cancel_delayed_work(struct delayed_work *dwork);  //取消延迟工作
bool cancel_delayed_work_sync(struct delayed_work *dwork); // 取消延迟工作

 

参考：

1. linux内核对中断的处理方式

2. tasklet_workqueue

3. 工作队列

4. kworker内核工作队列详解