workqueue

• How to queue a work
• Delay of a queued work

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

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

    /* ID is needed to determine kthread name */
    id = ida_simple_get(&pool->worker_ida, 0, 0, GFP_KERNEL);
    if (id < 0)
        goto fail;

    worker = alloc_worker(pool->node);
    if (!worker)
        goto fail;

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

    /*
      The pattern of naming a kworker
     */
    if (pool->cpu >= 0)
        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);
    if (IS_ERR(worker->task))
        goto fail;

    set_user_nice(worker->task, pool->attrs->nice);
    kthread_bind_mask(worker->task, pool->attrs->cpumask);

    /* successful, attach the worker to the pool */
    worker_attach_to_pool(worker, pool);

    /* start the newly created worker */
    spin_lock_irq(&pool->lock);
    worker->pool->nr_workers++;
    worker_enter_idle(worker);
    wake_up_process(worker->task);
    spin_unlock_irq(&pool->lock);

    return worker;

fail:
    if (id >= 0)
        ida_simple_remove(&pool->worker_ida, id);
    kfree(worker);
    return NULL;
}

            PENDING            -->       pwq | PENDING | PWQ | color
            queue_work_on()              insert_work()
            irq disabled                 irq disabled | pool->lock held
                                         on worklist

                        ^                |                     
                        |                v 
                        pool id               
                        process_one_work()
                        irq disabled | pool->lock held
                        off worklist
                
                        worker->current_func()


Note: the on/off worklist is the state after get out of pool->lock critical section

static int try_to_grab_pending(struct work_struct *work, bool is_dwork,
                   unsigned long *flags)
{
    struct worker_pool *pool;
    struct pool_workqueue *pwq;

    local_irq_save(*flags);

    /* try to steal the timer if it exists */
    >>>>
    /* try to claim PENDING the normal way */
    if (!test_and_set_bit(WORK_STRUCT_PENDING_BIT, work_data_bits(work)))
        return 0;

    /*
     * The queueing is in progress, or it is already queued. Try to
     * steal it from ->worklist without clearing WORK_STRUCT_PENDING.
     */
    pool = get_work_pool(work);
    if (!pool)
        goto fail;

    spin_lock(&pool->lock);
    /*
     * work->data is guaranteed to point to pwq only while the work
     * item is queued on pwq->wq, and both updating work->data to point
     * to pwq on queueing and to pool on dequeueing are done under
     * pwq->pool->lock.  This in turn guarantees that, if work->data
     * points to pwq which is associated with a locked pool, the work
     * item is currently queued on that pool.
     */
    pwq = get_work_pwq(work);
    if (pwq && pwq->pool == pool) {
        >>>>
        list_del_init(&work->entry);
        pwq_dec_nr_in_flight(pwq, get_work_color(work));

        /* work->data points to pwq iff queued, point to pool */
        set_work_pool_and_keep_pending(work, pool->id);

        spin_unlock(&pool->lock);
        return 1;
    }
    spin_unlock(&pool->lock);
fail:
    local_irq_restore(*flags);
    if (work_is_canceling(work))
        return -ENOENT;
    cpu_relax();
    return -EAGAIN;
}

 A bound pool is either associated or disassociated with its CPU. While associated 
 (!DISASSOCIATED), all workers are bound to the CPU and none has %WORKER_UNBOUND 
 set and concurrency management is in effect.
While DISASSOCIATED, the cpu may be offline and all workers have %WORKER_UNBOUND
set and concurrency management disabled, and may be executing on any CPU.  The 
pool behaves as an unbound one.

        mutex_lock(&pool->attach_mutex);
        spin_lock_irq(&pool->lock);

        for_each_pool_worker(worker, pool)
            worker->flags |= WORKER_UNBOUND;

        pool->flags |= POOL_DISASSOCIATED;

        spin_unlock_irq(&pool->lock);
        mutex_unlock(&pool->attach_mutex);

        /*
         * Call schedule() so that we cross rq->lock and thus can
         * guarantee sched callbacks see the %WORKER_UNBOUND flag.
         * This is necessary as scheduler callbacks may be invoked
         * from other cpus.
         */
        schedule(); If just synchronize with
        wq_worker_waking_up, synchronize_rcu_sched should be ok,
        wq_worker_waking_up is under preempt disable critical section.

        /*
         * Sched callbacks are disabled now.  Zap nr_running.
         * After this, nr_running stays zero and need_more_worker()
         * and keep_working() are always true as long as the
         * worklist is not empty.  This pool now behaves as an
         * unbound (in terms of concurrency management) pool which
         * are served by workers tied to the pool. 
         */
        atomic_set(&pool->nr_running, 0);

        /*
         * With concurrency management just turned off, a busy
         * worker blocking could lead to lengthy stalls.  Kick off
         * unbound chain execution of currently pending work items.
         */
        spin_lock_irq(&pool->lock);
        wake_up_worker(pool);
        spin_unlock_irq(&pool->lock);

    worker_clr_flags(worker, WORKER_PREP | WORKER_REBOUND);

