Commit 8864b4e5 authored by Tejun Heo's avatar Tejun Heo
Browse files

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: default avatarTejun Heo <tj@kernel.org>
Reviewed-by: default avatarLai Jiangshan <laijs@cn.fujitsu.com>
parent d2c1d404

kernel/workqueue.c

+114 −23
Original line number Diff line number Diff line
@@ -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;

	}



	if (!(wq->flags & WQ_UNBOUND)) {

		/*

	 * We're the sole accessor of @wq at this point.  Directly access

	 * the first pwq and put its pool.

		 * 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.

		 */

	if (wq->flags & WQ_UNBOUND) {

		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);

CRA Git | Maintained and supported by SUSTech CRA and CCSE