Re: [RFC PATCH 07/10] rcu: Wake NOCB rcuog kthreads on expedited grace period completion

[Frederic Weisbecker <frederic@kernel.org>]

Le Fri, Apr 17, 2026 at 04:11:55PM -0700, Puranjay Mohan a écrit :
> When an expedited grace period completes, rcu_exp_wait_wake() wakes
> waiters on rnp->exp_wq[] but does not notify NOCB rcuog kthreads.  These
> kthreads may be sleeping waiting for a grace period to complete.
> Without this wakeup, callbacks on offloaded CPUs that could benefit from
> the expedited GP must wait until the rcuog kthread wakes for some other
> reason (e.g., next normal GP completion or a timer).
> 
> Add rcu_exp_wake_nocb() which wakes rcuog kthreads for leaf-node CPUs,
> deduplicating via rdp->nocb_gp_rdp since multiple CPUs share one rcuog
> kthread.  Uses for_each_leaf_node_possible_cpu() because offline CPUs
> can have pending callbacks. The function is defined in tree_nocb.h with
> an empty stub for CONFIG_RCU_NOCB_CPU=n builds.
> 
> Reviewed-by: Paul E. McKenney <paulmck@kernel.org>
> Signed-off-by: Puranjay Mohan <puranjay@kernel.org>
> ---
>  kernel/rcu/tree.h      |  1 +
>  kernel/rcu/tree_exp.h  |  1 +
>  kernel/rcu/tree_nocb.h | 29 +++++++++++++++++++++++++++++
>  3 files changed, 31 insertions(+)
> 
> diff --git a/kernel/rcu/tree.h b/kernel/rcu/tree.h
> index 7dfc57e9adb1..40f778453591 100644
> --- a/kernel/rcu/tree.h
> +++ b/kernel/rcu/tree.h
> @@ -500,6 +500,7 @@ static struct swait_queue_head *rcu_nocb_gp_get(struct rcu_node *rnp);
>  static void rcu_nocb_gp_cleanup(struct swait_queue_head *sq);
>  static void rcu_init_one_nocb(struct rcu_node *rnp);
>  static bool wake_nocb_gp(struct rcu_data *rdp);
> +static void rcu_exp_wake_nocb(struct rcu_node *rnp);
>  static bool rcu_nocb_flush_bypass(struct rcu_data *rdp, struct rcu_head *rhp,
>  				  unsigned long j, bool lazy);
>  static void call_rcu_nocb(struct rcu_data *rdp, struct rcu_head *head,
> diff --git a/kernel/rcu/tree_exp.h b/kernel/rcu/tree_exp.h
> index 82cada459e5d..0df1009c6e97 100644
> --- a/kernel/rcu/tree_exp.h
> +++ b/kernel/rcu/tree_exp.h
> @@ -708,6 +708,7 @@ static void rcu_exp_wait_wake(unsigned long s)
>  		}
>  		smp_mb(); /* All above changes before wakeup. */
>  		wake_up_all(&rnp->exp_wq[rcu_seq_ctr(s) & 0x3]);
> +		rcu_exp_wake_nocb(rnp);
>  	}
>  	trace_rcu_exp_grace_period(rcu_state.name, s, TPS("endwake"));
>  	mutex_unlock(&rcu_state.exp_wake_mutex);
> diff --git a/kernel/rcu/tree_nocb.h b/kernel/rcu/tree_nocb.h
> index 7462cd5e2507..f37ee56d62a9 100644
> --- a/kernel/rcu/tree_nocb.h
> +++ b/kernel/rcu/tree_nocb.h
> @@ -190,6 +190,31 @@ static void rcu_init_one_nocb(struct rcu_node *rnp)
>  	init_swait_queue_head(&rnp->nocb_gp_wq[1]);
>  }
>  
> +/*
> + * Wake NOCB rcuog kthreads for leaf-node CPUs so that they can advance
> + * callbacks that were waiting for the just-completed expedited GP.
> + * Deduplicate via nocb_gp_rdp since multiple CPUs share one rcuog
> + * kthread.  Use for_each_leaf_node_possible_cpu() because offline CPUs
> + * may have pending callbacks.
> + */
> +static void rcu_exp_wake_nocb(struct rcu_node *rnp)

Please consolidate the naming to match rcu_nocb_gp_cleanup().