The cpu has been set onlined, but not active, could it be migrated ?
Yes,  kernel threads are allowed on online && !active CPUs

static void process_one_work(struct worker *worker, struct work_struct *work)
{
>>>>
    /* claim and dequeue */
    debug_work_deactivate(work);
    hash_add(pool->busy_hash, &worker->hentry, (unsigned long)work);
    worker->current_work = work;
    worker->current_func = work->func;
    worker->current_pwq = pwq;
    work_color = get_work_color(work);

    list_del_init(&work->entry);
    >>>>
    /*
     * Record the last pool and clear PENDING which should be the last
     * update to @work.  Also, do this inside @pool->lock so that
     * PENDING and queued state changes happen together while IRQ is
     * disabled.
     */
    set_work_pool_and_clear_pending(work, pool->id);

    spin_unlock_irq(&pool->lock);
>>>>
}

    /*
     * If @work was previously on a different pool, it might still be
     * running there, in which case the work needs to be queued on that
     * pool to guarantee non-reentrancy.
     */
    last_pool = get_work_pool(work); //depends on the pool is set in process_one_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);
        }
    }

    /*
     * A single work shouldn't be executed concurrently by
     * multiple workers on a single cpu.  Check whether anyone is
     * already processing the work.  If so, defer the work to the
     * currently executing one.
     */
    collision = find_worker_executing_work(pool, work);
    if (unlikely(collision)) {
        move_linked_works(work, &collision->scheduled, NULL);
        return;
    }

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

        pool->watchdog_ts = jiffies;

        if (likely(!(*work_data_bits(work) & WORK_STRUCT_LINKED))) {
            /* optimization path, not strictly necessary */
            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));
Note: pool->lock will ensure the safe. 

/* Setup a clean context for our children to inherit. */
    set_task_comm(tsk, "kthreadd");
    ignore_signals(tsk);
    set_cpus_allowed_ptr(tsk, cpu_all_mask);
    set_mems_allowed(node_states[N_MEMORY]);

    current->flags |= PF_NOFREEZE;
    cgroup_init_kthreadd();

void freeze_workqueues_begin(void)
{
    struct workqueue_struct *wq;
    struct pool_workqueue *pwq;

    mutex_lock(&wq_pool_mutex);

    WARN_ON_ONCE(workqueue_freezing);
    workqueue_freezing = true;

    list_for_each_entry(wq, &workqueues, list) {
        mutex_lock(&wq->mutex);
        for_each_pwq(pwq, wq)
            pwq_adjust_max_active(pwq);
        mutex_unlock(&wq->mutex);
    }

    mutex_unlock(&wq_pool_mutex);
}
pwq_adjust_max_active() will clear pwq->max_active to zero when it is freezing. 

__queue_work()
...
    if (likely(pwq->nr_active < pwq->max_active)) {
        trace_workqueue_activate_work(work);
        pwq->nr_active++;
        worklist = &pwq->pool->worklist;
        if (list_empty(worklist))
            pwq->pool->watchdog_ts = jiffies;
    } else {
        work_flags |= WORK_STRUCT_DELAYED;
        worklist = &pwq->delayed_works;
    }
....
Note: this delayed_works list is not related to deleyed work, __queue_delayed_work just
add a timer which will invoke __queue_work when expires.

process_one_work()
    -> pwq_dec_nr_in_flight()
    >>>> //under pool->lock
    pwq->nr_active--;
    if (!list_empty(&pwq->delayed_works)) {
        /* one down, submit a delayed one */
        if (pwq->nr_active < pwq->max_active)
            pwq_activate_first_delayed(pwq);
    }
    >>>>
        -> pwq_activate_delayed_work()

static void pwq_activate_delayed_work(struct work_struct *work)
{
    struct pool_workqueue *pwq = get_work_pwq(work);

    trace_workqueue_activate_work(work);
    if (list_empty(&pwq->pool->worklist))
        pwq->pool->watchdog_ts = jiffies;
    move_linked_works(work, &pwq->pool->worklist, NULL);
    __clear_bit(WORK_STRUCT_DELAYED_BIT, work_data_bits(work));
    pwq->nr_active++;
}

    bdi_wq = alloc_workqueue("writeback", WQ_MEM_RECLAIM | WQ_FREEZABLE |
                          WQ_UNBOUND | WQ_SYSFS, 0);

queue_delayed_work(wq, dw, 10/*ms*/)
/*5ms later*/
queue_delayed_work(wq, dw, 0/*ms*/)

bool mod_delayed_work_on(int cpu, struct workqueue_struct *wq,
             struct delayed_work *dwork, unsigned long delay)
{
    unsigned long flags;
    int ret;

    do {
        ret = try_to_grab_pending(&dwork->work, true, &flags);
    } while (unlikely(ret == -EAGAIN));

