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