> +{
> +	struct rcu_data *last_rdp_gp = NULL;
> +	int cpu;
> +
> +	if (!rcu_is_leaf_node(rnp))
> +		return;
> +
> +	for_each_leaf_node_possible_cpu(rnp, cpu) {
> +		struct rcu_data *rdp = per_cpu_ptr(&rcu_data, cpu);
> +
> +		if (rdp->nocb_gp_rdp == last_rdp_gp)
> +			continue;
> +		last_rdp_gp = rdp->nocb_gp_rdp;
> +		wake_nocb_gp(rdp);
> +	}

There are two waitqueues for rcuog wake-ups:

1) rdp->rdp_gp->nocb_gp_wq: to wait for callbacks on the queue
2) rnp->nocb_gp_wq: to wait for grace periods

So you're waking up the wrong one.

Something like the below? (untested)

diff --git a/kernel/rcu/tree.c b/kernel/rcu/tree.c
index 0e43866dc4cd..436e12e313c2 100644
--- a/kernel/rcu/tree.c
+++ b/kernel/rcu/tree.c
@@ -2193,8 +2193,13 @@ static noinline void rcu_gp_cleanup(void)
 			dump_blkd_tasks(rnp, 10);
 		WARN_ON_ONCE(rnp->qsmask);
 		WRITE_ONCE(rnp->gp_seq, new_gp_seq);
-		if (!rnp->parent)
-			smp_mb(); // Order against failing poll_state_synchronize_rcu_full().
+		if (!rnp->parent) {
+			/*
+			 * Order against failing poll_state_synchronize_rcu_full().
+			 * and also rcu_nocb_cleanup_wake() -> swait_active()
+			 */
+			smp_mb();
+		}
 		rdp = this_cpu_ptr(&rcu_data);
 		if (rnp == rdp->mynode)
 			needgp = __note_gp_changes(rnp, rdp) || needgp;
diff --git a/kernel/rcu/tree.h b/kernel/rcu/tree.h
index 40f778453591..8f272cb4e4f3 100644
--- a/kernel/rcu/tree.h
+++ b/kernel/rcu/tree.h
@@ -253,7 +253,7 @@ struct rcu_data {
 	u8 nocb_gp_sleep;		/* Is the nocb GP thread asleep? */
 	u8 nocb_gp_bypass;		/* Found a bypass on last scan? */
 	u8 nocb_gp_gp;			/* GP to wait for on last scan? */
-	unsigned long nocb_gp_seq;	/*  If so, ->gp_seq to wait for. */
+	struct rcu_gp_oldstate nocb_gp_seq; /*  If so, ->gp_seq to wait for. */
 	unsigned long nocb_gp_loops;	/* # passes through wait code. */
 	struct swait_queue_head nocb_gp_wq; /* For nocb kthreads to sleep on. */
 	bool nocb_cb_sleep;		/* Is the nocb CB thread asleep? */
@@ -498,9 +498,9 @@ static bool rcu_preempt_need_deferred_qs(struct task_struct *t);
 static void zero_cpu_stall_ticks(struct rcu_data *rdp);
 static struct swait_queue_head *rcu_nocb_gp_get(struct rcu_node *rnp);
 static void rcu_nocb_gp_cleanup(struct swait_queue_head *sq);
+static void rcu_nocb_exp_cleanup(struct rcu_node *rnp);
 static void rcu_init_one_nocb(struct rcu_node *rnp);
 static bool wake_nocb_gp(struct rcu_data *rdp);
-static void rcu_exp_wake_nocb(struct rcu_node *rnp);
 static bool rcu_nocb_flush_bypass(struct rcu_data *rdp, struct rcu_head *rhp,
 				  unsigned long j, bool lazy);
 static void call_rcu_nocb(struct rcu_data *rdp, struct rcu_head *head,
diff --git a/kernel/rcu/tree_exp.h b/kernel/rcu/tree_exp.h
index 0df1009c6e97..43c167a8a145 100644
--- a/kernel/rcu/tree_exp.h
+++ b/kernel/rcu/tree_exp.h
@@ -708,7 +708,8 @@ static void rcu_exp_wait_wake(unsigned long s)
 		}
 		smp_mb(); /* All above changes before wakeup. */
 		wake_up_all(&rnp->exp_wq[rcu_seq_ctr(s) & 0x3]);
-		rcu_exp_wake_nocb(rnp);
+		if (rcu_is_leaf_node(rnp))
+			rcu_nocb_exp_cleanup(rnp);
 	}
 	trace_rcu_exp_grace_period(rcu_state.name, s, TPS("endwake"));
 	mutex_unlock(&rcu_state.exp_wake_mutex);
