9a4bc36bb3419c5752517fe824deb61ace50d4b1
[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         sem_init(sem, 0);
120         init_rcu_head_on_stack(&b->h);
121         b->sem = sem;
122         call_rcu(&b->h, __sync_cb);
123         sem_down(sem);
124 }
125
126 static inline bool gp_in_progress(struct rcu_state *rsp)
127 {
128         unsigned long completed = READ_ONCE(rsp->completed);
129         unsigned long gpnum = READ_ONCE(rsp->gpnum);
130
131         assert(gpnum - completed <= 1);
132         return completed != gpnum;
133 }
134
135 /* Wakes the kthread to run a grace period if it isn't already running.
136  *
137  * If 'force', we'll make sure it runs a fresh GP, which will catch all CBs
138  * registered before this call.  That's not 100% true.  It might be possible on
139  * some non-x86 architectures for the writes that wake the ktask are reordered
140  * before the read of gpnum that our caller made.  Thus the caller could have a
141  * CB in a later GP.  Worst case, they'll wait an extra GP timeout.  Not too
142  * concerned, though I probably should be. */
143 static void wake_gp_ktask(struct rcu_state *rsp, bool force)
144 {
145         if (!force && gp_in_progress(rsp))
146                 return;
147         rsp->gp_ktask_ctl = 1;
148         rendez_wakeup(&rsp->gp_ktask_rv);
149 }
150
151 static void rcu_exec_cb(struct rcu_head *head)
152 {
153         if (__is_kfree_rcu_offset((unsigned long)head->func))
154                 kfree((void*)head - (unsigned long)head->func);
155         else
156                 head->func(head);
157 }
158
159 static void __early_call_rcu_kmsg(uint32_t srcid, long a0, long a1, long a2)
160 {
161         rcu_exec_cb((struct rcu_head*)a0);
162 }
163
164 void __early_call_rcu(struct rcu_head *head)
165 {
166         extern bool booting;
167
168         assert(booting);
169         assert(core_id() == 0);
170         send_kernel_message(0, __early_call_rcu_kmsg, (long)head, 0, 0,
171                             KMSG_ROUTINE);
172 }
173
174 /* This could be called from a remote core, e.g. rcu_barrier().  Returns the
175  * number of enqueued CBs, including the one we pass in. */
176 static int __call_rcu_rpi(struct rcu_state *rsp, struct rcu_pcpui *rpi,
177                            struct rcu_head *head, rcu_callback_t func)
178 {
179         unsigned int nr_cbs;
180
181         head->func = func;
182
183         if (!rpi->booted) {
184                 __early_call_rcu(head);
185                 return 0;
186         }
187         /* rsp->gpnum is the one we're either working on (if > completed) or the one
188          * we already did.  Either way, it's a GP that may have already been ACKed
189          * during a core's QS, and that core could have started a read-side critical
190          * section that must complete before CB runs.  That requires another GP. */
191         head->gpnum = READ_ONCE(rsp->gpnum) + 1;
192         spin_lock_irqsave(&rpi->lock);
193         list_add_tail(&head->link, &rpi->cbs);
194         nr_cbs = ++rpi->nr_cbs;
195         spin_unlock_irqsave(&rpi->lock);
196         /* rcu_barrier requires that the write to ->nr_cbs be visible before any
197          * future writes.  unlock orders the write inside, but doesn't prevent other
198          * writes from moving in.  Technically, our lock implementations do that,
199          * but it's not part of our definition.  Maybe it should be.  Til then: */
200         wmb();
201         return nr_cbs;
202 }
203
204 /* Minus the kfree offset check */
205 static void __call_rcu(struct rcu_head *head, rcu_callback_t func)
206 {
207         struct rcu_pcpui *rpi = PERCPU_VARPTR(rcu_pcpui);
208         struct rcu_state *rsp = rpi->rsp;
209         unsigned int thresh;
210
211         thresh = __call_rcu_rpi(rsp, rpi, head, func);
212         if (thresh > RCU_CB_THRESH)
213                 wake_gp_ktask(rpi->rsp, false);
214 }
215
216 void call_rcu(struct rcu_head *head, rcu_callback_t func)
217 {
218         assert(!__is_kfree_rcu_offset((unsigned long)func));
219         __call_rcu(head, func);
220 }
221
222 void rcu_barrier(void)
223 {
224         struct rcu_state *rsp = PERCPU_VAR(rcu_pcpui).rsp;
225         struct rcu_pcpui *rpi;
226         struct semaphore sem[1];
227         struct sync_cb_blob *b;
228         int nr_sent = 0;
229
230         /* TODO: if we have concurrent rcu_barriers, we might be able to share the
231          * CBs.  Say we have 1 CB on a core, then N rcu_barriers.  We'll have N
232          * call_rcus in flight, though we could share.  Linux does this with a mtx
233          * and some accounting, I think. */
234
235         b = kzmalloc(sizeof(struct sync_cb_blob) * num_cores, MEM_WAIT);
236         /* Remember, you block when sem is <= 0.  We'll get nr_sent ups, and we'll
237          * down 1 for each.  This is just like the synchronize_rcu() case; there,
238          * nr_sent == 1. */
239         sem_init(sem, 0);
240         /* Order any signal we received from someone who called call_rcu() before
241          * our rpi->nr_cbs reads. */
242         rmb();
243         for_each_core(i) {
244                 rpi = _PERCPU_VARPTR(rcu_pcpui, i);
245                 /* Lockless peek at nr_cbs.  Two things to note here:
246                  * - We look at nr_cbs and not the list, since there could be CBs on the
247                  *   stack-local work list or that have blocked.
248                  * - The guarantee is that we wait for any CBs from call_rcus that can
249                  *   be proved to happen before rcu_barrier.  That means call_rcu had to
250                  *   return, which means it had to set the nr_cbs. */
251                 if (!rpi->nr_cbs)
252                         continue;
253                 init_rcu_head_on_stack(&b[i].h);
254                 b[i].sem = sem;
255                 __call_rcu_rpi(rsp, rpi, &b[i].h, __sync_cb);
256                 nr_sent++;
257         }
258         if (!nr_sent) {
259                 kfree(b);
260                 return;
261         }
262         wake_gp_ktask(rpi->rsp, true);
263         /* sem_down_bulk is currently slow.  Even with some fixes, we actually want
264          * a barrier, which you could imagine doing with a tree.  sem_down_bulk()
265          * doesn't have the info that we have: that the wakeups are coming from N
266          * cores on the leaves of the tree. */
267         sem_down_bulk(sem, nr_sent);
268         kfree(b);
269 }
270
271 void rcu_force_quiescent_state(void)
272 {
273         /* It's unclear if we want to block until the QS has passed */
274         wake_gp_ktask(PERCPU_VAR(rcu_pcpui).rsp, true);
275 }
276
277 void kfree_call_rcu(struct rcu_head *head, rcu_callback_t off)
278 {
279         __call_rcu(head, off);
280 }
281
282 /* Clears the bits core(s) in grpmask present in rnp, trickling up to the root.
283  * Note that a 1 in qsmask means you haven't checked in - like a todo list.
284  * Last one out kicks the GP kthread. */
285 static void __mark_qs(struct rcu_state *rsp, struct rcu_node *rnp,
286                       unsigned long grpmask)
287 {
288         unsigned long new_qsm;
289
290         new_qsm = __sync_and_and_fetch(&rnp->qsmask, ~grpmask);
291         /* I don't fully understand this, but we need some form of transitive
292          * barrier across the entire tree.  Linux does this when they lock/unlock.
293          * Our equivalent is the atomic op. */
294         smp_mb__after_unlock_lock();
295         /* Only one thread will get 0 back - the last one to check in */
296         if (new_qsm)
297                 return;
298         if (rnp->parent)
299                 __mark_qs(rsp, rnp->parent, rnp->grpmask);
300         else
301                 rendez_wakeup(&rsp->gp_ktask_rv);
302 }
303
304 static void rcu_report_qs_rpi(struct rcu_state *rsp, struct rcu_pcpui *rpi)
305 {
306         /* Note we don't check ->completed == ->gpnum (gp_in_progress()).  We only
307          * care if our core hasn't reported in for a GP.  This time is a subset of
308          * gp_in_progress. */
309         if (rpi->gp_acked == READ_ONCE(rsp->gpnum)) {
310                 /* If a GP starts right afterwards, oh well.  Catch it next time. */
311                 return;
312         }
313         /* Lock ensures we only report a QS once per GP. */
314         spin_lock_irqsave(&rpi->lock);
315         if (rpi->gp_acked == READ_ONCE(rsp->gpnum)) {
316                 spin_unlock_irqsave(&rpi->lock);
317                 return;
318         }
319         /* A gp can start concurrently, but once started, we should never be behind
320          * by more than 1. */
321         assert(rpi->gp_acked + 1 == READ_ONCE(rsp->gpnum));
322         /* Up our gp_acked before actually marking it.  I don't want to hold the
323          * lock too long (e.g. some debug code in rendez_wakeup() calls call_rcu).
324          * So we've unlocked, but haven't actually checked in yet - that's fine.  No
325          * one else will attempt to check in until the next GP, which can't happen
326          * until after we check in for this GP. */
327         rpi->gp_acked++;
328         spin_unlock_irqsave(&rpi->lock);
329         __mark_qs(rsp, rpi->my_node, rpi->grpmask);
330 }
331
332 /* Cores advertise when they are in QSs.  If the core already reported in, or if
333  * we're not in a GP, this is a quick check (given a global read of ->gpnum). */
334 void rcu_report_qs(void)
335 {
336         rcu_report_qs_rpi(&rcu_state, PERCPU_VARPTR(rcu_pcpui));
337 }
338
339 /* For debugging checks on large trees.  Keep this in sync with
340  * rcu_init_fake_cores(). */
341 static void rcu_report_qs_fake_cores(struct rcu_state *rsp)
342 {
343         struct rcu_node *rnp;
344
345         rnp = rsp->level[rcu_num_lvls - 1];
346         for (int i = num_cores; i < rcu_num_cores; i++) {
347                 while (i > rnp->grphi)
348                         rnp++;
349                 if (rcu_debug_tardy && (i % 2))
350                         continue;
351                 __mark_qs(rsp, rnp, 1 << (i - rnp->grplo));
352         }
353 }
354
355 static void rcu_report_qs_remote_core(struct rcu_state *rsp, int coreid)
356 {
357         int cpu_state = READ_ONCE(pcpui_var(coreid, cpu_state));
358         struct rcu_pcpui *rpi = _PERCPU_VARPTR(rcu_pcpui, coreid);
359
360         /* Lockless peek.  If we ever saw them idle/user after a GP started, we
361          * know they had a QS, and we know we're still in the original GP. */
362         if (cpu_state == CPU_STATE_IDLE || cpu_state == CPU_STATE_USER)
363                 rcu_report_qs_rpi(rsp, rpi);
364 }
365
366 /* Checks every core, remotely via the cpu state, to see if it is in a QS.
367  * This is like an expedited grace period. */
368 static void rcu_report_qs_remote_cores(struct rcu_state *rsp)
369 {
370         for_each_core(i) {
371                 if (rcu_debug_tardy && (i % 2))
372                         continue;
373                 rcu_report_qs_remote_core(rsp, i);
374         }
375 }
376
377 static void rcu_report_qs_tardy_cores(struct rcu_state *rsp)
378 {
379         struct rcu_node *rnp;
380         unsigned long qsmask;
381         int i;
382
383         rcu_for_each_leaf_node(rsp, rnp) {
384                 qsmask = READ_ONCE(rnp->qsmask);
385                 if (!qsmask)
386                         continue;
387                 for_each_set_bit(i, &qsmask, BITS_PER_LONG) {
388                         /* Fake cores */
389                         if (i + rnp->grplo >= num_cores) {
390                                 __mark_qs(rsp, rnp, 1 << i);
391                                 continue;
392                         }
393                         rcu_report_qs_remote_core(rsp, i + rnp->grplo);
394                 }
395         }
396 }
397
398 static int root_qsmask_empty(void *arg)
399 {
400         struct rcu_state *rsp = arg;
401
402         return READ_ONCE(rsp->node[0].qsmask) == 0 ? 1 : 0;
403 }
404
405 static void rcu_run_gp(struct rcu_state *rsp)
406 {
407         struct rcu_node *rnp;
408
409         assert(rsp->gpnum == rsp->completed);
410         /* Initialize the tree for accumulating QSs.  We know there are no users on
411          * the tree.  The only time a core looks at the tree is when reporting a QS
412          * for a GP.  The previous GP is done, thus all cores reported their GP
413          * already (for the previous GP), and they won't try again until we
414          * advertise the next GP. */
415         rcu_for_each_node_breadth_first(rsp, rnp)
416                 rnp->qsmask = rnp->qsmaskinit;
417         /* Need the tree set for reporting QSs before advertising the GP */
418         wmb();
419         WRITE_ONCE(rsp->gpnum, rsp->gpnum + 1);
420         /* At this point, the cores can start reporting in. */
421         /* Fake cores help test a tree larger than num_cores. */
422         rcu_report_qs_fake_cores(rsp);
423         /* Expediting aggressively.  We could also wait briefly and then check the
424          * tardy cores. */
425         rcu_report_qs_remote_cores(rsp);
426         /* Note that even when we expedite the GP by checking remote cores, there's
427          * a race where a core halted but we didn't see it.  (they report QS, decide
428          * to halt, pause, we start GP, see they haven't halted, etc.  They could
429          * report the QS after setting the state, but I didn't want to . */
430         do {
431                 rendez_sleep_timeout(&rsp->gp_ktask_rv, root_qsmask_empty, rsp,
432                                      RCU_GP_TARDY_PERIOD);
433                 rcu_report_qs_tardy_cores(rsp);
434         } while (!root_qsmask_empty(rsp));
435         /* Not sure if we need any barriers here.  Once we post 'completed', the CBs
436          * can start running.  But no one should touch the tree til gpnum is
437          * incremented. */
438         WRITE_ONCE(rsp->completed, rsp->gpnum);
439 }
440
441 static int should_wake_ctl(void *arg)
442 {
443         int *ctl = arg;
444
445         return *ctl != 0 ? 1 : 0;
446 }
447
448 static void wake_mgmt_ktasks(struct rcu_state *rsp)
449 {
450         struct rcu_pcpui *rpi;
451
452         /* TODO: For each mgmt core */
453         rpi = _PERCPU_VARPTR(rcu_pcpui, 0);
454         rpi->mgmt_ktask_ctl = 1;
455         rendez_wakeup(&rpi->mgmt_ktask_rv);
456 }
457
458 static void rcu_gp_ktask(void *arg)
459 {
460         struct rcu_state *rsp = arg;
461
462         while (1) {
463                 rendez_sleep_timeout(&rsp->gp_ktask_rv, should_wake_ctl,
464                                      &rsp->gp_ktask_ctl, RCU_GP_MIN_PERIOD);
465                 rsp->gp_ktask_ctl = 0;
466                 /* Our write of 0 must happen before starting the GP.  If rcu_barrier's
467                  * CBs miss the start of the GP (and thus are in an unscheduled GP),
468                  * their write of 1 must happen after our write of 0 so that we rerun.
469                  * This is the post-and-poke pattern.  It's not a huge deal, since we'll
470                  * catch it after the GP period timeout. */
471                 wmb();
472                 rcu_run_gp(rsp);
473                 wake_mgmt_ktasks(rsp);
474         };
475 }
476
477 static void run_rcu_cbs(struct rcu_state *rsp, int coreid)
478 {
479         struct rcu_pcpui *rpi = _PERCPU_VARPTR(rcu_pcpui, coreid);
480         struct list_head work = LIST_HEAD_INIT(work);
481         struct rcu_head *head, *temp, *last_for_gp = NULL;
482         int nr_cbs = 0;
483         unsigned long completed;
484
485         /* We'll run the CBs for any GP completed so far, but not any GP that could
486          * be completed concurrently.  "CBs for a GP" means callbacks that must wait
487          * for that GP to complete. */
488         completed = READ_ONCE(rsp->completed);
489
490         /* This lockless peek is an optimization.  We're guaranteed to not miss the
491          * CB for the given GP: If the core had a CB for this GP, it must have
492          * put it on the list before checking in, before the GP completes, and
493          * before we run. */
494         if (list_empty(&rpi->cbs))
495                 return;
496
497         spin_lock_irqsave(&rpi->lock);
498         list_for_each_entry(head, &rpi->cbs, link) {
499                 if (ULONG_CMP_LT(completed, head->gpnum))
500                         break;
501                 nr_cbs++;
502                 last_for_gp = head;
503         }
504         if (last_for_gp)
505                 list_cut_position(&work, &rpi->cbs, &last_for_gp->link);
506         spin_unlock_irqsave(&rpi->lock);
507
508         if (!nr_cbs) {
509                 assert(list_empty(&work));
510                 return;
511         }
512         list_for_each_entry_safe(head, temp, &work, link) {
513                 list_del(&head->link);
514                 rcu_exec_cb(head);
515         }
516
517         /* We kept nr_cbs in place until the CBs, which could block, completed.
518          * This allows other readers (rcu_barrier()) of our pcpui to tell if we have
519          * any CBs pending.  This relies on us being the only consumer/runner of CBs
520          * for this core. */
521         spin_lock_irqsave(&rpi->lock);
522         rpi->nr_cbs -= nr_cbs;
523         spin_unlock_irqsave(&rpi->lock);
524 }
525
526 static void rcu_mgmt_ktask(void *arg)
527 {
528         struct rcu_pcpui *rpi = arg;
529         struct rcu_state *rsp = rpi->rsp;
530
531         while (1) {
532                 rendez_sleep(&rpi->mgmt_ktask_rv, should_wake_ctl,
533                              &rpi->mgmt_ktask_ctl);
534                 rpi->mgmt_ktask_ctl = 0;
535                 /* TODO: given the number of mgmt kthreads, we need to assign cores */
536                 for_each_core(i)
537                         run_rcu_cbs(rsp, i);
538         };
539 }
540
541 void rcu_init_pcpui(struct rcu_state *rsp, struct rcu_pcpui *rpi, int coreid)
542 {
543         struct rcu_node *rnp = rpi->my_node;
544
545         rpi->rsp = rsp;
546         assert(rnp->grplo <= coreid);
547         assert(coreid <= rnp->grphi);
548         rpi->coreid = coreid;
549         rpi->grpnum = coreid - rnp->grplo;
550         rpi->grpmask = 1 << rpi->grpnum;
551         rpi->booted = false;
552
553         /* We're single threaded now, so this is OK. */
554         rnp->qsmaskinit |= rpi->grpmask;
555
556         spinlock_init_irqsave(&rpi->lock);
557         INIT_LIST_HEAD(&rpi->cbs);
558         rpi->nr_cbs = 0;
559         rpi->gp_acked = rsp->completed;
560
561         /* TODO: For each mgmt core only */
562         if (coreid == 0) {
563                 rendez_init(&rpi->mgmt_ktask_rv);
564                 rpi->mgmt_ktask_ctl = 0;
565         }
566 }
567
568 /* Initializes the fake cores.  Works with rcu_report_qs_fake_cores() */
569 static void rcu_init_fake_cores(struct rcu_state *rsp)
570 {
571         struct rcu_node *rnp;
572
573         rnp = rsp->level[rcu_num_lvls - 1];
574         for (int i = num_cores; i < rcu_num_cores; i++) {
575                 while (i > rnp->grphi)
576                         rnp++;
577                 rnp->qsmaskinit |= 1 << (i - rnp->grplo);
578         }
579 }
580
581 void rcu_init(void)
582 {
583         struct rcu_state *rsp = &rcu_state;
584         struct rcu_pcpui *rpi;
585
586         rcu_init_geometry();
587         rcu_init_one(rsp);
588         rcu_init_fake_cores(rsp);
589         rcu_dump_rcu_node_tree(rsp);
590
591         ktask("rcu_gp", rcu_gp_ktask, rsp);
592         /* TODO: For each mgmt core */
593         ktask("rcu_mgmt_0", rcu_mgmt_ktask, _PERCPU_VARPTR(rcu_pcpui, 0));
594
595         /* If we have a call_rcu before percpu_init, we might be using the spot in
596          * the actual __percpu .section.  We'd be core 0, so that'd be OK, since all
597          * we're using it for is reading 'booted'. */
598         for_each_core(i) {
599                 rpi = _PERCPU_VARPTR(rcu_pcpui, i);
600                 rpi->booted = true;
601         }
602 }