• 万字整理 | 深入理解工作队列

    伟林，中年码农，从事过电信、手机、安全、芯片等行业，目前依旧从事Linux方向开发工作，个人爱好Linux相关知识分享，个人微博CSDN pwl999，欢迎大家关注！

    • 1.1 worker_pool

      • 1.1.1 normal worker_pool

      • 1.1.2 unbound worker_pool

    • 1.2 worker

      • 1.2.1 worker处理work

      • 1.2.2 worker_pool动态管理worker

      • 1.2.3 cpu hotplug处理

    • 1.3 workqueue

      • 1.3.1 系统workqueue

      • 1.3.2 workqueue创建

      • 1.3.3 flush_workqueue()

    • 1.4 pool_workqueue

    • 1.5 work

      • 1.5.1 queue_work()

      • 1.5.2 flush_work()

    • 2.1 schedule_work()

    • 2.2 sschedule_work_on()

    • 2.3 schedule_delayed_work()

    workqueue是内核里面很重要的一个机制，特别是内核驱动，一般的小型任务(work)都不会自己起一个线程来处理，而是扔到workqueu中处理。workqueue的主要工作就是用进程上下文来处理内核中大量的小任务。

    所以workqueue的主要设计思想：一个是并行，多个work不要相互阻塞；另外一个是节省资源，多个work尽量共享资源(进程、调度、内存)，不要造成系统过多的资源浪费。

    为了实现的设计思想，workqueue的设计实现也更新了很多版本。最新的workqueue实现叫做CMWQ(Concurrency Managed Workqueue)，也就是用更加智能的算法来实现“并行和节省”。新版本的workque创建函数改成alloc_workqueue()，旧版本的函数create_*workqueue()逐渐会被被废弃。

    本文的代码分析基于linux kernel 3.18.22，最好的学习方法还是"read the fucking source code"

    1.CMWQ的几个基本概念

    关于workqueue中几个概念都是work相关的数据结构非常容易混淆，大概可以这样来理解：

    • work ：工作。

    • workqueue ：工作的集合。workqueue和work是一对多的关系。

    • worker ：工人。在代码中worker对应一个work_thread()内核线程。

    • worker_pool：工人的集合。worker_pool和worker是一对多的关系。

    • pwq(pool_workqueue)：中间人/中介，负责建立起workqueue和worker_pool之间的关系。workqueue和pwq是一对多的关系，pwq和worker_pool是一对一的关系。

    be21b93db66dab346566f0783fa72b6e.png

    最终的目的还是把work(工作)传递给worker(工人)去执行，中间的数据结构和各种关系目的是把这件事组织的更加清晰高效。

    1.1 worker_pool

    每个执行work的线程叫做worker，一组worker的集合叫做worker_pool。CMWQ的精髓就在worker_pool里面worker的动态增减管理上manage_workers()。

    CMWQ对worker_pool分成两类：

    • normal worker_pool，给通用的workqueue使用；

    • unbound worker_pool，给WQ_UNBOUND类型的的workqueue使用；

    1.1.1 normal worker_pool

    默认work是在normal worker_pool中处理的。系统的规划是每个cpu创建两个normal worker_pool：一个normal优先级(nice=0)、一个高优先级(nice=HIGHPRI_NICE_LEVEL)，对应创建出来的worker的进程nice不一样。

    每个worker对应一个worker_thread()内核线程，一个worker_pool包含一个或者多个worker，worker_pool中worker的数量是根据worker_pool中work的负载来动态增减的。

    我们可以通过“ps|grep kworker”命令来查看所有worker对应的内核线程，normal worker_pool对应内核线程(worker_thread())的命名规则是这样的：

    1. snprintf(id_buf, sizeof(id_buf), "%d:%d%s", pool->cpu, id,
    2.    pool->attrs->nice < 0  ? "H" : "");
    3.  
    4.  worker->task = kthread_create_on_node(worker_thread, worker, pool->node,
    5.            "kworker/%s", id_buf);

    so类似名字是normal worker_pool：

    1. shell@PRO5:/ $ ps | grep "kworker"
    2. root      14    2     0      0     worker_thr 0000000000 S kworker/1:0H  // cpu1 高优先级worker_pool的第0个worker进程
    3. root      17    2     0      0     worker_thr 0000000000 S kworker/2:0  // cpu2 低优先级worker_pool的第0个worker进程
    4. root      18    2     0      0     worker_thr 0000000000 S kworker/2:0H  // cpu2 高优先级worker_pool的第0个worker进程
    5. root      23699 2     0      0     worker_thr 0000000000 S kworker/0:1  // cpu0 低优先级worker_pool的第1个worker进程
    85f50da19fa5f3efc3aecef1bde62804.png

    对应的拓扑图如下：

    b456b607050676cf7e7df208d47a927e.png

    以下是normal worker_pool详细的创建过程代码分析：

    • kernel/workqueue.c:

    • init_workqueues() -> init_worker_pool()/create_worker()

    1. static int __init init_workqueues(void)
    2. {
    3.  int std_nice[NR_STD_WORKER_POOLS] = { 0, HIGHPRI_NICE_LEVEL };
    4.  int i, cpu;
    5.  
    6.  // (1)给每个cpu创建对应的worker_pool
    7.  /* initialize CPU pools */
    8.  for_each_possible_cpu(cpu) {
    9.   struct worker_pool *pool;
    10.  
    11.   i = 0;
    12.   for_each_cpu_worker_pool(pool, cpu) {
    13.    BUG_ON(init_worker_pool(pool));
    14.    // 指定cpu
    15.    pool->cpu = cpu;
    16.    cpumask_copy(pool->attrs->cpumask, cpumask_of(cpu));
    17.    // 指定进程优先级nice
    18.    pool->attrs->nice = std_nice[i++];
    19.    pool->node = cpu_to_node(cpu);
    20.  
    21.    /* alloc pool ID */
    22.    mutex_lock(&wq_pool_mutex);
    23.    BUG_ON(worker_pool_assign_id(pool));
    24.    mutex_unlock(&wq_pool_mutex);
    25.   }
    26.  }
    27.  
    28.  // (2)给每个worker_pool创建第一个worker
    29.  /* create the initial worker */
    30.  for_each_online_cpu(cpu) {
    31.   struct worker_pool *pool;
    32.  
    33.   for_each_cpu_worker_pool(pool, cpu) {
    34.    pool->flags &= ~POOL_DISASSOCIATED;
    35.    BUG_ON(!create_worker(pool));
    36.   }
    37.  }
    38.  
    39. }
    40. | →
    41. static int init_worker_pool(struct worker_pool *pool)
    42. {
    43.  spin_lock_init(&pool->lock);
    44.  pool->id = -1;
    45.  pool->cpu = -1;
    46.  pool->node = NUMA_NO_NODE;
    47.  pool->flags |= POOL_DISASSOCIATED;
    48.  // (1.1) worker_pool的work list，各个workqueue把work挂载到这个链表上，
    49.  // 让worker_pool对应的多个worker来执行
    50.  INIT_LIST_HEAD(&pool->worklist);
    51.  // (1.2) worker_pool的idle worker list，
    52.  // worker没有活干时，不会马上销毁，先进入idle状态备选
    53.  INIT_LIST_HEAD(&pool->idle_list);
    54.  // (1.3) worker_pool的busy worker list，
    55.  // worker正在干活，在执行work
    56.  hash_init(pool->busy_hash);
    57.  
    58.  // (1.4) 检查idle状态worker是否需要destroy的timer
    59.  init_timer_deferrable(&pool->idle_timer);
    60.  pool->idle_timer.function = idle_worker_timeout;
    61.  pool->idle_timer.data = (unsigned long)pool;
    62.  
    63.  // (1.5) 在worker_pool创建新的worker时，检查是否超时的timer
    64.  setup_timer(&pool->mayday_timer, pool_mayday_timeout,
    65.       (unsigned long)pool);
    66.  
    67.  mutex_init(&pool->manager_arb);
    68.  mutex_init(&pool->attach_mutex);
    69.  INIT_LIST_HEAD(&pool->workers);
    70.  
    71.  ida_init(&pool->worker_ida);
    72.  INIT_HLIST_NODE(&pool->hash_node);
    73.  pool->refcnt = 1;
    74.  
    75.  /* shouldn't fail above this point */
    76.  pool->attrs = alloc_workqueue_attrs(GFP_KERNEL);
    77.  if (!pool->attrs)
    78.   return -ENOMEM;
    79.  return 0;
    80. }
    81. | →
    82. static struct worker *create_worker(struct worker_pool *pool)
    83. {
    84.  struct worker *worker = NULL;
    85.  int id = -1;
    86.  char id_buf[16];
    87.  
    88.  /* ID is needed to determine kthread name */
    89.  id = ida_simple_get(&pool->worker_ida, 0, 0, GFP_KERNEL);
    90.  if (id < 0)
    91.   goto fail;
    92.  
    93.  worker = alloc_worker(pool->node);
    94.  if (!worker)
    95.   goto fail;
    96.  
    97.  worker->pool = pool;
    98.  worker->id = id;
    99.  
    100.  if (pool->cpu >= 0)
    101.   // (2.1) 给normal worker_pool的worker构造进程名
    102.   snprintf(id_buf, sizeof(id_buf), "%d:%d%s", pool->cpu, id,
    103.     pool->attrs->nice < 0  ? "H" : "");
    104.  else
    105.   // (2.2) 给unbound worker_pool的worker构造进程名
    106.   snprintf(id_buf, sizeof(id_buf), "u%d:%d", pool->id, id);
    107.  
    108.  // (2.3) 创建worker对应的内核进程
    109.  worker->task = kthread_create_on_node(worker_thread, worker, pool->node,
    110.            "kworker/%s", id_buf);
    111.  if (IS_ERR(worker->task))
    112.   goto fail;
    113.  
    114.  // (2.4) 设置内核进程对应的优先级nice
    115.  set_user_nice(worker->task, pool->attrs->nice);
    116.  
    117.  /* prevent userland from meddling with cpumask of workqueue workers */
    118.  worker->task->flags |= PF_NO_SETAFFINITY;
    119.  
    120.  // (2.5) 将worker和worker_pool绑定
    121.  /* successful, attach the worker to the pool */
    122.  worker_attach_to_pool(worker, pool);
    123.  
    124.  // (2.6) 将worker初始状态设置成idle，
    125.  // wake_up_process以后，worker自动leave idle状态
    126.  /* start the newly created worker */
    127.  spin_lock_irq(&pool->lock);
    128.  worker->pool->nr_workers++;
    129.  worker_enter_idle(worker);
    130.  wake_up_process(worker->task);
    131.  spin_unlock_irq(&pool->lock);
    132.  
    133.  return worker;
    134.  
    135. fail:
    136.  if (id >= 0)
    137.   ida_simple_remove(&pool->worker_ida, id);
    138.  kfree(worker);
    139.  return NULL;
    140. }
    141. || →
    142. static void worker_attach_to_pool(struct worker *worker,
    143.        struct worker_pool *pool)
    144. {
    145.  mutex_lock(&pool->attach_mutex);
    146.  
    147.  // (2.5.1) 将worker线程和cpu绑定
    148.  /*
    149.   * set_cpus_allowed_ptr() will fail if the cpumask doesn't have any
    150.   * online CPUs.  It'll be re-applied when any of the CPUs come up.
    151.   */
    152.  set_cpus_allowed_ptr(worker->task, pool->attrs->cpumask);
    153.  
    154.  /*
    155.   * The pool->attach_mutex ensures %POOL_DISASSOCIATED remains
    156.   * stable across this function.  See the comments above the
    157.   * flag definition for details.
    158.   */
    159.  if (pool->flags & POOL_DISASSOCIATED)
    160.   worker->flags |= WORKER_UNBOUND;
    161.  
    162.  // (2.5.2) 将worker加入worker_pool链表
    163.  list_add_tail(&worker->node, &pool->workers);
    164.  
    165.  mutex_unlock(&pool->attach_mutex);
    166. }

    1.1.2 unbound worker_pool

    大部分的work都是通过normal worker_pool来执行的(例如通过schedule_work()、schedule_work_on()压入到系统workqueue(system_wq)中的work)，最后都是通过normal worker_pool中的worker来执行的。这些worker是和某个cpu绑定的，work一旦被worker开始执行，都是一直运行到某个cpu上的不会切换cpu。

    unbound worker_pool相对应的意思，就是worker可以在多个cpu上调度的。但是他其实也是绑定的，只不过它绑定的单位不是cpu而是node。所谓的node是对NUMA(Non Uniform Memory Access Architecture)系统来说的，NUMA可能存在多个node，每个node可能包含一个或者多个cpu。

    unbound worker_pool对应内核线程(worker_thread())的命名规则是这样的：

    1. snprintf(id_buf, sizeof(id_buf), "u%d:%d", pool->id, id);
    2.  
    3.  worker->task = kthread_create_on_node(worker_thread, worker, pool->node,
    4.            "kworker/%s", id_buf);

    so类似名字是unbound worker_pool：

    1. shell@PRO5:/ $ ps | grep "kworker"
    2. root      23906 2     0      0     worker_thr 0000000000 S kworker/u20:2 // unbound pool 20的第2个worker进程
    3. root      24564 2     0      0     worker_thr 0000000000 S kworker/u20:0 // unbound pool 20的第0个worker进程
    4. root      24622 2     0      0     worker_thr 0000000000 S kworker/u21:1 // unbound pool 21的第1个worker进程

    unbound worker_pool也分成两类：

    • unbound_std_wq。每个node对应一个worker_pool，多个node就对应多个worker_pool;

    99a85596c1472823617534e21c16b5bc.png

    对应的拓扑图如下：

    133c3287040bb2709c18454448d8a923.png

    • ordered_wq。所有node对应一个default worker_pool；

    a9dbd75eaffb507d31465cc9fcf7c9f3.png

    对应的拓扑图如下：

    5686c7e7a9e5d390dd6b58709c4eb3b6.png

    以下是unbound worker_pool详细的创建过程代码分析：

    • kernel/workqueue.c:

    • init_workqueues() -> unbound_std_wq_attrs/ordered_wq_attrs

    1. static int __init init_workqueues(void)
    2. {
    3.  
    4.  // (1) 初始化normal和high nice对应的unbound attrs
    5.  /* create default unbound and ordered wq attrs */
    6.  for (i = 0; i < NR_STD_WORKER_POOLS; i++) {
    7.   struct workqueue_attrs *attrs;
    8.  
    9.   // (2) unbound_std_wq_attrs
    10.   BUG_ON(!(attrs = alloc_workqueue_attrs(GFP_KERNEL)));
    11.   attrs->nice = std_nice[i];
    12.   unbound_std_wq_attrs[i] = attrs;
    13.   /*
    14.    * An ordered wq should have only one pwq as ordering is
    15.    * guaranteed by max_active which is enforced by pwqs.
    16.    * Turn off NUMA so that dfl_pwq is used for all nodes.
    17.    */
    18.   // (3) ordered_wq_attrs，no_numa = true;
    19.   BUG_ON(!(attrs = alloc_workqueue_attrs(GFP_KERNEL)));
    20.   attrs->nice = std_nice[i];
    21.   attrs->no_numa = true;
    22.   ordered_wq_attrs[i] = attrs;
    23.  }
    24.  
    25.  
    26. }

    • kernel/workqueue.c:

    • __alloc_workqueue_key() -> alloc_and_link_pwqs() -> apply_workqueue_attrs() -> alloc_unbound_pwq()/numa_pwq_tbl_install()

    1. struct workqueue_struct *__alloc_workqueue_key(const char *fmt,
    2.             unsigned int flags,
    3.             int max_active,
    4.             struct lock_class_key *key,
    5.             const char *lock_name, ...)
    6. {
    7.  size_t tbl_size = 0;
    8.  va_list args;
    9.  struct workqueue_struct *wq;
    10.  struct pool_workqueue *pwq;
    11.  
    12.  /* see the comment above the definition of WQ_POWER_EFFICIENT */
    13.  if ((flags & WQ_POWER_EFFICIENT) && wq_power_efficient)
    14.   flags |= WQ_UNBOUND;
    15.  
    16.  /* allocate wq and format name */
    17.  if (flags & WQ_UNBOUND)
    18.   tbl_size = nr_node_ids * sizeof(wq->numa_pwq_tbl[0]);
    19.  
    20.  // (1) 分配workqueue_struct数据结构
    21.  wq = kzalloc(sizeof(*wq) + tbl_size, GFP_KERNEL);
    22.  if (!wq)
    23.   return NULL;
    24.  
    25.  if (flags & WQ_UNBOUND) {
    26.   wq->unbound_attrs = alloc_workqueue_attrs(GFP_KERNEL);
    27.   if (!wq->unbound_attrs)
    28.    goto err_free_wq;
    29.  }
    30.  
    31.  va_start(args, lock_name);
    32.  vsnprintf(wq->name, sizeof(wq->name), fmt, args);
    33.  va_end(args);
    34.  
    35.  // (2) pwq最多放到worker_pool中的work数
    36.  max_active = max_active ?: WQ_DFL_ACTIVE;
    37.  max_active = wq_clamp_max_active(max_active, flags, wq->name);
    38.  
    39.  /* init wq */
    40.  wq->flags = flags;
    41.  wq->saved_max_active = max_active;
    42.  mutex_init(&wq->mutex);
    43.  atomic_set(&wq->nr_pwqs_to_flush, 0);
    44.  INIT_LIST_HEAD(&wq->pwqs);
    45.  INIT_LIST_HEAD(&wq->flusher_queue);
    46.  INIT_LIST_HEAD(&wq->flusher_overflow);
    47.  INIT_LIST_HEAD(&wq->maydays);
    48.  
    49.  lockdep_init_map(&wq->lockdep_map, lock_name, key, 0);
    50.  INIT_LIST_HEAD(&wq->list);
    51.  
    52.  // (3) 给workqueue分配对应的pool_workqueue
    53.  // pool_workqueue将workqueue和worker_pool链接起来
    54.  if (alloc_and_link_pwqs(wq) < 0)
    55.   goto err_free_wq;
    56.  
    57.  // (4) 如果是WQ_MEM_RECLAIM类型的workqueue
    58.  // 创建对应的rescuer_thread()内核进程
    59.  /*
    60.   * Workqueues which may be used during memory reclaim should
    61.   * have a rescuer to guarantee forward progress.
    62.   */
    63.  if (flags & WQ_MEM_RECLAIM) {
    64.   struct worker *rescuer;
    65.  
    66.   rescuer = alloc_worker(NUMA_NO_NODE);
    67.   if (!rescuer)
    68.    goto err_destroy;
    69.  
    70.   rescuer->rescue_wq = wq;
    71.   rescuer->task = kthread_create(rescuer_thread, rescuer, "%s",
    72.             wq->name);
    73.   if (IS_ERR(rescuer->task)) {
    74.    kfree(rescuer);
    75.    goto err_destroy;
    76.   }
    77.  
    78.   wq->rescuer = rescuer;
    79.   rescuer->task->flags |= PF_NO_SETAFFINITY;
    80.   wake_up_process(rescuer->task);
    81.  }
    82.  
    83.  // (5) 如果是需要，创建workqueue对应的sysfs文件
    84.  if ((wq->flags & WQ_SYSFS) && workqueue_sysfs_register(wq))
    85.   goto err_destroy;
    86.  
    87.  /*
    88.   * wq_pool_mutex protects global freeze state and workqueues list.
    89.   * Grab it, adjust max_active and add the new @wq to workqueues
    90.   * list.
    91.   */
    92.  mutex_lock(&wq_pool_mutex);
    93.  
    94.  mutex_lock(&wq->mutex);
    95.  for_each_pwq(pwq, wq)
    96.   pwq_adjust_max_active(pwq);
    97.  mutex_unlock(&wq->mutex);
    98.  
    99.  // (6) 将新的workqueue加入到全局链表workqueues中
    100.  list_add(&wq->list, &workqueues);
    101.  
    102.  mutex_unlock(&wq_pool_mutex);
    103.  
    104.  return wq;
    105.  
    106. err_free_wq:
    107.  free_workqueue_attrs(wq->unbound_attrs);
    108.  kfree(wq);
    109.  return NULL;
    110. err_destroy:
    111.  destroy_workqueue(wq);
    112.  return NULL;
    113. }
    114. | →
    115. static int alloc_and_link_pwqs(struct workqueue_struct *wq)
    116. {
    117.  bool highpri = wq->flags & WQ_HIGHPRI;
    118.  int cpu, ret;
    119.  
    120.  // (3.1) normal workqueue
    121.  // pool_workqueue链接workqueue和worker_pool的过程
    122.  if (!(wq->flags & WQ_UNBOUND)) {
    123.   // 给workqueue的每个cpu分配对应的pool_workqueue，赋值给wq->cpu_pwqs
    124.   wq->cpu_pwqs = alloc_percpu(struct pool_workqueue);
    125.   if (!wq->cpu_pwqs)
    126.    return -ENOMEM;
    127.  
    128.   for_each_possible_cpu(cpu) {
    129.    struct pool_workqueue *pwq =
    130.     per_cpu_ptr(wq->cpu_pwqs, cpu);
    131.    struct worker_pool *cpu_pools =
    132.     per_cpu(cpu_worker_pools, cpu);
    133.  
    134.    // 将初始化时已经创建好的normal worker_pool，赋值给pool_workqueue
    135.    init_pwq(pwq, wq, &cpu_pools[highpri]);
    136.  
    137.    mutex_lock(&wq->mutex);
    138.    // 将pool_workqueue和workqueue链接起来
    139.    link_pwq(pwq);
    140.    mutex_unlock(&wq->mutex);
    141.   }
    142.   return 0;
    143.  } else if (wq->flags & __WQ_ORDERED) {
    144.  // (3.2) unbound ordered_wq workqueue
    145.  // pool_workqueue链接workqueue和worker_pool的过程
    146.   ret = apply_workqueue_attrs(wq, ordered_wq_attrs[highpri]);
    147.   /* there should only be single pwq for ordering guarantee */
    148.   WARN(!ret && (wq->pwqs.next != &wq->dfl_pwq->pwqs_node ||
    149.          wq->pwqs.prev != &wq->dfl_pwq->pwqs_node),
    150.        "ordering guarantee broken for workqueue %s\n", wq->name);
    151.   return ret;
    152.  } else {
    153.  // (3.3) unbound unbound_std_wq workqueue
    154.  // pool_workqueue链接workqueue和worker_pool的过程
    155.   return apply_workqueue_attrs(wq, unbound_std_wq_attrs[highpri]);
    156.  }
    157. }
    158. || →
    159. int apply_workqueue_attrs(struct workqueue_struct *wq,
    160.      const struct workqueue_attrs *attrs)
    161. {
    162.  
    163.  // (3.2.1) 根据的ubound的ordered_wq_attrs/unbound_std_wq_attrs
    164.  // 创建对应的pool_workqueue和worker_pool
    165.  // 其中worker_pool不是默认创建好的，是需要动态创建的，对应的worker内核进程也要重新创建
    166.  // 创建好的pool_workqueue赋值给pwq_tbl[node]
    167.  /*
    168.   * If something goes wrong during CPU up/down, we'll fall back to
    169.   * the default pwq covering whole @att- kernel/workqueue.c:  
    170. - __alloc_workqueue_key() -> alloc_and_link_pwqs() -> apply_workqueue_attrs() -> alloc_unbound_pwq()/numa_pwq_tbl_install()rs->cpumask.  Always create
    171.   * it even if we don't use it immediately.
    172.   */
    173.  dfl_pwq = alloc_unbound_pwq(wq, new_attrs);
    174.  if (!dfl_pwq)
    175.   goto enomem_pwq;
    176.  
    177.  for_each_node(node) {
    178.   if (wq_calc_node_cpumask(attrs, node, -1, tmp_attrs->cpumask)) {
    179.    pwq_tbl[node] = alloc_unbound_pwq(wq, tmp_attrs);
    180.    if (!pwq_tbl[node])
    181.     goto enomem_pwq;
    182.   } else {
    183.    dfl_pwq->refcnt++;
    184.    pwq_tbl[node] = dfl_pwq;
    185.   }
    186.  }
    187.  
    188.  
    189.  /* save the previous pwq and install the new one */
    190.  // (3.2.2) 将临时pwq_tbl[node]赋值给wq->numa_pwq_tbl[node]
    191.  for_each_node(node)
    192.   pwq_tbl[node] = numa_pwq_tbl_install(wq, node, pwq_tbl[node]);
    193.  
    194. }
    195. ||| →
    196. static struct pool_workqueue *alloc_unbound_pwq(struct workqueue_struct *wq,
    197.      const struct workqueue_attrs *attrs)
    198. {
    199.  struct worker_pool *pool;
    200.  struct pool_workqueue *pwq;
    201.  
    202.  lockdep_assert_held(&wq_pool_mutex);
    203.  
    204.  // (3.2.1.1) 如果对应attrs已经创建多对应的unbound_pool，则使用已有的unbound_pool
    205.  // 否则根据attrs创建新的unbound_pool
    206.  pool = get_unbound_pool(attrs);
    207.  if (!pool)
    208.   return NULL;
    209.  
    210.  pwq = kmem_cache_alloc_node(pwq_cache, GFP_KERNEL, pool->node);
    211.  if (!pwq) {
    212.   put_unbound_pool(pool);
    213.   return NULL;
    214.  }
    215.  
    216.  init_pwq(pwq, wq, pool);
    217.  return pwq;
    218. }

    1.2 worker

    每个worker对应一个worker_thread()内核线程，一个worker_pool对应一个或者多个worker。多个worker从同一个链表中worker_pool->worklist获取work进行处理。

    所以这其中有几个重点：

    • worker怎么处理work；

    • worker_pool怎么动态管理worker的数量；

    1.2.1 worker处理work

    处理work的过程主要在worker_thread() -> process_one_work()中处理，我们具体看看代码的实现过程。

    • kernel/workqueue.c:

    • worker_thread() -> process_one_work()

    1. static int worker_thread(void *__worker)
    2. {
    3.  struct worker *worker = __worker;
    4.  struct worker_pool *pool = worker->pool;
    5.  
    6.  /* tell the scheduler that this is a workqueue worker */
    7.  worker->task->flags |= PF_WQ_WORKER;
    8. woke_up:
    9.  spin_lock_irq(&pool->lock);
    10.  
    11.  // (1) 是否die
    12.  /* am I supposed to die? */
    13.  if (unlikely(worker->flags & WORKER_DIE)) {
    14.   spin_unlock_irq(&pool->lock);
    15.   WARN_ON_ONCE(!list_empty(&worker->entry));
    16.   worker->task->flags &= ~PF_WQ_WORKER;
    17.  
    18.   set_task_comm(worker->task, "kworker/dying");
    19.   ida_simple_remove(&pool->worker_ida, worker->id);
    20.   worker_detach_from_pool(worker, pool);
    21.   kfree(worker);
    22.   return 0;
    23.  }
    24.  
    25.  // (2) 脱离idle状态
    26.  // 被唤醒之前worker都是idle状态
    27.  worker_leave_idle(worker);
    28. recheck:
    29.  
    30.  // (3) 如果需要本worker继续执行则继续，否则进入idle状态
    31.  // need more worker的条件： (pool->worklist != 0) && (pool->nr_running == 0)
    32.  // worklist上有work需要执行，并且现在没有处于running的work
    33.  /* no more worker necessary? */
    34.  if (!need_more_worker(pool))
    35.   goto sleep;
    36.  
    37.  // (4) 如果(pool->nr_idle == 0)，则启动创建更多的worker
    38.  // 说明idle队列中已经没有备用worker了，先创建 一些worker备用
    39.  /* do we need to manage? */
    40.  if (unlikely(!may_start_working(pool)) && manage_workers(worker))
    41.   goto recheck;
    42.  
    43.  /*
    44.   * ->scheduled list can only be filled while a worker is
    45.   * preparing to process a work or actually processing it.
    46.   * Make sure nobody diddled with it while I was sleeping.
    47.   */
    48.  WARN_ON_ONCE(!list_empty(&worker->scheduled));
    49.  
    50.  /*
    51.   * Finish PREP stage.  We're guaranteed to have at least one idle
    52.   * worker or that someone else has already assumed the manager
    53.   * role.  This is where @worker starts participating in concurrency
    54.   * management if applicable and concurrency management is restored
    55.   * after being rebound.  See rebind_workers() for details.
    56.   */
    57.  worker_clr_flags(worker, WORKER_PREP | WORKER_REBOUND);
    58.  
    59.  do {
    60.   // (5) 如果pool->worklist不为空，从其中取出一个work进行处理
    61.   struct work_struct *work =
    62.    list_first_entry(&pool->worklist,
    63.       struct work_struct, entry);
    64.  
    65.   if (likely(!(*work_data_bits(work) & WORK_STRUCT_LINKED))) {
    66.    /* optimization path, not strictly necessary */
    67.    // (6) 执行正常的work
    68.    process_one_work(worker, work);
    69.    if (unlikely(!list_empty(&worker->scheduled)))
    70.     process_scheduled_works(worker);
    71.   } else {
    72.    // (7) 执行系统特意scheduled给某个worker的work
    73.    // 普通的work是放在池子的公共list中的pool->worklist
    74.    // 只有一些特殊的work被特意派送给某个worker的worker->scheduled
    75.    // 包括：1、执行flush_work时插入的barrier work；
    76.    // 2、collision时从其他worker推送到本worker的work
    77.    move_linked_works(work, &worker->scheduled, NULL);
    78.    process_scheduled_works(worker);
    79.   }
    80.  // (8) worker keep_working的条件：
    81.  // pool->worklist不为空 && (pool->nr_running <= 1)
    82.  } while (keep_working(pool));
    83.  
    84.  worker_set_flags(worker, WORKER_PREP);supposed
    85. sleep:
    86.  // (9) worker进入idle状态
    87.  /*
    88.   * pool->lock is held and there's no work to process and no need to
    89.   * manage, sleep.  Workers are woken up only while holding
    90.   * pool->lock or from local cpu, so setting the current state
    91.   * before releasing pool->lock is enough to prevent losing any
    92.   * event.
    93.   */
    94.  worker_enter_idle(worker);
    95.  __set_current_state(TASK_INTERRUPTIBLE);
    96.  spin_unlock_irq(&pool->lock);
    97.  schedule();
    98.  goto woke_up;
    99. }
    100. | →
    101. static void process_one_work(struct worker *worker, struct work_struct *work)
    102. __releases(&pool->lock)
    103. __acquires(&pool->lock)
    104. {
    105.  struct pool_workqueue *pwq = get_work_pwq(work);
    106.  struct worker_pool *pool = worker->pool;
    107.  bool cpu_intensive = pwq->wq->flags & WQ_CPU_INTENSIVE;
    108.  int work_color;
    109.  struct worker *collision;
    110. #ifdef CONFIG_LOCKDEP
    111.  /*
    112.   * It is permissible to free the struct work_struct from
    113.   * inside the function that is called from it, this we need to
    114.   * take into account for lockdep too.  To avoid bogus "held
    115.   * lock freed" warnings as well as problems when looking into
    116.   * work->lockdep_map, make a copy and use that here.
    117.   */
    118.  struct lockdep_map lockdep_map;
    119.  
    120.  lockdep_copy_map(&lockdep_map, &work->lockdep_map);
    121. #endif
    122.  /* ensure we're on the correct CPU */
    123.  WARN_ON_ONCE(!(pool->flags & POOL_DISASSOCIATED) &&
    124.        raw_smp_processor_id() != pool->cpu);
    125.  
    126.  // (8.1) 如果work已经在worker_pool的其他worker上执行，
    127.  // 将work放入对应worker的scheduled队列中延后执行
    128.  /*
    129.   * A single work shouldn't be executed concurrently by
    130.   * multiple workers on a single cpu.  Check whether anyone is
    131.   * already processing the work.  If so, defer the work to the
    132.   * currently executing one.
    133.   */
    134.  collision = find_worker_executing_work(pool, work);
    135.  if (unlikely(collision)) {
    136.   move_linked_works(work, &collision->scheduled, NULL);
    137.   return;
    138.  }
    139.  
    140.  // (8.2) 将worker加入busy队列pool->busy_hash
    141.  /* claim and dequeue */
    142.  debug_work_deactivate(work);
    143.  hash_add(pool->busy_hash, &worker->hentry, (unsigned long)work);
    144.  worker->current_work = work;
    145.  worker->current_func = work->func;
    146.  worker->current_pwq = pwq;
    147.  work_color = get_work_color(work);
    148.  
    149.  list_del_init(&work->entry);
    150.  
    151.  // (8.3) 如果work所在的wq是cpu密集型的WQ_CPU_INTENSIVE
    152.  // 则当前work的执行脱离worker_pool的动态调度，成为一个独立的线程
    153.  /*
    154.   * CPU intensive works don't participate in concurrency management.
    155.   * They're the scheduler's responsibility.  This takes @worker out
    156.   * of concurrency management and the next code block will chain
    157.   * execution of the pending work items.
    158.   */
    159.  if (unlikely(cpu_intensive))
    160.   worker_set_flags(worker, WORKER_CPU_INTENSIVE);
    161.  
    162.  // (8.4) 在UNBOUND或者CPU_INTENSIVE work中判断是否需要唤醒idle worker
    163.  // 普通work不会执行这个操作
    164.  /*
    165.   * Wake up another worker if necessary.  The condition is always
    166.   * false for normal per-cpu workers since nr_running would always
    167.   * be >= 1 at this point.  This is used to chain execution of the
    168.   * pending work items for WORKER_NOT_RUNNING workers such as the
    169.   * UNBOUND and CPU_INTENSIVE ones.
    170.   */
    171.  if (need_more_worker(pool))
    172.   wake_up_worker(pool);
    173.  
    174.  /*
    175.   * Record the last pool and clear PENDING which should be the last
    176.   * update to @work.  Also, do this inside @pool->lock so that
    177.   * PENDING and queued state changes happen together while IRQ is
    178.   * disabled.
    179.   */
    180.  set_work_pool_and_clear_pending(work, pool->id);
    181.  
    182.  spin_unlock_irq(&pool->lock);
    183.  
    184.  lock_map_acquire_read(&pwq->wq->lockdep_map);
    185.  lock_map_acquire(&lockdep_map);
    186.  trace_workqueue_execute_start(work);
    187.  // (8.5) 执行work函数
    188.  worker->current_func(work);
    189.  /*
    190.   * While we must be careful to not use "work" after this, the trace
    191.   * point will only record its address.
    192.   */
    193.  trace_workqueue_execute_end(work);
    194.  lock_map_release(&lockdep_map);
    195.  lock_map_release(&pwq->wq->lockdep_map);
    196.  
    197.  if (unlikely(in_atomic() || lockdep_depth(current) > 0)) {
    198.   pr_err("BUG: workqueue leaked lock or atomic: %s/0x%08x/%d\n"
    199.          "     last function: %pf\n",
    200.          current->comm, preempt_count(), task_pid_nr(current),
    201.          worker->current_func);
    202.   debug_show_held_locks(current);
    203.   dump_stack();
    204.  }
    205.  
    206.  /*
    207.   * The following prevents a kworker from hogging CPU on !PREEMPT
    208.   * kernels, where a requeueing work item waiting for something to
    209.   * happen could deadlock with stop_machine as such work item could
    210.   * indefinitely requeue itself while all other CPUs are trapped in
    211.   * stop_machine. At the same time, report a quiescent RCU state so
    212.   * the same condition doesn't freeze RCU.
    213.   */
    214.  cond_resched_rcu_qs();
    215.  
    216.  spin_lock_irq(&pool->lock);
    217.  
    218.  /* clear cpu intensive status */
    219.  if (unlikely(cpu_intensive))
    220.   worker_clr_flags(worker, WORKER_CPU_INTENSIVE);
    221.  
    222.  /* we're done with it, release */
    223.  hash_del(&worker->hentry);
    224.  worker->current_work = NULL;
    225.  worker->current_func = NULL;
    226.  worker->current_pwq = NULL;
    227.  worker->desc_valid = false;
    228.  pwq_dec_nr_in_flight(pwq, work_color);
    229. }

    1.2.2 worker_pool动态管理worker

    worker_pool怎么来动态增减worker，这部分的算法是CMWQ的核心。其思想如下：

    • worker_pool中的worker有3种状态：idle、running、suspend；

    • 如果worker_pool中有work需要处理，保持至少一个runn- kernel/workqueue.c:

    • worker_thread() -> process_one_work() ing worker来处理；

    • running worker在处理work的过程中进入了阻塞suspend状态，为了保持其他work的执行，需要唤醒新的idle worker来处理work；

    • 如果有work需要执行且running worker大于1个，会让多余的running worker进入idle状态；

    • 如果没有work需要执行，会让所有worker进入idle状态；

    • 如果创建的worker过多，destroy_worker在300s(IDLE_WORKER_TIMEOUT)时间内没有再次运行的idle worker。

    26531075eb59f3cda2ef8778cb33eb2d.png

    详细代码可以参考上节worker_thread() -> process_one_work()的分析。

    为了追踪worker的running和suspend状态，用来动态调整worker的数量。wq使用在进程调度中加钩子函数的技巧：

    • 追踪worker从suspend进入running状态：ttwu_activate() -> wq_worker_waking_up()

    1. void wq_worker_waking_up(struct task_struct *task, int cpu)
    2. {
    3.  struct worker *worker = kthread_data(task);
    4.  
    5.  if (!(worker->flags & WORKER_NOT_RUNNING)) {
    6.   WARN_ON_ONCE(worker->pool->cpu != cpu);
    7.   // 增加worker_pool中running的worker数量
    8.   atomic_inc(&worker->pool->nr_running);
    9.  }
    10. }

    • 追踪worker从running进入suspend状态：__schedule() -> wq_worker_sleeping()

    1. struct task_struct *wq_worker_sleeping(struct task_struct *task, int cpu)
    2. {
    3.  struct worker *worker = kthread_data(task), *to_wakeup = NULL;
    4.  struct worker_pool *pool;
    5.  
    6.  /*
    7.   * Rescuers, which may not have all the fields set up like normal
    8.   * workers, also reach here, let's not access anything before
    9.   * checking NOT_RUNNING.
    10.   */
    11.  if (worker->flags & WORKER_NOT_RUNNING)
    12.   return NULL;
    13.  
    14.  pool = worker->pool;
    15.  
    16.  /* this can only happen on the local cpu */
    17.  if (WARN_ON_ONCE(cpu != raw_smp_processor_id() || pool->cpu != cpu))
    18.   return NULL;
    19.  
    20.  /*
    21.   * The counterpart of the following dec_and_test, implied mb,
    22.   * worklist not empty test sequence is in insert_work().
    23.   * Please read comment there.
    24.   *
    25.   * NOT_RUNNING is clear.  This means that we're bound to and
    26.   * running on the local cpu w/ rq lock held and preemption
    27.   * disabled, which in turn means that none else could be
    28.   * manipulating idle_list, so dereferencing idle_list without pool
    29.   * lock is safe.
    30.   */
    31.  // 减少worker_pool中running的worker数量
    32.  // 如果worklist还有work需要处理，唤醒第一个idle worker进行处理
    33.  if (atomic_dec_and_test(&pool->nr_running) &&
    34.      !list_empty(&pool->worklist))
    35.   to_wakeup = first_idle_worker(pool);
    36.  return to_wakeup ? to_wakeup->task : NULL;
    37. }

    这里worker_pool的调度思想是：如果有work需要处理，保持一个running状态的worker处理，不多也不少。

    但是这里有一个问题如果work是cpu密集型的，它虽然也没有进入suspend状态，但是会长时间的占用cpu，让后续的work阻塞太长时间。

    为了解决这个问题，CMWQ设计了WQ_CPU_INTENSIVE，如果一个wq声明自己是CPU_INTENSIVE，则让当前worker脱离动态调度，像是进入了suspend状态，那么CMWQ会创建新的worker，后续的work会得到执行。

    • kernel/workqueue.c:

    • worker_thread() -> process_one_work()

    1. static void process_one_work(struct worker *worker, struct work_struct *work)
    2. __releases(&pool->lock)
    3. __acquires(&pool->lock)
    4. {
    5.  
    6.  bool cpu_intensive = pwq->wq->flags & WQ_CPU_INTENSIVE;
    7.  
    8.  
    9.  // (1) 设置当前worker的WORKER_CPU_INTENSIVE标志
    10.  // nr_running会被减1
    11.  // 对worker_pool来说，当前worker相当于进入了suspend状态
    12.  /*
    13.   * CPU intensive works don't participate in concurrency management.
    14.   * They're the scheduler's responsibility.  This takes @worker out
    15.   * of concurrency management and the next code block will chain
    16.   * execution of the pending work items.
    17.   */
    18.  if (unlikely(cpu_intensive))
    19.   worker_set_flags(worker, WORKER_CPU_INTENSIVE);
    20.  
    21.  // (2) 接上一步，判断是否需要唤醒新的worker来处理work
    22.  /*
    23.   * Wake up another worker if necessary.  The condition is always
    24.   * false for normal per-cpu workers since nr_running would always
    25.   * be >= 1 at this point.  This is used to chain execution of the
    26.   * pending work items for WORKER_NOT_RUNNING workers such as the
    27.   * UNBOUND and CPU_INTENSIVE ones.
    28.   */
    29.  if (need_more_worker(pool))
    30.   wake_up_worker(pool);
    31.  
    32.  // (3) 执行work
    33.  worker->current_func(work);
    34.  
    35.  
    36.  // (4) 执行完，清理当前worker的WORKER_CPU_INTENSIVE标志
    37.  // 当前worker重新进入running状态
    38.  /* clear cpu intensive status */
    39.  if (unlikely(cpu_intensive))
    40.   worker_clr_flags(worker, WORKER_CPU_INTENSIVE);
    41.  
    42.  
    43. }
    44.  
    45.  
    46.  WORKER_NOT_RUNNING = WORKER_PREP | WORKER_CPU_INTENSIVE |
    47.       WORKER_UNBOUND | WORKER_REBOUND,
    48.  
    49.  
    50. static inline void worker_set_flags(struct worker *worker, unsigned int flags)
    51. {
    52.  struct worker_pool *pool = worker->pool;
    53.  
    54.  WARN_ON_ONCE(worker->task != current);
    55.  
    56.  /* If transitioning into NOT_RUNNING, adjust nr_running. */
    57.  if ((flags & WORKER_NOT_RUNNING) &&
    58.      !(worker->flags & WORKER_NOT_RUNNING)) {
    59.   atomic_dec(&pool->nr_running);
    60.  }
    61.  
    62.  worker->flags |= flags;
    63. }
    64.  
    65.  
    66. static inline void worker_clr_flags(struct worker *worker, unsigned int flags)
    67. {
    68.  struct worker_pool *pool = worker->pool;
    69.  unsigned int oflags = worker->flags;
    70.  
    71.  WARN_ON_ONCE(worker->task != current);
    72.  
    73.  worker->flags &= ~flags;
    74.  
    75.  /*
    76.   * If transitioning out of NOT_RUNNING, increment nr_running.  Note
    77.   * that the nested NOT_RUNNING is not a noop.  NOT_RUNNING is mask
    78.   * of multiple flags, not a single flag.
    79.   */
    80.  if ((flags & WORKER_NOT_RUNNING) && (oflags & WORKER_NOT_RUNNING))
    81.   if (!(worker->flags & WORKER_NOT_RUNNING))
    82.    atomic_inc(&pool->nr_running);
    83. }

    1.2.3 cpu hotplug处理

    从上几节可以看到，系统会创建和cpu绑定的normal worker_pool和不绑定cpu的unbound worker_pool，worker_pool又会动态的创建worker。

    那么在cpu hotplug的时候，会怎么样动态的处理worker_pool和worker呢？来看具体的代码分析：

    • kernel/workqueue.c:

    • workqueue_cpu_up_callback()/workqueue_cpu_down_callback()

    1. static int __init init_workqueues(void)
    2. {
    3.  
    4.  cpu_notifier(workqueue_cpu_up_callback, CPU_PRI_WORKQUEUE_UP);
    5.  hotcpu_notifier(workqueue_cpu_down_callback, CPU_PRI_WORKQUEUE_DOWN);
    6.  
    7. }
    8. | →
    9. static int workqueue_cpu_down_callback(struct notifier_block *nfb,
    10.        unsigned long action,
    11.        void *hcpu)
    12. {
    13.  int cpu = (unsigned long)hcpu;
    14.  struct work_struct unbind_work;
    15.  struct workqueue_struct *wq;
    16.  
    17.  switch (action & ~CPU_TASKS_FROZEN) {
    18.  case CPU_DOWN_PREPARE:
    19.   /* unbinding per-cpu workers should happen on the local CPU */
    20.   INIT_WORK_ONSTACK(&unbind_work, wq_unbind_fn);
    21.   // (1) cpu down_prepare
    22.   // 把和当前cpu绑定的normal worker_pool上的worker停工
    23.   // 随着当前cpu被down掉，这些worker会迁移到其他cpu上
    24.   queue_work_on(cpu, system_highpri_wq, &unbind_work);
    25.  
    26.   // (2) unbound wq对cpu变化的更新
    27.   /* update NUMA affinity of unbound workqueues */
    28.   mutex_lock(&wq_pool_mutex);
    29.   list_for_each_entry(wq, &workqueues, list)
    30.    wq_update_unbound_numa(wq, cpu, false);
    31.   mutex_unlock(&wq_pool_mutex);
    32.  
    33.   /* wait for per-cpu unbinding to finish */
    34.   flush_work(&unbind_work);
    35.   destroy_work_on_stack(&unbind_work);
    36.   break;
    37.  }
    38.  return NOTIFY_OK;
    39. }
    40. | →
    41. static int workqueue_cpu_up_callback(struct notifier_block *nfb,
    42.             unsigned long action,
    43.             void *hcpu)
    44. {
    45.  int cpu = (unsigned long)hcpu;
    46.  struct worker_pool *pool;
    47.  struct workqueue_struct *wq;
    48.  int pi;
    49.  
    50.  switch (action & ~CPU_TASKS_FROZEN) {
    51.  case CPU_UP_PREPARE:
    52.   for_each_cpu_worker_pool(pool, cpu) {
    53.    if (pool->nr_workers)
    54.     continue;
    55.    if (!create_worker(pool))
    56.     return NOTIFY_BAD;
    57.   }
    58.   break;
    59.  
    60.  case CPU_DOWN_FAILED:
    61.  case CPU_ONLINE:
    62.   mutex_lock(&wq_pool_mutex);
    63.   
    64.   // (3) cpu up
    65.   for_each_pool(pool, pi) {
    66.    mutex_lock(&pool->attach_mutex);
    67.  
    68.    // 如果和当前cpu绑定的normal worker_pool上，有WORKER_UNBOUND停工的worker
    69.    // 重新绑定worker到worker_pool
    70.    // 让这些worker开工，并绑定到当前cpu
    71.    if (pool->cpu == cpu)
    72.     rebind_workers(pool);
    73.    else if (pool->cpu < 0)
    74.     restore_unbound_workers_cpumask(pool, cpu);
    75.  
    76.    mutex_unlock(&pool->attach_mutex);
    77.   }
    78.  
    79.   /* update NUMA affinity of unbound workqueues */
    80.   list_for_each_entry(wq, &workqueues, list)
    81.    wq_update_unbound_numa(wq, cpu, true);
    82.  
    83.   mutex_unlock(&wq_pool_mutex);
    84.   break;
    85.  }
    86.  return NOTIFY_OK;
    87. }

    1.3 workqueue

    workqueue就是存放一组work的集合，基本可以分为两类：一类系统创建的workqueue，一类是用户自己创建的workqueue。

    不论是系统还是用户workqueue，如果没有指定WQ_UNBOUND，默认都是和normal worker_pool绑定。

    1.3.1 系统workqueue

    系统在初始化时创建了一批默认的workqueue：system_wq、system_highpri_wq、system_long_wq、system_unbound_wq、system_freezable_wq、system_power_efficient_wq、system_freezable_power_efficient_wq。

    像system_wq，就是schedule_work()默认使用的。

    • kernel/workqueue.c:

    • init_workqueues()

    1. static int __init init_workqueues(void)
    2. {
    3.  
    4.  system_wq = alloc_workqueue("events", 0, 0);
    5.  system_highpri_wq = alloc_workqueue("events_highpri", WQ_HIGHPRI, 0);
    6.  system_long_wq = alloc_workqueue("events_long", 0, 0);
    7.  system_unbound_wq = alloc_workqueue("events_unbound", WQ_UNBOUND,
    8.          WQ_UNBOUND_MAX_ACTIVE);
    9.  system_freezable_wq = alloc_workqueue("events_freezable",
    10.            WQ_FREEZABLE, 0);
    11.  system_power_efficient_wq = alloc_workqueue("events_power_efficient",
    12.            WQ_POWER_EFFICIENT, 0);
    13.  system_freezable_power_efficient_wq = alloc_workqueue("events_freezable_power_efficient",
    14.            WQ_FREEZABLE | WQ_POWER_EFFICIENT,
    15.            0);
    16.  
    17. }

    1.3.2 workqueue创建

    详细过程见上几节的代码分析：alloc_workqueue() -> __alloc_workqueue_key() -> alloc_and_link_pwqs()。

    1.3.3 flush_workqueue()

    这一部分的逻辑，wq->work_color、wq->flush_color换来换去的逻辑实在看的头晕。看不懂暂时不想看，放着以后看吧，或者有谁看懂了教我一下。：）

    1.4 pool_workqueue

    pool_workqueue只是一个中介角色。

    详细过程见上几节的代码分析：alloc_workqueue() -> __alloc_workqueue_key() -> alloc_and_link_pwqs()。

    1.5 work

    描述一份待执行的工作。

    1.5.1 queue_work()

    将work压入到workqueue当中。

    • kernel/workqueue.c:

    • queue_work() -> queue_work_on() -> __queue_work()

    1. static void __queue_work(int cpu, struct workqueue_struct *wq,
    2.     struct work_struct *work)
    3. {
    4.  struct pool_workqueue *pwq;
    5.  struct worker_pool *last_pool;
    6.  struct list_head *worklist;
    7.  unsigned int work_flags;
    8.  unsigned int req_cpu = cpu;
    9.  
    10.  /*
    11.   * While a work item is PENDING && off queue, a task trying to
    12.   * steal the PENDING will busy-loop waiting for it to either get
    13.   * queued or lose PENDING.  Grabbing PENDING and queueing should
    14.   * happen with IRQ disabled.
    15.   */
    16.  WARN_ON_ONCE(!irqs_disabled());
    17.  
    18.  debug_work_activate(work);
    19.  
    20.  /* if draining, only works from the same workqueue are allowed */
    21.  if (unlikely(wq->flags & __WQ_DRAINING) &&
    22.      WARN_ON_ONCE(!is_chained_work(wq)))
    23.   return;
    24. retry:
    25.  // (1) 如果没有指定cpu，则使用当前cpu
    26.  if (req_cpu == WORK_CPU_UNBOUND)
    27.   cpu = raw_smp_processor_id();
    28.  
    29.  /* pwq which will be used unless @work is executing elsewhere */
    30.  if (!(wq->flags & WQ_UNBOUND))
    31.   // (2) 对于normal wq，使用当前cpu对应的normal worker_pool
    32.   pwq = per_cpu_ptr(wq->cpu_pwqs, cpu);
    33.  else
    34.   // (3) 对于unbound wq，使用当前cpu对应node的worker_pool
    35.   pwq = unbound_pwq_by_node(wq, cpu_to_node(cpu));
    36.  
    37.  // (4) 如果work在其他worker上正在被执行，把work压到对应的worker上去
    38.  // 避免work出现重入的问题
    39.  /*
    40.   * If @work was previously on a different pool, it might still be
    41.   * running there, in which case the work needs to be queued on that
    42.   * pool to guarantee non-reentrancy.
    43.   */
    44.  last_pool = get_work_pool(work);
    45.  if (last_pool && last_pool != pwq->pool) {
    46.   struct worker *worker;
    47.  
    48.   spin_lock(&last_pool->lock);
    49.  
    50.   worker = find_worker_executing_work(last_pool, work);
    51.  
    52.   if (worker && worker->current_pwq->wq == wq) {
    53.    pwq = worker->current_pwq;
    54.   } else {
    55.    /* meh... not running there, queue here */
    56.    spin_unlock(&last_pool->lock);
    57.    spin_lock(&pwq->pool->lock);
    58.   }
    59.  } else {
    60.   spin_lock(&pwq->pool->lock);
    61.  }
    62.  
    63.  /*
    64.   * pwq is determined and locked.  For unbound pools, we could have
    65.   * raced with pwq release and it could already be dead.  If its
    66.   * refcnt is zero, repeat pwq selection.  Note that pwqs never die
    67.   * without another pwq replacing it in the numa_pwq_tbl or while
    68.   * work items are executing on it, so the retrying is guaranteed to
    69.   * make forward-progress.
    70.   */
    71.  if (unlikely(!pwq->refcnt)) {
    72.   if (wq->flags & WQ_UNBOUND) {
    73.    spin_unlock(&pwq->pool->lock);
    74.    cpu_relax();
    75.    goto retry;
    76.   }
    77.   /* oops */
    78.   WARN_ONCE(true, "workqueue: per-cpu pwq for %s on cpu%d has 0 refcnt",
    79.      wq->name, cpu);
    80.  }
    81.  
    82.  /* pwq determined, queue */
    83.  trace_workqueue_queue_work(req_cpu, pwq, work);
    84.  
    85.  if (WARN_ON(!list_empty(&work->entry))) {
    86.   spin_unlock(&pwq->pool->lock);
    87.   return;
    88.  }
    89.  
    90.  pwq->nr_in_flight[pwq->work_color]++;
    91.  work_flags = work_color_to_flags(pwq->work_color);
    92.  
    93.  // (5) 如果还没有达到max_active，将work挂载到pool->worklist
    94.  if (likely(pwq->nr_active < pwq->max_active)) {
    95.   trace_workqueue_activate_work(work);
    96.   pwq->nr_active++;
    97.   worklist = &pwq->pool->worklist;
    98.  // 否则，将work挂载到临时队列pwq->delayed_works
    99.  } else {
    100.   work_flags |= WORK_STRUCT_DELAYED;
    101.   worklist = &pwq->delayed_works;
    102.  }
    103.  
    104.  // (6) 将work压入worklist当中
    105.  insert_work(pwq, work, worklist, work_flags);
    106.  
    107.  spin_unlock(&pwq->pool->lock);
    108. }

    1.5.2 flush_work()

    flush某个work，确保work执行完成。

    怎么判断异步的work已经执行完成？这里面使用了一个技巧：在目标work的后面插入一个新的work wq_barrier，如果wq_barrier执行完成，那么目标work肯定已经执行完成。

    • kernel/workqueue.c:

    • queue_work() -> queue_work_on() -> __queue_work()

    1. /**
    2.  * flush_work - wait for a work to finish executing the last queueing instance
    3.  * @work: the work to flush
    4.  *
    5.  * Wait until @work has finished execution.  @work is guaranteed to be idle
    6.  * on return if it hasn't been requeued since flush started.
    7.  *
    8.  * Return:
    9.  * %true if flush_work() waited for the work to finish execution,
    10.  * %false if it was already idle.
    11.  */
    12. bool flush_work(struct work_struct *work)
    13. {
    14.  struct wq_barrier barr;
    15.  
    16.  lock_map_acquire(&work->lockdep_map);
    17.  lock_map_release(&work->lockdep_map);
    18.  
    19.  if (start_flush_work(work, &barr)) {
    20.   // 等待barr work执行完成的信号
    21.   wait_for_completion(&barr.done);
    22.   destroy_work_on_stack(&barr.work);
    23.   return true;
    24.  } else {
    25.   return false;
    26.  }
    27. }
    28. | →
    29. static bool start_flush_work(struct work_struct *work, struct wq_barrier *barr)
    30. {
    31.  struct worker *worker = NULL;
    32.  struct worker_pool *pool;
    33.  struct pool_workqueue *pwq;
    34.  
    35.  might_sleep();
    36.  
    37.  // (1) 如果work所在worker_pool为NULL，说明work已经执行完
    38.  local_irq_disable();
    39.  pool = get_work_pool(work);
    40.  if (!pool) {
    41.   local_irq_enable();
    42.   return false;
    43.  }
    44.  
    45.  spin_lock(&pool->lock);
    46.  /* see the comment in try_to_grab_pending() with the same code */
    47.  pwq = get_work_pwq(work);
    48.  if (pwq) {
    49.   // (2) 如果work所在pwq指向的worker_pool不等于上一步得到的worker_pool，说明work已经执行完
    50.   if (unlikely(pwq->pool != pool))
    51.    goto already_gone;
    52.  } else {
    53.   // (3) 如果work所在pwq为NULL，并且也没有在当前执行的work中，说明work已经执行完
    54.   worker = find_worker_executing_work(pool, work);
    55.   if (!worker)
    56.    goto already_gone;
    57.   pwq = worker->current_pwq;
    58.  }
    59.  
    60.  // (4) 如果work没有执行完，向work的后面插入barr work
    61.  insert_wq_barrier(pwq, barr, work, worker);
    62.  spin_unlock_irq(&pool->lock);
    63.  
    64.  /*
    65.   * If @max_active is 1 or rescuer is in use, flushing another work
    66.   * item on the same workqueue may lead to deadlock.  Make sure the
    67.   * flusher is not running on the same workqueue by verifying write
    68.   * access.
    69.   */
    70.  if (pwq->wq->saved_max_active == 1 || pwq->wq->rescuer)
    71.   lock_map_acquire(&pwq->wq->lockdep_map);
    72.  else
    73.   lock_map_acquire_read(&pwq->wq->lockdep_map);
    74.  lock_map_release(&pwq->wq->lockdep_map);
    75.  
    76.  return true;
    77. already_gone:
    78.  spin_unlock_irq(&pool->lock);
    79.  return false;
    80. }
    81. || →
    82. static void insert_wq_barrier(struct pool_workqueue *pwq,
    83.          struct wq_barrier *barr,
    84.          struct work_struct *target, struct worker *worker)
    85. {
    86.  struct list_head *head;
    87.  unsigned int linked = 0;
    88.  
    89.  /*
    90.   * debugobject calls are safe here even with pool->lock locked
    91.   * as we know for sure that this will not trigger any of the
    92.   * checks and call back into the fixup functions where we
    93.   * might deadlock.
    94.   */
    95.  // (4.1) barr work的执行函数wq_barrier_func()
    96.  INIT_WORK_ONSTACK(&barr->work, wq_barrier_func);
    97.  __set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(&barr->work));
    98.  init_completion(&barr->done);
    99.  
    100.  /*
    101.   * If @target is currently being executed, schedule the
    102.   * barrier to the worker; otherwise, put it after @target.
    103.   */
    104.  // (4.2) 如果work当前在worker中执行，则barr work插入scheduled队列
    105.  if (worker)
    106.   head = worker->scheduled.next;
    107.  // 否则，则barr work插入正常的worklist队列中，插入位置在目标work后面
    108.  // 并且置上WORK_STRUCT_LINKED标志
    109.  else {
    110.   unsigned long *bits = work_data_bits(target);
    111.  
    112.   head = target->entry.next;
    113.   /* there can already be other linked works, inherit and set */
    114.   linked = *bits & WORK_STRUCT_LINKED;
    115.   __set_bit(WORK_STRUCT_LINKED_BIT, bits);
    116.  }
    117.  
    118.  debug_work_activate(&barr->work);
    119.  insert_work(pwq, &barr->work, head,
    120.       work_color_to_flags(WORK_NO_COLOR) | linked);
    121. }
    122. ||| →
    123. static void wq_barrier_func(struct work_struct *work)
    124. {
    125.  struct wq_barrier *barr = container_of(work, struct wq_barrier, work);
    126.  // (4.1.1) barr work执行完成，发出complete信号。
    127.  complete(&barr->done);
    128. }

    2.Workqueue对外接口函数

    CMWQ实现的workqueue机制，被包装成相应的对外接口函数。

    2.1 schedule_work()

    把work压入系统默认wq system_wq，WORK_CPU_UNBOUND指定worker为当前cpu绑定的normal worker_pool创建的worker。

    • kernel/workqueue.c:

    • schedule_work() -> queue_work_on() -> __queue_work()

    1. static inline bool schedule_work(struct work_struct *work)
    2. {
    3.  return queue_work(system_wq, work);
    4. }
    5. | →
    6. static inline bool queue_work(struct workqueue_struct *wq,
    7.          struct work_struct *work)
    8. {
    9.  return queue_work_on(WORK_CPU_UNBOUND, wq, work);
    10. }

    2.2 sschedule_work_on()

    在schedule_work()基础上，可以指定work运行的cpu。

    • kernel/workqueue.c:

    • schedule_work_on() -> queue_work_on() -> __queue_work()

    1. static inline bool schedule_work_on(int cpu, struct work_struct *work)
    2. {
    3.  return queue_work_on(cpu, system_wq, work);
    4. }

    2.3 schedule_delayed_work()

    启动一个timer，在timer定时到了以后调用delayed_work_timer_fn()把work压入系统默认wq system_wq。

    • kernel/workqueue.c:

    • schedule_work_on() -> queue_work_on() -> __queue_work()

    1. static inline bool schedule_delayed_work(struct delayed_work *dwork,
    2.       unsigned long delay)
    3. {
    4.  return queue_delayed_work(system_wq, dwork, delay);
    5. }
    6. | →
    7. static inline bool queue_delayed_work(struct workqueue_struct *wq,
    8.           struct delayed_work *dwork,
    9.           unsigned long delay)
    10. {
    11.  return queue_delayed_work_on(WORK_CPU_UNBOUND, wq, dwork, delay);
    12. }
    13. || →
    14. bool queue_delayed_work_on(int cpu, struct workqueue_struct *wq,
    15.       struct delayed_work *dwork, unsigned long delay)
    16. {
    17.  struct work_struct *work = &dwork->work;
    18.  bool ret = false;
    19.  unsigned long flags;
    20.  
    21.  /* read the comment in __queue_work() */
    22.  local_irq_save(flags);
    23.  
    24.  if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work))) {
    25.   __queue_delayed_work(cpu, wq, dwork, delay);
    26.   ret = true;
    27.  }
    28.  
    29.  local_irq_restore(flags);
    30.  return ret;
    31. }
    32. ||| →
    33. static void __queue_delayed_work(int cpu, struct workqueue_struct *wq,
    34.     struct delayed_work *dwork, unsigned long delay)
    35. {
    36.  struct timer_list *timer = &dwork->timer;
    37.  struct work_struct *work = &dwork->work;
    38.  
    39.  WARN_ON_ONCE(timer->function != delayed_work_timer_fn ||
    40.        timer->data != (unsigned long)dwork);
    41.  WARN_ON_ONCE(timer_pending(timer));
    42.  WARN_ON_ONCE(!list_empty(&work->entry));
    43.  
    44.  /*
    45.   * If @delay is 0, queue @dwork->work immediately.  This is for
    46.   * both optimization and correctness.  The earliest @timer can
    47.   * expire is on the closest next tick and delayed_work users depend
    48.   * on that there's no such delay when @delay is 0.
    49.   */
    50.  if (!delay) {
    51.   __queue_work(cpu, wq, &dwork->work);
    52.   return;
    53.  }
    54.  
    55.  timer_stats_timer_set_start_info(&dwork->timer);
    56.  
    57.  dwork->wq = wq;
    58.  dwork->cpu = cpu;
    59.  timer->expires = jiffies + delay;
    60.  
    61.  if (unlikely(cpu != WORK_CPU_UNBOUND))
    62.   add_timer_on(timer, cpu);
    63.  else
    64.   add_timer(timer);
    65. }
    66. |||| →
    67. void delayed_work_timer_fn(unsigned long __data)
    68. {
    69.  struct delayed_work *dwork = (struct delayed_work *)__data;
    70.  
    71.  /* should have been called from irqsafe timer with irq already off */
    72.  __queue_work(dwork->cpu, dwork->wq, &dwork->work);
    73. }

    参考资料

    1. Documentation/workqueue.txt

    b5fb81b9d619b6fcd1c00645f14077dc.jpeg

  • 相关阅读:
    python发送邮件和企业微信
    Linux中安装Redis
    Code Review：提升代码质量与团队能力的利器
    【技术分享】NetLogon于域内提权漏洞(CVE-2020-1472)
    IIS发布.net网站(配置Nginx以及HTTP和HTTPS)
    神器必会！特别好使的编辑器Source Insight
    初学Flutter：swiper实现
    2022深圳xxx校招Java笔试题目（选择题+简答题）
    web前端网页设计期末课程大作业：旅游网页主题网站设计——紫色的旅游开发景点网站静态模板(4页)HTML+CSS+JavaScript
    MyBatisPlus（二十）防全表更新与删除
  • 原文地址：https://blog.csdn.net/melody157398/article/details/126457865