diff --git a/kernel/rcu/tree_nocb.h b/kernel/rcu/tree_nocb.h
index f37ee56d62a9..60d4182b9509 100644
--- a/kernel/rcu/tree_nocb.h
+++ b/kernel/rcu/tree_nocb.h
@@ -170,13 +170,59 @@ static void rcu_lockdep_assert_cblist_protected(struct rcu_data *rdp)
 		lockdep_assert_held(&rdp->nocb_lock);
 }
 
+static void rcu_nocb_cleanup_wake(struct swait_queue_head *sq)
+{
+	if (swait_active(sq))
+		swake_up_all(sq);
+}
+
 /*
  * Wake up any no-CBs CPUs' kthreads that were waiting on the just-ended
  * grace period.
  */
 static void rcu_nocb_gp_cleanup(struct swait_queue_head *sq)
 {
-	swake_up_all(sq);
+	/*
+	 * swait active() can be checked first because:
+	 *
+	 * rcu_gp_cleanup()                              nocb_gp_wait()
+	 * ---------------                               --------------
+	 * WRITE_ONCE(root->gp_seq, new_gp_seq);         swait_event_interruptible_exclusive(sq)
+	 * smp_mb()                                         prepare_to_swait()
+	 * if swait_active(sq)                                 list_add_tail(&wait->task_list, &q->task_list);
+	 *    swake_up_all(sq)                                 set_current_state()
+	 *                                                       smp_mb()
+	 *                                                  if (poll_state_synchronize_rcu_full())
+	 *                                                     if (rcu_seq_done_exact(&root->gp_seq, rgosp->rgos_norm))
+	 *                                                        ...
+	 */
+	rcu_nocb_cleanup_wake(sq);
+}
+
+/*
+ * Wake NOCB rcuog kthreads for leaf-node CPUs so that they can advance
+ * callbacks that were waiting for the just-completed expedited GP.
+ * Wake-up waitqueues for both even and odd GP numbers because exp and
+ * normal sequences don't match.
+ */
+static void rcu_nocb_exp_cleanup(struct rcu_node *rnp)
+{
+/*
+ * swait active() can be checked first because:
+ *
+ * rcu_exp_wait_wake()                           nocb_gp_wait()
+ * ---------------                               --------------
+ * rcu_seq_end(&rcu_state.expedited_sequence);   swait_event_interruptible_exclusive(sq)
+ * smp_mb()                                         prepare_to_swait()
+ * if swait_active(sq)                                 list_add_tail(&wait->task_list, &q->task_list);
+ *    swake_up_all(sq)                                 set_current_state()
+ *                                                        smp_mb()
+ *                                                  if (poll_state_synchronize_rcu_full())
+ *                                                     if (rcu_seq_done_exact(&rcu_state.expedited_sequence, rgosp->rgos_exp))
+ *                                                        ...
+ */
+	rcu_nocb_cleanup_wake(&rnp->nocb_gp_wq[0]);
+	rcu_nocb_cleanup_wake(&rnp->nocb_gp_wq[1]);
 }
 
 static struct swait_queue_head *rcu_nocb_gp_get(struct rcu_node *rnp)
@@ -190,31 +236,6 @@ static void rcu_init_one_nocb(struct rcu_node *rnp)
 	init_swait_queue_head(&rnp->nocb_gp_wq[1]);
 }
 
