[PATCH wq/for-7.5 v4 2/2] workqueue: Keep the limits of both max_active domains current

[Breno Leitao <leitao@debian.org>]

Commit 27db9dd7f84f3a ("workqueue: Give percpu workqueues their own
max_active") gave the percpu backend a limit of its own. A workqueue
carries max_active and min_active for the unbound backend and
percpu_max_active for the percpu one. Only the backend it is created on
gets its limits set. The other stays at zero.

The planned PERCPU affinity scope will let a workqueue switch backends
through its attributes, and a switch should land on a sensible limit.
Derive the other domain whenever the limits of the backend in use are
set.

The two numbers cannot simply be kept equal. percpu_max_active caps every
CPU on its own, while max_active is a workqueue wide budget distributed
among the nodes, with min_active as the per-node floor. A per-CPU limit
of 16 on a 64 CPU machine is not a workqueue wide limit of 16.

Convert between them as Tejun suggested [1]:

  U = unbound max_active, the system-wide target
  L = unbound min_active, the floor on each node
  P = percpu_max_active, the limit on each CPU
  N = online CPUs the workqueue is spread over, at least 1

  Setting P:
    U = min(P * N, WQ_MAX_ACTIVE)
    L = P

  Setting U:
    L = min(L, U)
    P = max(DIV_ROUND_UP(U, N), L)

  Setting L:
    L = clamp(L, 0, U)
    P = max(DIV_ROUND_UP(U, N), L)

For an unbound workqueue, N counts the online CPUs in its effective
cpumask, which wq_update_node_max_active() distributes max_active over.
For a percpu workqueue, N counts the online CPUs in wq_unbound_cpumask,
which an unbound scope would spread it over.

wq_rescale_max_active() does the conversion and publishes the limits
through wq_adjust_max_active(). The setters call it, and so does
apply_wqattrs_commit(), which runs when a workqueue is created and when
its attributes or the unbound cpumask change. workqueue_init_topology()
calls it too, as workqueues created before smp_init() counted only the
boot CPU.

N is not resampled on CPU hotplug, nor for a percpu workqueue when
wq_unbound_cpumask changes. Nothing reads the derived domain yet, and a
scope change goes through apply_wqattrs_commit(), which rederives it.

For a percpu workqueue, saved_max_active now holds the derived unbound
limit. start_flush_work() and max_active_show() want the limit the
workqueue was configured with, so they read it through
wq_user_max_active(). Both read it locklessly, so the saved limits are
written with WRITE_ONCE().

Suggested-by: Tejun Heo <tj@kernel.org>
Link: https://lore.kernel.org/all/ec6c22eddc8ce0b1ef2434ed62acd560@kernel.org/ [1]
Signed-off-by: Breno Leitao <leitao@debian.org>
---
 kernel/workqueue.c | 152 ++++++++++++++++++++++++++++++++++++++++++-----------
 1 file changed, 120 insertions(+), 32 deletions(-)

diff --git a/kernel/workqueue.c b/kernel/workqueue.c
index 826a64c6055d0c..492ff2133c41a2 100644
--- a/kernel/workqueue.c
+++ b/kernel/workqueue.c
@@ -393,6 +393,7 @@ struct workqueue_struct {
 	int			min_active;	/* WO: min active works */
 	int			saved_max_active; /* WQ: saved max_active */
 	int			saved_min_active; /* WQ: saved min_active */
+	int			saved_percpu_max_active; /* WQ: percpu limit */
 
 	struct workqueue_attrs	*attrs;	/* PW: workqueue attributes */
 	struct pool_workqueue __rcu *dfl_pwq;   /* PW: only for unbound wqs */
@@ -421,6 +422,14 @@ struct workqueue_struct {
 	struct wq_node_nr_active *node_nr_active[]; /* I: per-node nr_active */
 };
 
+static int wq_user_max_active(struct workqueue_struct *wq)
+{
+	if (wq->flags & WQ_UNBOUND)
+		return READ_ONCE(wq->saved_max_active);
+
+	return READ_ONCE(wq->saved_percpu_max_active);
+}
+
 /*
  * Each pod type describes how CPUs should be grouped for unbound workqueues.
  * See the comment above workqueue_attrs->affn_scope.
@@ -569,6 +578,7 @@ static int worker_thread(void *__worker);
 static void workqueue_sysfs_unregister(struct workqueue_struct *wq);
 static void show_pwq(struct pool_workqueue *pwq);
 static void show_one_worker_pool(struct worker_pool *pool);
+static void wq_rescale_max_active(struct workqueue_struct *wq);
 
 #define CREATE_TRACE_POINTS
 #include <trace/events/workqueue.h>
@@ -4549,7 +4559,7 @@ static bool start_flush_work(struct work_struct *work, struct wq_barrier *barr,
 	 * workqueues the deadlock happens when the rescuer stalls, blocking
 	 * forward progress.
 	 */
-	if (!from_cancel && (wq->saved_max_active == 1 || wq->rescuer))
+	if (!from_cancel && (wq_user_max_active(wq) == 1 || wq->rescuer))
 		touch_wq_lockdep_map(wq);
 
 	rcu_read_unlock();
@@ -5800,6 +5810,9 @@ static void apply_wqattrs_commit(struct apply_wqattrs_ctx *ctx)
 	if (ctx->wq->flags & WQ_UNBOUND)
 		wq_update_node_max_active(ctx->wq, -1);
 
+	/* derive the other max_active domain for the new effective cpumask */
+	wq_rescale_max_active(ctx->wq);
+
 	mutex_unlock(&ctx->wq->mutex);
 }
 
@@ -6031,6 +6044,42 @@ static int init_rescuer(struct workqueue_struct *wq)
 	return 0;
 }
 
