Linux-workqueue讲解

31 March 2022

1. workqueue 是什么？

workqueue是对内核线程封装的用于处理各种工作项的一种处理方法，由于处理对象是用链表拼接一个个工作项，依次取出来处理，然后从链表删除，就像一个队列排好队依次处理一样，所以也称工作队列，

所谓封装可以简单理解一个中转站，一边指向“合适”的内核线程，一边接受你丢过来的工作项，用结构体 workqueue_srtuct表示，而所谓工作项也是个结构体 work_struct，里面有个成员指针，指向你最终要实现的函数，

struct workqueue_struct {
    struct list_head    pwqs;        /* WR: all pwqs of this wq */
    struct list_head    list;        /* PL: list of all workqueues */

    struct workqueue_attrs    *unbound_attrs;    /* WQ: only for unbound wqs */
    struct pool_workqueue    *dfl_pwq;    /* WQ: only for unbound wqs */

    char            name[WQ_NAME_LEN]; /* I: workqueue name */

    unsigned int        flags ____cacheline_aligned; /* WQ: WQ_* flags */
    struct pool_workqueue __percpu *cpu_pwqs; /* I: per-cpu pwqs */
    struct pool_workqueue __rcu *numa_pwq_tbl[]; /* FR: unbound pwqs indexed by node */
};

struct work_struct {
    atomic_long_t data;    //函数的参数
    struct list_head entry;    //挂到链表
    work_func_t func; //函数指针，指向你实现的函数功能
};

当然使用者在实现自己函数功能后可以直接调用，或者通过kthread_create()把函数当做新线程的主代码，或者add_timer添加到一个定时器延时处理。

那为何要弄个work_struct工作项先封装函数，然后再丢到workqueue_srtuct处理呢？这就看使用场景了，如果是一个大函数，处理事项比较多，且需要重复处理，可以单独开辟一个内核线程处理；对延时敏感的可以用定时器；

如果只是简单的一个函数功能，且函数里面有延时动作的，就适合放到工作队列来处理了，毕竟定时器处理的函数是在中断上下文，不能delay或者引发进程切换的API，而且开辟一个内核线程是耗时且耗费资源的，一般用于函数需要while(1) 不断循环处理的，不然处理一次函数后退出，线程又被销毁，简直就是浪费！

2. 怎么用？

一个简单示例：

#include <linux/module.h>
#include <linux/kernel.h>
#include <linux/delay.h>
#include <linux/workqueue.h>

struct workqueue_struct *workqueue_test;

struct work_struct work_test;

void work_test_func(struct work_struct *work)
{
    printk("%s()\n", __func__);

    //mdelay(1000);
    //queue_work(workqueue_test, &work_test);
}


static int test_init(void)
{
    printk("Hello,world!\n");

    /* 1. 自己创建一个workqueue， 中间参数为0，默认配置 */
    workqueue_test = alloc_workqueue("workqueue_test", 0, 0);

    /* 2. 初始化一个工作项，并添加自己实现的函数 */
    INIT_WORK(&work_test, work_test_func);

    /* 3. 将自己的工作项添加到指定的工作队列去， 同时唤醒相应线程处理 */
    queue_work(workqueue_test, &work_test);

    return 0;
}

static void test_exit(void)
{
    printk("Goodbye,cruel world!\n");
    destroy_workqueue(workqueue_test);
}

module_init(test_init);
module_exit(test_exit);

MODULE_AUTHOR("ByteJog<bytejog@163.com>");
MODULE_LICENSE("GPL");

只需三步就可以了，当然内核已经为我们创建了几个工作队列，我们可以直接将自己的工作项挂到相应的队列即可：

/* kernel/workqueue.c */

system_wq = alloc_workqueue("events", 0, 0);
system_highpri_wq = alloc_workqueue("events_highpri", WQ_HIGHPRI, 0);
system_long_wq = alloc_workqueue("events_long", 0, 0);
system_unbound_wq = alloc_workqueue("events_unbound", WQ_UNBOUND,
						WQ_UNBOUND_MAX_ACTIVE);
system_freezable_wq = alloc_workqueue("events_freezable",
						WQ_FREEZABLE, 0);
system_power_efficient_wq = alloc_workqueue("events_power_efficient",
					    WQ_POWER_EFFICIENT, 0);
system_freezable_power_efficient_wq =
						alloc_workqueue("events_freezable_power_efficient",
					    WQ_FREEZABLE | WQ_POWER_EFFICIENT,0);

所以代码可以改为：

static int test_init(void)
{
    printk("Hello,world!\n");

    /* 2. 初始化一个工作项，并添加自己实现的函数 */
    INIT_WORK(&work_test, work_test_func);

    /* 3. 将自己的工作项添加到指定的工作队列去， 同时唤醒相应线程处理 */
    queue_work(system_wq, &work_test);

    return 0;
}

如果workqueue对象是 system_wq，可以使用另一个封装函数schedule_work(&work_test)

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

将自己的工作项挂到已有的工作队列需要注意的是由于这些队列是共享的，各个驱动都有可能将自己的工作项放到同个队列，会导致队列的项拥挤，当有些项写的代码耗时久或者调用delay()延时特别久，你的项将会迟迟得不到执行！

所以早期很多驱动开发人员都是自己创建workqueue，添加自己的work。在Linux-2.XXX时代，创建workqueue时会创建属于workqueue自己的内核线程，这些线程是“私有的”，虽然是方便了驱动开发人员，但每个驱动都“一言不合”就

创建workqueue导致太多线程，严重占用系统资源和效率，所以在Linux-3.XXX时代，社区开发人员将workqueue和内核线程剥离！内核会自己事先创建相应数量的线程（后面详解），被所有驱动共享使用。用户调用alloc_workqueue()只是创建workqueue这个空壳，其主要作用：

a. 兼容Linux-2.XXX时代代码

b. 新增flag字段表明这个workqueue的属性（普通优先级还是高优先级等），方便在queue_work()时寻找“合适的”线程，因为事先创建的线程分普通优先级、高优先级、绑定CPU线程，非绑定CPU线程等。

当然这对驱动开发人员是透明的，驱动人员只需关注调用queue_work()让线程执行自己的工作项，至于是这个workqueue的私有线程还是现在的共享线程，不重要！这样就限制了系统工作线程的暴涨，唯一的缺点就是前面提到的，跟别人共享会增加自己的工作项被执行的不确定性。只能说各个驱动开发人员自我约束，尽量使得工作项函数简短快速，如果我们需要等自己的工作项被执行完才能处理其他事情，可以调用flush_work() 等待work被执行完：

/**
 * flush_work - wait for a work to finish executing the last queueing instance
 * @work: the work to flush
 *
 * Wait until @work has finished execution.  @work is guaranteed to be idle
 * on return if it hasn't been requeued since flush started.
 *
 * RETURNS:
 * %true if flush_work() waited for the work to finish execution,
 * %false if it was already idle.
 */
bool flush_work(struct work_struct *work)
{
    struct wq_barrier barr;

    lock_map_acquire(&work->lockdep_map);
    lock_map_release(&work->lockdep_map);

    if (start_flush_work(work, &barr)) {
        wait_for_completion(&barr.done);
        destroy_work_on_stack(&barr.work);
        return true;
    } else {
        return false;
    }
}
EXPORT_SYMBOL_GPL(flush_work);

参考：

https://www.cnblogs.com/vedic/p/11069249.html