-/*
- * Wake NOCB rcuog kthreads for leaf-node CPUs so that they can advance
- * callbacks that were waiting for the just-completed expedited GP.
- * Deduplicate via nocb_gp_rdp since multiple CPUs share one rcuog
- * kthread.  Use for_each_leaf_node_possible_cpu() because offline CPUs
- * may have pending callbacks.
- */
-static void rcu_exp_wake_nocb(struct rcu_node *rnp)
-{
-	struct rcu_data *last_rdp_gp = NULL;
-	int cpu;
-
-	if (!rcu_is_leaf_node(rnp))
-		return;
-
-	for_each_leaf_node_possible_cpu(rnp, cpu) {
-		struct rcu_data *rdp = per_cpu_ptr(&rcu_data, cpu);
-
-		if (rdp->nocb_gp_rdp == last_rdp_gp)
-			continue;
-		last_rdp_gp = rdp->nocb_gp_rdp;
-		wake_nocb_gp(rdp);
-	}
-}
-
 /* Clear any pending deferred wakeup timer (nocb_gp_lock must be held). */
 static void nocb_defer_wakeup_cancel(struct rcu_data *rdp_gp)
 {
@@ -684,7 +705,7 @@ static void nocb_gp_wait(struct rcu_data *my_rdp)
 {
 	bool bypass = false;
 	int __maybe_unused cpu = my_rdp->cpu;
-	struct rcu_gp_oldstate cur_gp_seq_full;
+	struct rcu_gp_oldstate wait_gp_seq = {0}; //remove uninitialized warning
 	unsigned long flags;
 	bool gotcbs = false;
 	unsigned long j = jiffies;
@@ -694,7 +715,6 @@ static void nocb_gp_wait(struct rcu_data *my_rdp)
 	bool needwake_gp;
 	struct rcu_data *rdp, *rdp_toggling = NULL;
 	struct rcu_node *rnp;
-	unsigned long wait_gp_seq = 0; // Suppress "use uninitialized" warning.
 	bool wasempty = false;
 
 	/*
@@ -718,6 +738,7 @@ static void nocb_gp_wait(struct rcu_data *my_rdp)
 	 * won't be ignored for long.
 	 */
 	list_for_each_entry(rdp, &my_rdp->nocb_head_rdp, nocb_entry_rdp) {
+		struct rcu_gp_oldstate cur_gp_seq;
 		long bypass_ncbs;
 		bool flush_bypass = false;
 		long lazy_ncbs;
@@ -755,8 +776,8 @@ static void nocb_gp_wait(struct rcu_data *my_rdp)
 		needwake_gp = false;
 		if (!rcu_segcblist_restempty(&rdp->cblist,
 					     RCU_NEXT_READY_TAIL) ||
-		    (rcu_segcblist_nextgp(&rdp->cblist, &cur_gp_seq_full) &&
-		     poll_state_synchronize_rcu_full(&cur_gp_seq_full))) {
+		    (rcu_segcblist_nextgp(&rdp->cblist, &cur_gp_seq) &&
+		     poll_state_synchronize_rcu_full(&cur_gp_seq))) {
 			raw_spin_lock_rcu_node(rnp); /* irqs disabled. */
 			needwake_gp = rcu_advance_cbs(rnp, rdp);
 			wasempty = rcu_segcblist_restempty(&rdp->cblist,
@@ -777,11 +798,15 @@ static void nocb_gp_wait(struct rcu_data *my_rdp)
 		 * numbers from rcu_accelerate_cbs() inside
 		 * rcu_advance_cbs() and will be handled on the next pass.
 		 */
-		if (rcu_segcblist_nextgp(&rdp->cblist, &cur_gp_seq_full) &&
-		    !poll_state_synchronize_rcu_full(&cur_gp_seq_full)) {
+		if (rcu_segcblist_nextgp(&rdp->cblist, &cur_gp_seq) &&
+		    !poll_state_synchronize_rcu_full(&cur_gp_seq)) {
+			if (!needwait_gp ||
+			    ULONG_CMP_LT(cur_gp_seq.rgos_norm, wait_gp_seq.rgos_norm))
+				wait_gp_seq.rgos_norm = cur_gp_seq.rgos_norm;
 			if (!needwait_gp ||
-			    ULONG_CMP_LT(cur_gp_seq_full.rgos_norm, wait_gp_seq))
-				wait_gp_seq = cur_gp_seq_full.rgos_norm;
+			    ULONG_CMP_LT(cur_gp_seq.rgos_exp, wait_gp_seq.rgos_exp))
+				wait_gp_seq.rgos_exp = cur_gp_seq.rgos_exp;
+
 			needwait_gp = true;
 			trace_rcu_nocb_wake(rcu_state.name, rdp->cpu,
 					    TPS("NeedWaitGP"));
@@ -803,7 +828,8 @@ static void nocb_gp_wait(struct rcu_data *my_rdp)
 
 	my_rdp->nocb_gp_bypass = bypass;
 	my_rdp->nocb_gp_gp = needwait_gp;
-	my_rdp->nocb_gp_seq = needwait_gp ? wait_gp_seq : 0;
+	if (needwait_gp)
+		my_rdp->nocb_gp_seq = wait_gp_seq;
 
 	// At least one child with non-empty ->nocb_bypass, so set
 	// timer in order to avoid stranding its callbacks.
@@ -838,12 +864,12 @@ static void nocb_gp_wait(struct rcu_data *my_rdp)
 		nocb_gp_sleep(my_rdp, cpu);
 	} else {
 		rnp = my_rdp->mynode;
-		trace_rcu_this_gp(rnp, my_rdp, wait_gp_seq, TPS("StartWait"));
+		trace_rcu_this_gp(rnp, my_rdp, wait_gp_seq.rgos_norm, TPS("StartWait"));
 		swait_event_interruptible_exclusive(
-			rnp->nocb_gp_wq[rcu_seq_ctr(wait_gp_seq) & 0x1],
-			rcu_seq_done(&rnp->gp_seq, wait_gp_seq) ||
+			rnp->nocb_gp_wq[rcu_seq_ctr(wait_gp_seq.rgos_norm) & 0x1],
+			poll_state_synchronize_rcu_full(&wait_gp_seq) ||
 			!READ_ONCE(my_rdp->nocb_gp_sleep));
-		trace_rcu_this_gp(rnp, my_rdp, wait_gp_seq, TPS("EndWait"));
+		trace_rcu_this_gp(rnp, my_rdp, wait_gp_seq.rgos_norm, TPS("EndWait"));
 	}
 
 	if (!rcu_nocb_poll) {
@@ -877,7 +903,8 @@ static void nocb_gp_wait(struct rcu_data *my_rdp)
 		swake_up_one(&rdp_toggling->nocb_state_wq);
 	}
 
-	my_rdp->nocb_gp_seq = -1;
+	my_rdp->nocb_gp_seq.rgos_norm = -1;
+	my_rdp->nocb_gp_seq.rgos_exp = -1;
 	WARN_ON(signal_pending(current));
 }
 
@@ -1561,7 +1588,7 @@ static void show_rcu_nocb_gp_state(struct rcu_data *rdp)
 {
 	struct rcu_node *rnp = rdp->mynode;
 
-	pr_info("nocb GP %d %c%c%c%c%c %c[%c%c] %c%c:%ld rnp %d:%d %lu %c CPU %d%s\n",
+	pr_info("nocb GP %d %c%c%c%c%c %c[%c%c] %c%c:%ld/%ld rnp %d:%d %lu %c CPU %d%s\n",
 		rdp->cpu,
 		"kK"[!!rdp->nocb_gp_kthread],
 		"lL"[raw_spin_is_locked(&rdp->nocb_gp_lock)],
@@ -1573,7 +1600,8 @@ static void show_rcu_nocb_gp_state(struct rcu_data *rdp)
 		".W"[swait_active(&rnp->nocb_gp_wq[1])],
 		".B"[!!rdp->nocb_gp_bypass],
 		".G"[!!rdp->nocb_gp_gp],
-		(long)rdp->nocb_gp_seq,
+		(long)rdp->nocb_gp_seq.rgos_norm,
+		(long)rdp->nocb_gp_seq.rgos_exp,
 		rnp->grplo, rnp->grphi, READ_ONCE(rdp->nocb_gp_loops),
 		rdp->nocb_gp_kthread ? task_state_to_char(rdp->nocb_gp_kthread) : '.',
 		rdp->nocb_gp_kthread ? (int)task_cpu(rdp->nocb_gp_kthread) : -1,
@@ -1684,16 +1712,16 @@ static void rcu_nocb_gp_cleanup(struct swait_queue_head *sq)
 {
 }
 
-static struct swait_queue_head *rcu_nocb_gp_get(struct rcu_node *rnp)
+static void rcu_nocb_exp_cleanup(struct rcu_node *rnp)
 {
-	return NULL;
 }
 
-static void rcu_init_one_nocb(struct rcu_node *rnp)
+static struct swait_queue_head *rcu_nocb_gp_get(struct rcu_node *rnp)
 {
+	return NULL;
 }
 
-static void rcu_exp_wake_nocb(struct rcu_node *rnp)
+static void rcu_init_one_nocb(struct rcu_node *rnp)
 {
 }