xref: /openbmc/linux/kernel/rcu/tree_nocb.h (revision 7b7fd0ac7dc1ffcaf24d9bca0f051b0168e43cd4)
1  /* SPDX-License-Identifier: GPL-2.0+ */
2  /*
3   * Read-Copy Update mechanism for mutual exclusion (tree-based version)
4   * Internal non-public definitions that provide either classic
5   * or preemptible semantics.
6   *
7   * Copyright Red Hat, 2009
8   * Copyright IBM Corporation, 2009
9   * Copyright SUSE, 2021
10   *
11   * Author: Ingo Molnar <mingo@elte.hu>
12   *	   Paul E. McKenney <paulmck@linux.ibm.com>
13   *	   Frederic Weisbecker <frederic@kernel.org>
14   */
15  
16  #ifdef CONFIG_RCU_NOCB_CPU
17  static cpumask_var_t rcu_nocb_mask; /* CPUs to have callbacks offloaded. */
18  static bool __read_mostly rcu_nocb_poll;    /* Offload kthread are to poll. */
rcu_lockdep_is_held_nocb(struct rcu_data * rdp)19  static inline int rcu_lockdep_is_held_nocb(struct rcu_data *rdp)
20  {
21  	return lockdep_is_held(&rdp->nocb_lock);
22  }
23  
rcu_current_is_nocb_kthread(struct rcu_data * rdp)24  static inline bool rcu_current_is_nocb_kthread(struct rcu_data *rdp)
25  {
26  	/* Race on early boot between thread creation and assignment */
27  	if (!rdp->nocb_cb_kthread || !rdp->nocb_gp_kthread)
28  		return true;
29  
30  	if (current == rdp->nocb_cb_kthread || current == rdp->nocb_gp_kthread)
31  		if (in_task())
32  			return true;
33  	return false;
34  }
35  
36  /*
37   * Offload callback processing from the boot-time-specified set of CPUs
38   * specified by rcu_nocb_mask.  For the CPUs in the set, there are kthreads
39   * created that pull the callbacks from the corresponding CPU, wait for
40   * a grace period to elapse, and invoke the callbacks.  These kthreads
41   * are organized into GP kthreads, which manage incoming callbacks, wait for
42   * grace periods, and awaken CB kthreads, and the CB kthreads, which only
43   * invoke callbacks.  Each GP kthread invokes its own CBs.  The no-CBs CPUs
44   * do a wake_up() on their GP kthread when they insert a callback into any
45   * empty list, unless the rcu_nocb_poll boot parameter has been specified,
46   * in which case each kthread actively polls its CPU.  (Which isn't so great
47   * for energy efficiency, but which does reduce RCU's overhead on that CPU.)
48   *
49   * This is intended to be used in conjunction with Frederic Weisbecker's
50   * adaptive-idle work, which would seriously reduce OS jitter on CPUs
51   * running CPU-bound user-mode computations.
52   *
53   * Offloading of callbacks can also be used as an energy-efficiency
54   * measure because CPUs with no RCU callbacks queued are more aggressive
55   * about entering dyntick-idle mode.
56   */
57  
58  
59  /*
60   * Parse the boot-time rcu_nocb_mask CPU list from the kernel parameters.
61   * If the list is invalid, a warning is emitted and all CPUs are offloaded.
62   */
rcu_nocb_setup(char * str)63  static int __init rcu_nocb_setup(char *str)
64  {
65  	alloc_bootmem_cpumask_var(&rcu_nocb_mask);
66  	if (*str == '=') {
67  		if (cpulist_parse(++str, rcu_nocb_mask)) {
68  			pr_warn("rcu_nocbs= bad CPU range, all CPUs set\n");
69  			cpumask_setall(rcu_nocb_mask);
70  		}
71  	}
72  	rcu_state.nocb_is_setup = true;
73  	return 1;
74  }
75  __setup("rcu_nocbs", rcu_nocb_setup);
76  
parse_rcu_nocb_poll(char * arg)77  static int __init parse_rcu_nocb_poll(char *arg)
78  {
79  	rcu_nocb_poll = true;
80  	return 1;
81  }
82  __setup("rcu_nocb_poll", parse_rcu_nocb_poll);
83  
84  /*
85   * Don't bother bypassing ->cblist if the call_rcu() rate is low.
86   * After all, the main point of bypassing is to avoid lock contention
87   * on ->nocb_lock, which only can happen at high call_rcu() rates.
88   */
89  static int nocb_nobypass_lim_per_jiffy = 16 * 1000 / HZ;
90  module_param(nocb_nobypass_lim_per_jiffy, int, 0);
91  
92  /*
93   * Acquire the specified rcu_data structure's ->nocb_bypass_lock.  If the
94   * lock isn't immediately available, perform minimal sanity check.
95   */
rcu_nocb_bypass_lock(struct rcu_data * rdp)96  static void rcu_nocb_bypass_lock(struct rcu_data *rdp)
97  	__acquires(&rdp->nocb_bypass_lock)
98  {
99  	lockdep_assert_irqs_disabled();
100  	if (raw_spin_trylock(&rdp->nocb_bypass_lock))
101  		return;
102  	/*
103  	 * Contention expected only when local enqueue collide with
104  	 * remote flush from kthreads.
105  	 */
106  	WARN_ON_ONCE(smp_processor_id() != rdp->cpu);
107  	raw_spin_lock(&rdp->nocb_bypass_lock);
108  }
109  
110  /*
111   * Conditionally acquire the specified rcu_data structure's
112   * ->nocb_bypass_lock.
113   */
rcu_nocb_bypass_trylock(struct rcu_data * rdp)114  static bool rcu_nocb_bypass_trylock(struct rcu_data *rdp)
115  {
116  	lockdep_assert_irqs_disabled();
117  	return raw_spin_trylock(&rdp->nocb_bypass_lock);
118  }
119  
120  /*
121   * Release the specified rcu_data structure's ->nocb_bypass_lock.
122   */
rcu_nocb_bypass_unlock(struct rcu_data * rdp)123  static void rcu_nocb_bypass_unlock(struct rcu_data *rdp)
124  	__releases(&rdp->nocb_bypass_lock)
125  {
126  	lockdep_assert_irqs_disabled();
127  	raw_spin_unlock(&rdp->nocb_bypass_lock);
128  }
129  
130  /*
131   * Acquire the specified rcu_data structure's ->nocb_lock, but only
132   * if it corresponds to a no-CBs CPU.
133   */
rcu_nocb_lock(struct rcu_data * rdp)134  static void rcu_nocb_lock(struct rcu_data *rdp)
135  {
136  	lockdep_assert_irqs_disabled();
137  	if (!rcu_rdp_is_offloaded(rdp))
138  		return;
139  	raw_spin_lock(&rdp->nocb_lock);
140  }
141  
142  /*
143   * Release the specified rcu_data structure's ->nocb_lock, but only
144   * if it corresponds to a no-CBs CPU.
145   */
rcu_nocb_unlock(struct rcu_data * rdp)146  static void rcu_nocb_unlock(struct rcu_data *rdp)
147  {
148  	if (rcu_rdp_is_offloaded(rdp)) {
149  		lockdep_assert_irqs_disabled();
150  		raw_spin_unlock(&rdp->nocb_lock);
151  	}
152  }
153  
154  /*
155   * Release the specified rcu_data structure's ->nocb_lock and restore
156   * interrupts, but only if it corresponds to a no-CBs CPU.
157   */
rcu_nocb_unlock_irqrestore(struct rcu_data * rdp,unsigned long flags)158  static void rcu_nocb_unlock_irqrestore(struct rcu_data *rdp,
159  				       unsigned long flags)
160  {
161  	if (rcu_rdp_is_offloaded(rdp)) {
162  		lockdep_assert_irqs_disabled();
163  		raw_spin_unlock_irqrestore(&rdp->nocb_lock, flags);
164  	} else {
165  		local_irq_restore(flags);
166  	}
167  }
168  
169  /* Lockdep check that ->cblist may be safely accessed. */
rcu_lockdep_assert_cblist_protected(struct rcu_data * rdp)170  static void rcu_lockdep_assert_cblist_protected(struct rcu_data *rdp)
171  {
172  	lockdep_assert_irqs_disabled();
173  	if (rcu_rdp_is_offloaded(rdp))
174  		lockdep_assert_held(&rdp->nocb_lock);
175  }
176  
177  /*
178   * Wake up any no-CBs CPUs' kthreads that were waiting on the just-ended
179   * grace period.
180   */
rcu_nocb_gp_cleanup(struct swait_queue_head * sq)181  static void rcu_nocb_gp_cleanup(struct swait_queue_head *sq)
182  {
183  	swake_up_all(sq);
184  }
185  
rcu_nocb_gp_get(struct rcu_node * rnp)186  static struct swait_queue_head *rcu_nocb_gp_get(struct rcu_node *rnp)
187  {
188  	return &rnp->nocb_gp_wq[rcu_seq_ctr(rnp->gp_seq) & 0x1];
189  }
190  
rcu_init_one_nocb(struct rcu_node * rnp)191  static void rcu_init_one_nocb(struct rcu_node *rnp)
192  {
193  	init_swait_queue_head(&rnp->nocb_gp_wq[0]);
194  	init_swait_queue_head(&rnp->nocb_gp_wq[1]);
195  }
196  
__wake_nocb_gp(struct rcu_data * rdp_gp,struct rcu_data * rdp,bool force,unsigned long flags)197  static bool __wake_nocb_gp(struct rcu_data *rdp_gp,
198  			   struct rcu_data *rdp,
199  			   bool force, unsigned long flags)
200  	__releases(rdp_gp->nocb_gp_lock)
201  {
202  	bool needwake = false;
203  
204  	if (!READ_ONCE(rdp_gp->nocb_gp_kthread)) {
205  		raw_spin_unlock_irqrestore(&rdp_gp->nocb_gp_lock, flags);
206  		trace_rcu_nocb_wake(rcu_state.name, rdp->cpu,
207  				    TPS("AlreadyAwake"));
208  		return false;
209  	}
210  
211  	if (rdp_gp->nocb_defer_wakeup > RCU_NOCB_WAKE_NOT) {
212  		WRITE_ONCE(rdp_gp->nocb_defer_wakeup, RCU_NOCB_WAKE_NOT);
213  		del_timer(&rdp_gp->nocb_timer);
214  	}
215  
216  	if (force || READ_ONCE(rdp_gp->nocb_gp_sleep)) {
217  		WRITE_ONCE(rdp_gp->nocb_gp_sleep, false);
218  		needwake = true;
219  	}
220  	raw_spin_unlock_irqrestore(&rdp_gp->nocb_gp_lock, flags);
221  	if (needwake) {
222  		trace_rcu_nocb_wake(rcu_state.name, rdp->cpu, TPS("DoWake"));
223  		if (cpu_is_offline(raw_smp_processor_id()))
224  			swake_up_one_online(&rdp_gp->nocb_gp_wq);
225  		else
226  			wake_up_process(rdp_gp->nocb_gp_kthread);
227  	}
228  
229  	return needwake;
230  }
231  
232  /*
233   * Kick the GP kthread for this NOCB group.
234   */
wake_nocb_gp(struct rcu_data * rdp,bool force)235  static bool wake_nocb_gp(struct rcu_data *rdp, bool force)
236  {
237  	unsigned long flags;
238  	struct rcu_data *rdp_gp = rdp->nocb_gp_rdp;
239  
240  	raw_spin_lock_irqsave(&rdp_gp->nocb_gp_lock, flags);
241  	return __wake_nocb_gp(rdp_gp, rdp, force, flags);
242  }
243  
244  /*
245   * LAZY_FLUSH_JIFFIES decides the maximum amount of time that
246   * can elapse before lazy callbacks are flushed. Lazy callbacks
247   * could be flushed much earlier for a number of other reasons
248   * however, LAZY_FLUSH_JIFFIES will ensure no lazy callbacks are
249   * left unsubmitted to RCU after those many jiffies.
250   */
251  #define LAZY_FLUSH_JIFFIES (10 * HZ)
252  static unsigned long jiffies_till_flush = LAZY_FLUSH_JIFFIES;
253  
254  #ifdef CONFIG_RCU_LAZY
255  // To be called only from test code.
rcu_lazy_set_jiffies_till_flush(unsigned long jif)256  void rcu_lazy_set_jiffies_till_flush(unsigned long jif)
257  {
258  	jiffies_till_flush = jif;
259  }
260  EXPORT_SYMBOL(rcu_lazy_set_jiffies_till_flush);
261  
rcu_lazy_get_jiffies_till_flush(void)262  unsigned long rcu_lazy_get_jiffies_till_flush(void)
263  {
264  	return jiffies_till_flush;
265  }
266  EXPORT_SYMBOL(rcu_lazy_get_jiffies_till_flush);
267  #endif
268  
269  /*
270   * Arrange to wake the GP kthread for this NOCB group at some future
271   * time when it is safe to do so.
272   */
wake_nocb_gp_defer(struct rcu_data * rdp,int waketype,const char * reason)273  static void wake_nocb_gp_defer(struct rcu_data *rdp, int waketype,
274  			       const char *reason)
275  {
276  	unsigned long flags;
277  	struct rcu_data *rdp_gp = rdp->nocb_gp_rdp;
278  
279  	raw_spin_lock_irqsave(&rdp_gp->nocb_gp_lock, flags);
280  
281  	/*
282  	 * Bypass wakeup overrides previous deferments. In case of
283  	 * callback storms, no need to wake up too early.
284  	 */
285  	if (waketype == RCU_NOCB_WAKE_LAZY &&
286  	    rdp->nocb_defer_wakeup == RCU_NOCB_WAKE_NOT) {
287  		mod_timer(&rdp_gp->nocb_timer, jiffies + jiffies_till_flush);
288  		WRITE_ONCE(rdp_gp->nocb_defer_wakeup, waketype);
289  	} else if (waketype == RCU_NOCB_WAKE_BYPASS) {
290  		mod_timer(&rdp_gp->nocb_timer, jiffies + 2);
291  		WRITE_ONCE(rdp_gp->nocb_defer_wakeup, waketype);
292  	} else {
293  		if (rdp_gp->nocb_defer_wakeup < RCU_NOCB_WAKE)
294  			mod_timer(&rdp_gp->nocb_timer, jiffies + 1);
295  		if (rdp_gp->nocb_defer_wakeup < waketype)
296  			WRITE_ONCE(rdp_gp->nocb_defer_wakeup, waketype);
297  	}
298  
299  	raw_spin_unlock_irqrestore(&rdp_gp->nocb_gp_lock, flags);
300  
301  	trace_rcu_nocb_wake(rcu_state.name, rdp->cpu, reason);
302  }
303  
304  /*
305   * Flush the ->nocb_bypass queue into ->cblist, enqueuing rhp if non-NULL.
306   * However, if there is a callback to be enqueued and if ->nocb_bypass
307   * proves to be initially empty, just return false because the no-CB GP
308   * kthread may need to be awakened in this case.
309   *
310   * Return true if there was something to be flushed and it succeeded, otherwise
311   * false.
312   *
313   * Note that this function always returns true if rhp is NULL.
314   */
rcu_nocb_do_flush_bypass(struct rcu_data * rdp,struct rcu_head * rhp_in,unsigned long j,bool lazy)315  static bool rcu_nocb_do_flush_bypass(struct rcu_data *rdp, struct rcu_head *rhp_in,
316  				     unsigned long j, bool lazy)
317  {
318  	struct rcu_cblist rcl;
319  	struct rcu_head *rhp = rhp_in;
320  
321  	WARN_ON_ONCE(!rcu_rdp_is_offloaded(rdp));
322  	rcu_lockdep_assert_cblist_protected(rdp);
323  	lockdep_assert_held(&rdp->nocb_bypass_lock);
324  	if (rhp && !rcu_cblist_n_cbs(&rdp->nocb_bypass)) {
325  		raw_spin_unlock(&rdp->nocb_bypass_lock);
326  		return false;
327  	}
328  	/* Note: ->cblist.len already accounts for ->nocb_bypass contents. */
329  	if (rhp)
330  		rcu_segcblist_inc_len(&rdp->cblist); /* Must precede enqueue. */
331  
332  	/*
333  	 * If the new CB requested was a lazy one, queue it onto the main
334  	 * ->cblist so that we can take advantage of the grace-period that will
335  	 * happen regardless. But queue it onto the bypass list first so that
336  	 * the lazy CB is ordered with the existing CBs in the bypass list.
337  	 */
338  	if (lazy && rhp) {
339  		rcu_cblist_enqueue(&rdp->nocb_bypass, rhp);
340  		rhp = NULL;
341  	}
342  	rcu_cblist_flush_enqueue(&rcl, &rdp->nocb_bypass, rhp);
343  	WRITE_ONCE(rdp->lazy_len, 0);
344  
345  	rcu_segcblist_insert_pend_cbs(&rdp->cblist, &rcl);
346  	WRITE_ONCE(rdp->nocb_bypass_first, j);
347  	rcu_nocb_bypass_unlock(rdp);
348  	return true;
349  }
350  
351  /*
352   * Flush the ->nocb_bypass queue into ->cblist, enqueuing rhp if non-NULL.
353   * However, if there is a callback to be enqueued and if ->nocb_bypass
354   * proves to be initially empty, just return false because the no-CB GP
355   * kthread may need to be awakened in this case.
356   *
357   * Note that this function always returns true if rhp is NULL.
358   */
rcu_nocb_flush_bypass(struct rcu_data * rdp,struct rcu_head * rhp,unsigned long j,bool lazy)359  static bool rcu_nocb_flush_bypass(struct rcu_data *rdp, struct rcu_head *rhp,
360  				  unsigned long j, bool lazy)
361  {
362  	if (!rcu_rdp_is_offloaded(rdp))
363  		return true;
364  	rcu_lockdep_assert_cblist_protected(rdp);
365  	rcu_nocb_bypass_lock(rdp);
366  	return rcu_nocb_do_flush_bypass(rdp, rhp, j, lazy);
367  }
368  
369  /*
370   * If the ->nocb_bypass_lock is immediately available, flush the
371   * ->nocb_bypass queue into ->cblist.
372   */
rcu_nocb_try_flush_bypass(struct rcu_data * rdp,unsigned long j)373  static void rcu_nocb_try_flush_bypass(struct rcu_data *rdp, unsigned long j)
374  {
375  	rcu_lockdep_assert_cblist_protected(rdp);
376  	if (!rcu_rdp_is_offloaded(rdp) ||
377  	    !rcu_nocb_bypass_trylock(rdp))
378  		return;
379  	WARN_ON_ONCE(!rcu_nocb_do_flush_bypass(rdp, NULL, j, false));
380  }
381  
382  /*
383   * See whether it is appropriate to use the ->nocb_bypass list in order
384   * to control contention on ->nocb_lock.  A limited number of direct
385   * enqueues are permitted into ->cblist per jiffy.  If ->nocb_bypass
386   * is non-empty, further callbacks must be placed into ->nocb_bypass,
387   * otherwise rcu_barrier() breaks.  Use rcu_nocb_flush_bypass() to switch
388   * back to direct use of ->cblist.  However, ->nocb_bypass should not be
389   * used if ->cblist is empty, because otherwise callbacks can be stranded
390   * on ->nocb_bypass because we cannot count on the current CPU ever again
391   * invoking call_rcu().  The general rule is that if ->nocb_bypass is
392   * non-empty, the corresponding no-CBs grace-period kthread must not be
393   * in an indefinite sleep state.
394   *
395   * Finally, it is not permitted to use the bypass during early boot,
396   * as doing so would confuse the auto-initialization code.  Besides
397   * which, there is no point in worrying about lock contention while
398   * there is only one CPU in operation.
399   */
rcu_nocb_try_bypass(struct rcu_data * rdp,struct rcu_head * rhp,bool * was_alldone,unsigned long flags,bool lazy)400  static bool rcu_nocb_try_bypass(struct rcu_data *rdp, struct rcu_head *rhp,
401  				bool *was_alldone, unsigned long flags,
402  				bool lazy)
403  {
404  	unsigned long c;
405  	unsigned long cur_gp_seq;
406  	unsigned long j = jiffies;
407  	long ncbs = rcu_cblist_n_cbs(&rdp->nocb_bypass);
408  	bool bypass_is_lazy = (ncbs == READ_ONCE(rdp->lazy_len));
409  
410  	lockdep_assert_irqs_disabled();
411  
412  	// Pure softirq/rcuc based processing: no bypassing, no
413  	// locking.
414  	if (!rcu_rdp_is_offloaded(rdp)) {
415  		*was_alldone = !rcu_segcblist_pend_cbs(&rdp->cblist);
416  		return false;
417  	}
418  
419  	// In the process of (de-)offloading: no bypassing, but
420  	// locking.
421  	if (!rcu_segcblist_completely_offloaded(&rdp->cblist)) {
422  		rcu_nocb_lock(rdp);
423  		*was_alldone = !rcu_segcblist_pend_cbs(&rdp->cblist);
424  		return false; /* Not offloaded, no bypassing. */
425  	}
426  
427  	// Don't use ->nocb_bypass during early boot.
428  	if (rcu_scheduler_active != RCU_SCHEDULER_RUNNING) {
429  		rcu_nocb_lock(rdp);
430  		WARN_ON_ONCE(rcu_cblist_n_cbs(&rdp->nocb_bypass));
431  		*was_alldone = !rcu_segcblist_pend_cbs(&rdp->cblist);
432  		return false;
433  	}
434  
435  	// If we have advanced to a new jiffy, reset counts to allow
436  	// moving back from ->nocb_bypass to ->cblist.
437  	if (j == rdp->nocb_nobypass_last) {
438  		c = rdp->nocb_nobypass_count + 1;
439  	} else {
440  		WRITE_ONCE(rdp->nocb_nobypass_last, j);
441  		c = rdp->nocb_nobypass_count - nocb_nobypass_lim_per_jiffy;
442  		if (ULONG_CMP_LT(rdp->nocb_nobypass_count,
443  				 nocb_nobypass_lim_per_jiffy))
444  			c = 0;
445  		else if (c > nocb_nobypass_lim_per_jiffy)
446  			c = nocb_nobypass_lim_per_jiffy;
447  	}
448  	WRITE_ONCE(rdp->nocb_nobypass_count, c);
449  
450  	// If there hasn't yet been all that many ->cblist enqueues
451  	// this jiffy, tell the caller to enqueue onto ->cblist.  But flush
452  	// ->nocb_bypass first.
453  	// Lazy CBs throttle this back and do immediate bypass queuing.
454  	if (rdp->nocb_nobypass_count < nocb_nobypass_lim_per_jiffy && !lazy) {
455  		rcu_nocb_lock(rdp);
456  		*was_alldone = !rcu_segcblist_pend_cbs(&rdp->cblist);
457  		if (*was_alldone)
458  			trace_rcu_nocb_wake(rcu_state.name, rdp->cpu,
459  					    TPS("FirstQ"));
460  
461  		WARN_ON_ONCE(!rcu_nocb_flush_bypass(rdp, NULL, j, false));
462  		WARN_ON_ONCE(rcu_cblist_n_cbs(&rdp->nocb_bypass));
463  		return false; // Caller must enqueue the callback.
464  	}
465  
466  	// If ->nocb_bypass has been used too long or is too full,
467  	// flush ->nocb_bypass to ->cblist.
468  	if ((ncbs && !bypass_is_lazy && j != READ_ONCE(rdp->nocb_bypass_first)) ||
469  	    (ncbs &&  bypass_is_lazy &&
470  	     (time_after(j, READ_ONCE(rdp->nocb_bypass_first) + jiffies_till_flush))) ||
471  	    ncbs >= qhimark) {
472  		rcu_nocb_lock(rdp);
473  		*was_alldone = !rcu_segcblist_pend_cbs(&rdp->cblist);
474  
475  		if (!rcu_nocb_flush_bypass(rdp, rhp, j, lazy)) {
476  			if (*was_alldone)
477  				trace_rcu_nocb_wake(rcu_state.name, rdp->cpu,
478  						    TPS("FirstQ"));
479  			WARN_ON_ONCE(rcu_cblist_n_cbs(&rdp->nocb_bypass));
480  			return false; // Caller must enqueue the callback.
481  		}
482  		if (j != rdp->nocb_gp_adv_time &&
483  		    rcu_segcblist_nextgp(&rdp->cblist, &cur_gp_seq) &&
484  		    rcu_seq_done(&rdp->mynode->gp_seq, cur_gp_seq)) {
485  			rcu_advance_cbs_nowake(rdp->mynode, rdp);
486  			rdp->nocb_gp_adv_time = j;
487  		}
488  
489  		// The flush succeeded and we moved CBs into the regular list.
490  		// Don't wait for the wake up timer as it may be too far ahead.
491  		// Wake up the GP thread now instead, if the cblist was empty.
492  		__call_rcu_nocb_wake(rdp, *was_alldone, flags);
493  
494  		return true; // Callback already enqueued.
495  	}
496  
497  	// We need to use the bypass.
498  	rcu_nocb_bypass_lock(rdp);
499  	ncbs = rcu_cblist_n_cbs(&rdp->nocb_bypass);
500  	rcu_segcblist_inc_len(&rdp->cblist); /* Must precede enqueue. */
501  	rcu_cblist_enqueue(&rdp->nocb_bypass, rhp);
502  
503  	if (lazy)
504  		WRITE_ONCE(rdp->lazy_len, rdp->lazy_len + 1);
505  
506  	if (!ncbs) {
507  		WRITE_ONCE(rdp->nocb_bypass_first, j);
508  		trace_rcu_nocb_wake(rcu_state.name, rdp->cpu, TPS("FirstBQ"));
509  	}
510  	rcu_nocb_bypass_unlock(rdp);
511  	smp_mb(); /* Order enqueue before wake. */
512  	// A wake up of the grace period kthread or timer adjustment
513  	// needs to be done only if:
514  	// 1. Bypass list was fully empty before (this is the first
515  	//    bypass list entry), or:
516  	// 2. Both of these conditions are met:
517  	//    a. The bypass list previously had only lazy CBs, and:
518  	//    b. The new CB is non-lazy.
519  	if (!ncbs || (bypass_is_lazy && !lazy)) {
520  		// No-CBs GP kthread might be indefinitely asleep, if so, wake.
521  		rcu_nocb_lock(rdp); // Rare during call_rcu() flood.
522  		if (!rcu_segcblist_pend_cbs(&rdp->cblist)) {
523  			trace_rcu_nocb_wake(rcu_state.name, rdp->cpu,
524  					    TPS("FirstBQwake"));
525  			__call_rcu_nocb_wake(rdp, true, flags);
526  		} else {
527  			trace_rcu_nocb_wake(rcu_state.name, rdp->cpu,
528  					    TPS("FirstBQnoWake"));
529  			rcu_nocb_unlock(rdp);
530  		}
531  	}
532  	return true; // Callback already enqueued.
533  }
534  
535  /*
536   * Awaken the no-CBs grace-period kthread if needed, either due to it
537   * legitimately being asleep or due to overload conditions.
538   *
539   * If warranted, also wake up the kthread servicing this CPUs queues.
540   */
__call_rcu_nocb_wake(struct rcu_data * rdp,bool was_alldone,unsigned long flags)541  static void __call_rcu_nocb_wake(struct rcu_data *rdp, bool was_alldone,
542  				 unsigned long flags)
543  				 __releases(rdp->nocb_lock)
544  {
545  	long bypass_len;
546  	unsigned long cur_gp_seq;
547  	unsigned long j;
548  	long lazy_len;
549  	long len;
550  	struct task_struct *t;
551  
552  	// If we are being polled or there is no kthread, just leave.
553  	t = READ_ONCE(rdp->nocb_gp_kthread);
554  	if (rcu_nocb_poll || !t) {
555  		rcu_nocb_unlock(rdp);
556  		trace_rcu_nocb_wake(rcu_state.name, rdp->cpu,
557  				    TPS("WakeNotPoll"));
558  		return;
559  	}
560  	// Need to actually to a wakeup.
561  	len = rcu_segcblist_n_cbs(&rdp->cblist);
562  	bypass_len = rcu_cblist_n_cbs(&rdp->nocb_bypass);
563  	lazy_len = READ_ONCE(rdp->lazy_len);
564  	if (was_alldone) {
565  		rdp->qlen_last_fqs_check = len;
566  		// Only lazy CBs in bypass list
567  		if (lazy_len && bypass_len == lazy_len) {
568  			rcu_nocb_unlock(rdp);
569  			wake_nocb_gp_defer(rdp, RCU_NOCB_WAKE_LAZY,
570  					   TPS("WakeLazy"));
571  		} else if (!irqs_disabled_flags(flags) && cpu_online(rdp->cpu)) {
572  			/* ... if queue was empty ... */
573  			rcu_nocb_unlock(rdp);
574  			wake_nocb_gp(rdp, false);
575  			trace_rcu_nocb_wake(rcu_state.name, rdp->cpu,
576  					    TPS("WakeEmpty"));
577  		} else {
578  			/*
579  			 * Don't do the wake-up upfront on fragile paths.
580  			 * Also offline CPUs can't call swake_up_one_online() from
581  			 * (soft-)IRQs. Rely on the final deferred wake-up from
582  			 * rcutree_report_cpu_dead()
583  			 */
584  			rcu_nocb_unlock(rdp);
585  			wake_nocb_gp_defer(rdp, RCU_NOCB_WAKE,
586  					   TPS("WakeEmptyIsDeferred"));
587  		}
588  	} else if (len > rdp->qlen_last_fqs_check + qhimark) {
589  		/* ... or if many callbacks queued. */
590  		rdp->qlen_last_fqs_check = len;
591  		j = jiffies;
592  		if (j != rdp->nocb_gp_adv_time &&
593  		    rcu_segcblist_nextgp(&rdp->cblist, &cur_gp_seq) &&
594  		    rcu_seq_done(&rdp->mynode->gp_seq, cur_gp_seq)) {
595  			rcu_advance_cbs_nowake(rdp->mynode, rdp);
596  			rdp->nocb_gp_adv_time = j;
597  		}
598  		smp_mb(); /* Enqueue before timer_pending(). */
599  		if ((rdp->nocb_cb_sleep ||
600  		     !rcu_segcblist_ready_cbs(&rdp->cblist)) &&
601  		    !timer_pending(&rdp->nocb_timer)) {
602  			rcu_nocb_unlock(rdp);
603  			wake_nocb_gp_defer(rdp, RCU_NOCB_WAKE_FORCE,
604  					   TPS("WakeOvfIsDeferred"));
605  		} else {
606  			rcu_nocb_unlock(rdp);
607  			trace_rcu_nocb_wake(rcu_state.name, rdp->cpu, TPS("WakeNot"));
608  		}
609  	} else {
610  		rcu_nocb_unlock(rdp);
611  		trace_rcu_nocb_wake(rcu_state.name, rdp->cpu, TPS("WakeNot"));
612  	}
613  }
614  
nocb_gp_toggle_rdp(struct rcu_data * rdp,bool * wake_state)615  static int nocb_gp_toggle_rdp(struct rcu_data *rdp,
616  			       bool *wake_state)
617  {
618  	struct rcu_segcblist *cblist = &rdp->cblist;
619  	unsigned long flags;
620  	int ret;
621  
622  	rcu_nocb_lock_irqsave(rdp, flags);
623  	if (rcu_segcblist_test_flags(cblist, SEGCBLIST_OFFLOADED) &&
624  	    !rcu_segcblist_test_flags(cblist, SEGCBLIST_KTHREAD_GP)) {
625  		/*
626  		 * Offloading. Set our flag and notify the offload worker.
627  		 * We will handle this rdp until it ever gets de-offloaded.
628  		 */
629  		rcu_segcblist_set_flags(cblist, SEGCBLIST_KTHREAD_GP);
630  		if (rcu_segcblist_test_flags(cblist, SEGCBLIST_KTHREAD_CB))
631  			*wake_state = true;
632  		ret = 1;
633  	} else if (!rcu_segcblist_test_flags(cblist, SEGCBLIST_OFFLOADED) &&
634  		   rcu_segcblist_test_flags(cblist, SEGCBLIST_KTHREAD_GP)) {
635  		/*
636  		 * De-offloading. Clear our flag and notify the de-offload worker.
637  		 * We will ignore this rdp until it ever gets re-offloaded.
638  		 */
639  		rcu_segcblist_clear_flags(cblist, SEGCBLIST_KTHREAD_GP);
640  		if (!rcu_segcblist_test_flags(cblist, SEGCBLIST_KTHREAD_CB))
641  			*wake_state = true;
642  		ret = 0;
643  	} else {
644  		WARN_ON_ONCE(1);
645  		ret = -1;
646  	}
647  
648  	rcu_nocb_unlock_irqrestore(rdp, flags);
649  
650  	return ret;
651  }
652  
nocb_gp_sleep(struct rcu_data * my_rdp,int cpu)653  static void nocb_gp_sleep(struct rcu_data *my_rdp, int cpu)
654  {
655  	trace_rcu_nocb_wake(rcu_state.name, cpu, TPS("Sleep"));
656  	swait_event_interruptible_exclusive(my_rdp->nocb_gp_wq,
657  					!READ_ONCE(my_rdp->nocb_gp_sleep));
658  	trace_rcu_nocb_wake(rcu_state.name, cpu, TPS("EndSleep"));
659  }
660  
661  /*
662   * No-CBs GP kthreads come here to wait for additional callbacks to show up
663   * or for grace periods to end.
664   */
nocb_gp_wait(struct rcu_data * my_rdp)665  static void nocb_gp_wait(struct rcu_data *my_rdp)
666  {
667  	bool bypass = false;
668  	int __maybe_unused cpu = my_rdp->cpu;
669  	unsigned long cur_gp_seq;
670  	unsigned long flags;
671  	bool gotcbs = false;
672  	unsigned long j = jiffies;
673  	bool lazy = false;
674  	bool needwait_gp = false; // This prevents actual uninitialized use.
675  	bool needwake;
676  	bool needwake_gp;
677  	struct rcu_data *rdp, *rdp_toggling = NULL;
678  	struct rcu_node *rnp;
679  	unsigned long wait_gp_seq = 0; // Suppress "use uninitialized" warning.
680  	bool wasempty = false;
681  
682  	/*
683  	 * Each pass through the following loop checks for CBs and for the
684  	 * nearest grace period (if any) to wait for next.  The CB kthreads
685  	 * and the global grace-period kthread are awakened if needed.
686  	 */
687  	WARN_ON_ONCE(my_rdp->nocb_gp_rdp != my_rdp);
688  	/*
689  	 * An rcu_data structure is removed from the list after its
690  	 * CPU is de-offloaded and added to the list before that CPU is
691  	 * (re-)offloaded.  If the following loop happens to be referencing
692  	 * that rcu_data structure during the time that the corresponding
693  	 * CPU is de-offloaded and then immediately re-offloaded, this
694  	 * loop's rdp pointer will be carried to the end of the list by
695  	 * the resulting pair of list operations.  This can cause the loop
696  	 * to skip over some of the rcu_data structures that were supposed
697  	 * to have been scanned.  Fortunately a new iteration through the
698  	 * entire loop is forced after a given CPU's rcu_data structure
699  	 * is added to the list, so the skipped-over rcu_data structures
700  	 * won't be ignored for long.
701  	 */
702  	list_for_each_entry(rdp, &my_rdp->nocb_head_rdp, nocb_entry_rdp) {
703  		long bypass_ncbs;
704  		bool flush_bypass = false;
705  		long lazy_ncbs;
706  
707  		trace_rcu_nocb_wake(rcu_state.name, rdp->cpu, TPS("Check"));
708  		rcu_nocb_lock_irqsave(rdp, flags);
709  		lockdep_assert_held(&rdp->nocb_lock);
710  		bypass_ncbs = rcu_cblist_n_cbs(&rdp->nocb_bypass);
711  		lazy_ncbs = READ_ONCE(rdp->lazy_len);
712  
713  		if (bypass_ncbs && (lazy_ncbs == bypass_ncbs) &&
714  		    (time_after(j, READ_ONCE(rdp->nocb_bypass_first) + jiffies_till_flush) ||
715  		     bypass_ncbs > 2 * qhimark)) {
716  			flush_bypass = true;
717  		} else if (bypass_ncbs && (lazy_ncbs != bypass_ncbs) &&
718  		    (time_after(j, READ_ONCE(rdp->nocb_bypass_first) + 1) ||
719  		     bypass_ncbs > 2 * qhimark)) {
720  			flush_bypass = true;
721  		} else if (!bypass_ncbs && rcu_segcblist_empty(&rdp->cblist)) {
722  			rcu_nocb_unlock_irqrestore(rdp, flags);
723  			continue; /* No callbacks here, try next. */
724  		}
725  
726  		if (flush_bypass) {
727  			// Bypass full or old, so flush it.
728  			(void)rcu_nocb_try_flush_bypass(rdp, j);
729  			bypass_ncbs = rcu_cblist_n_cbs(&rdp->nocb_bypass);
730  			lazy_ncbs = READ_ONCE(rdp->lazy_len);
731  		}
732  
733  		if (bypass_ncbs) {
734  			trace_rcu_nocb_wake(rcu_state.name, rdp->cpu,
735  					    bypass_ncbs == lazy_ncbs ? TPS("Lazy") : TPS("Bypass"));
736  			if (bypass_ncbs == lazy_ncbs)
737  				lazy = true;
738  			else
739  				bypass = true;
740  		}
741  		rnp = rdp->mynode;
742  
743  		// Advance callbacks if helpful and low contention.
744  		needwake_gp = false;
745  		if (!rcu_segcblist_restempty(&rdp->cblist,
746  					     RCU_NEXT_READY_TAIL) ||
747  		    (rcu_segcblist_nextgp(&rdp->cblist, &cur_gp_seq) &&
748  		     rcu_seq_done(&rnp->gp_seq, cur_gp_seq))) {
749  			raw_spin_lock_rcu_node(rnp); /* irqs disabled. */
750  			needwake_gp = rcu_advance_cbs(rnp, rdp);
751  			wasempty = rcu_segcblist_restempty(&rdp->cblist,
752  							   RCU_NEXT_READY_TAIL);
753  			raw_spin_unlock_rcu_node(rnp); /* irqs disabled. */
754  		}
755  		// Need to wait on some grace period?
756  		WARN_ON_ONCE(wasempty &&
757  			     !rcu_segcblist_restempty(&rdp->cblist,
758  						      RCU_NEXT_READY_TAIL));
759  		if (rcu_segcblist_nextgp(&rdp->cblist, &cur_gp_seq)) {
760  			if (!needwait_gp ||
761  			    ULONG_CMP_LT(cur_gp_seq, wait_gp_seq))
762  				wait_gp_seq = cur_gp_seq;
763  			needwait_gp = true;
764  			trace_rcu_nocb_wake(rcu_state.name, rdp->cpu,
765  					    TPS("NeedWaitGP"));
766  		}
767  		if (rcu_segcblist_ready_cbs(&rdp->cblist)) {
768  			needwake = rdp->nocb_cb_sleep;
769  			WRITE_ONCE(rdp->nocb_cb_sleep, false);
770  			smp_mb(); /* CB invocation -after- GP end. */
771  		} else {
772  			needwake = false;
773  		}
774  		rcu_nocb_unlock_irqrestore(rdp, flags);
775  		if (needwake) {
776  			swake_up_one(&rdp->nocb_cb_wq);
777  			gotcbs = true;
778  		}
779  		if (needwake_gp)
780  			rcu_gp_kthread_wake();
781  	}
782  
783  	my_rdp->nocb_gp_bypass = bypass;
784  	my_rdp->nocb_gp_gp = needwait_gp;
785  	my_rdp->nocb_gp_seq = needwait_gp ? wait_gp_seq : 0;
786  
787  	// At least one child with non-empty ->nocb_bypass, so set
788  	// timer in order to avoid stranding its callbacks.
789  	if (!rcu_nocb_poll) {
790  		// If bypass list only has lazy CBs. Add a deferred lazy wake up.
791  		if (lazy && !bypass) {
792  			wake_nocb_gp_defer(my_rdp, RCU_NOCB_WAKE_LAZY,
793  					TPS("WakeLazyIsDeferred"));
794  		// Otherwise add a deferred bypass wake up.
795  		} else if (bypass) {
796  			wake_nocb_gp_defer(my_rdp, RCU_NOCB_WAKE_BYPASS,
797  					TPS("WakeBypassIsDeferred"));
798  		}
799  	}
800  
801  	if (rcu_nocb_poll) {
802  		/* Polling, so trace if first poll in the series. */
803  		if (gotcbs)
804  			trace_rcu_nocb_wake(rcu_state.name, cpu, TPS("Poll"));
805  		if (list_empty(&my_rdp->nocb_head_rdp)) {
806  			raw_spin_lock_irqsave(&my_rdp->nocb_gp_lock, flags);
807  			if (!my_rdp->nocb_toggling_rdp)
808  				WRITE_ONCE(my_rdp->nocb_gp_sleep, true);
809  			raw_spin_unlock_irqrestore(&my_rdp->nocb_gp_lock, flags);
810  			/* Wait for any offloading rdp */
811  			nocb_gp_sleep(my_rdp, cpu);
812  		} else {
813  			schedule_timeout_idle(1);
814  		}
815  	} else if (!needwait_gp) {
816  		/* Wait for callbacks to appear. */
817  		nocb_gp_sleep(my_rdp, cpu);
818  	} else {
819  		rnp = my_rdp->mynode;
820  		trace_rcu_this_gp(rnp, my_rdp, wait_gp_seq, TPS("StartWait"));
821  		swait_event_interruptible_exclusive(
822  			rnp->nocb_gp_wq[rcu_seq_ctr(wait_gp_seq) & 0x1],
823  			rcu_seq_done(&rnp->gp_seq, wait_gp_seq) ||
824  			!READ_ONCE(my_rdp->nocb_gp_sleep));
825  		trace_rcu_this_gp(rnp, my_rdp, wait_gp_seq, TPS("EndWait"));
826  	}
827  
828  	if (!rcu_nocb_poll) {
829  		raw_spin_lock_irqsave(&my_rdp->nocb_gp_lock, flags);
830  		// (De-)queue an rdp to/from the group if its nocb state is changing
831  		rdp_toggling = my_rdp->nocb_toggling_rdp;
832  		if (rdp_toggling)
833  			my_rdp->nocb_toggling_rdp = NULL;
834  
835  		if (my_rdp->nocb_defer_wakeup > RCU_NOCB_WAKE_NOT) {
836  			WRITE_ONCE(my_rdp->nocb_defer_wakeup, RCU_NOCB_WAKE_NOT);
837  			del_timer(&my_rdp->nocb_timer);
838  		}
839  		WRITE_ONCE(my_rdp->nocb_gp_sleep, true);
840  		raw_spin_unlock_irqrestore(&my_rdp->nocb_gp_lock, flags);
841  	} else {
842  		rdp_toggling = READ_ONCE(my_rdp->nocb_toggling_rdp);
843  		if (rdp_toggling) {
844  			/*
845  			 * Paranoid locking to make sure nocb_toggling_rdp is well
846  			 * reset *before* we (re)set SEGCBLIST_KTHREAD_GP or we could
847  			 * race with another round of nocb toggling for this rdp.
848  			 * Nocb locking should prevent from that already but we stick
849  			 * to paranoia, especially in rare path.
850  			 */
851  			raw_spin_lock_irqsave(&my_rdp->nocb_gp_lock, flags);
852  			my_rdp->nocb_toggling_rdp = NULL;
853  			raw_spin_unlock_irqrestore(&my_rdp->nocb_gp_lock, flags);
854  		}
855  	}
856  
857  	if (rdp_toggling) {
858  		bool wake_state = false;
859  		int ret;
860  
861  		ret = nocb_gp_toggle_rdp(rdp_toggling, &wake_state);
862  		if (ret == 1)
863  			list_add_tail(&rdp_toggling->nocb_entry_rdp, &my_rdp->nocb_head_rdp);
864  		else if (ret == 0)
865  			list_del(&rdp_toggling->nocb_entry_rdp);
866  		if (wake_state)
867  			swake_up_one(&rdp_toggling->nocb_state_wq);
868  	}
869  
870  	my_rdp->nocb_gp_seq = -1;
871  	WARN_ON(signal_pending(current));
872  }
873  
874  /*
875   * No-CBs grace-period-wait kthread.  There is one of these per group
876   * of CPUs, but only once at least one CPU in that group has come online
877   * at least once since boot.  This kthread checks for newly posted
878   * callbacks from any of the CPUs it is responsible for, waits for a
879   * grace period, then awakens all of the rcu_nocb_cb_kthread() instances
880   * that then have callback-invocation work to do.
881   */
rcu_nocb_gp_kthread(void * arg)882  static int rcu_nocb_gp_kthread(void *arg)
883  {
884  	struct rcu_data *rdp = arg;
885  
886  	for (;;) {
887  		WRITE_ONCE(rdp->nocb_gp_loops, rdp->nocb_gp_loops + 1);
888  		nocb_gp_wait(rdp);
889  		cond_resched_tasks_rcu_qs();
890  	}
891  	return 0;
892  }
893  
nocb_cb_can_run(struct rcu_data * rdp)894  static inline bool nocb_cb_can_run(struct rcu_data *rdp)
895  {
896  	u8 flags = SEGCBLIST_OFFLOADED | SEGCBLIST_KTHREAD_CB;
897  
898  	return rcu_segcblist_test_flags(&rdp->cblist, flags);
899  }
900  
nocb_cb_wait_cond(struct rcu_data * rdp)901  static inline bool nocb_cb_wait_cond(struct rcu_data *rdp)
902  {
903  	return nocb_cb_can_run(rdp) && !READ_ONCE(rdp->nocb_cb_sleep);
904  }
905  
906  /*
907   * Invoke any ready callbacks from the corresponding no-CBs CPU,
908   * then, if there are no more, wait for more to appear.
909   */
nocb_cb_wait(struct rcu_data * rdp)910  static void nocb_cb_wait(struct rcu_data *rdp)
911  {
912  	struct rcu_segcblist *cblist = &rdp->cblist;
913  	unsigned long cur_gp_seq;
914  	unsigned long flags;
915  	bool needwake_state = false;
916  	bool needwake_gp = false;
917  	bool can_sleep = true;
918  	struct rcu_node *rnp = rdp->mynode;
919  
920  	do {
921  		swait_event_interruptible_exclusive(rdp->nocb_cb_wq,
922  						    nocb_cb_wait_cond(rdp));
923  
924  		// VVV Ensure CB invocation follows _sleep test.
925  		if (smp_load_acquire(&rdp->nocb_cb_sleep)) { // ^^^
926  			WARN_ON(signal_pending(current));
927  			trace_rcu_nocb_wake(rcu_state.name, rdp->cpu, TPS("WokeEmpty"));
928  		}
929  	} while (!nocb_cb_can_run(rdp));
930  
931  
932  	local_irq_save(flags);
933  	rcu_momentary_dyntick_idle();
934  	local_irq_restore(flags);
935  	/*
936  	 * Disable BH to provide the expected environment.  Also, when
937  	 * transitioning to/from NOCB mode, a self-requeuing callback might
938  	 * be invoked from softirq.  A short grace period could cause both
939  	 * instances of this callback would execute concurrently.
940  	 */
941  	local_bh_disable();
942  	rcu_do_batch(rdp);
943  	local_bh_enable();
944  	lockdep_assert_irqs_enabled();
945  	rcu_nocb_lock_irqsave(rdp, flags);
946  	if (rcu_segcblist_nextgp(cblist, &cur_gp_seq) &&
947  	    rcu_seq_done(&rnp->gp_seq, cur_gp_seq) &&
948  	    raw_spin_trylock_rcu_node(rnp)) { /* irqs already disabled. */
949  		needwake_gp = rcu_advance_cbs(rdp->mynode, rdp);
950  		raw_spin_unlock_rcu_node(rnp); /* irqs remain disabled. */
951  	}
952  
953  	if (rcu_segcblist_test_flags(cblist, SEGCBLIST_OFFLOADED)) {
954  		if (!rcu_segcblist_test_flags(cblist, SEGCBLIST_KTHREAD_CB)) {
955  			rcu_segcblist_set_flags(cblist, SEGCBLIST_KTHREAD_CB);
956  			if (rcu_segcblist_test_flags(cblist, SEGCBLIST_KTHREAD_GP))
957  				needwake_state = true;
958  		}
959  		if (rcu_segcblist_ready_cbs(cblist))
960  			can_sleep = false;
961  	} else {
962  		/*
963  		 * De-offloading. Clear our flag and notify the de-offload worker.
964  		 * We won't touch the callbacks and keep sleeping until we ever
965  		 * get re-offloaded.
966  		 */
967  		WARN_ON_ONCE(!rcu_segcblist_test_flags(cblist, SEGCBLIST_KTHREAD_CB));
968  		rcu_segcblist_clear_flags(cblist, SEGCBLIST_KTHREAD_CB);
969  		if (!rcu_segcblist_test_flags(cblist, SEGCBLIST_KTHREAD_GP))
970  			needwake_state = true;
971  	}
972  
973  	WRITE_ONCE(rdp->nocb_cb_sleep, can_sleep);
974  
975  	if (rdp->nocb_cb_sleep)
976  		trace_rcu_nocb_wake(rcu_state.name, rdp->cpu, TPS("CBSleep"));
977  
978  	rcu_nocb_unlock_irqrestore(rdp, flags);
979  	if (needwake_gp)
980  		rcu_gp_kthread_wake();
981  
982  	if (needwake_state)
983  		swake_up_one(&rdp->nocb_state_wq);
984  }
985  
986  /*
987   * Per-rcu_data kthread, but only for no-CBs CPUs.  Repeatedly invoke
988   * nocb_cb_wait() to do the dirty work.
989   */
rcu_nocb_cb_kthread(void * arg)990  static int rcu_nocb_cb_kthread(void *arg)
991  {
992  	struct rcu_data *rdp = arg;
993  
994  	// Each pass through this loop does one callback batch, and,
995  	// if there are no more ready callbacks, waits for them.
996  	for (;;) {
997  		nocb_cb_wait(rdp);
998  		cond_resched_tasks_rcu_qs();
999  	}
1000  	return 0;
1001  }
1002  
1003  /* Is a deferred wakeup of rcu_nocb_kthread() required? */
rcu_nocb_need_deferred_wakeup(struct rcu_data * rdp,int level)1004  static int rcu_nocb_need_deferred_wakeup(struct rcu_data *rdp, int level)
1005  {
1006  	return READ_ONCE(rdp->nocb_defer_wakeup) >= level;
1007  }
1008  
1009  /* Do a deferred wakeup of rcu_nocb_kthread(). */
do_nocb_deferred_wakeup_common(struct rcu_data * rdp_gp,struct rcu_data * rdp,int level,unsigned long flags)1010  static bool do_nocb_deferred_wakeup_common(struct rcu_data *rdp_gp,
1011  					   struct rcu_data *rdp, int level,
1012  					   unsigned long flags)
1013  	__releases(rdp_gp->nocb_gp_lock)
1014  {
1015  	int ndw;
1016  	int ret;
1017  
1018  	if (!rcu_nocb_need_deferred_wakeup(rdp_gp, level)) {
1019  		raw_spin_unlock_irqrestore(&rdp_gp->nocb_gp_lock, flags);
1020  		return false;
1021  	}
1022  
1023  	ndw = rdp_gp->nocb_defer_wakeup;
1024  	ret = __wake_nocb_gp(rdp_gp, rdp, ndw == RCU_NOCB_WAKE_FORCE, flags);
1025  	trace_rcu_nocb_wake(rcu_state.name, rdp->cpu, TPS("DeferredWake"));
1026  
1027  	return ret;
1028  }
1029  
1030  /* Do a deferred wakeup of rcu_nocb_kthread() from a timer handler. */
do_nocb_deferred_wakeup_timer(struct timer_list * t)1031  static void do_nocb_deferred_wakeup_timer(struct timer_list *t)
1032  {
1033  	unsigned long flags;
1034  	struct rcu_data *rdp = from_timer(rdp, t, nocb_timer);
1035  
1036  	WARN_ON_ONCE(rdp->nocb_gp_rdp != rdp);
1037  	trace_rcu_nocb_wake(rcu_state.name, rdp->cpu, TPS("Timer"));
1038  
1039  	raw_spin_lock_irqsave(&rdp->nocb_gp_lock, flags);
1040  	smp_mb__after_spinlock(); /* Timer expire before wakeup. */
1041  	do_nocb_deferred_wakeup_common(rdp, rdp, RCU_NOCB_WAKE_BYPASS, flags);
1042  }
1043  
1044  /*
1045   * Do a deferred wakeup of rcu_nocb_kthread() from fastpath.
1046   * This means we do an inexact common-case check.  Note that if
1047   * we miss, ->nocb_timer will eventually clean things up.
1048   */
do_nocb_deferred_wakeup(struct rcu_data * rdp)1049  static bool do_nocb_deferred_wakeup(struct rcu_data *rdp)
1050  {
1051  	unsigned long flags;
1052  	struct rcu_data *rdp_gp = rdp->nocb_gp_rdp;
1053  
1054  	if (!rdp_gp || !rcu_nocb_need_deferred_wakeup(rdp_gp, RCU_NOCB_WAKE))
1055  		return false;
1056  
1057  	raw_spin_lock_irqsave(&rdp_gp->nocb_gp_lock, flags);
1058  	return do_nocb_deferred_wakeup_common(rdp_gp, rdp, RCU_NOCB_WAKE, flags);
1059  }
1060  
rcu_nocb_flush_deferred_wakeup(void)1061  void rcu_nocb_flush_deferred_wakeup(void)
1062  {
1063  	do_nocb_deferred_wakeup(this_cpu_ptr(&rcu_data));
1064  }
1065  EXPORT_SYMBOL_GPL(rcu_nocb_flush_deferred_wakeup);
1066  
rdp_offload_toggle(struct rcu_data * rdp,bool offload,unsigned long flags)1067  static int rdp_offload_toggle(struct rcu_data *rdp,
1068  			       bool offload, unsigned long flags)
1069  	__releases(rdp->nocb_lock)
1070  {
1071  	struct rcu_segcblist *cblist = &rdp->cblist;
1072  	struct rcu_data *rdp_gp = rdp->nocb_gp_rdp;
1073  	bool wake_gp = false;
1074  
1075  	rcu_segcblist_offload(cblist, offload);
1076  
1077  	if (rdp->nocb_cb_sleep)
1078  		rdp->nocb_cb_sleep = false;
1079  	rcu_nocb_unlock_irqrestore(rdp, flags);
1080  
1081  	/*
1082  	 * Ignore former value of nocb_cb_sleep and force wake up as it could
1083  	 * have been spuriously set to false already.
1084  	 */
1085  	swake_up_one(&rdp->nocb_cb_wq);
1086  
1087  	raw_spin_lock_irqsave(&rdp_gp->nocb_gp_lock, flags);
1088  	// Queue this rdp for add/del to/from the list to iterate on rcuog
1089  	WRITE_ONCE(rdp_gp->nocb_toggling_rdp, rdp);
1090  	if (rdp_gp->nocb_gp_sleep) {
1091  		rdp_gp->nocb_gp_sleep = false;
1092  		wake_gp = true;
1093  	}
1094  	raw_spin_unlock_irqrestore(&rdp_gp->nocb_gp_lock, flags);
1095  
1096  	return wake_gp;
1097  }
1098  
rcu_nocb_rdp_deoffload(void * arg)1099  static long rcu_nocb_rdp_deoffload(void *arg)
1100  {
1101  	struct rcu_data *rdp = arg;
1102  	struct rcu_segcblist *cblist = &rdp->cblist;
1103  	unsigned long flags;
1104  	int wake_gp;
1105  	struct rcu_data *rdp_gp = rdp->nocb_gp_rdp;
1106  
1107  	/*
1108  	 * rcu_nocb_rdp_deoffload() may be called directly if
1109  	 * rcuog/o[p] spawn failed, because at this time the rdp->cpu
1110  	 * is not online yet.
1111  	 */
1112  	WARN_ON_ONCE((rdp->cpu != raw_smp_processor_id()) && cpu_online(rdp->cpu));
1113  
1114  	pr_info("De-offloading %d\n", rdp->cpu);
1115  
1116  	rcu_nocb_lock_irqsave(rdp, flags);
1117  	/*
1118  	 * Flush once and for all now. This suffices because we are
1119  	 * running on the target CPU holding ->nocb_lock (thus having
1120  	 * interrupts disabled), and because rdp_offload_toggle()
1121  	 * invokes rcu_segcblist_offload(), which clears SEGCBLIST_OFFLOADED.
1122  	 * Thus future calls to rcu_segcblist_completely_offloaded() will
1123  	 * return false, which means that future calls to rcu_nocb_try_bypass()
1124  	 * will refuse to put anything into the bypass.
1125  	 */
1126  	WARN_ON_ONCE(!rcu_nocb_flush_bypass(rdp, NULL, jiffies, false));
1127  	/*
1128  	 * Start with invoking rcu_core() early. This way if the current thread
1129  	 * happens to preempt an ongoing call to rcu_core() in the middle,
1130  	 * leaving some work dismissed because rcu_core() still thinks the rdp is
1131  	 * completely offloaded, we are guaranteed a nearby future instance of
1132  	 * rcu_core() to catch up.
1133  	 */
1134  	rcu_segcblist_set_flags(cblist, SEGCBLIST_RCU_CORE);
1135  	invoke_rcu_core();
1136  	wake_gp = rdp_offload_toggle(rdp, false, flags);
1137  
1138  	mutex_lock(&rdp_gp->nocb_gp_kthread_mutex);
1139  	if (rdp_gp->nocb_gp_kthread) {
1140  		if (wake_gp)
1141  			wake_up_process(rdp_gp->nocb_gp_kthread);
1142  
1143  		/*
1144  		 * If rcuo[p] kthread spawn failed, directly remove SEGCBLIST_KTHREAD_CB.
1145  		 * Just wait SEGCBLIST_KTHREAD_GP to be cleared by rcuog.
1146  		 */
1147  		if (!rdp->nocb_cb_kthread) {
1148  			rcu_nocb_lock_irqsave(rdp, flags);
1149  			rcu_segcblist_clear_flags(&rdp->cblist, SEGCBLIST_KTHREAD_CB);
1150  			rcu_nocb_unlock_irqrestore(rdp, flags);
1151  		}
1152  
1153  		swait_event_exclusive(rdp->nocb_state_wq,
1154  					!rcu_segcblist_test_flags(cblist,
1155  					  SEGCBLIST_KTHREAD_CB | SEGCBLIST_KTHREAD_GP));
1156  	} else {
1157  		/*
1158  		 * No kthread to clear the flags for us or remove the rdp from the nocb list
1159  		 * to iterate. Do it here instead. Locking doesn't look stricly necessary
1160  		 * but we stick to paranoia in this rare path.
1161  		 */
1162  		rcu_nocb_lock_irqsave(rdp, flags);
1163  		rcu_segcblist_clear_flags(&rdp->cblist,
1164  				SEGCBLIST_KTHREAD_CB | SEGCBLIST_KTHREAD_GP);
1165  		rcu_nocb_unlock_irqrestore(rdp, flags);
1166  
1167  		list_del(&rdp->nocb_entry_rdp);
1168  	}
1169  	mutex_unlock(&rdp_gp->nocb_gp_kthread_mutex);
1170  
1171  	/*
1172  	 * Lock one last time to acquire latest callback updates from kthreads
1173  	 * so we can later handle callbacks locally without locking.
1174  	 */
1175  	rcu_nocb_lock_irqsave(rdp, flags);
1176  	/*
1177  	 * Theoretically we could clear SEGCBLIST_LOCKING after the nocb
1178  	 * lock is released but how about being paranoid for once?
1179  	 */
1180  	rcu_segcblist_clear_flags(cblist, SEGCBLIST_LOCKING);
1181  	/*
1182  	 * Without SEGCBLIST_LOCKING, we can't use
1183  	 * rcu_nocb_unlock_irqrestore() anymore.
1184  	 */
1185  	raw_spin_unlock_irqrestore(&rdp->nocb_lock, flags);
1186  
1187  	/* Sanity check */
1188  	WARN_ON_ONCE(rcu_cblist_n_cbs(&rdp->nocb_bypass));
1189  
1190  
1191  	return 0;
1192  }
1193  
rcu_nocb_cpu_deoffload(int cpu)1194  int rcu_nocb_cpu_deoffload(int cpu)
1195  {
1196  	struct rcu_data *rdp = per_cpu_ptr(&rcu_data, cpu);
1197  	int ret = 0;
1198  
1199  	cpus_read_lock();
1200  	mutex_lock(&rcu_state.barrier_mutex);
1201  	if (rcu_rdp_is_offloaded(rdp)) {
1202  		if (cpu_online(cpu)) {
1203  			ret = work_on_cpu(cpu, rcu_nocb_rdp_deoffload, rdp);
1204  			if (!ret)
1205  				cpumask_clear_cpu(cpu, rcu_nocb_mask);
1206  		} else {
1207  			pr_info("NOCB: Cannot CB-deoffload offline CPU %d\n", rdp->cpu);
1208  			ret = -EINVAL;
1209  		}
1210  	}
1211  	mutex_unlock(&rcu_state.barrier_mutex);
1212  	cpus_read_unlock();
1213  
1214  	return ret;
1215  }
1216  EXPORT_SYMBOL_GPL(rcu_nocb_cpu_deoffload);
1217  
rcu_nocb_rdp_offload(void * arg)1218  static long rcu_nocb_rdp_offload(void *arg)
1219  {
1220  	struct rcu_data *rdp = arg;
1221  	struct rcu_segcblist *cblist = &rdp->cblist;
1222  	unsigned long flags;
1223  	int wake_gp;
1224  	struct rcu_data *rdp_gp = rdp->nocb_gp_rdp;
1225  
1226  	WARN_ON_ONCE(rdp->cpu != raw_smp_processor_id());
1227  	/*
1228  	 * For now we only support re-offload, ie: the rdp must have been
1229  	 * offloaded on boot first.
1230  	 */
1231  	if (!rdp->nocb_gp_rdp)
1232  		return -EINVAL;
1233  
1234  	if (WARN_ON_ONCE(!rdp_gp->nocb_gp_kthread))
1235  		return -EINVAL;
1236  
1237  	pr_info("Offloading %d\n", rdp->cpu);
1238  
1239  	/*
1240  	 * Can't use rcu_nocb_lock_irqsave() before SEGCBLIST_LOCKING
1241  	 * is set.
1242  	 */
1243  	raw_spin_lock_irqsave(&rdp->nocb_lock, flags);
1244  
1245  	/*
1246  	 * We didn't take the nocb lock while working on the
1247  	 * rdp->cblist with SEGCBLIST_LOCKING cleared (pure softirq/rcuc mode).
1248  	 * Every modifications that have been done previously on
1249  	 * rdp->cblist must be visible remotely by the nocb kthreads
1250  	 * upon wake up after reading the cblist flags.
1251  	 *
1252  	 * The layout against nocb_lock enforces that ordering:
1253  	 *
1254  	 *  __rcu_nocb_rdp_offload()   nocb_cb_wait()/nocb_gp_wait()
1255  	 * -------------------------   ----------------------------
1256  	 *      WRITE callbacks           rcu_nocb_lock()
1257  	 *      rcu_nocb_lock()           READ flags
1258  	 *      WRITE flags               READ callbacks
1259  	 *      rcu_nocb_unlock()         rcu_nocb_unlock()
1260  	 */
1261  	wake_gp = rdp_offload_toggle(rdp, true, flags);
1262  	if (wake_gp)
1263  		wake_up_process(rdp_gp->nocb_gp_kthread);
1264  	swait_event_exclusive(rdp->nocb_state_wq,
1265  			      rcu_segcblist_test_flags(cblist, SEGCBLIST_KTHREAD_CB) &&
1266  			      rcu_segcblist_test_flags(cblist, SEGCBLIST_KTHREAD_GP));
1267  
1268  	/*
1269  	 * All kthreads are ready to work, we can finally relieve rcu_core() and
1270  	 * enable nocb bypass.
1271  	 */
1272  	rcu_nocb_lock_irqsave(rdp, flags);
1273  	rcu_segcblist_clear_flags(cblist, SEGCBLIST_RCU_CORE);
1274  	rcu_nocb_unlock_irqrestore(rdp, flags);
1275  
1276  	return 0;
1277  }
1278  
rcu_nocb_cpu_offload(int cpu)1279  int rcu_nocb_cpu_offload(int cpu)
1280  {
1281  	struct rcu_data *rdp = per_cpu_ptr(&rcu_data, cpu);
1282  	int ret = 0;
1283  
1284  	cpus_read_lock();
1285  	mutex_lock(&rcu_state.barrier_mutex);
1286  	if (!rcu_rdp_is_offloaded(rdp)) {
1287  		if (cpu_online(cpu)) {
1288  			ret = work_on_cpu(cpu, rcu_nocb_rdp_offload, rdp);
1289  			if (!ret)
1290  				cpumask_set_cpu(cpu, rcu_nocb_mask);
1291  		} else {
1292  			pr_info("NOCB: Cannot CB-offload offline CPU %d\n", rdp->cpu);
1293  			ret = -EINVAL;
1294  		}
1295  	}
1296  	mutex_unlock(&rcu_state.barrier_mutex);
1297  	cpus_read_unlock();
1298  
1299  	return ret;
1300  }
1301  EXPORT_SYMBOL_GPL(rcu_nocb_cpu_offload);
1302  
1303  #ifdef CONFIG_RCU_LAZY
1304  static unsigned long
lazy_rcu_shrink_count(struct shrinker * shrink,struct shrink_control * sc)1305  lazy_rcu_shrink_count(struct shrinker *shrink, struct shrink_control *sc)
1306  {
1307  	int cpu;
1308  	unsigned long count = 0;
1309  
1310  	if (WARN_ON_ONCE(!cpumask_available(rcu_nocb_mask)))
1311  		return 0;
1312  
1313  	/*  Protect rcu_nocb_mask against concurrent (de-)offloading. */
1314  	if (!mutex_trylock(&rcu_state.barrier_mutex))
1315  		return 0;
1316  
1317  	/* Snapshot count of all CPUs */
1318  	for_each_cpu(cpu, rcu_nocb_mask) {
1319  		struct rcu_data *rdp = per_cpu_ptr(&rcu_data, cpu);
1320  
1321  		count +=  READ_ONCE(rdp->lazy_len);
1322  	}
1323  
1324  	mutex_unlock(&rcu_state.barrier_mutex);
1325  
1326  	return count ? count : SHRINK_EMPTY;
1327  }
1328  
1329  static unsigned long
lazy_rcu_shrink_scan(struct shrinker * shrink,struct shrink_control * sc)1330  lazy_rcu_shrink_scan(struct shrinker *shrink, struct shrink_control *sc)
1331  {
1332  	int cpu;
1333  	unsigned long flags;
1334  	unsigned long count = 0;
1335  
1336  	if (WARN_ON_ONCE(!cpumask_available(rcu_nocb_mask)))
1337  		return 0;
1338  	/*
1339  	 * Protect against concurrent (de-)offloading. Otherwise nocb locking
1340  	 * may be ignored or imbalanced.
1341  	 */
1342  	if (!mutex_trylock(&rcu_state.barrier_mutex)) {
1343  		/*
1344  		 * But really don't insist if barrier_mutex is contended since we
1345  		 * can't guarantee that it will never engage in a dependency
1346  		 * chain involving memory allocation. The lock is seldom contended
1347  		 * anyway.
1348  		 */
1349  		return 0;
1350  	}
1351  
1352  	/* Snapshot count of all CPUs */
1353  	for_each_cpu(cpu, rcu_nocb_mask) {
1354  		struct rcu_data *rdp = per_cpu_ptr(&rcu_data, cpu);
1355  		int _count;
1356  
1357  		if (WARN_ON_ONCE(!rcu_rdp_is_offloaded(rdp)))
1358  			continue;
1359  
1360  		if (!READ_ONCE(rdp->lazy_len))
1361  			continue;
1362  
1363  		rcu_nocb_lock_irqsave(rdp, flags);
1364  		/*
1365  		 * Recheck under the nocb lock. Since we are not holding the bypass
1366  		 * lock we may still race with increments from the enqueuer but still
1367  		 * we know for sure if there is at least one lazy callback.
1368  		 */
1369  		_count = READ_ONCE(rdp->lazy_len);
1370  		if (!_count) {
1371  			rcu_nocb_unlock_irqrestore(rdp, flags);
1372  			continue;
1373  		}
1374  		rcu_nocb_try_flush_bypass(rdp, jiffies);
1375  		rcu_nocb_unlock_irqrestore(rdp, flags);
1376  		wake_nocb_gp(rdp, false);
1377  		sc->nr_to_scan -= _count;
1378  		count += _count;
1379  		if (sc->nr_to_scan <= 0)
1380  			break;
1381  	}
1382  
1383  	mutex_unlock(&rcu_state.barrier_mutex);
1384  
1385  	return count ? count : SHRINK_STOP;
1386  }
1387  
1388  static struct shrinker lazy_rcu_shrinker = {
1389  	.count_objects = lazy_rcu_shrink_count,
1390  	.scan_objects = lazy_rcu_shrink_scan,
1391  	.batch = 0,
1392  	.seeks = DEFAULT_SEEKS,
1393  };
1394  #endif // #ifdef CONFIG_RCU_LAZY
1395  
rcu_init_nohz(void)1396  void __init rcu_init_nohz(void)
1397  {
1398  	int cpu;
1399  	struct rcu_data *rdp;
1400  	const struct cpumask *cpumask = NULL;
1401  
1402  #if defined(CONFIG_NO_HZ_FULL)
1403  	if (tick_nohz_full_running && !cpumask_empty(tick_nohz_full_mask))
1404  		cpumask = tick_nohz_full_mask;
1405  #endif
1406  
1407  	if (IS_ENABLED(CONFIG_RCU_NOCB_CPU_DEFAULT_ALL) &&
1408  	    !rcu_state.nocb_is_setup && !cpumask)
1409  		cpumask = cpu_possible_mask;
1410  
1411  	if (cpumask) {
1412  		if (!cpumask_available(rcu_nocb_mask)) {
1413  			if (!zalloc_cpumask_var(&rcu_nocb_mask, GFP_KERNEL)) {
1414  				pr_info("rcu_nocb_mask allocation failed, callback offloading disabled.\n");
1415  				return;
1416  			}
1417  		}
1418  
1419  		cpumask_or(rcu_nocb_mask, rcu_nocb_mask, cpumask);
1420  		rcu_state.nocb_is_setup = true;
1421  	}
1422  
1423  	if (!rcu_state.nocb_is_setup)
1424  		return;
1425  
1426  #ifdef CONFIG_RCU_LAZY
1427  	if (register_shrinker(&lazy_rcu_shrinker, "rcu-lazy"))
1428  		pr_err("Failed to register lazy_rcu shrinker!\n");
1429  #endif // #ifdef CONFIG_RCU_LAZY
1430  
1431  	if (!cpumask_subset(rcu_nocb_mask, cpu_possible_mask)) {
1432  		pr_info("\tNote: kernel parameter 'rcu_nocbs=', 'nohz_full', or 'isolcpus=' contains nonexistent CPUs.\n");
1433  		cpumask_and(rcu_nocb_mask, cpu_possible_mask,
1434  			    rcu_nocb_mask);
1435  	}
1436  	if (cpumask_empty(rcu_nocb_mask))
1437  		pr_info("\tOffload RCU callbacks from CPUs: (none).\n");
1438  	else
1439  		pr_info("\tOffload RCU callbacks from CPUs: %*pbl.\n",
1440  			cpumask_pr_args(rcu_nocb_mask));
1441  	if (rcu_nocb_poll)
1442  		pr_info("\tPoll for callbacks from no-CBs CPUs.\n");
1443  
1444  	for_each_cpu(cpu, rcu_nocb_mask) {
1445  		rdp = per_cpu_ptr(&rcu_data, cpu);
1446  		if (rcu_segcblist_empty(&rdp->cblist))
1447  			rcu_segcblist_init(&rdp->cblist);
1448  		rcu_segcblist_offload(&rdp->cblist, true);
1449  		rcu_segcblist_set_flags(&rdp->cblist, SEGCBLIST_KTHREAD_CB | SEGCBLIST_KTHREAD_GP);
1450  		rcu_segcblist_clear_flags(&rdp->cblist, SEGCBLIST_RCU_CORE);
1451  	}
1452  	rcu_organize_nocb_kthreads();
1453  }
1454  
1455  /* Initialize per-rcu_data variables for no-CBs CPUs. */
rcu_boot_init_nocb_percpu_data(struct rcu_data * rdp)1456  static void __init rcu_boot_init_nocb_percpu_data(struct rcu_data *rdp)
1457  {
1458  	init_swait_queue_head(&rdp->nocb_cb_wq);
1459  	init_swait_queue_head(&rdp->nocb_gp_wq);
1460  	init_swait_queue_head(&rdp->nocb_state_wq);
1461  	raw_spin_lock_init(&rdp->nocb_lock);
1462  	raw_spin_lock_init(&rdp->nocb_bypass_lock);
1463  	raw_spin_lock_init(&rdp->nocb_gp_lock);
1464  	timer_setup(&rdp->nocb_timer, do_nocb_deferred_wakeup_timer, 0);
1465  	rcu_cblist_init(&rdp->nocb_bypass);
1466  	WRITE_ONCE(rdp->lazy_len, 0);
1467  	mutex_init(&rdp->nocb_gp_kthread_mutex);
1468  }
1469  
1470  /*
1471   * If the specified CPU is a no-CBs CPU that does not already have its
1472   * rcuo CB kthread, spawn it.  Additionally, if the rcuo GP kthread
1473   * for this CPU's group has not yet been created, spawn it as well.
1474   */
rcu_spawn_cpu_nocb_kthread(int cpu)1475  static void rcu_spawn_cpu_nocb_kthread(int cpu)
1476  {
1477  	struct rcu_data *rdp = per_cpu_ptr(&rcu_data, cpu);
1478  	struct rcu_data *rdp_gp;
1479  	struct task_struct *t;
1480  	struct sched_param sp;
1481  
1482  	if (!rcu_scheduler_fully_active || !rcu_state.nocb_is_setup)
1483  		return;
1484  
1485  	/* If there already is an rcuo kthread, then nothing to do. */
1486  	if (rdp->nocb_cb_kthread)
1487  		return;
1488  
1489  	/* If we didn't spawn the GP kthread first, reorganize! */
1490  	sp.sched_priority = kthread_prio;
1491  	rdp_gp = rdp->nocb_gp_rdp;
1492  	mutex_lock(&rdp_gp->nocb_gp_kthread_mutex);
1493  	if (!rdp_gp->nocb_gp_kthread) {
1494  		t = kthread_run(rcu_nocb_gp_kthread, rdp_gp,
1495  				"rcuog/%d", rdp_gp->cpu);
1496  		if (WARN_ONCE(IS_ERR(t), "%s: Could not start rcuo GP kthread, OOM is now expected behavior\n", __func__)) {
1497  			mutex_unlock(&rdp_gp->nocb_gp_kthread_mutex);
1498  			goto end;
1499  		}
1500  		WRITE_ONCE(rdp_gp->nocb_gp_kthread, t);
1501  		if (kthread_prio)
1502  			sched_setscheduler_nocheck(t, SCHED_FIFO, &sp);
1503  	}
1504  	mutex_unlock(&rdp_gp->nocb_gp_kthread_mutex);
1505  
1506  	/* Spawn the kthread for this CPU. */
1507  	t = kthread_run(rcu_nocb_cb_kthread, rdp,
1508  			"rcuo%c/%d", rcu_state.abbr, cpu);
1509  	if (WARN_ONCE(IS_ERR(t), "%s: Could not start rcuo CB kthread, OOM is now expected behavior\n", __func__))
1510  		goto end;
1511  
1512  	if (IS_ENABLED(CONFIG_RCU_NOCB_CPU_CB_BOOST) && kthread_prio)
1513  		sched_setscheduler_nocheck(t, SCHED_FIFO, &sp);
1514  
1515  	WRITE_ONCE(rdp->nocb_cb_kthread, t);
1516  	WRITE_ONCE(rdp->nocb_gp_kthread, rdp_gp->nocb_gp_kthread);
1517  	return;
1518  end:
1519  	mutex_lock(&rcu_state.barrier_mutex);
1520  	if (rcu_rdp_is_offloaded(rdp)) {
1521  		rcu_nocb_rdp_deoffload(rdp);
1522  		cpumask_clear_cpu(cpu, rcu_nocb_mask);
1523  	}
1524  	mutex_unlock(&rcu_state.barrier_mutex);
1525  }
1526  
1527  /* How many CB CPU IDs per GP kthread?  Default of -1 for sqrt(nr_cpu_ids). */
1528  static int rcu_nocb_gp_stride = -1;
1529  module_param(rcu_nocb_gp_stride, int, 0444);
1530  
1531  /*
1532   * Initialize GP-CB relationships for all no-CBs CPU.
1533   */
rcu_organize_nocb_kthreads(void)1534  static void __init rcu_organize_nocb_kthreads(void)
1535  {
1536  	int cpu;
1537  	bool firsttime = true;
1538  	bool gotnocbs = false;
1539  	bool gotnocbscbs = true;
1540  	int ls = rcu_nocb_gp_stride;
1541  	int nl = 0;  /* Next GP kthread. */
1542  	struct rcu_data *rdp;
1543  	struct rcu_data *rdp_gp = NULL;  /* Suppress misguided gcc warn. */
1544  
1545  	if (!cpumask_available(rcu_nocb_mask))
1546  		return;
1547  	if (ls == -1) {
1548  		ls = nr_cpu_ids / int_sqrt(nr_cpu_ids);
1549  		rcu_nocb_gp_stride = ls;
1550  	}
1551  
1552  	/*
1553  	 * Each pass through this loop sets up one rcu_data structure.
1554  	 * Should the corresponding CPU come online in the future, then
1555  	 * we will spawn the needed set of rcu_nocb_kthread() kthreads.
1556  	 */
1557  	for_each_possible_cpu(cpu) {
1558  		rdp = per_cpu_ptr(&rcu_data, cpu);
1559  		if (rdp->cpu >= nl) {
1560  			/* New GP kthread, set up for CBs & next GP. */
1561  			gotnocbs = true;
1562  			nl = DIV_ROUND_UP(rdp->cpu + 1, ls) * ls;
1563  			rdp_gp = rdp;
1564  			INIT_LIST_HEAD(&rdp->nocb_head_rdp);
1565  			if (dump_tree) {
1566  				if (!firsttime)
1567  					pr_cont("%s\n", gotnocbscbs
1568  							? "" : " (self only)");
1569  				gotnocbscbs = false;
1570  				firsttime = false;
1571  				pr_alert("%s: No-CB GP kthread CPU %d:",
1572  					 __func__, cpu);
1573  			}
1574  		} else {
1575  			/* Another CB kthread, link to previous GP kthread. */
1576  			gotnocbscbs = true;
1577  			if (dump_tree)
1578  				pr_cont(" %d", cpu);
1579  		}
1580  		rdp->nocb_gp_rdp = rdp_gp;
1581  		if (cpumask_test_cpu(cpu, rcu_nocb_mask))
1582  			list_add_tail(&rdp->nocb_entry_rdp, &rdp_gp->nocb_head_rdp);
1583  	}
1584  	if (gotnocbs && dump_tree)
1585  		pr_cont("%s\n", gotnocbscbs ? "" : " (self only)");
1586  }
1587  
1588  /*
1589   * Bind the current task to the offloaded CPUs.  If there are no offloaded
1590   * CPUs, leave the task unbound.  Splat if the bind attempt fails.
1591   */
rcu_bind_current_to_nocb(void)1592  void rcu_bind_current_to_nocb(void)
1593  {
1594  	if (cpumask_available(rcu_nocb_mask) && !cpumask_empty(rcu_nocb_mask))
1595  		WARN_ON(sched_setaffinity(current->pid, rcu_nocb_mask));
1596  }
1597  EXPORT_SYMBOL_GPL(rcu_bind_current_to_nocb);
1598  
1599  // The ->on_cpu field is available only in CONFIG_SMP=y, so...
1600  #ifdef CONFIG_SMP
show_rcu_should_be_on_cpu(struct task_struct * tsp)1601  static char *show_rcu_should_be_on_cpu(struct task_struct *tsp)
1602  {
1603  	return tsp && task_is_running(tsp) && !tsp->on_cpu ? "!" : "";
1604  }
1605  #else // #ifdef CONFIG_SMP
show_rcu_should_be_on_cpu(struct task_struct * tsp)1606  static char *show_rcu_should_be_on_cpu(struct task_struct *tsp)
1607  {
1608  	return "";
1609  }
1610  #endif // #else #ifdef CONFIG_SMP
1611  
1612  /*
1613   * Dump out nocb grace-period kthread state for the specified rcu_data
1614   * structure.
1615   */
show_rcu_nocb_gp_state(struct rcu_data * rdp)1616  static void show_rcu_nocb_gp_state(struct rcu_data *rdp)
1617  {
1618  	struct rcu_node *rnp = rdp->mynode;
1619  
1620  	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",
1621  		rdp->cpu,
1622  		"kK"[!!rdp->nocb_gp_kthread],
1623  		"lL"[raw_spin_is_locked(&rdp->nocb_gp_lock)],
1624  		"dD"[!!rdp->nocb_defer_wakeup],
1625  		"tT"[timer_pending(&rdp->nocb_timer)],
1626  		"sS"[!!rdp->nocb_gp_sleep],
1627  		".W"[swait_active(&rdp->nocb_gp_wq)],
1628  		".W"[swait_active(&rnp->nocb_gp_wq[0])],
1629  		".W"[swait_active(&rnp->nocb_gp_wq[1])],
1630  		".B"[!!rdp->nocb_gp_bypass],
1631  		".G"[!!rdp->nocb_gp_gp],
1632  		(long)rdp->nocb_gp_seq,
1633  		rnp->grplo, rnp->grphi, READ_ONCE(rdp->nocb_gp_loops),
1634  		rdp->nocb_gp_kthread ? task_state_to_char(rdp->nocb_gp_kthread) : '.',
1635  		rdp->nocb_gp_kthread ? (int)task_cpu(rdp->nocb_gp_kthread) : -1,
1636  		show_rcu_should_be_on_cpu(rdp->nocb_gp_kthread));
1637  }
1638  
1639  /* Dump out nocb kthread state for the specified rcu_data structure. */
show_rcu_nocb_state(struct rcu_data * rdp)1640  static void show_rcu_nocb_state(struct rcu_data *rdp)
1641  {
1642  	char bufw[20];
1643  	char bufr[20];
1644  	struct rcu_data *nocb_next_rdp;
1645  	struct rcu_segcblist *rsclp = &rdp->cblist;
1646  	bool waslocked;
1647  	bool wassleep;
1648  
1649  	if (rdp->nocb_gp_rdp == rdp)
1650  		show_rcu_nocb_gp_state(rdp);
1651  
1652  	nocb_next_rdp = list_next_or_null_rcu(&rdp->nocb_gp_rdp->nocb_head_rdp,
1653  					      &rdp->nocb_entry_rdp,
1654  					      typeof(*rdp),
1655  					      nocb_entry_rdp);
1656  
1657  	sprintf(bufw, "%ld", rsclp->gp_seq[RCU_WAIT_TAIL]);
1658  	sprintf(bufr, "%ld", rsclp->gp_seq[RCU_NEXT_READY_TAIL]);
1659  	pr_info("   CB %d^%d->%d %c%c%c%c%c F%ld L%ld C%d %c%c%s%c%s%c%c q%ld %c CPU %d%s\n",
1660  		rdp->cpu, rdp->nocb_gp_rdp->cpu,
1661  		nocb_next_rdp ? nocb_next_rdp->cpu : -1,
1662  		"kK"[!!rdp->nocb_cb_kthread],
1663  		"bB"[raw_spin_is_locked(&rdp->nocb_bypass_lock)],
1664  		"lL"[raw_spin_is_locked(&rdp->nocb_lock)],
1665  		"sS"[!!rdp->nocb_cb_sleep],
1666  		".W"[swait_active(&rdp->nocb_cb_wq)],
1667  		jiffies - rdp->nocb_bypass_first,
1668  		jiffies - rdp->nocb_nobypass_last,
1669  		rdp->nocb_nobypass_count,
1670  		".D"[rcu_segcblist_ready_cbs(rsclp)],
1671  		".W"[!rcu_segcblist_segempty(rsclp, RCU_WAIT_TAIL)],
1672  		rcu_segcblist_segempty(rsclp, RCU_WAIT_TAIL) ? "" : bufw,
1673  		".R"[!rcu_segcblist_segempty(rsclp, RCU_NEXT_READY_TAIL)],
1674  		rcu_segcblist_segempty(rsclp, RCU_NEXT_READY_TAIL) ? "" : bufr,
1675  		".N"[!rcu_segcblist_segempty(rsclp, RCU_NEXT_TAIL)],
1676  		".B"[!!rcu_cblist_n_cbs(&rdp->nocb_bypass)],
1677  		rcu_segcblist_n_cbs(&rdp->cblist),
1678  		rdp->nocb_cb_kthread ? task_state_to_char(rdp->nocb_cb_kthread) : '.',
1679  		rdp->nocb_cb_kthread ? (int)task_cpu(rdp->nocb_cb_kthread) : -1,
1680  		show_rcu_should_be_on_cpu(rdp->nocb_cb_kthread));
1681  
1682  	/* It is OK for GP kthreads to have GP state. */
1683  	if (rdp->nocb_gp_rdp == rdp)
1684  		return;
1685  
1686  	waslocked = raw_spin_is_locked(&rdp->nocb_gp_lock);
1687  	wassleep = swait_active(&rdp->nocb_gp_wq);
1688  	if (!rdp->nocb_gp_sleep && !waslocked && !wassleep)
1689  		return;  /* Nothing untoward. */
1690  
1691  	pr_info("   nocb GP activity on CB-only CPU!!! %c%c%c %c\n",
1692  		"lL"[waslocked],
1693  		"dD"[!!rdp->nocb_defer_wakeup],
1694  		"sS"[!!rdp->nocb_gp_sleep],
1695  		".W"[wassleep]);
1696  }
1697  
1698  #else /* #ifdef CONFIG_RCU_NOCB_CPU */
1699  
rcu_lockdep_is_held_nocb(struct rcu_data * rdp)1700  static inline int rcu_lockdep_is_held_nocb(struct rcu_data *rdp)
1701  {
1702  	return 0;
1703  }
1704  
rcu_current_is_nocb_kthread(struct rcu_data * rdp)1705  static inline bool rcu_current_is_nocb_kthread(struct rcu_data *rdp)
1706  {
1707  	return false;
1708  }
1709  
1710  /* No ->nocb_lock to acquire.  */
rcu_nocb_lock(struct rcu_data * rdp)1711  static void rcu_nocb_lock(struct rcu_data *rdp)
1712  {
1713  }
1714  
1715  /* No ->nocb_lock to release.  */
rcu_nocb_unlock(struct rcu_data * rdp)1716  static void rcu_nocb_unlock(struct rcu_data *rdp)
1717  {
1718  }
1719  
1720  /* No ->nocb_lock to release.  */
rcu_nocb_unlock_irqrestore(struct rcu_data * rdp,unsigned long flags)1721  static void rcu_nocb_unlock_irqrestore(struct rcu_data *rdp,
1722  				       unsigned long flags)
1723  {
1724  	local_irq_restore(flags);
1725  }
1726  
1727  /* Lockdep check that ->cblist may be safely accessed. */
rcu_lockdep_assert_cblist_protected(struct rcu_data * rdp)1728  static void rcu_lockdep_assert_cblist_protected(struct rcu_data *rdp)
1729  {
1730  	lockdep_assert_irqs_disabled();
1731  }
1732  
rcu_nocb_gp_cleanup(struct swait_queue_head * sq)1733  static void rcu_nocb_gp_cleanup(struct swait_queue_head *sq)
1734  {
1735  }
1736  
rcu_nocb_gp_get(struct rcu_node * rnp)1737  static struct swait_queue_head *rcu_nocb_gp_get(struct rcu_node *rnp)
1738  {
1739  	return NULL;
1740  }
1741  
rcu_init_one_nocb(struct rcu_node * rnp)1742  static void rcu_init_one_nocb(struct rcu_node *rnp)
1743  {
1744  }
1745  
wake_nocb_gp(struct rcu_data * rdp,bool force)1746  static bool wake_nocb_gp(struct rcu_data *rdp, bool force)
1747  {
1748  	return false;
1749  }
1750  
rcu_nocb_flush_bypass(struct rcu_data * rdp,struct rcu_head * rhp,unsigned long j,bool lazy)1751  static bool rcu_nocb_flush_bypass(struct rcu_data *rdp, struct rcu_head *rhp,
1752  				  unsigned long j, bool lazy)
1753  {
1754  	return true;
1755  }
1756  
rcu_nocb_try_bypass(struct rcu_data * rdp,struct rcu_head * rhp,bool * was_alldone,unsigned long flags,bool lazy)1757  static bool rcu_nocb_try_bypass(struct rcu_data *rdp, struct rcu_head *rhp,
1758  				bool *was_alldone, unsigned long flags, bool lazy)
1759  {
1760  	return false;
1761  }
1762  
__call_rcu_nocb_wake(struct rcu_data * rdp,bool was_empty,unsigned long flags)1763  static void __call_rcu_nocb_wake(struct rcu_data *rdp, bool was_empty,
1764  				 unsigned long flags)
1765  {
1766  	WARN_ON_ONCE(1);  /* Should be dead code! */
1767  }
1768  
rcu_boot_init_nocb_percpu_data(struct rcu_data * rdp)1769  static void __init rcu_boot_init_nocb_percpu_data(struct rcu_data *rdp)
1770  {
1771  }
1772  
rcu_nocb_need_deferred_wakeup(struct rcu_data * rdp,int level)1773  static int rcu_nocb_need_deferred_wakeup(struct rcu_data *rdp, int level)
1774  {
1775  	return false;
1776  }
1777  
do_nocb_deferred_wakeup(struct rcu_data * rdp)1778  static bool do_nocb_deferred_wakeup(struct rcu_data *rdp)
1779  {
1780  	return false;
1781  }
1782  
rcu_spawn_cpu_nocb_kthread(int cpu)1783  static void rcu_spawn_cpu_nocb_kthread(int cpu)
1784  {
1785  }
1786  
show_rcu_nocb_state(struct rcu_data * rdp)1787  static void show_rcu_nocb_state(struct rcu_data *rdp)
1788  {
1789  }
1790  
1791  #endif /* #else #ifdef CONFIG_RCU_NOCB_CPU */
1792