workqueue: implement get/put_pwq()

Add pool_workqueue->refcnt along with get/put_pwq().  Both per-cpu and
unbound pwqs have refcnts and any work item inserted on a pwq
increments the refcnt which is dropped when the work item finishes.

For per-cpu pwqs the base ref is never dropped and destroy_workqueue()
frees the pwqs as before.  For unbound ones, destroy_workqueue()
simply drops the base ref on the first pwq.  When the refcnt reaches
zero, pwq_unbound_release_workfn() is scheduled on system_wq, which
unlinks the pwq, puts the associated pool and frees the pwq and wq as
necessary.  This needs to be done from a work item as put_pwq() needs
to be protected by pool->lock but release can't happen with the lock
held - e.g. put_unbound_pool() involves blocking operations.

Unbound pool->locks are marked with lockdep subclas 1 as put_pwq()
will schedule the release work item on system_wq while holding the
unbound pool's lock and triggers recursive locking warning spuriously.

This will be used to implement dynamic creation and destruction of
unbound pwqs.

Signed-off-by: Tejun Heo <tj@kernel.org>
Reviewed-by: Lai Jiangshan <laijs@cn.fujitsu.com>

diff --git a/kernel/workqueue.c b/kernel/workqueue.c
index 5ac846e0085ee1ce0c07d65f2d5d566dbe768a21..7dd8e7bcec51f9e361de0a4a6c46ff3e911797ca 100644 (file)
--- a/kernel/workqueue.c
+++ b/kernel/workqueue.c
@@ -179,6 +179,7 @@ struct pool_workqueue {
        struct workqueue_struct *wq;            /* I: the owning workqueue */
        int                     work_color;     /* L: current color */
        int                     flush_color;    /* L: flushing color */
+       int                     refcnt;         /* L: reference count */
        int                     nr_in_flight[WORK_NR_COLORS];
                                                /* L: nr of in_flight works */
        int                     nr_active;      /* L: nr of active works */
@@ -186,6 +187,15 @@ struct pool_workqueue {
        struct list_head        delayed_works;  /* L: delayed works */
        struct list_head        pwqs_node;      /* R: node on wq->pwqs */
        struct list_head        mayday_node;    /* W: node on wq->maydays */
+
+       /*
+        * Release of unbound pwq is punted to system_wq.  See put_pwq()
+        * and pwq_unbound_release_workfn() for details.  pool_workqueue
+        * itself is also sched-RCU protected so that the first pwq can be
+        * determined without grabbing workqueue_lock.
+        */
+       struct work_struct      unbound_release_work;
+       struct rcu_head         rcu;
 } __aligned(1 << WORK_STRUCT_FLAG_BITS);
 
 /*
@@ -939,6 +949,45 @@ static void move_linked_works(struct work_struct *work, struct list_head *head,
                *nextp = n;
 }
 
+/**
+ * get_pwq - get an extra reference on the specified pool_workqueue
+ * @pwq: pool_workqueue to get
+ *
+ * Obtain an extra reference on @pwq.  The caller should guarantee that
+ * @pwq has positive refcnt and be holding the matching pool->lock.
+ */
+static void get_pwq(struct pool_workqueue *pwq)
+{
+       lockdep_assert_held(&pwq->pool->lock);
+       WARN_ON_ONCE(pwq->refcnt <= 0);
+       pwq->refcnt++;
+}
+
+/**
+ * put_pwq - put a pool_workqueue reference
+ * @pwq: pool_workqueue to put
+ *
+ * Drop a reference of @pwq.  If its refcnt reaches zero, schedule its
+ * destruction.  The caller should be holding the matching pool->lock.
+ */
+static void put_pwq(struct pool_workqueue *pwq)
+{
+       lockdep_assert_held(&pwq->pool->lock);
+       if (likely(--pwq->refcnt))
+               return;
+       if (WARN_ON_ONCE(!(pwq->wq->flags & WQ_UNBOUND)))
+               return;
+       /*
+        * @pwq can't be released under pool->lock, bounce to
+        * pwq_unbound_release_workfn().  This never recurses on the same
+        * pool->lock as this path is taken only for unbound workqueues and
+        * the release work item is scheduled on a per-cpu workqueue.  To
+        * avoid lockdep warning, unbound pool->locks are given lockdep
+        * subclass of 1 in get_unbound_pool().
+        */
+       schedule_work(&pwq->unbound_release_work);
+}
+
 static void pwq_activate_delayed_work(struct work_struct *work)
 {
        struct pool_workqueue *pwq = get_work_pwq(work);
@@ -970,9 +1019,9 @@ static void pwq_activate_first_delayed(struct pool_workqueue *pwq)
  */
 static void pwq_dec_nr_in_flight(struct pool_workqueue *pwq, int color)
 {
-       /* ignore uncolored works */
+       /* uncolored work items don't participate in flushing or nr_active */
        if (color == WORK_NO_COLOR)
-               return;
+               goto out_put;
 
        pwq->nr_in_flight[color]--;
 
@@ -985,11 +1034,11 @@ static void pwq_dec_nr_in_flight(struct pool_workqueue *pwq, int color)
 
        /* is flush in progress and are we at the flushing tip? */
        if (likely(pwq->flush_color != color))
-               return;
+               goto out_put;
 
        /* are there still in-flight works? */
        if (pwq->nr_in_flight[color])
-               return;
+               goto out_put;
 
        /* this pwq is done, clear flush_color */
        pwq->flush_color = -1;
@@ -1000,6 +1049,8 @@ static void pwq_dec_nr_in_flight(struct pool_workqueue *pwq, int color)
         */
        if (atomic_dec_and_test(&pwq->wq->nr_pwqs_to_flush))
                complete(&pwq->wq->first_flusher->done);
+out_put:
+       put_pwq(pwq);
 }
 
 /**
@@ -1122,6 +1173,7 @@ static void insert_work(struct pool_workqueue *pwq, struct work_struct *work,
        /* we own @work, set data and link */
        set_work_pwq(work, pwq, extra_flags);
        list_add_tail(&work->entry, head);
+       get_pwq(pwq);
 
        /*
         * Ensure either worker_sched_deactivated() sees the above
@@ -3301,6 +3353,7 @@ static struct worker_pool *get_unbound_pool(const struct workqueue_attrs *attrs)
        if (!pool || init_worker_pool(pool) < 0)
                goto fail;
 
+       lockdep_set_subclass(&pool->lock, 1);   /* see put_pwq() */
        copy_workqueue_attrs(pool->attrs, attrs);
 
        if (worker_pool_assign_id(pool) < 0)
@@ -3329,7 +3382,41 @@ fail:
        return NULL;
 }
 
-/* initialize @pwq which interfaces with @pool for @wq and link it in */
+static void rcu_free_pwq(struct rcu_head *rcu)
+{
+       kmem_cache_free(pwq_cache,
+                       container_of(rcu, struct pool_workqueue, rcu));
+}
+
+/*
+ * Scheduled on system_wq by put_pwq() when an unbound pwq hits zero refcnt
+ * and needs to be destroyed.
+ */
+static void pwq_unbound_release_workfn(struct work_struct *work)
+{
+       struct pool_workqueue *pwq = container_of(work, struct pool_workqueue,
+                                                 unbound_release_work);
+       struct workqueue_struct *wq = pwq->wq;
+       struct worker_pool *pool = pwq->pool;
+
+       if (WARN_ON_ONCE(!(wq->flags & WQ_UNBOUND)))
+               return;
+
+       spin_lock_irq(&workqueue_lock);
+       list_del_rcu(&pwq->pwqs_node);
+       spin_unlock_irq(&workqueue_lock);
+
+       put_unbound_pool(pool);
+       call_rcu_sched(&pwq->rcu, rcu_free_pwq);
+
+       /*
+        * If we're the last pwq going away, @wq is already dead and no one
+        * is gonna access it anymore.  Free it.
+        */
+       if (list_empty(&wq->pwqs))
+               kfree(wq);
+}
+
 static void init_and_link_pwq(struct pool_workqueue *pwq,
                              struct workqueue_struct *wq,
                              struct worker_pool *pool)
@@ -3339,9 +3426,11 @@ static void init_and_link_pwq(struct pool_workqueue *pwq,
        pwq->pool = pool;
        pwq->wq = wq;
        pwq->flush_color = -1;
+       pwq->refcnt = 1;
        pwq->max_active = wq->saved_max_active;
        INIT_LIST_HEAD(&pwq->delayed_works);
        INIT_LIST_HEAD(&pwq->mayday_node);
+       INIT_WORK(&pwq->unbound_release_work, pwq_unbound_release_workfn);
 
        list_add_tail_rcu(&pwq->pwqs_node, &wq->pwqs);
 }
@@ -3384,15 +3473,6 @@ static int alloc_and_link_pwqs(struct workqueue_struct *wq)
        return 0;
 }
 