+static int wq_nr_online_cpus(const struct cpumask *mask)
+{
+	return max(cpumask_weight_and(mask, cpu_online_mask), 1);
+}
+
+/* online CPUs an unbound backend of @wq would spread max_active over */
+static int wq_nr_effective_cpus(struct workqueue_struct *wq)
+{
+	lockdep_assert_held(&wq->mutex);
+
+	if (!(wq->flags & WQ_UNBOUND))
+		return wq_nr_online_cpus(wq_unbound_cpumask);
+
+	return wq_nr_online_cpus(unbound_effective_cpumask(wq));
+}
+
+static int wq_scale_to_unbound(int percpu_max, int nr_cpus)
+{
+	return min_t(s64, (s64)percpu_max * nr_cpus, WQ_MAX_ACTIVE);
+}
+
+static int wq_scale_to_percpu(int max_active, int min_active, int nr_cpus)
+{
+	return max(DIV_ROUND_UP(max_active, nr_cpus), min_active);
+}
+
+static void wq_init_bh_max_active(struct workqueue_struct *wq)
+{
+	wq->max_active = INT_MAX;
+	wq->min_active = INT_MAX;
+	wq->percpu_max_active = INT_MAX;
+	wq->saved_max_active = INT_MAX;
+	wq->saved_min_active = INT_MAX;
+	wq->saved_percpu_max_active = INT_MAX;
+}
+
 static void wq_init_max_active(struct workqueue_struct *wq, int max_active)
 {
 	int effective_max_active;
@@ -6040,65 +6089,69 @@ static void wq_init_max_active(struct workqueue_struct *wq, int max_active)
 		 * BH workqueues always share a single execution context per CPU
 		 * and don't impose any max_active limit.
 		 */
-		effective_max_active = INT_MAX;
-	} else {
-		effective_max_active = max_active ?: WQ_DFL_ACTIVE;
-		effective_max_active = wq_clamp_max_active(effective_max_active,
-							   wq->flags, wq->name);
+		wq_init_bh_max_active(wq);
+		return;
 	}
 
+	effective_max_active = max_active ?: WQ_DFL_ACTIVE;
+	effective_max_active = wq_clamp_max_active(effective_max_active,
+						   wq->flags, wq->name);
+
+	/* apply_wqattrs_commit() derives the other domain */
 	if (wq->flags & WQ_UNBOUND) {
 		wq->max_active = effective_max_active;
 		wq->min_active = min(effective_max_active, WQ_DFL_MIN_ACTIVE);
 		wq->saved_min_active = wq->min_active;
+		wq->saved_max_active = effective_max_active;
 	} else {
 		wq->percpu_max_active = effective_max_active;
+		wq->saved_percpu_max_active = effective_max_active;
 	}
-	wq->saved_max_active = effective_max_active;
 }
 
 /**
  * wq_adjust_max_active - update a wq's max_active to the current setting
  * @wq: target workqueue
  *
- * If @wq isn't freezing, set the limit that applies to @wq's backing to the
- * saved_max_active and activate inactive work items accordingly. If @wq is
- * freezing, clear it to zero.
+ * If @wq isn't freezing, publish the saved limits of both domains and activate
+ * inactive work items accordingly. If @wq is freezing, clear them to zero.
  */
 static void wq_adjust_max_active(struct workqueue_struct *wq)
 {
+	int new_max, new_min, new_percpu_max;
 	bool activated;
-	int new_max, new_min;
 
 	lockdep_assert_held(&wq->mutex);
 
 	if ((wq->flags & WQ_FREEZABLE) && workqueue_freezing) {
 		new_max = 0;
 		new_min = 0;
+		new_percpu_max = 0;
 	} else {
 		new_max = wq->saved_max_active;
 		new_min = wq->saved_min_active;
+		new_percpu_max = wq->saved_percpu_max_active;
 	}
 
 	/*
-	 * Update the limit and then kick inactive work items if more active
+	 * Update both domains and then kick inactive work items if more active
 	 * work items are allowed. This doesn't break work item ordering
 	 * because new work items are always queued behind existing inactive
 	 * work items if there are any.
+	 *
+	 * Both domains are kept current so that a pwq on either kind of pool
+	 * has a valid limit.
 	 */
-	if (wq->flags & WQ_UNBOUND) {
-		if (wq->max_active == new_max && wq->min_active == new_min)
-			return;
+	if (wq->max_active == new_max && wq->min_active == new_min &&
+	    wq->percpu_max_active == new_percpu_max)
+		return;
 
-		WRITE_ONCE(wq->max_active, new_max);
-		WRITE_ONCE(wq->min_active, new_min);
-		wq_update_node_max_active(wq, -1);
-	} else {
-		if (wq->percpu_max_active == new_max)
-			return;
+	WRITE_ONCE(wq->max_active, new_max);
+	WRITE_ONCE(wq->min_active, new_min);
+	WRITE_ONCE(wq->percpu_max_active, new_percpu_max);
 
-		WRITE_ONCE(wq->percpu_max_active, new_max);
-	}
+	if (wq->flags & WQ_UNBOUND)
+		wq_update_node_max_active(wq, -1);
 
 	if (new_max == 0)
 		return;
@@ -6125,6 +6178,37 @@ static void wq_adjust_max_active(struct workqueue_struct *wq)
 	} while (activated);
 }
 
