[PATCHSET wq/for-6.12] workqueue: Use wq_online_cpumask consistently and make CPU hotplug unbound pwq updates asynchronous

[PATCHSET wq/for-6.12] workqueue: Use wq_online_cpumask consistently and make CPU hotplug unbound pwq updates asynchronous

[Tejun Heo]

Hello,

Make wq_online_cpumask more consistently which also simplifies the code, and
make unbound pwq updates from CPU hotplug path asynchronous so that we don't
add additional dependency on fork from CPU hotplug.

 0001-workqueue-Make-wq_affn_dfl_set-use-wq_online_cpumask.patch
 0002-workqueue-Make-wq_update_node_max_active-use-wq_onli.patch
 0003-workqueue-s-unbound_wq_update_pwq-update_unbound_pwq.patch
 0004-workqueue-Make-unbound-pwq-updates-from-CPU-hotplug-.patch

The patches are also available in

 git://git.kernel.org/pub/scm/linux/kernel/git/tj/wq.git use-wq_online_cpumask

diffstat follows. Thanks.

 kernel/workqueue.c |  121 ++++++++++++++++++++++++++++++++++-----------------------------------------
 1 file changed, 56 insertions(+), 65 deletions(-)

--
tejun

[PATCH 1/4] workqueue: Make wq_affn_dfl_set() use wq_online_cpumask

[Tejun Heo]

Other unbound pwq update paths are already using wq_online_cpumask which is
protected by wq_pool_mutex. Make wq_affn_dfl_set() to use wq_online_cpumask
too for synchronization and consistency.

Signed-off-by: Tejun Heo <tj@kernel.org>
---
 kernel/workqueue.c | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/kernel/workqueue.c b/kernel/workqueue.c
index d56bd2277e58..6571e1f3c835 100644
--- a/kernel/workqueue.c
+++ b/kernel/workqueue.c
@@ -6909,18 +6909,16 @@ static int wq_affn_dfl_set(const char *val, const struct kernel_param *kp)
 	if (affn == WQ_AFFN_DFL)
 		return -EINVAL;
 
-	cpus_read_lock();
 	mutex_lock(&wq_pool_mutex);
 
 	wq_affn_dfl = affn;
 
 	list_for_each_entry(wq, &workqueues, list) {
-		for_each_online_cpu(cpu)
+		for_each_cpu(cpu, wq_online_cpumask)
 			unbound_wq_update_pwq(wq, cpu);
 	}
 
 	mutex_unlock(&wq_pool_mutex);
-	cpus_read_unlock();
 
 	return 0;
 }
-- 
2.45.2

Re: [PATCH 1/4] workqueue: Make wq_affn_dfl_set() use wq_online_cpumask

[Lai Jiangshan]

Hello

On Sat, Aug 3, 2024 at 5:18 AM Tejun Heo <tj@kernel.org> wrote:
>
> Other unbound pwq update paths are already using wq_online_cpumask which is
> protected by wq_pool_mutex. Make wq_affn_dfl_set() to use wq_online_cpumask
> too for synchronization and consistency.
>
> Signed-off-by: Tejun Heo <tj@kernel.org>
> ---
>  kernel/workqueue.c | 4 +---
>  1 file changed, 1 insertion(+), 3 deletions(-)
>
> diff --git a/kernel/workqueue.c b/kernel/workqueue.c
> index d56bd2277e58..6571e1f3c835 100644
> --- a/kernel/workqueue.c
> +++ b/kernel/workqueue.c
> @@ -6909,18 +6909,16 @@ static int wq_affn_dfl_set(const char *val, const struct kernel_param *kp)
>         if (affn == WQ_AFFN_DFL)
>                 return -EINVAL;
>
> -       cpus_read_lock();
>         mutex_lock(&wq_pool_mutex);
>
>         wq_affn_dfl = affn;
>
>         list_for_each_entry(wq, &workqueues, list) {
> -               for_each_online_cpu(cpu)
> +               for_each_cpu(cpu, wq_online_cpumask)
>                         unbound_wq_update_pwq(wq, cpu);
>         }

I think it should be for_each_possible_cpu() for updating the pwqs.

For all the 4 patches:

Reviewed-by: Lai Jiangshan <jiangshanlai@gmail.com>

Thanks
Lai

Re: [PATCH 1/4] workqueue: Make wq_affn_dfl_set() use wq_online_cpumask