-static void free_pwqs(struct workqueue_struct *wq)
-{
-       if (!(wq->flags & WQ_UNBOUND))
-               free_percpu(wq->cpu_pwqs);
-       else if (!list_empty(&wq->pwqs))
-               kmem_cache_free(pwq_cache, list_first_entry(&wq->pwqs,
-                                       struct pool_workqueue, pwqs_node));
-}
-
 static int wq_clamp_max_active(int max_active, unsigned int flags,
                               const char *name)
 {
@@ -3524,7 +3604,8 @@ void destroy_workqueue(struct workqueue_struct *wq)
                        }
                }
 
-               if (WARN_ON(pwq->nr_active) ||
+               if (WARN_ON(pwq->refcnt > 1) ||
+                   WARN_ON(pwq->nr_active) ||
                    WARN_ON(!list_empty(&pwq->delayed_works))) {
                        spin_unlock_irq(&workqueue_lock);
                        return;
@@ -3545,17 +3626,27 @@ void destroy_workqueue(struct workqueue_struct *wq)
                wq->rescuer = NULL;
        }
 
-       /*
-        * We're the sole accessor of @wq at this point.  Directly access
-        * the first pwq and put its pool.
-        */
-       if (wq->flags & WQ_UNBOUND) {
+       if (!(wq->flags & WQ_UNBOUND)) {
+               /*
+                * The base ref is never dropped on per-cpu pwqs.  Directly
+                * free the pwqs and wq.
+                */
+               free_percpu(wq->cpu_pwqs);
+               kfree(wq);
+       } else {
+               /*
+                * We're the sole accessor of @wq at this point.  Directly
+                * access the first pwq and put the base ref.  As both pwqs
+                * and pools are sched-RCU protected, the lock operations
+                * are safe.  @wq will be freed when the last pwq is
+                * released.
+                */
                pwq = list_first_entry(&wq->pwqs, struct pool_workqueue,
                                       pwqs_node);
-               put_unbound_pool(pwq->pool);
+               spin_lock_irq(&pwq->pool->lock);
+               put_pwq(pwq);
+               spin_unlock_irq(&pwq->pool->lock);
        }
-       free_pwqs(wq);
-       kfree(wq);
 }
 EXPORT_SYMBOL_GPL(destroy_workqueue);