    if (likely(ret >= 0)) {
        __queue_delayed_work(cpu, wq, dwork, delay);
        local_irq_restore(flags);
    }

    /* -ENOENT from try_to_grab_pending() becomes %true */
    return ret;
}

commit ae943d20624de0a6aac7dd0597616dce2c498029
Author: Bart Van Assche 
Date:   Fri Jan 19 08:58:55 2018 -0800

    blk-mq: Avoid that blk_mq_delay_run_hw_queue() introduces unintended delays
    
    Make sure that calling blk_mq_run_hw_queue() or
    blk_mq_kick_requeue_list() triggers a queue run without delay even
    if blk_mq_delay_run_hw_queue() has been called recently and if its
    delay has not yet expired.
    
    Reviewed-by: Mike Snitzer 
    Signed-off-by: Bart Van Assche 
    Signed-off-by: Jens Axboe 

worker_thread
  -> process_one_work
    -> pwq_dec_nr_in_flight //under pool->lock
    ---
        pwq->nr_active--;
        if (!list_empty(&pwq->delayed_works)) {
            /* one down, submit a delayed one */
            if (pwq->nr_active < pwq->max_active)
                pwq_activate_first_delayed(pwq);
        }
    ---
      -> pwq_activate_delayed_work

worker_thread
---

    // under pool->lock

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

        pool->watchdog_ts = jiffies;

        if (likely(!(*work_data_bits(work) & WORK_STRUCT_LINKED))) {
            /* optimization path, not strictly necessary */
            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));
---
   
static bool keep_working(struct worker_pool *pool)
{
    return !list_empty(&pool->worklist) &&

        atomic_read(&pool->nr_running) <= 1;

}

worker_thread
---

    // under spinlock worker_pool->lock

    do {

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

        pool->watchdog_ts = jiffies;

        if (likely(!(*work_data_bits(work) & WORK_STRUCT_LINKED))) {
            /* optimization path, not strictly necessary */
            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));
---

• When to create

  worker_thread
  ---
      if (unlikely(!may_start_working(pool)) && manage_workers(worker))
  ---
  
  static bool may_start_working(struct worker_pool *pool)
  {
      return pool->nr_idle;
  }
  
  maybe_create_worker guarantees that there is at least one idle kworker when it
  returns.
  
  This idle kworker could be wakedup when one working one falls into sleep in
  __schedule
  ---
  
              /*
               * If a worker went to sleep, notify and ask workqueue
               * whether it wants to wake up a task to maintain
               * concurrency.
               */
  
              if (prev->flags & PF_WQ_WORKER) {
                  struct task_struct *to_wakeup;
  
                  to_wakeup = wq_worker_sleeping(prev);
                  if (to_wakeup)
                      try_to_wake_up_local(to_wakeup, &rf);
              }
  
  ---
  
  

• When to destroy

  idle_worker_timeout
  ---
      spin_lock_irq(&pool->lock);
  
      while (too_many_workers(pool)) {
  
              ---
              return nr_idle > 2 && (nr_idle - 2) * MAX_IDLE_WORKERS_RATIO >= nr_busy;
              ---
  
          struct worker *worker;
          unsigned long expires;
  
          /* idle_list is kept in LIFO order, check the last one */
          worker = list_entry(pool->idle_list.prev, struct worker, entry);
          expires = worker->last_active + IDLE_WORKER_TIMEOUT;300 * HZ
  
          if (time_before(jiffies, expires)) {
              mod_timer(&pool->idle_timer, expires);
              break;
          }
  
          destroy_worker(worker);
          ---
  
          // pool->lock could serialize everything.
  
          pool->nr_workers--;
          pool->nr_idle--;
  
          list_del_init(&worker->entry);
          worker->flags |= WORKER_DIE;
          wake_up_process(worker->task);
          ---
      }
  
      spin_unlock_irq(&pool->lock);
  ---
  

• which to wake up

  When a work is queued on a pool, the first idle kworker will be waked up if
  there is no running one.
  
  insert_work
  ---
      set_work_pwq(work, pwq, extra_flags);
      list_add_tail(&work->entry, head);
      get_pwq(pwq);
  
      /*
       * Ensure either wq_worker_sleeping() sees the above
       * list_add_tail() or we see zero nr_running to avoid workers lying
       * around lazily while there are works to be processed.
       */
      smp_mb();
  
      if (__need_more_worker(pool)) //pool->nr_running
          wake_up_worker(pool);
  ---
  

there are 2 kworkers at least in one pool. When the 1st one gets to sleep,
the other one will be waked up. Before the second one starts to handle the work,
it will try to create another kworker to ensure there is one idle task for
backup.

In the other word, works queued to one worker pool will be handled one by one if
none of them gets to sleep. If one of them is cpu-intensive, the latter ones
maybe delayed. This indicates there could be extra delay in workqueue than a
specific kthread.