workqueue: Don't hold any lock while rcuwait'ing for !POOL_MANAGER_ACTIVE

put_unbound_pool() currently passes wq_manager_inactive() as exit condition to rcuwait_wait_event(), which grabs pool->lock to check for pool->flags & POOL_MANAGER_ACTIVE A later patch will require destroy_worker() to be invoked with wq_pool_attach_mutex held, which needs to be acquired before pool->lock. A mutex cannot be acquired within rcuwait_wait_event(), as it could clobber the task state set by rcuwait_wait_event() Instead, restructure the waiting logic to acquire any necessary lock outside of rcuwait_wait_event(). Since further work cannot be inserted into unbound pwqs that have reached ->refcnt==0, this is bound to make forward progress as eventually the worklist will be drained and need_more_worker(pool) will remain false, preventing any worker from stealing the manager position from us.

Valentin Schneider committed 3 years ago

Commit 9ab03be1 parent 3f959aa1 file · +19 −17Browse files

Showing the changes to kernel/workqueue.c (1 of 1 file). Show all

Unified · Split

kernel/workqueue.c+19−17

@@ -3608,18 +3608,6 @@ static void rcu_free_pool(struct rcu_head *rcu) · 3607 unmodified lines
3608	3608	kfree(pool);
3609	3609	}
3610	3610
3611		−/* This returns with the lock held on success (pool manager is inactive). */
3612		−static bool wq_manager_inactive(struct worker_pool *pool)
3613		−{
3614		− raw_spin_lock_irq(&pool->lock);
3615		−
3616		− if (pool->flags & POOL_MANAGER_ACTIVE) {
3617		− raw_spin_unlock_irq(&pool->lock);
3618		− return false;
3619		− }
3620		− return true;
3621		−}
3622		−
3623	3611	/**
3624	3612	* put_unbound_pool - put a worker_pool
3625	3613	* @pool: worker_pool to put
@@ -3655,12 +3643,26 @@ static void put_unbound_pool(struct worker_pool *pool) · 29 unmodified lines
3655	3643	* Become the manager and destroy all workers. This prevents
3656	3644	* @pool's workers from blocking on attach_mutex. We're the last
3657	3645	* manager and @pool gets freed with the flag set.
3658		− * Because of how wq_manager_inactive() works, we will hold the
3659		− * spinlock after a successful wait.
	3646	+ *
	3647	+ * Having a concurrent manager is quite unlikely to happen as we can
	3648	+ * only get here with
	3649	+ * pwq->refcnt == pool->refcnt == 0
	3650	+ * which implies no work queued to the pool, which implies no worker can
	3651	+ * become the manager. However a worker could have taken the role of
	3652	+ * manager before the refcnts dropped to 0, since maybe_create_worker()
	3653	+ * drops pool->lock
3660	3654	*/
3661		− rcuwait_wait_event(&manager_wait, wq_manager_inactive(pool),
3662		− TASK_UNINTERRUPTIBLE);
3663		− pool->flags \|= POOL_MANAGER_ACTIVE;
	3655	+ while (true) {
	3656	+ rcuwait_wait_event(&manager_wait,
	3657	+ !(pool->flags & POOL_MANAGER_ACTIVE),
	3658	+ TASK_UNINTERRUPTIBLE);
	3659	+ raw_spin_lock_irq(&pool->lock);
	3660	+ if (!(pool->flags & POOL_MANAGER_ACTIVE)) {
	3661	+ pool->flags \|= POOL_MANAGER_ACTIVE;
	3662	+ break;
	3663	+ }
	3664	+ raw_spin_unlock_irq(&pool->lock);
	3665	+ }
3664	3666
3665	3667	while ((worker = first_idle_worker(pool)))
3666	3668	destroy_worker(worker);