Concurrency Managed Workqueue(四)workqueue如何处理work

1、前言

本文主要讲述下面两部分的内容：

一、将work挂入workqueue的处理过程

二、如何处理挂入workqueue的work

 

2、用户将一个work挂入workqueue

一、queue_work_on函数

使用workqueue机制的模块能够调用queue_work_on（有其余变种的接口，这里略过，其实思路是一致的）将一个定义好的work挂入workqueue，具体代码以下：

bool queue_work_on(int cpu, struct workqueue_struct *wq, struct work_struct *work)
{
    ……

    if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work))) {
        __queue_work(cpu, wq, work);－－－挂入work list并通知worker thread pool来处理
        ret = true;
    }

    ……
}

work_struct的data member中的WORK_STRUCT_PENDING_BIT这个bit标识了该work是处于pending状态仍是正在处理中，pending状态的work只会挂入一次。大部分的逻辑都是在__queue_work函数中，下面的小节都是描述该函数的执行过程。

二、__WQ_DRAINING的解释

__queue_work函数一开始会校验__WQ_DRAINING这个flag，以下：

if (unlikely(wq->flags & __WQ_DRAINING) && WARN_ON_ONCE(!is_chained_work(wq)))
        return;

__WQ_DRAINING这个flag表示该workqueue正在进行draining的操做，这多半是发送在销毁workqueue的时候，既然要销毁，那么挂入该workqueue的全部的work都要处理完毕，才容许它消亡。当想要将一个workqueue中全部的work都清空的时候，若是还有work挂入怎么办？通常而言，这时候固然是不容许新的work挂入了，毕竟如今的目标是清空workqueue中的work。可是有一种特例（经过is_chained_work断定），也就是正在清空的work（隶属于该workqueue）又触发了一个queue work的操做（也就是所谓chained work），这时候该work容许挂入。

三、选择pool workqueue

if (req_cpu == WORK_CPU_UNBOUND)
        cpu = raw_smp_processor_id();

if (!(wq->flags & WQ_UNBOUND))
        pwq = per_cpu_ptr(wq->cpu_pwqs, cpu);
    else
        pwq = unbound_pwq_by_node(wq, cpu_to_node(cpu));

WORK_CPU_UNBOUND表示并不指定cpu，这时候，选择当前代码运行的那个cpu了。一旦肯定了cpu了，对于非unbound的workqueue，固然使用per cpu的pool workqueue。若是是unbound的workqueue，那么要根据numa node id来选择。cpu_to_node能够从cpu id获取node id。须要注意的是：这里选择的pool wq只是备选的，可能用也可能不用，它有可能会被替换掉，具体参考下一节描述。

四、选择worker thread pool

与其说挂入workqueue，不如说挂入worker thread pool，由于毕竟是线程池来处理具体的work。pool_workqueue有一个相关联的worker thread pool（struct pool_workqueue的pool成员），所以看起来选择了pool wq也就选定了worker pool了，可是，不是当前选定的那个pool wq对应的worker pool就适合该work，由于有时候该work可能正在其余的worker thread上执行中，在这种状况下，为了确保work的callback function不会重入，该work最好仍是挂在那个worker thread pool上，具体代码以下：

last_pool = get_work_pool(work);
    if (last_pool && last_pool != pwq->pool) {
        struct worker *worker;

        spin_lock(&last_pool->lock);

        worker = find_worker_executing_work(last_pool, work);

        if (worker && worker->current_pwq->wq == wq) {
            pwq = worker->current_pwq;
        } else {
            /* meh... not running there, queue here */
            spin_unlock(&last_pool->lock);
            spin_lock(&pwq->pool->lock);
        }
    } else {
        spin_lock(&pwq->pool->lock);
    }

last_pool记录了上一次该work是被哪个worker pool处理的，若是last_pool就是pool wq对应的worker pool，那么皆大欢喜，不然只能使用last pool了。使用last pool的例子比较复杂一些，由于这时候须要根据last worker pool找到对应的pool workqueue。find_worker_executing_work函数能够找到具体哪个worker线程正在处理该work，若是没有找到，那么仍是使用第3节中选定的pool wq吧，不然，选择该worker线程当前的那个pool workqueue（其实也就是选定了线程池）。

五、选择work挂入的队列

队列有两个，一个是被推迟执行的队列（pwq->delayed_works），一个是线程池要处理的队列（pwq->pool->worklist），若是挂入线程池要处理的队列，也就意味着该work进入active状态，线程池会马上启动处理流程，若是挂入推迟执行的队列，那么该work仍是pending状态：

    pwq->nr_in_flight[pwq->work_color]++;
    work_flags = work_color_to_flags(pwq->work_color);

    if (likely(pwq->nr_active < pwq->max_active)) {
        pwq->nr_active++;
        worklist = &pwq->pool->worklist;
    } else {
        work_flags |= WORK_STRUCT_DELAYED;
        worklist = &pwq->delayed_works;
    }

    insert_work(pwq, work, worklist, work_flags);

具体的挂入队列的动做是在insert_work函数中完成的。

六、唤醒idle的worker来处理该work

在insert_work函数中有下面的代码：

if (__need_more_worker(pool))
        wake_up_worker(pool);

当线程池中正在运行状态的worker线程数目等于0的时候，说明须要wakeup线程池中处于idle状态的的worker线程来处理work。

 

3、线程池如何建立worker线程？

一、per cpu worker pool何时建立worker线程？

对于per-CPU workqueue，每一个cpu有两个线程池，一个是normal，一个是high priority的。在初始化函数init_workqueues中有对这两个线程池的初始化：

for_each_online_cpu(cpu) {
    struct worker_pool *pool;

    for_each_cpu_worker_pool(pool, cpu) {
        pool->flags &= ~POOL_DISASSOCIATED;
        BUG_ON(!create_worker(pool));
    }
}