[Tejun Heo]

Hello,

On Thu, Aug 08, 2024 at 12:23:32PM +0800, Lai Jiangshan wrote:
> > @@ -6909,18 +6909,16 @@ static int wq_affn_dfl_set(const char *val, const struct kernel_param *kp)
> >         if (affn == WQ_AFFN_DFL)
> >                 return -EINVAL;
> >
> > -       cpus_read_lock();
> >         mutex_lock(&wq_pool_mutex);
> >
> >         wq_affn_dfl = affn;
> >
> >         list_for_each_entry(wq, &workqueues, list) {
> > -               for_each_online_cpu(cpu)
> > +               for_each_cpu(cpu, wq_online_cpumask)
> >                         unbound_wq_update_pwq(wq, cpu);
> >         }
> 
> I think it should be for_each_possible_cpu() for updating the pwqs.

Can you send a patch?

Thanks.

-- 
tejun

[PATCH 2/4] workqueue: Make wq_update_node_max_active() use wq_online_cpumask

[Tejun Heo]

As, while a CPU is going down, wq_update_node_max_active() is called with
the CPU still online, it takes @off_cpu so that the CPU can be treated as
offline. This is cumbersome and no longer necessary with the recently added
wq_online_cpumask which is cleared before wq_update_node_max_active() is
called.

Make wq_update_node_max_active() use wq_online_cpumask and drop @off_cpu.

Signed-off-by: Tejun Heo <tj@kernel.org>
---
 kernel/workqueue.c | 29 ++++++++++++-----------------
 1 file changed, 12 insertions(+), 17 deletions(-)

diff --git a/kernel/workqueue.c b/kernel/workqueue.c
index 6571e1f3c835..bb96414f68b8 100644
--- a/kernel/workqueue.c
+++ b/kernel/workqueue.c
@@ -1566,13 +1566,12 @@ static struct wq_node_nr_active *wq_node_nr_active(struct workqueue_struct *wq,
 /**
  * wq_update_node_max_active - Update per-node max_actives to use
  * @wq: workqueue to update
- * @off_cpu: CPU that's going down, -1 if a CPU is not going down
  *
  * Update @wq->node_nr_active[]->max. @wq must be unbound. max_active is
  * distributed among nodes according to the proportions of numbers of online
  * cpus. The result is always between @wq->min_active and max_active.
  */
-static void wq_update_node_max_active(struct workqueue_struct *wq, int off_cpu)
+static void wq_update_node_max_active(struct workqueue_struct *wq)
 {
 	struct cpumask *effective = unbound_effective_cpumask(wq);
 	int min_active = READ_ONCE(wq->min_active);
@@ -1584,12 +1583,7 @@ static void wq_update_node_max_active(struct workqueue_struct *wq, int off_cpu)
 	if (!wq_topo_initialized)
 		return;
 
-	if (off_cpu >= 0 && !cpumask_test_cpu(off_cpu, effective))
-		off_cpu = -1;
-
-	total_cpus = cpumask_weight_and(effective, cpu_online_mask);
-	if (off_cpu >= 0)
-		total_cpus--;
+	total_cpus = cpumask_weight_and(effective, wq_online_cpumask);
 
 	/* If all CPUs of the wq get offline, use the default values */
 	if (unlikely(!total_cpus)) {
@@ -1601,11 +1595,12 @@ static void wq_update_node_max_active(struct workqueue_struct *wq, int off_cpu)
 	}
 
 	for_each_node(node) {
-		int node_cpus;
+		int cpu, node_cpus = 0;
 
-		node_cpus = cpumask_weight_and(effective, cpumask_of_node(node));
-		if (off_cpu >= 0 && cpu_to_node(off_cpu) == node)
-			node_cpus--;
+		/* count the number of effective online CPUs in the node */
+		for_each_cpu_and(cpu, effective, cpumask_of_node(node))
+			if (cpumask_test_cpu(cpu, wq_online_cpumask))
+				node_cpus++;
 
 		wq_node_nr_active(wq, node)->max =
 			clamp(DIV_ROUND_UP(max_active * node_cpus, total_cpus),
@@ -5298,7 +5293,7 @@ static void apply_wqattrs_commit(struct apply_wqattrs_ctx *ctx)
 	ctx->dfl_pwq = install_unbound_pwq(ctx->wq, -1, ctx->dfl_pwq);
 
 	/* update node_nr_active->max */
-	wq_update_node_max_active(ctx->wq, -1);
+	wq_update_node_max_active(ctx->wq);
 
 	/* rescuer needs to respect wq cpumask changes */
 	if (ctx->wq->rescuer)
@@ -5583,7 +5578,7 @@ static void wq_adjust_max_active(struct workqueue_struct *wq)
 	WRITE_ONCE(wq->min_active, new_min);
 
 	if (wq->flags & WQ_UNBOUND)
-		wq_update_node_max_active(wq, -1);
+		wq_update_node_max_active(wq);
 
 	if (new_max == 0)
 		return;
@@ -6597,7 +6592,7 @@ int workqueue_online_cpu(unsigned int cpu)
 				unbound_wq_update_pwq(wq, tcpu);
 
 			mutex_lock(&wq->mutex);
-			wq_update_node_max_active(wq, -1);
+			wq_update_node_max_active(wq);
 			mutex_unlock(&wq->mutex);
 		}
 	}
@@ -6632,7 +6627,7 @@ int workqueue_offline_cpu(unsigned int cpu)
 				unbound_wq_update_pwq(wq, tcpu);
 
 			mutex_lock(&wq->mutex);
-			wq_update_node_max_active(wq, cpu);
+			wq_update_node_max_active(wq);
 			mutex_unlock(&wq->mutex);
 		}
 	}
