workqueue: Provide a handshake for canceling BH workers

While a BH work item is canceled, the core code spins until it
determines that the item completed. On PREEMPT_RT the spinning relies on
a lock in local_bh_disable() to avoid a live lock if the canceling
thread has higher priority than the BH-worker and preempts it. This lock
ensures that the BH-worker makes progress by PI-boosting it.

This lock in local_bh_disable() is a central per-CPU BKL and about to be
removed.

To provide the required synchronisation add a per pool lock. The lock is
acquired by the bh_worker at the begin while the individual callbacks
are invoked. To enforce progress in case of interruption, __flush_work()
needs to acquire the lock.
This will flush all BH-work items assigned to that pool.

Signed-off-by: Sebastian Andrzej Siewior <bigeasy@linutronix.de>
Signed-off-by: Tejun Heo <tj@kernel.org>

authored by

Sebastian Andrzej Siewior and committed by

Tejun Heo 9 months ago ad7c7f4b cda2b2d6

+41 -9

1 changed file

expand all

kernel

workqueue.c

+41 -9

kernel/workqueue.c

··· 222 222 struct workqueue_attrs *attrs; /* I: worker attributes */ 223 223 struct hlist_node hash_node; /* PL: unbound_pool_hash node */ 224 224 int refcnt; /* PL: refcnt for unbound pools */ 225 - 225 + #ifdef CONFIG_PREEMPT_RT 226 + spinlock_t cb_lock; /* BH worker cancel lock */ 227 + #endif 226 228 /* 227 229 * Destruction of pool is RCU protected to allow dereferences 228 230 * from get_work_pool(). ··· 3080 3078 goto restart; 3081 3079 } 3082 3080 3081 + #ifdef CONFIG_PREEMPT_RT 3082 + static void worker_lock_callback(struct worker_pool *pool) 3083 + { 3084 + spin_lock(&pool->cb_lock); 3085 + } 3086 + 3087 + static void worker_unlock_callback(struct worker_pool *pool) 3088 + { 3089 + spin_unlock(&pool->cb_lock); 3090 + } 3091 + 3092 + static void workqueue_callback_cancel_wait_running(struct worker_pool *pool) 3093 + { 3094 + spin_lock(&pool->cb_lock); 3095 + spin_unlock(&pool->cb_lock); 3096 + } 3097 + 3098 + #else 3099 + 3100 + static void worker_lock_callback(struct worker_pool *pool) { } 3101 + static void worker_unlock_callback(struct worker_pool *pool) { } 3102 + static void workqueue_callback_cancel_wait_running(struct worker_pool *pool) { } 3103 + 3104 + #endif 3105 + 3083 3106 /** 3084 3107 * manage_workers - manage worker pool 3085 3108 * @worker: self ··· 3584 3557 int nr_restarts = BH_WORKER_RESTARTS; 3585 3558 unsigned long end = jiffies + BH_WORKER_JIFFIES; 3586 3559 3560 + worker_lock_callback(pool); 3587 3561 raw_spin_lock_irq(&pool->lock); 3588 3562 worker_leave_idle(worker); 3589 3563 ··· 3613 3585 worker_enter_idle(worker); 3614 3586 kick_pool(pool); 3615 3587 raw_spin_unlock_irq(&pool->lock); 3588 + worker_unlock_callback(pool); 3616 3589 } 3617 3590 3618 3591 /* ··· 4251 4222 (data & WORK_OFFQ_BH)) { 4252 4223 /* 4253 4224 * On RT, prevent a live lock when %current preempted 4254 - * soft interrupt processing or prevents ksoftirqd from 4255 - * running by keeping flipping BH. If the BH work item 4256 - * runs on a different CPU then this has no effect other 4257 - * than doing the BH disable/enable dance for nothing. 4258 - * This is copied from 4259 - * kernel/softirq.c::tasklet_unlock_spin_wait(). 4225 + * soft interrupt processing by blocking on lock which 4226 + * is owned by the thread invoking the callback. 4260 4227 */ 4261 4228 while (!try_wait_for_completion(&barr.done)) { 4262 4229 if (IS_ENABLED(CONFIG_PREEMPT_RT)) { 4263 - local_bh_disable(); 4264 - local_bh_enable(); 4230 + struct worker_pool *pool; 4231 + 4232 + guard(rcu)(); 4233 + pool = get_work_pool(work); 4234 + if (pool) 4235 + workqueue_callback_cancel_wait_running(pool); 4265 4236 } else { 4266 4237 cpu_relax(); 4267 4238 } ··· 4811 4782 ida_init(&pool->worker_ida); 4812 4783 INIT_HLIST_NODE(&pool->hash_node); 4813 4784 pool->refcnt = 1; 4785 + #ifdef CONFIG_PREEMPT_RT 4786 + spin_lock_init(&pool->cb_lock); 4787 + #endif 4814 4788 4815 4789 /* shouldn't fail above this point */ 4816 4790 pool->attrs = alloc_workqueue_attrs();

Configure Feed

Configure Feed