+/*
+ * Derive the saved limits of the max_active domain @wq's backend doesn't use
+ * from those of the one it does, with the current number of usable CPUs, and
+ * publish both.
+ */
+static void wq_rescale_max_active(struct workqueue_struct *wq)
+{
+	int nr_cpus;
+
+	lockdep_assert_held(&wq->mutex);
+
+	/* BH workqueues have no max_active limit in either domain */
+	if (wq->flags & WQ_BH)
+		return;
+
+	nr_cpus = wq_nr_effective_cpus(wq);
+
+	if (wq->flags & WQ_UNBOUND) {
+		WRITE_ONCE(wq->saved_percpu_max_active,
+			   wq_scale_to_percpu(wq->saved_max_active,
+					      wq->saved_min_active, nr_cpus));
+	} else {
+		WRITE_ONCE(wq->saved_max_active,
+			   wq_scale_to_unbound(wq->saved_percpu_max_active,
+					       nr_cpus));
+		wq->saved_min_active = wq->saved_percpu_max_active;
+	}
+
+	wq_adjust_max_active(wq);
+}
+
 __printf(1, 0)
 static struct workqueue_struct *__alloc_workqueue(const char *fmt,
 						  unsigned int flags,
@@ -6459,11 +6543,14 @@ void workqueue_set_max_active(struct workqueue_struct *wq, int max_active)
 
 	mutex_lock(&wq->mutex);
 
-	wq->saved_max_active = max_active;
-	if (wq->flags & WQ_UNBOUND)
+	if (wq->flags & WQ_UNBOUND) {
+		WRITE_ONCE(wq->saved_max_active, max_active);
 		wq->saved_min_active = min(wq->saved_min_active, max_active);
+	} else {
+		WRITE_ONCE(wq->saved_percpu_max_active, max_active);
+	}
 
-	wq_adjust_max_active(wq);
+	wq_rescale_max_active(wq);
 
 	mutex_unlock(&wq->mutex);
 }
@@ -6492,7 +6579,7 @@ void workqueue_set_min_active(struct workqueue_struct *wq, int min_active)
 
 	mutex_lock(&wq->mutex);
 	wq->saved_min_active = clamp(min_active, 0, wq->saved_max_active);
-	wq_adjust_max_active(wq);
+	wq_rescale_max_active(wq);
 	mutex_unlock(&wq->mutex);
 }
 
@@ -7587,7 +7674,7 @@ static ssize_t max_active_show(struct device *dev,
 {
 	struct workqueue_struct *wq = dev_to_wq(dev);
 
-	return scnprintf(buf, PAGE_SIZE, "%d\n", wq->saved_max_active);
+	return scnprintf(buf, PAGE_SIZE, "%d\n", wq_user_max_active(wq));
 }
 
 static ssize_t max_active_store(struct device *dev,
@@ -8822,11 +8909,12 @@ void __init workqueue_init_topology(void)
 	list_for_each_entry(wq, &workqueues, list) {
 		for_each_online_cpu(cpu)
 			unbound_wq_update_pwq(wq, cpu);
-		if (wq->flags & WQ_UNBOUND) {
-			mutex_lock(&wq->mutex);
+		mutex_lock(&wq->mutex);
+		if (wq->flags & WQ_UNBOUND)
 			wq_update_node_max_active(wq, -1);
-			mutex_unlock(&wq->mutex);
-		}
+		/* early workqueues counted only the boot CPU */
+		wq_rescale_max_active(wq);
+		mutex_unlock(&wq->mutex);
 	}
 
 	mutex_unlock(&wq_pool_mutex);

-- 
2.53.0-Meta