@@ -7935,7 +7930,7 @@ void __init workqueue_init_topology(void)
 			unbound_wq_update_pwq(wq, cpu);
 		if (wq->flags & WQ_UNBOUND) {
 			mutex_lock(&wq->mutex);
-			wq_update_node_max_active(wq, -1);
+			wq_update_node_max_active(wq);
 			mutex_unlock(&wq->mutex);
 		}
 	}
-- 
2.45.2

[PATCH 3/4] workqueue: s/unbound_wq_update_pwq/update_unbound_pwqs/

[Tejun Heo]

unbound_wq_update_pwq() reads odd and is unwieldy. Rename to
update_unbound_pwqs().

Signed-off-by: Tejun Heo <tj@kernel.org>
---
 kernel/workqueue.c | 24 ++++++++++++------------
 1 file changed, 12 insertions(+), 12 deletions(-)

diff --git a/kernel/workqueue.c b/kernel/workqueue.c
index bb96414f68b8..0f593f2884e0 100644
--- a/kernel/workqueue.c
+++ b/kernel/workqueue.c
@@ -433,8 +433,8 @@ static struct kmem_cache *pwq_cache;
 static struct wq_pod_type wq_pod_types[WQ_AFFN_NR_TYPES];
 static enum wq_affn_scope wq_affn_dfl = WQ_AFFN_CACHE;
 
-/* buf for wq_update_unbound_pod_attrs(), protected by CPU hotplug exclusion */
-static struct workqueue_attrs *unbound_wq_update_pwq_attrs_buf;
+/* buf for update_unbound_pwqs_attrs(), protected by CPU hotplug exclusion */
+static struct workqueue_attrs *update_unbound_pwqs_attrs_buf;
 
 static DEFINE_MUTEX(wq_pool_mutex);	/* protects pools and workqueues list */
 static DEFINE_MUTEX(wq_pool_attach_mutex); /* protects worker attach/detach */
@@ -5351,7 +5351,7 @@ int apply_workqueue_attrs(struct workqueue_struct *wq,
 }
 
 /**
- * unbound_wq_update_pwq - update a pwq slot for CPU hot[un]plug
+ * update_unbound_pwqs - update a pwq slot for CPU hot[un]plug
  * @wq: the target workqueue
  * @cpu: the CPU to update the pwq slot for
  *
@@ -5369,7 +5369,7 @@ int apply_workqueue_attrs(struct workqueue_struct *wq,
  * CPU_DOWN. If a workqueue user wants strict affinity, it's the user's
  * responsibility to flush the work item from CPU_DOWN_PREPARE.
  */