所以，在系统初始化的时候，per cpu workqueue共享的那些线程池（2 x cpu nr）就会经过create_worker建立一个initial worker。

一旦initial worker启动，该线程会执行worker_thread函数来处理work，在处理过程当中，若是有须要， worker会建立新的线程。

二、unbound thread pool何时建立worker线程？

咱们先看看unbound thread pool的创建，和per-CPU不一样的是unbound thread pool是全局共享的，所以，每当建立不一样属性的unbound workqueue的时候，都须要建立pool_workqueue及其对应的worker pool，这时候就会调用get_unbound_pool函数在当前系统中现存的线程池中找是否有匹配的worker pool，若是没有就须要建立新的线程池。在建立新的线程池以后，会马上调用create_worker建立一个initial worker。和per cpu worker pool同样，一旦initial worker启动，随着work不断的挂入以及worker处理work的具体状况，线程池会动态建立worker。

三、如何建立worker。代码以下：

static struct worker *create_worker(struct worker_pool *pool)
{
    struct worker *worker = NULL;
    int id = -1;
    char id_buf[16];

    id = ida_simple_get(&pool->worker_ida, 0, 0, GFP_KERNEL);－－－－分配ID

    worker = alloc_worker(pool->node);－－－－－分配worker struct的内存

    worker->pool = pool;
    worker->id = id;

    if (pool->cpu >= 0)－－－－－－－－－worker的名字
        snprintf(id_buf, sizeof(id_buf), "%d:%d%s", pool->cpu, id,  pool->attrs->nice < 0  ? "H" : "");
    else
        snprintf(id_buf, sizeof(id_buf), "u%d:%d", pool->id, id);

worker->task = kthread_create_on_node(worker_thread, worker, pool->node,   "kworker/%s", id_buf);

    set_user_nice(worker->task, pool->attrs->nice); －－－建立task并设定nice value
    worker->task->flags |= PF_NO_SETAFFINITY; 
    worker_attach_to_pool(worker, pool); －－－－－创建worker和线程池的关系

    spin_lock_irq(&pool->lock);
    worker->pool->nr_workers++;
    worker_enter_idle(worker);
    wake_up_process(worker->task);－－－－－－让worker运行起来
    spin_unlock_irq(&pool->lock);

    return worker;
}

代码不复杂，经过线程池（struct worker_pool）绑定的cpu信息（struct worker_pool的cpu成员）能够知道该pool是per-CPU仍是unbound，对于per-CPU线程池，pool->cpu是大于等于0的。对于对于per-CPU线程池，其worker线程的名字是kworker/cpu：worker id，若是是high priority的，后面还跟着一个H字符。对于unbound线程池，其worker线程的名字是kworker/u pool id：worker id。

 

4、work的处理

本章主要描述worker_thread函数的执行流程，部分代码有删节，保留主干部分。

一、PF_WQ_WORKER标记

worker线程函数一开始就会经过PF_WQ_WORKER来标注本身：

worker->task->flags |= PF_WQ_WORKER;

有了这样一个flag，调度器在调度当前进程sleep的时候能够检查这个准备sleep的进程是不是一个worker线程，若是是的话，那么调度器不能鲁莽的调度到其余的进程，这时候，还须要找到该worker对应的线程池，唤醒一个idle的worker线程。经过workqueue模块和调度器模块的交互，当work A被阻塞后（处理该work的worker线程进入sleep），调度器会唤醒其余的worker线程来处理其余的work B，work C……

二、管理线程池中的线程

recheck:
    if (!need_more_worker(pool))
        goto sleep;

    if (unlikely(!may_start_working(pool)) && manage_workers(worker))
        goto recheck;

如何判断是否须要建立更多的worker线程呢？原则以下：

（1）有事情作：挂在worker pool中的work list不能是空的，若是是空的，那么固然sleep就行了

（2）比较忙：worker pool的nr_running成员表示线程池中当前正在干活（running状态）的worker线程有多少个，当nr_running等于0表示全部的worker线程在处理work的时候阻塞了，这时候，必需要启动新的worker线程来处理worker pool上处于active状态的work链表上的work们。

三、worker线程开始处理work

worker_clr_flags(worker, WORKER_PREP | WORKER_REBOUND);

do {
    struct work_struct *work =   list_first_entry(&pool->worklist,  struct work_struct, entry);

    if (likely(!(*work_data_bits(work) & WORK_STRUCT_LINKED))) {
        process_one_work(worker, work);
        if (unlikely(!list_empty(&worker->scheduled)))
            process_scheduled_works(worker);
    } else {
        move_linked_works(work, &worker->scheduled, NULL);
        process_scheduled_works(worker);
    }
} while (keep_working(pool));

worker_set_flags(worker, WORKER_PREP);

按理说worker线程处理work应该比较简单，从线程池的worklist中取一个work，而后调用process_one_work处理之就OK了，不过现实稍微复杂一些，work和work之间并非独立的，也就是说，work A和work B多是linked work，这些linked work应该被一个worker来处理。WORK_STRUCT_LINKED标记了work是属于linked work，若是是linked work，worker并不直接处理，而是将其挂入scheduled work list，而后调用process_scheduled_works来处理。毫无疑问，process_scheduled_works也是调用process_one_work来处理一个一个scheduled work list上的work。

scheduled work list并不是仅仅应用在linked work，在worker处理work的时候，有一个原则要保证：同一个work不能被同一个cpu上的多个worker同时执行。这时候，若是worker发现本身要处理的work正在被另一个worker线程处理，那么本worker线程将不处理该work，只须要挂入正在执行该work的worker线程的scheduled work list便可。