23a8a6cf9383735e322504c6fb98ed2820f49758
[akaros.git] / kern / src / rcu.c
1 /* Copyright (c) 2018 Google Inc
2  * Barret Rhoden <brho@cs.berkeley.edu>
3  * See LICENSE for details.
4  *
5  * RCU.  We borrow a few things from Linux - mostly the header bits and the
6  * tree-rcu structure.
7  *
8  * Acronyms/definitions:
9  * - CB: RCU callbacks (call_rcu)
10  * - QS: quiescent state - a time when we know a core isn't in an RCU read-side
11  *   critical section.
12  * - GP: grace period.  Some quotes from Linux/Paul:
13  *   - "A time period during which all such pre-existing readers complete is
14  *   called a 'grace period'."
15  *   - "Anything outside of an RCU read-side critical section is a quiescent
16  *   state, and a grace period is any time period in which every CPU (or task,
17  *   for
18  * - gpnum: number of the current grace period we are working on
19  * - completed: number of the grace periods completed
20  *
21  * We differ in a few ways from Linux's implementation:
22  *
23  * - Callbacks run on management cores (a.k.a, LL cores, e.g. core 0).  This way
24  *   we don't have to kick idle or user space cores to run their CBs, and those
25  *   CBs don't interfere with a possibly unrelated process.
26  *
27  * - Our RCU is most similar to rcu_sched (classic RCU), and not the preemptible
28  *   RCU.  Our kthreads don't get preempted, so we don't need to worry about
29  *   read side critical sections being interrupted.
30  *
31  * - There is no softirq processing to note the passing of GPs or to run CBs.
32  *
33  * - Our tree uses atomic ops to trace grace periods within the rcu_nodes.
34  *   Linux's tree-rcu uses locks.  They need the locks since under some
35  *   circumstances, a QS would be marked during a read-side critical section,
36  *   and the QS marking needed to track the gpnum to keep the QS matched to the
37  *   GP.  See
38  *   https://www.kernel.org/doc/Documentation/RCU/Design/Data-Structures/Data-Structures.html
39  *   and grep "Come on".  We don't need to worry about this since we only mark a
40  *   QS under two situations:
41  *
42  *   - The core knows it is does not hold an rcu_read_lock, so we can always
43  *   mark QS.
44  *   - The GP kthread saw the core either idle or in userspace after the gp
45  *   started.  That means we know that core had a QS after the GP started.
46  *
47  *   So any time we mark a QS is actually a QS.  I think Linux has times where
48  *   they note a QS for an older GP, and set a note to mark that QS *for that
49  *   GP* in the future.  Their locks make sure they are marking for the right
50  *   gpnum.  There might be some element of the rnps not knowing about the
51  *   latest GP yet too.
52  *
53  * - We do use locking at the per-core level to decide whether or not to start
54  *   mark a QS for a given GP.  (lock, compare gp_acked to gpnum, etc).  This
55  *   ensures only one thread (the core or the GP kth) marks the core for a given
56  *   GP.  We actually could handle it if the both did, (make the trickle-up
57  *   idempotent, which we do for the interior nodes) but we could run into
58  *   situations where a core checks in for a GP before the global gpnum was set.
59  *   This could happen when the GP kth is resetting the tree for the next GP.
60  *   I think it'd be OK, but not worth the hassle and confusion.
61  *
62  * - We have a kthread for GP management, like Linux.  Callbacks are enqueued
63  *   locally (on the core that calls call_rcu), like Linux.  We have a kthread
64  *   per management core to process the callbacks, and these threads will handle
65  *   the callbacks of *all* cores.  Each core has a specific mgmt kthread that
66  *   will run its callbacks.  It is important that a particular core's callbacks
67  *   are processed by the same thread - I rely on this to implement rcu_barrier
68  *   easily.  In that case, we just need to schedule a CB on every core that has
69  *   CBs, and when those N CBs are done, our barrier passed.  This relies on CBs
70  *   being processed in order for a given core.  We could do the barrier in
71  *   other ways, but it doesn't seem like a big deal.
72  *
73  * - I kept around some seq counter and locking stuff in rcu_helper.h.  We might
74  *   use that in the future.
75  */
76
77 #include <rcu.h>
78 #include <kthread.h>
79 #include <smp.h>
80 #include <kmalloc.h>
81
82 /* How many CBs to queue up before we trigger a GP */
83 #define RCU_CB_THRESH 10
84 /* How long (usec) we wait between running a GP if we weren't triggered. */
85 #define RCU_GP_MIN_PERIOD 25000
86 /* How long (usec) we wait for cores to check in. */
87 #define RCU_GP_TARDY_PERIOD 1000
88
89 /* In rcu_tree_helper.c */
90 extern int rcu_num_cores;
91 extern int rcu_num_lvls;
92
93 /* Controls whether we skip cores when we expedite, which forces tardy cores. */
94 static bool rcu_debug_tardy;
95
96 /* Externed in rcu_tree_helper.c */
97 struct rcu_state rcu_state;
98
99
100 DEFINE_PERCPU(struct rcu_pcpui, rcu_pcpui);
101
102 struct sync_cb_blob {
103         struct rcu_head h;
104         struct semaphore *sem;
105 };
106
107 static void __sync_cb(struct rcu_head *head)
108 {
109         struct sync_cb_blob *b = container_of(head, struct sync_cb_blob, h);
110
111         sem_up(b->sem);
112 }
113
114 void synchronize_rcu(void)
115 {
116         struct sync_cb_blob b[1];
117         struct semaphore sem[1];
118
119         if (in_rcu_cb_ctx(this_pcpui_ptr()))
120                 panic("Attempted synchronize_rcu() from an RCU callback!");
121         if (is_rcu_ktask(current_kthread))
122                 panic("Attempted synchronize_rcu() from an RCU thread!");
123         sem_init(sem, 0);
124         init_rcu_head_on_stack(&b->h);
125         b->sem = sem;
126         call_rcu(&b->h, __sync_cb);
127         sem_down(sem);
128 }
129
130 static inline bool gp_in_progress(struct rcu_state *rsp)
131 {
132         unsigned long completed = READ_ONCE(rsp->completed);
133         unsigned long gpnum = READ_ONCE(rsp->gpnum);
134
135         assert(gpnum - completed <= 1);
136         return completed != gpnum;
137 }
138
139 /* Wakes the kthread to run a grace period if it isn't already running.
140  *
141  * If 'force', we'll make sure it runs a fresh GP, which will catch all CBs
142  * registered before this call.  That's not 100% true.  It might be possible on
143  * some non-x86 architectures for the writes that wake the ktask are reordered
144  * before the read of gpnum that our caller made.  Thus the caller could have a
145  * CB in a later GP.  Worst case, they'll wait an extra GP timeout.  Not too
146  * concerned, though I probably should be. */
147 static void wake_gp_ktask(struct rcu_state *rsp, bool force)
148 {
149         if (!force && gp_in_progress(rsp))
150                 return;
151         rsp->gp_ktask_ctl = 1;
152         rendez_wakeup(&rsp->gp_ktask_rv);
153 }
154
155 static void rcu_exec_cb(struct rcu_head *head)
156 {
157         if (__is_kfree_rcu_offset((unsigned long)head->func))
158                 kfree((void*)head - (unsigned long)head->func);
159         else
160                 head->func(head);
161 }
162
163 void __early_call_rcu(struct rcu_head *head)
164 {
165         extern bool booting;
166
167         assert(booting);
168         assert(core_id() == 0);
169         run_as_rkm(rcu_exec_cb, head);
170 }
171
172 /* This could be called from a remote core, e.g. rcu_barrier().  Returns the
173  * number of enqueued CBs, including the one we pass in. */
174 static int __call_rcu_rpi(struct rcu_state *rsp, struct rcu_pcpui *rpi,
175                            struct rcu_head *head, rcu_callback_t func)
176 {
177         unsigned int nr_cbs;
178
179         head->func = func;
180
181         if (!rpi->booted) {
182                 __early_call_rcu(head);
183                 return 0;
184         }
185         /* rsp->gpnum is the one we're either working on (if > completed) or the one
186          * we already did.  Either way, it's a GP that may have already been ACKed
187          * during a core's QS, and that core could have started a read-side critical
188          * section that must complete before CB runs.  That requires another GP. */
189         head->gpnum = READ_ONCE(rsp->gpnum) + 1;
190         spin_lock_irqsave(&rpi->lock);
191         list_add_tail(&head->link, &rpi->cbs);
192         nr_cbs = ++rpi->nr_cbs;
193         spin_unlock_irqsave(&rpi->lock);
194         /* rcu_barrier requires that the write to ->nr_cbs be visible before any
195          * future writes.  unlock orders the write inside, but doesn't prevent other
196          * writes from moving in.  Technically, our lock implementations do that,
197          * but it's not part of our definition.  Maybe it should be.  Til then: */
198         wmb();
199         return nr_cbs;
200 }
201
202 /* Minus the kfree offset check */
203 static void __call_rcu(struct rcu_head *head, rcu_callback_t func)
204 {
205         struct rcu_pcpui *rpi = PERCPU_VARPTR(rcu_pcpui);
206         struct rcu_state *rsp = rpi->rsp;
207         unsigned int thresh;
208
209         thresh = __call_rcu_rpi(rsp, rpi, head, func);
210         if (thresh > RCU_CB_THRESH)
211                 wake_gp_ktask(rpi->rsp, false);
212 }
213
214 void call_rcu(struct rcu_head *head, rcu_callback_t func)
215 {
216         assert(!__is_kfree_rcu_offset((unsigned long)func));
217         __call_rcu(head, func);
218 }
219
220 void rcu_barrier(void)
221 {
222         struct rcu_state *rsp = PERCPU_VAR(rcu_pcpui).rsp;
223         struct rcu_pcpui *rpi;
224         struct semaphore sem[1];
225         struct sync_cb_blob *b;
226         int nr_sent = 0;
227
228         if (in_rcu_cb_ctx(this_pcpui_ptr()))
229                 panic("Attempted rcu_barrier() from an RCU callback!");
230         if (is_rcu_ktask(current_kthread))
231                 panic("Attempted rcu_barrier() from an RCU thread!");
232         /* TODO: if we have concurrent rcu_barriers, we might be able to share the
233          * CBs.  Say we have 1 CB on a core, then N rcu_barriers.  We'll have N
234          * call_rcus in flight, though we could share.  Linux does this with a mtx
235          * and some accounting, I think. */
236
237         b = kzmalloc(sizeof(struct sync_cb_blob) * num_cores, MEM_WAIT);
238         /* Remember, you block when sem is <= 0.  We'll get nr_sent ups, and we'll
239          * down 1 for each.  This is just like the synchronize_rcu() case; there,
240          * nr_sent == 1. */
241         sem_init(sem, 0);
242         /* Order any signal we received from someone who called call_rcu() before
243          * our rpi->nr_cbs reads. */
244         rmb();
245         for_each_core(i) {
246                 rpi = _PERCPU_VARPTR(rcu_pcpui, i);
247                 /* Lockless peek at nr_cbs.  Two things to note here:
248                  * - We look at nr_cbs and not the list, since there could be CBs on the
249                  *   stack-local work list or that have blocked.
250                  * - The guarantee is that we wait for any CBs from call_rcus that can
251                  *   be proved to happen before rcu_barrier.  That means call_rcu had to
252                  *   return, which means it had to set the nr_cbs. */
253                 if (!rpi->nr_cbs)
254                         continue;
255                 init_rcu_head_on_stack(&b[i].h);
256                 b[i].sem = sem;
257                 __call_rcu_rpi(rsp, rpi, &b[i].h, __sync_cb);
258                 nr_sent++;
259         }
260         if (!nr_sent) {
261                 kfree(b);
262                 return;
263         }
264         wake_gp_ktask(rpi->rsp, true);
265         /* sem_down_bulk is currently slow.  Even with some fixes, we actually want
266          * a barrier, which you could imagine doing with a tree.  sem_down_bulk()
267          * doesn't have the info that we have: that the wakeups are coming from N
268          * cores on the leaves of the tree. */
269         sem_down_bulk(sem, nr_sent);
270         kfree(b);
271 }
272
273 void rcu_force_quiescent_state(void)
274 {
275         /* It's unclear if we want to block until the QS has passed */
276         wake_gp_ktask(PERCPU_VAR(rcu_pcpui).rsp, true);
277 }
278
279 void kfree_call_rcu(struct rcu_head *head, rcu_callback_t off)
280 {
281         __call_rcu(head, off);
282 }
283
284 /* Clears the bits core(s) in grpmask present in rnp, trickling up to the root.
285  * Note that a 1 in qsmask means you haven't checked in - like a todo list.
286  * Last one out kicks the GP kthread. */
287 static void __mark_qs(struct rcu_state *rsp, struct rcu_node *rnp,
288                       unsigned long grpmask)
289 {
290         unsigned long new_qsm;
291
292         new_qsm = __sync_and_and_fetch(&rnp->qsmask, ~grpmask);
293         /* I don't fully understand this, but we need some form of transitive
294          * barrier across the entire tree.  Linux does this when they lock/unlock.
295          * Our equivalent is the atomic op. */
296         smp_mb__after_unlock_lock();
297         /* Only one thread will get 0 back - the last one to check in */
298         if (new_qsm)
299                 return;
300         if (rnp->parent)
301                 __mark_qs(rsp, rnp->parent, rnp->grpmask);
302         else
303                 rendez_wakeup(&rsp->gp_ktask_rv);
304 }
305
306 static void rcu_report_qs_rpi(struct rcu_state *rsp, struct rcu_pcpui *rpi)
307 {
308         /* Note we don't check ->completed == ->gpnum (gp_in_progress()).  We only
309          * care if our core hasn't reported in for a GP.  This time is a subset of
310          * gp_in_progress. */
311         if (rpi->gp_acked == READ_ONCE(rsp->gpnum)) {
312                 /* If a GP starts right afterwards, oh well.  Catch it next time. */
313                 return;
314         }
315         /* Lock ensures we only report a QS once per GP. */
316         spin_lock_irqsave(&rpi->lock);
317         if (rpi->gp_acked == READ_ONCE(rsp->gpnum)) {
318                 spin_unlock_irqsave(&rpi->lock);
319                 return;
320         }
321         /* A gp can start concurrently, but once started, we should never be behind
322          * by more than 1. */
323         assert(rpi->gp_acked + 1 == READ_ONCE(rsp->gpnum));
324         /* Up our gp_acked before actually marking it.  I don't want to hold the
325          * lock too long (e.g. some debug code in rendez_wakeup() calls call_rcu).
326          * So we've unlocked, but haven't actually checked in yet - that's fine.  No
327          * one else will attempt to check in until the next GP, which can't happen
328          * until after we check in for this GP. */
329         rpi->gp_acked++;
330         spin_unlock_irqsave(&rpi->lock);
331         __mark_qs(rsp, rpi->my_node, rpi->grpmask);
332 }
333
334 /* Cores advertise when they are in QSs.  If the core already reported in, or if
335  * we're not in a GP, this is a quick check (given a global read of ->gpnum). */
336 void rcu_report_qs(void)
337 {
338         rcu_report_qs_rpi(&rcu_state, PERCPU_VARPTR(rcu_pcpui));
339 }
340
341 /* For debugging checks on large trees.  Keep this in sync with
342  * rcu_init_fake_cores(). */
343 static void rcu_report_qs_fake_cores(struct rcu_state *rsp)
344 {
345         struct rcu_node *rnp;
346
347         rnp = rsp->level[rcu_num_lvls - 1];
348         for (int i = num_cores; i < rcu_num_cores; i++) {
349                 while (i > rnp->grphi)
350                         rnp++;
351                 if (rcu_debug_tardy && (i % 2))
352                         continue;
353                 __mark_qs(rsp, rnp, 1 << (i - rnp->grplo));
354         }
355 }
356
357 static void rcu_report_qs_remote_core(struct rcu_state *rsp, int coreid)
358 {
359         int cpu_state = READ_ONCE(pcpui_var(coreid, cpu_state));
360         struct rcu_pcpui *rpi = _PERCPU_VARPTR(rcu_pcpui, coreid);
361
362         /* Lockless peek.  If we ever saw them idle/user after a GP started, we
363          * know they had a QS, and we know we're still in the original GP. */
364         if (cpu_state == CPU_STATE_IDLE || cpu_state == CPU_STATE_USER)
365                 rcu_report_qs_rpi(rsp, rpi);
366 }
367
368 /* Checks every core, remotely via the cpu state, to see if it is in a QS.
369  * This is like an expedited grace period. */
370 static void rcu_report_qs_remote_cores(struct rcu_state *rsp)
371 {
372         for_each_core(i) {
373                 if (rcu_debug_tardy && (i % 2))
374                         continue;
375                 rcu_report_qs_remote_core(rsp, i);
376         }
377 }
378
379 static void rcu_report_qs_tardy_cores(struct rcu_state *rsp)
380 {
381         struct rcu_node *rnp;
382         unsigned long qsmask;
383         int i;
384
385         rcu_for_each_leaf_node(rsp, rnp) {
386                 qsmask = READ_ONCE(rnp->qsmask);
387                 if (!qsmask)
388                         continue;
389                 for_each_set_bit(i, &qsmask, BITS_PER_LONG) {
390                         /* Fake cores */
391                         if (i + rnp->grplo >= num_cores) {
392                                 __mark_qs(rsp, rnp, 1 << i);
393                                 continue;
394                         }
395                         rcu_report_qs_remote_core(rsp, i + rnp->grplo);
396                 }
397         }
398 }
399
400 static int root_qsmask_empty(void *arg)
401 {
402         struct rcu_state *rsp = arg;
403
404         return READ_ONCE(rsp->node[0].qsmask) == 0 ? 1 : 0;
405 }
406
407 static void rcu_run_gp(struct rcu_state *rsp)
408 {
409         struct rcu_node *rnp;
410
411         assert(rsp->gpnum == rsp->completed);
412         /* Initialize the tree for accumulating QSs.  We know there are no users on
413          * the tree.  The only time a core looks at the tree is when reporting a QS
414          * for a GP.  The previous GP is done, thus all cores reported their GP
415          * already (for the previous GP), and they won't try again until we
416          * advertise the next GP. */
417         rcu_for_each_node_breadth_first(rsp, rnp)
418                 rnp->qsmask = rnp->qsmaskinit;
419         /* Need the tree set for reporting QSs before advertising the GP */
420         wmb();
421         WRITE_ONCE(rsp->gpnum, rsp->gpnum + 1);
422         /* At this point, the cores can start reporting in. */
423         /* Fake cores help test a tree larger than num_cores. */
424         rcu_report_qs_fake_cores(rsp);
425         /* Expediting aggressively.  We could also wait briefly and then check the
426          * tardy cores. */
427         rcu_report_qs_remote_cores(rsp);
428         /* Note that even when we expedite the GP by checking remote cores, there's
429          * a race where a core halted but we didn't see it.  (they report QS, decide
430          * to halt, pause, we start GP, see they haven't halted, etc.  They could
431          * report the QS after setting the state, but I didn't want to . */
432         do {
433                 rendez_sleep_timeout(&rsp->gp_ktask_rv, root_qsmask_empty, rsp,
434                                      RCU_GP_TARDY_PERIOD);
435                 rcu_report_qs_tardy_cores(rsp);
436         } while (!root_qsmask_empty(rsp));
437         /* Not sure if we need any barriers here.  Once we post 'completed', the CBs
438          * can start running.  But no one should touch the tree til gpnum is
439          * incremented. */
440         WRITE_ONCE(rsp->completed, rsp->gpnum);
441 }
442
443 static int should_wake_ctl(void *arg)
444 {
445         int *ctl = arg;
446
447         return *ctl != 0 ? 1 : 0;
448 }
449
450 static void wake_mgmt_ktasks(struct rcu_state *rsp)
451 {
452         struct rcu_pcpui *rpi;
453
454         /* TODO: For each mgmt core */
455         rpi = _PERCPU_VARPTR(rcu_pcpui, 0);
456         rpi->mgmt_ktask_ctl = 1;
457         rendez_wakeup(&rpi->mgmt_ktask_rv);
458 }
459
460 static void rcu_gp_ktask(void *arg)
461 {
462         struct rcu_state *rsp = arg;
463
464         current_kthread->flags |= KTH_IS_RCU_KTASK;
465         while (1) {
466                 rendez_sleep_timeout(&rsp->gp_ktask_rv, should_wake_ctl,
467                                      &rsp->gp_ktask_ctl, RCU_GP_MIN_PERIOD);
468                 rsp->gp_ktask_ctl = 0;
469                 /* Our write of 0 must happen before starting the GP.  If rcu_barrier's
470                  * CBs miss the start of the GP (and thus are in an unscheduled GP),
471                  * their write of 1 must happen after our write of 0 so that we rerun.
472                  * This is the post-and-poke pattern.  It's not a huge deal, since we'll
473                  * catch it after the GP period timeout. */
474                 wmb();
475                 rcu_run_gp(rsp);
476                 wake_mgmt_ktasks(rsp);
477         };
478 }
479
480 static void run_rcu_cbs(struct rcu_state *rsp, int coreid)
481 {
482         struct rcu_pcpui *rpi = _PERCPU_VARPTR(rcu_pcpui, coreid);
483         struct list_head work = LIST_HEAD_INIT(work);
484         struct rcu_head *head, *temp, *last_for_gp = NULL;
485         int nr_cbs = 0;
486         unsigned long completed;
487
488         /* We'll run the CBs for any GP completed so far, but not any GP that could
489          * be completed concurrently.  "CBs for a GP" means callbacks that must wait
490          * for that GP to complete. */
491         completed = READ_ONCE(rsp->completed);
492
493         /* This lockless peek is an optimization.  We're guaranteed to not miss the
494          * CB for the given GP: If the core had a CB for this GP, it must have
495          * put it on the list before checking in, before the GP completes, and
496          * before we run. */
497         if (list_empty(&rpi->cbs))
498                 return;
499
500         spin_lock_irqsave(&rpi->lock);
501         list_for_each_entry(head, &rpi->cbs, link) {
502                 if (ULONG_CMP_LT(completed, head->gpnum))
503                         break;
504                 nr_cbs++;
505                 last_for_gp = head;
506         }
507         if (last_for_gp)
508                 list_cut_position(&work, &rpi->cbs, &last_for_gp->link);
509         spin_unlock_irqsave(&rpi->lock);
510
511         if (!nr_cbs) {
512                 assert(list_empty(&work));
513                 return;
514         }
515         /* When we're in an RCU callback, we can't block.  In our non-preemptive
516          * world, not blocking also means our kthread won't migrate from this core,
517          * such that the pcpui pointer (and thus the specific __ctx_depth) won't
518          * change. */
519         set_rcu_cb(this_pcpui_ptr());
520         list_for_each_entry_safe(head, temp, &work, link) {
521                 list_del(&head->link);
522                 rcu_exec_cb(head);
523         }
524         clear_rcu_cb(this_pcpui_ptr());
525
526         /* We kept nr_cbs in place until the CBs, which could block, completed.
527          * This allows other readers (rcu_barrier()) of our pcpui to tell if we have
528          * any CBs pending.  This relies on us being the only consumer/runner of CBs
529          * for this core. */
530         spin_lock_irqsave(&rpi->lock);
531         rpi->nr_cbs -= nr_cbs;
532         spin_unlock_irqsave(&rpi->lock);
533 }
534
535 static void rcu_mgmt_ktask(void *arg)
536 {
537         struct rcu_pcpui *rpi = arg;
538         struct rcu_state *rsp = rpi->rsp;
539
540         current_kthread->flags |= KTH_IS_RCU_KTASK;
541         while (1) {
542                 rendez_sleep(&rpi->mgmt_ktask_rv, should_wake_ctl,
543                              &rpi->mgmt_ktask_ctl);
544                 rpi->mgmt_ktask_ctl = 0;
545                 /* TODO: given the number of mgmt kthreads, we need to assign cores */
546                 for_each_core(i)
547                         run_rcu_cbs(rsp, i);
548         };
549 }
550
551 void rcu_init_pcpui(struct rcu_state *rsp, struct rcu_pcpui *rpi, int coreid)
552 {
553         struct rcu_node *rnp = rpi->my_node;
554
555         rpi->rsp = rsp;
556         assert(rnp->grplo <= coreid);
557         assert(coreid <= rnp->grphi);
558         rpi->coreid = coreid;
559         rpi->grpnum = coreid - rnp->grplo;
560         rpi->grpmask = 1 << rpi->grpnum;
561         rpi->booted = false;
562
563         /* We're single threaded now, so this is OK. */
564         rnp->qsmaskinit |= rpi->grpmask;
565
566         spinlock_init_irqsave(&rpi->lock);
567         INIT_LIST_HEAD(&rpi->cbs);
568         rpi->nr_cbs = 0;
569         rpi->gp_acked = rsp->completed;
570
571         /* TODO: For each mgmt core only */
572         if (coreid == 0) {
573                 rendez_init(&rpi->mgmt_ktask_rv);
574                 rpi->mgmt_ktask_ctl = 0;
575         }
576 }
577
578 /* Initializes the fake cores.  Works with rcu_report_qs_fake_cores() */
579 static void rcu_init_fake_cores(struct rcu_state *rsp)
580 {
581         struct rcu_node *rnp;
582
583         rnp = rsp->level[rcu_num_lvls - 1];
584         for (int i = num_cores; i < rcu_num_cores; i++) {
585                 while (i > rnp->grphi)
586                         rnp++;
587                 rnp->qsmaskinit |= 1 << (i - rnp->grplo);
588         }
589 }
590
591 void rcu_init(void)
592 {
593         struct rcu_state *rsp = &rcu_state;
594         struct rcu_pcpui *rpi;
595
596         rcu_init_geometry();
597         rcu_init_one(rsp);
598         rcu_init_fake_cores(rsp);
599         rcu_dump_rcu_node_tree(rsp);
600
601         ktask("rcu_gp", rcu_gp_ktask, rsp);
602         /* TODO: For each mgmt core */
603         ktask("rcu_mgmt_0", rcu_mgmt_ktask, _PERCPU_VARPTR(rcu_pcpui, 0));
604
605         /* If we have a call_rcu before percpu_init, we might be using the spot in
606          * the actual __percpu .section.  We'd be core 0, so that'd be OK, since all
607          * we're using it for is reading 'booted'. */
608         for_each_core(i) {
609                 rpi = _PERCPU_VARPTR(rcu_pcpui, i);
610                 rpi->booted = true;
611         }
612 }