4d90887f9524092d9a86949b62d82b085bed7bb7
[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 (!can_block(this_pcpui_ptr()))
120                 panic("Attempted %s() from an unblockable context!", __func__);
121         if (is_rcu_ktask(current_kthread))
122                 panic("Attempted %s() from an RCU thread!", __func__);
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 (!can_block(this_pcpui_ptr()))
229                 panic("Attempted %s() from an unblockable context!", __func__);
230         if (is_rcu_ktask(current_kthread))
231                 panic("Attempted %s() from an RCU thread!", __func__);
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         b = kzmalloc(sizeof(struct sync_cb_blob) * num_cores, MEM_WAIT);
237         /* Remember, you block when sem is <= 0.  We'll get nr_sent ups, and we'll
238          * down 1 for each.  This is just like the synchronize_rcu() case; there,
239          * nr_sent == 1. */
240         sem_init(sem, 0);
241         /* Order any signal we received from someone who called call_rcu() before
242          * our rpi->nr_cbs reads. */
243         rmb();
244         for_each_core(i) {
245                 rpi = _PERCPU_VARPTR(rcu_pcpui, i);
246                 /* Lockless peek at nr_cbs.  Two things to note here:
247                  * - We look at nr_cbs and not the list, since there could be CBs on the
248                  *   stack-local work list or that have blocked.
249                  * - The guarantee is that we wait for any CBs from call_rcus that can
250                  *   be proved to happen before rcu_barrier.  That means call_rcu had to
251                  *   return, which means it had to set the nr_cbs. */
252                 if (!rpi->nr_cbs)
253                         continue;
254                 init_rcu_head_on_stack(&b[i].h);
255                 b[i].sem = sem;
256                 __call_rcu_rpi(rsp, rpi, &b[i].h, __sync_cb);
257                 nr_sent++;
258         }
259         if (!nr_sent) {
260                 kfree(b);
261                 return;
262         }
263         wake_gp_ktask(rpi->rsp, true);
264         /* sem_down_bulk is currently slow.  Even with some fixes, we actually want
265          * a barrier, which you could imagine doing with a tree.  sem_down_bulk()
266          * doesn't have the info that we have: that the wakeups are coming from N
267          * cores on the leaves of the tree. */
268         sem_down_bulk(sem, nr_sent);
269         kfree(b);
270 }
271
272 void rcu_force_quiescent_state(void)
273 {
274         /* It's unclear if we want to block until the QS has passed */
275         wake_gp_ktask(PERCPU_VAR(rcu_pcpui).rsp, true);
276 }
277
278 void kfree_call_rcu(struct rcu_head *head, rcu_callback_t off)
279 {
280         __call_rcu(head, off);
281 }
282
283 /* Clears the bits core(s) in grpmask present in rnp, trickling up to the root.
284  * Note that a 1 in qsmask means you haven't checked in - like a todo list.
285  * Last one out kicks the GP kthread. */
286 static void __mark_qs(struct rcu_state *rsp, struct rcu_node *rnp,
287                       unsigned long grpmask)
288 {
289         unsigned long new_qsm;
290
291         new_qsm = __sync_and_and_fetch(&rnp->qsmask, ~grpmask);
292         /* I don't fully understand this, but we need some form of transitive
293          * barrier across the entire tree.  Linux does this when they lock/unlock.
294          * Our equivalent is the atomic op. */
295         smp_mb__after_unlock_lock();
296         /* Only one thread will get 0 back - the last one to check in */
297         if (new_qsm)
298                 return;
299         if (rnp->parent)
300                 __mark_qs(rsp, rnp->parent, rnp->grpmask);
301         else
302                 rendez_wakeup(&rsp->gp_ktask_rv);
303 }
304
305 static void rcu_report_qs_rpi(struct rcu_state *rsp, struct rcu_pcpui *rpi)
306 {
307         /* Note we don't check ->completed == ->gpnum (gp_in_progress()).  We only
308          * care if our core hasn't reported in for a GP.  This time is a subset of
309          * gp_in_progress. */
310         if (rpi->gp_acked == READ_ONCE(rsp->gpnum)) {
311                 /* If a GP starts right afterwards, oh well.  Catch it next time. */
312                 return;
313         }
314         /* Lock ensures we only report a QS once per GP. */
315         spin_lock_irqsave(&rpi->lock);
316         if (rpi->gp_acked == READ_ONCE(rsp->gpnum)) {
317                 spin_unlock_irqsave(&rpi->lock);
318                 return;
319         }
320         /* A gp can start concurrently, but once started, we should never be behind
321          * by more than 1. */
322         assert(rpi->gp_acked + 1 == READ_ONCE(rsp->gpnum));
323         /* Up our gp_acked before actually marking it.  I don't want to hold the
324          * lock too long (e.g. some debug code in rendez_wakeup() calls call_rcu).
325          * So we've unlocked, but haven't actually checked in yet - that's fine.  No
326          * one else will attempt to check in until the next GP, which can't happen
327          * until after we check in for this GP. */
328         rpi->gp_acked++;
329         spin_unlock_irqsave(&rpi->lock);
330         __mark_qs(rsp, rpi->my_node, rpi->grpmask);
331 }
332
333 /* Cores advertise when they are in QSs.  If the core already reported in, or if
334  * we're not in a GP, this is a quick check (given a global read of ->gpnum). */
335 void rcu_report_qs(void)
336 {
337         rcu_report_qs_rpi(&rcu_state, PERCPU_VARPTR(rcu_pcpui));
338 }
339
340 /* For debugging checks on large trees.  Keep this in sync with
341  * rcu_init_fake_cores(). */
342 static void rcu_report_qs_fake_cores(struct rcu_state *rsp)
343 {
344         struct rcu_node *rnp;
345
346         rnp = rsp->level[rcu_num_lvls - 1];
347         for (int i = num_cores; i < rcu_num_cores; i++) {
348                 while (i > rnp->grphi)
349                         rnp++;
350                 if (rcu_debug_tardy && (i % 2))
351                         continue;
352                 __mark_qs(rsp, rnp, 1 << (i - rnp->grplo));
353         }
354 }
355
356 static void rcu_report_qs_remote_core(struct rcu_state *rsp, int coreid)
357 {
358         int cpu_state = READ_ONCE(pcpui_var(coreid, cpu_state));
359         struct rcu_pcpui *rpi = _PERCPU_VARPTR(rcu_pcpui, coreid);
360
361         /* Lockless peek.  If we ever saw them idle/user after a GP started, we
362          * know they had a QS, and we know we're still in the original GP. */
363         if (cpu_state == CPU_STATE_IDLE || cpu_state == CPU_STATE_USER)
364                 rcu_report_qs_rpi(rsp, rpi);
365 }
366
367 /* Checks every core, remotely via the cpu state, to see if it is in a QS.
368  * This is like an expedited grace period. */
369 static void rcu_report_qs_remote_cores(struct rcu_state *rsp)
370 {
371         for_each_core(i) {
372                 if (rcu_debug_tardy && (i % 2))
373                         continue;
374                 rcu_report_qs_remote_core(rsp, i);
375         }
376 }
377
378 static void rcu_report_qs_tardy_cores(struct rcu_state *rsp)
379 {
380         struct rcu_node *rnp;
381         unsigned long qsmask;
382         int i;
383
384         rcu_for_each_leaf_node(rsp, rnp) {
385                 qsmask = READ_ONCE(rnp->qsmask);
386                 if (!qsmask)
387                         continue;
388                 for_each_set_bit(i, &qsmask, BITS_PER_LONG) {
389                         /* Fake cores */
390                         if (i + rnp->grplo >= num_cores) {
391                                 __mark_qs(rsp, rnp, 1 << i);
392                                 continue;
393                         }
394                         rcu_report_qs_remote_core(rsp, i + rnp->grplo);
395                 }
396         }
397 }
398
399 static int root_qsmask_empty(void *arg)
400 {
401         struct rcu_state *rsp = arg;
402
403         return READ_ONCE(rsp->node[0].qsmask) == 0 ? 1 : 0;
404 }
405
406 static void rcu_run_gp(struct rcu_state *rsp)
407 {
408         struct rcu_node *rnp;
409
410         assert(rsp->gpnum == rsp->completed);
411         /* Initialize the tree for accumulating QSs.  We know there are no users on
412          * the tree.  The only time a core looks at the tree is when reporting a QS
413          * for a GP.  The previous GP is done, thus all cores reported their GP
414          * already (for the previous GP), and they won't try again until we
415          * advertise the next GP. */
416         rcu_for_each_node_breadth_first(rsp, rnp)
417                 rnp->qsmask = rnp->qsmaskinit;
418         /* Need the tree set for reporting QSs before advertising the GP */
419         wmb();
420         WRITE_ONCE(rsp->gpnum, rsp->gpnum + 1);
421         /* At this point, the cores can start reporting in. */
422         /* Fake cores help test a tree larger than num_cores. */
423         rcu_report_qs_fake_cores(rsp);
424         /* Expediting aggressively.  We could also wait briefly and then check the
425          * tardy cores. */
426         rcu_report_qs_remote_cores(rsp);
427         /* Note that even when we expedite the GP by checking remote cores, there's
428          * a race where a core halted but we didn't see it.  (they report QS, decide
429          * to halt, pause, we start GP, see they haven't halted, etc.  They could
430          * report the QS after setting the state, but I didn't want to . */
431         do {
432                 rendez_sleep_timeout(&rsp->gp_ktask_rv, root_qsmask_empty, rsp,
433                                      RCU_GP_TARDY_PERIOD);
434                 rcu_report_qs_tardy_cores(rsp);
435         } while (!root_qsmask_empty(rsp));
436         /* Not sure if we need any barriers here.  Once we post 'completed', the CBs
437          * can start running.  But no one should touch the tree til gpnum is
438          * incremented. */
439         WRITE_ONCE(rsp->completed, rsp->gpnum);
440 }
441
442 static int should_wake_ctl(void *arg)
443 {
444         int *ctl = arg;
445
446         return *ctl != 0 ? 1 : 0;
447 }
448
449 static void wake_mgmt_ktasks(struct rcu_state *rsp)
450 {
451         struct rcu_pcpui *rpi;
452
453         /* TODO: For each mgmt core */
454         rpi = _PERCPU_VARPTR(rcu_pcpui, 0);
455         rpi->mgmt_ktask_ctl = 1;
456         rendez_wakeup(&rpi->mgmt_ktask_rv);
457 }
458
459 static void rcu_gp_ktask(void *arg)
460 {
461         struct rcu_state *rsp = arg;
462
463         current_kthread->flags |= KTH_IS_RCU_KTASK;
464         while (1) {
465                 rendez_sleep_timeout(&rsp->gp_ktask_rv, should_wake_ctl,
466                                      &rsp->gp_ktask_ctl, RCU_GP_MIN_PERIOD);
467                 rsp->gp_ktask_ctl = 0;
468                 /* Our write of 0 must happen before starting the GP.  If rcu_barrier's
469                  * CBs miss the start of the GP (and thus are in an unscheduled GP),
470                  * their write of 1 must happen after our write of 0 so that we rerun.
471                  * This is the post-and-poke pattern.  It's not a huge deal, since we'll
472                  * catch it after the GP period timeout. */
473                 wmb();
474                 rcu_run_gp(rsp);
475                 wake_mgmt_ktasks(rsp);
476         };
477 }
478
479 static void run_rcu_cbs(struct rcu_state *rsp, int coreid)
480 {
481         struct rcu_pcpui *rpi = _PERCPU_VARPTR(rcu_pcpui, coreid);
482         struct list_head work = LIST_HEAD_INIT(work);
483         struct rcu_head *head, *temp, *last_for_gp = NULL;
484         int nr_cbs = 0;
485         unsigned long completed;
486
487         /* We'll run the CBs for any GP completed so far, but not any GP that could
488          * be completed concurrently.  "CBs for a GP" means callbacks that must wait
489          * for that GP to complete. */
490         completed = READ_ONCE(rsp->completed);
491
492         /* This lockless peek is an optimization.  We're guaranteed to not miss the
493          * CB for the given GP: If the core had a CB for this GP, it must have
494          * put it on the list before checking in, before the GP completes, and
495          * before we run. */
496         if (list_empty(&rpi->cbs))
497                 return;
498
499         spin_lock_irqsave(&rpi->lock);
500         list_for_each_entry(head, &rpi->cbs, link) {
501                 if (ULONG_CMP_LT(completed, head->gpnum))
502                         break;
503                 nr_cbs++;
504                 last_for_gp = head;
505         }
506         if (last_for_gp)
507                 list_cut_position(&work, &rpi->cbs, &last_for_gp->link);
508         spin_unlock_irqsave(&rpi->lock);
509
510         if (!nr_cbs) {
511                 assert(list_empty(&work));
512                 return;
513         }
514         /* When we're in an RCU callback, we can't block.  In our non-preemptive
515          * world, not blocking also means our kthread won't migrate from this core,
516          * such that the pcpui pointer (and thus the specific __ctx_depth) won't
517          * change. */
518         set_cannot_block(this_pcpui_ptr());
519         list_for_each_entry_safe(head, temp, &work, link) {
520                 list_del(&head->link);
521                 rcu_exec_cb(head);
522         }
523         clear_cannot_block(this_pcpui_ptr());
524
525         /* We kept nr_cbs in place until the CBs, which could block, completed.
526          * This allows other readers (rcu_barrier()) of our pcpui to tell if we have
527          * any CBs pending.  This relies on us being the only consumer/runner of CBs
528          * for this core. */
529         spin_lock_irqsave(&rpi->lock);
530         rpi->nr_cbs -= nr_cbs;
531         spin_unlock_irqsave(&rpi->lock);
532 }
533
534 static void rcu_mgmt_ktask(void *arg)
535 {
536         struct rcu_pcpui *rpi = arg;
537         struct rcu_state *rsp = rpi->rsp;
538
539         current_kthread->flags |= KTH_IS_RCU_KTASK;
540         while (1) {
541                 rendez_sleep(&rpi->mgmt_ktask_rv, should_wake_ctl,
542                              &rpi->mgmt_ktask_ctl);
543                 rpi->mgmt_ktask_ctl = 0;
544                 /* TODO: given the number of mgmt kthreads, we need to assign cores */
545                 for_each_core(i)
546                         run_rcu_cbs(rsp, i);
547         };
548 }
549
550 void rcu_init_pcpui(struct rcu_state *rsp, struct rcu_pcpui *rpi, int coreid)
551 {
552         struct rcu_node *rnp = rpi->my_node;
553
554         rpi->rsp = rsp;
555         assert(rnp->grplo <= coreid);
556         assert(coreid <= rnp->grphi);
557         rpi->coreid = coreid;
558         rpi->grpnum = coreid - rnp->grplo;
559         rpi->grpmask = 1 << rpi->grpnum;
560         rpi->booted = false;
561
562         /* We're single threaded now, so this is OK. */
563         rnp->qsmaskinit |= rpi->grpmask;
564
565         spinlock_init_irqsave(&rpi->lock);
566         INIT_LIST_HEAD(&rpi->cbs);
567         rpi->nr_cbs = 0;
568         rpi->gp_acked = rsp->completed;
569
570         /* TODO: For each mgmt core only */
571         if (coreid == 0) {
572                 rendez_init(&rpi->mgmt_ktask_rv);
573                 rpi->mgmt_ktask_ctl = 0;
574         }
575 }
576
577 /* Initializes the fake cores.  Works with rcu_report_qs_fake_cores() */
578 static void rcu_init_fake_cores(struct rcu_state *rsp)
579 {
580         struct rcu_node *rnp;
581
582         rnp = rsp->level[rcu_num_lvls - 1];
583         for (int i = num_cores; i < rcu_num_cores; i++) {
584                 while (i > rnp->grphi)
585                         rnp++;
586                 rnp->qsmaskinit |= 1 << (i - rnp->grplo);
587         }
588 }
589
590 void rcu_init(void)
591 {
592         struct rcu_state *rsp = &rcu_state;
593         struct rcu_pcpui *rpi;
594
595         rcu_init_geometry();
596         rcu_init_one(rsp);
597         rcu_init_fake_cores(rsp);
598         rcu_dump_rcu_node_tree(rsp);
599
600         ktask("rcu_gp", rcu_gp_ktask, rsp);
601         /* TODO: For each mgmt core */
602         ktask("rcu_mgmt_0", rcu_mgmt_ktask, _PERCPU_VARPTR(rcu_pcpui, 0));
603
604         /* If we have a call_rcu before percpu_init, we might be using the spot in
605          * the actual __percpu .section.  We'd be core 0, so that'd be OK, since all
606          * we're using it for is reading 'booted'. */
607         for_each_core(i) {
608                 rpi = _PERCPU_VARPTR(rcu_pcpui, i);
609                 rpi->booted = true;
610         }
611 }