-static void unbound_wq_update_pwq(struct workqueue_struct *wq, int cpu)
+static void update_unbound_pwqs(struct workqueue_struct *wq, int cpu)
 {
 	struct pool_workqueue *old_pwq = NULL, *pwq;
 	struct workqueue_attrs *target_attrs;
@@ -5384,7 +5384,7 @@ static void unbound_wq_update_pwq(struct workqueue_struct *wq, int cpu)
 	 * Let's use a preallocated one.  The following buf is protected by
 	 * CPU hotplug exclusion.
 	 */
-	target_attrs = unbound_wq_update_pwq_attrs_buf;
+	target_attrs = update_unbound_pwqs_attrs_buf;
 
 	copy_workqueue_attrs(target_attrs, wq->unbound_attrs);
 	wqattrs_actualize_cpumask(target_attrs, wq_unbound_cpumask);
@@ -6589,7 +6589,7 @@ int workqueue_online_cpu(unsigned int cpu)
 			int tcpu;
 
 			for_each_cpu(tcpu, pt->pod_cpus[pt->cpu_pod[cpu]])
-				unbound_wq_update_pwq(wq, tcpu);
+				update_unbound_pwqs(wq, tcpu);
 
 			mutex_lock(&wq->mutex);
 			wq_update_node_max_active(wq);
@@ -6624,7 +6624,7 @@ int workqueue_offline_cpu(unsigned int cpu)
 			int tcpu;
 
 			for_each_cpu(tcpu, pt->pod_cpus[pt->cpu_pod[cpu]])
-				unbound_wq_update_pwq(wq, tcpu);
+				update_unbound_pwqs(wq, tcpu);
 
 			mutex_lock(&wq->mutex);
 			wq_update_node_max_active(wq);
@@ -6910,7 +6910,7 @@ static int wq_affn_dfl_set(const char *val, const struct kernel_param *kp)
 
 	list_for_each_entry(wq, &workqueues, list) {
 		for_each_cpu(cpu, wq_online_cpumask)
-			unbound_wq_update_pwq(wq, cpu);
+			update_unbound_pwqs(wq, cpu);
 	}
 
 	mutex_unlock(&wq_pool_mutex);
@@ -7656,8 +7656,8 @@ void __init workqueue_init_early(void)
 
 	pwq_cache = KMEM_CACHE(pool_workqueue, SLAB_PANIC);
 
-	unbound_wq_update_pwq_attrs_buf = alloc_workqueue_attrs();
-	BUG_ON(!unbound_wq_update_pwq_attrs_buf);
+	update_unbound_pwqs_attrs_buf = alloc_workqueue_attrs();
+	BUG_ON(!update_unbound_pwqs_attrs_buf);
 
 	/*
 	 * If nohz_full is enabled, set power efficient workqueue as unbound.
@@ -7922,12 +7922,12 @@ void __init workqueue_init_topology(void)
 
 	/*
 	 * Workqueues allocated earlier would have all CPUs sharing the default
-	 * worker pool. Explicitly call unbound_wq_update_pwq() on all workqueue
+	 * worker pool. Explicitly call update_unbound_pwqs() on all workqueue
 	 * and CPU combinations to apply per-pod sharing.
 	 */
 	list_for_each_entry(wq, &workqueues, list) {
 		for_each_online_cpu(cpu)
-			unbound_wq_update_pwq(wq, cpu);
+			update_unbound_pwqs(wq, cpu);
 		if (wq->flags & WQ_UNBOUND) {
 			mutex_lock(&wq->mutex);
 			wq_update_node_max_active(wq);
-- 
2.45.2

[PATCH 4/4] workqueue: Make unbound pwq updates from CPU hotplug path asynchronous

[Tejun Heo]

84193c07105c ("workqueue: Generalize unbound CPU pods") and associated
commits added generalized affinity scope support. In the process, it added
synchronous calls to update_unbound_pwqs() from CPU hotplug path. When a CPU
comes online in an otherwise empty pod, this can lead to creation of a new
worker_pool involving creating new workers. Workqueues work correctly
without synchronous updates and there's no need to add additional fork
dependency to hotplug path. Make unbound pwq updates asynchronous.

This makes update_unbound_pwqs() called on all CPUs instead of the pod's
CPUs. However, when no-op, the call is pretty cheap and the additional calls
shouldn't be noticeable.

Signed-off-by: Tejun Heo <tj@kernel.org>
---
 kernel/workqueue.c | 72 ++++++++++++++++++++++------------------------
 1 file changed, 35 insertions(+), 37 deletions(-)

diff --git a/kernel/workqueue.c b/kernel/workqueue.c
index 0f593f2884e0..33c6635bcb87 100644
--- a/kernel/workqueue.c
+++ b/kernel/workqueue.c
@@ -6544,6 +6544,39 @@ static void restore_unbound_workers_cpumask(struct worker_pool *pool, int cpu)
 		WARN_ON_ONCE(set_cpus_allowed_ptr(worker->task, &cpumask) < 0);
 }
 
+static void hotplug_update_unbound_pwqs_workfn(struct work_struct *work)
+{
+	struct workqueue_struct *wq;
+
+	mutex_lock(&wq_pool_mutex);
+
+	/*
+	 * CPUs have gone online or offline. Update the pod affinities of all
+	 * unbound workqueues. While not optimal, workqueues operate correctly
+	 * without these updates, so this can be done asynchronously, which
+	 * avoids adding fork dependency from CPU hotplug path.
+	 */
+	list_for_each_entry(wq, &workqueues, list) {
+		struct workqueue_attrs *attrs = wq->unbound_attrs;
+
+		if (attrs) {
+			int cpu;
+
+			for_each_possible_cpu(cpu)
+				update_unbound_pwqs(wq, cpu);
+
+			mutex_lock(&wq->mutex);
+			wq_update_node_max_active(wq);
+			mutex_unlock(&wq->mutex);
+		}
+	}
+
+	mutex_unlock(&wq_pool_mutex);
+}
+
+static DECLARE_WORK(hotplug_update_unbound_pwqs_work,
+		    hotplug_update_unbound_pwqs_workfn);
+
 int workqueue_prepare_cpu(unsigned int cpu)
 {
 	struct worker_pool *pool;
@@ -6560,7 +6593,6 @@ int workqueue_prepare_cpu(unsigned int cpu)
 int workqueue_online_cpu(unsigned int cpu)
 {
 	struct worker_pool *pool;
-	struct workqueue_struct *wq;
 	int pi;
 
 	mutex_lock(&wq_pool_mutex);
@@ -6580,22 +6612,7 @@ int workqueue_online_cpu(unsigned int cpu)
 		mutex_unlock(&wq_pool_attach_mutex);
 	}
 
-	/* update pod affinity of unbound workqueues */
-	list_for_each_entry(wq, &workqueues, list) {
-		struct workqueue_attrs *attrs = wq->unbound_attrs;
-
-		if (attrs) {
-			const struct wq_pod_type *pt = wqattrs_pod_type(attrs);
-			int tcpu;
-
-			for_each_cpu(tcpu, pt->pod_cpus[pt->cpu_pod[cpu]])
-				update_unbound_pwqs(wq, tcpu);
-
-			mutex_lock(&wq->mutex);
-			wq_update_node_max_active(wq);
-			mutex_unlock(&wq->mutex);
-		}
-	}
+	schedule_work(&hotplug_update_unbound_pwqs_work);
 
 	mutex_unlock(&wq_pool_mutex);
 	return 0;
@@ -6603,34 +6620,15 @@ int workqueue_online_cpu(unsigned int cpu)
 
 int workqueue_offline_cpu(unsigned int cpu)
 {
-	struct workqueue_struct *wq;
-
 	/* unbinding per-cpu workers should happen on the local CPU */
 	if (WARN_ON(cpu != smp_processor_id()))
 		return -1;
 
 	unbind_workers(cpu);
 
-	/* update pod affinity of unbound workqueues */
 	mutex_lock(&wq_pool_mutex);
-
 	cpumask_clear_cpu(cpu, wq_online_cpumask);
-
-	list_for_each_entry(wq, &workqueues, list) {
-		struct workqueue_attrs *attrs = wq->unbound_attrs;
-
-		if (attrs) {
-			const struct wq_pod_type *pt = wqattrs_pod_type(attrs);
-			int tcpu;
-
-			for_each_cpu(tcpu, pt->pod_cpus[pt->cpu_pod[cpu]])
-				update_unbound_pwqs(wq, tcpu);
-
-			mutex_lock(&wq->mutex);
-			wq_update_node_max_active(wq);
-			mutex_unlock(&wq->mutex);
-		}
-	}
+	schedule_work(&hotplug_update_unbound_pwqs_work);
 	mutex_unlock(&wq_pool_mutex);
 
 	return 0;
-- 
2.45.2