1/*
2 * Preemptible tree RCU is what we do here
3 *
4 * The big invariant here: Grace periods have to outlive
5 * all read sections that began *before* it started
6 *
7 * How we do that:
8 *
9 * 1. CPUs report quiescent states for grace period N only after it's
10 * observed gp_seq == N. Read sections that begin on that CPU
11 * after the fact thus can't see pointers that the batch will free
12 *
13 * 2. CPUs never report while unregistered readers are still active,
14 * but once a reader switches out, it is registered on the leaf
15 *
16 * Quiescent states combine up a tree, allowing us to complete grace periods
17 * in O(CPUs / fanout + preempted readers)
18 *
19 * TODO: CPU hotplug, expedited grace periods, priority boosting blocked
20 * readers, and maybe a better watchdog integration under a debug flag
21 * to get more information about what sleeping readers are doing
22 */
23
24#include <atomic.h>
25#include <compiler/atomic.h>
26#include <console/printf.h>
27#include <enforce.h>
28#include <global.h>
29#include <irq/ipi.h>
30#include <irq/irq.h>
31#include <log.h>
32#include <math/align.h>
33#include <math/min_max.h>
34#include <mem/alloc.h>
35#include <mem/nofail.h>
36#include <sch/sched.h>
37#include <smp/core.h>
38#include <structures/list.h>
39#include <sync/rcu.h>
40#include <sync/semaphore.h>
41#include <sync/spinlock.h>
42#include <thread/thread.h>
43#include <time/spin_sleep.h>
44#include <time/time.h>
45
46#include "rcu_internal.h"
47
48LOG_SITE_DEFINE_PRINT(rcu);
49LOG_HANDLE_DEFINE_PRINT(rcu);
50static struct rcu_globals rcu_global;
51
52static inline struct rcu_node *rcu_leaf_for_cpu(cpu_id_t cpu) {
53 return &rcu_global.leaves[cpu / RCU_FANOUT];
54}
55
56static inline bool rcu_node_pending(const struct rcu_node *node) {
57 if (node->is_leaf)
58 return !cpu_mask_empty(m: &node->qs_cpus) || node->blocked_count != 0;
59
60 return node->qs_children != 0;
61}
62
63/* Send completed node up
64 *
65 * A node completes when it's empty, with no child mask bits,
66 * no readers for this GP, and completed_seq keeps this idempotent
67 */
68static void rcu_propagate_done(struct rcu_node *node, uint64_t seq,
69 enum irql irql) TSA_NO_ANALYSIS {
70 SPINLOCK_ASSERT_HELD(&node->lock);
71 while (true) {
72 if (node->gp_seq != seq || node->completed_seq == seq ||
73 rcu_node_pending(node)) {
74 spin_unlock(&node->lock, irql);
75 return;
76 }
77
78 enforce(seq > node->completed_seq);
79 node->completed_seq = seq;
80
81 struct rcu_node *parent = node->parent;
82 size_t child_index = node->child_index;
83
84 spin_unlock(&node->lock, irql);
85
86 if (!parent) {
87 /* Root is done, GP over */
88 atomic_store_release(&rcu_global.gp_completed, seq);
89 return;
90 }
91
92 irql = spin_lock_high(&parent->lock);
93 if (parent->gp_seq != seq) {
94 spin_unlock(&parent->lock, irql);
95 return;
96 }
97
98 bitmap_clear(map: &parent->qs_children, bit: child_index);
99 node = parent;
100 }
101}
102
103/* cpu has passed through a quiescent state */
104static void rcu_report_cpu_locked(struct rcu_node *leaf, cpu_id_t cpu,
105 uint64_t gp_seq_seen,
106 enum irql irql) TSA_NO_ANALYSIS {
107 SPINLOCK_ASSERT_HELD(&leaf->lock);
108 if (leaf->gp_seq != gp_seq_seen || leaf->completed_seq == gp_seq_seen) {
109 spin_unlock(&leaf->lock, irql);
110 return;
111 }
112
113 atomic_store_rlx(&rcu_global.cpus[cpu].reported_seq, gp_seq_seen);
114
115 cpu_mask_clear(m: &leaf->qs_cpus, cpu);
116 rcu_propagate_done(node: leaf, seq: gp_seq_seen, irql);
117}
118
119void rcu_read_lock(void) TSA_NO_ANALYSIS {
120 struct thread *t = thread_get_current();
121 if (cc_unlikely(!t))
122 return;
123
124 assume(!irq_in_nmi(), "RCU read section in NMI context");
125
126 uint32_t nesting = ca_read_once(t->rcu_nesting);
127 enforce(nesting != UINT32_MAX, OVERFLOW, "RCU nesting");
128
129 if (cc_likely(nesting != 0)) {
130 ca_write_once(t->rcu_nesting, nesting + 1);
131 } else {
132 uint64_t seq = atomic_load_acq(&rcu_global.gp_seq);
133 t->rcu_read_seq = seq;
134 ca_write_once(t->rcu_nesting, 1);
135 }
136
137 ca_barrier();
138 crash_unwind_enter_rcu();
139}
140
141/* Remove a reader off the leaf, and this can run on whatever CPU
142 * the reader happens to wake on */
143static void rcu_unregister_reader(struct thread *t) TSA_NO_ANALYSIS {
144 enum irql outer = irql_raise(new_level: IRQL_HIGH_LEVEL);
145
146 struct rcu_node *leaf = t->rcu_leaf;
147 if (leaf) {
148 enum irql irql = spin_lock_high(&leaf->lock);
149
150 list_node_delete_init(n: &t->rcu_list_node);
151 t->rcu_leaf = NULL;
152
153 uint64_t bseq = t->rcu_blocked_seq;
154 t->rcu_blocked_seq = 0;
155
156 if (bseq != 0 && bseq == leaf->gp_seq) {
157 enforce(leaf->blocked_count != 0, UNDERFLOW, "RCU blocked_count");
158 leaf->blocked_count--;
159 rcu_propagate_done(node: leaf, seq: bseq, irql);
160 } else {
161 spin_unlock(&leaf->lock, irql);
162 }
163 }
164
165 irql_lower(old_level: outer);
166}
167
168void rcu_read_unlock(void) TSA_NO_ANALYSIS {
169 struct thread *t = thread_get_current();
170 if (cc_unlikely(!t))
171 return;
172
173 uint32_t nesting = ca_read_once(t->rcu_nesting);
174 enforce(nesting != 0, UNDERFLOW, "RCU nesting");
175
176 if (cc_likely(nesting > 1)) {
177 ca_write_once(t->rcu_nesting, nesting - 1);
178 } else {
179 ca_barrier();
180 ca_write_once(t->rcu_nesting, 0);
181 ca_barrier();
182
183 if (cc_unlikely(t->rcu_leaf))
184 rcu_unregister_reader(t);
185 }
186
187 crash_unwind_exit_rcu();
188}
189
190static inline bool rcu_cpu_needs_qs(cpu_id_t cpu, uint64_t *seen) {
191 uint64_t s = atomic_load_acq(&rcu_global.gp_seq);
192 if (s == atomic_load_acq(&rcu_global.gp_completed))
193 return false; /* no GP in flight */
194
195 if (atomic_load_rlx(&rcu_global.cpus[cpu].reported_seq) == s)
196 return false; /* already reported */
197
198 *seen = s;
199 return true;
200}
201
202void rcu_note_context_switch(struct thread *outgoing) TSA_NO_ANALYSIS {
203 if (!cc_unlikely(rcu_global.ready))
204 return;
205
206 bool must_register =
207 outgoing && ca_read_once(outgoing->rcu_nesting) && !outgoing->rcu_leaf;
208
209 uint64_t seen;
210 if (cc_likely(!must_register) &&
211 !rcu_cpu_needs_qs(cpu: smp_id(cond: TOPC_IRQL), seen: &seen))
212 return;
213
214 enum irql outer = irql_raise(new_level: IRQL_HIGH_LEVEL);
215
216 cpu_id_t cpu = smp_id(cond: TOPC_IRQL);
217 struct rcu_node *leaf = rcu_leaf_for_cpu(cpu);
218 uint64_t gp_seq_seen = atomic_load_acq(&rcu_global.gp_seq);
219
220 enum irql irql = spin_lock_high(&leaf->lock);
221
222 uint64_t lseq = leaf->gp_seq;
223 bool leaf_active = leaf->completed_seq != lseq;
224
225 if (outgoing && ca_read_once(outgoing->rcu_nesting) &&
226 !outgoing->rcu_leaf) {
227 outgoing->rcu_leaf = leaf;
228 outgoing->rcu_blocked_seq = 0;
229 list_head_push_back(h: &leaf->blocked, n: &outgoing->rcu_list_node);
230
231 uint64_t rseq = outgoing->rcu_read_seq;
232 if (leaf_active && rseq < lseq) {
233 outgoing->rcu_blocked_seq = lseq;
234 leaf->blocked_count++;
235 }
236 }
237
238 rcu_report_cpu_locked(leaf, cpu, gp_seq_seen, irql);
239
240 irql_lower(old_level: outer);
241}
242
243/* This just lets the CPU answer IRQ_NOP */
244void rcu_note_irq_exit(void) TSA_NO_ANALYSIS {
245 if (!rcu_global.ready)
246 return;
247
248 enforce(irq_in_interrupt(), INVALID_CONTEXT);
249 struct thread *t = thread_get_current();
250
251 /* Just report when the interrupted context isn't holding an unregistered
252 * read side critical section, as readers that are accounted in a leaf
253 * don't need to be handled over here */
254 if (t && ca_read_once(t->rcu_nesting) && !t->rcu_leaf)
255 return;
256
257 uint64_t gp_seq_seen = atomic_load_acq(&rcu_global.gp_seq);
258 if (gp_seq_seen == atomic_load_acq(&rcu_global.gp_completed))
259 return;
260
261 cpu_id_t cpu = smp_id(cond: TOPC_IRQ);
262 if (atomic_load_rlx(&rcu_global.cpus[cpu].reported_seq) == gp_seq_seen)
263 return;
264
265 /* Technically the IRQL stuff here is a no-op, but we'll
266 * preserve it in case we move the irq.c callsite +
267 * gives us a bit of semantic/invariant clarity */
268 enum irql outer = irql_raise(new_level: IRQL_HIGH_LEVEL);
269
270 struct rcu_node *leaf = rcu_leaf_for_cpu(cpu);
271
272 enum irql irql = spin_lock_high(&leaf->lock);
273 rcu_report_cpu_locked(leaf, cpu, gp_seq_seen, irql);
274
275 irql_lower(old_level: outer);
276}
277
278void rcu_idle_enter(void) {
279 if (!rcu_global.ready)
280 return;
281
282 enforce(!irqs_enabled(), INVALID_CONTEXT);
283
284 atomic_store_release(&rcu_global.cpus[smp_id(TOPC_IFLAG)].idle, true);
285}
286
287void rcu_idle_exit(void) {
288 if (!rcu_global.ready)
289 return;
290
291 struct rcu_cpu *me = &rcu_global.cpus[smp_id(cond: TOPC_IFLAG)];
292 if (cc_likely(!atomic_load_rlx(&me->idle)))
293 return;
294
295 atomic_xchg_acq_rel(&me->idle, false);
296}
297
298void rcu_defer(struct rcu_cb *cc_nonnull_ptr cb, rcu_fn cc_nonnull_ptr func) {
299 enforce_precond(cb && func);
300
301 cb->fn = func;
302
303#ifdef TEST_ENABLED
304 cb->target_gen = 0;
305 cb->gen_when_called = 0;
306#endif
307
308 list_node_init(n: &cb->list);
309
310 /* Pin + disable interrupts for queue selection */
311 enum irql outer = irql_raise(new_level: IRQL_HIGH_LEVEL);
312
313#ifdef TEST_ENABLED
314 cb->enqueued_waiting_on_gen = atomic_load_rlx(&rcu_global.gp_seq);
315#endif
316
317 if (cc_unlikely(!rcu_global.ready)) {
318 /* Before rcu_init(), we have nothing, and the boot CPU
319 * is the only thing running RCU operations, so we can just do this */
320 irql_lower(old_level: outer);
321 cb->fn(cb);
322 return;
323 }
324
325 struct rcu_cpu *q = &rcu_global.cpus[smp_id(cond: TOPC_IRQL)];
326
327 enum irql irql = spin_lock_high(&q->lock);
328 list_head_push_back(h: &q->list, n: &cb->list);
329 spin_unlock(&q->lock, irql);
330
331 irql_lower(old_level: outer);
332
333 semaphore_post(s: &rcu_global.sem);
334}
335
336static void rcu_detach_callbacks(struct list_head *batch) {
337 for (cpu_id_t cpu = 0; cpu < global.core_count; cpu++) {
338 struct rcu_cpu *q = &rcu_global.cpus[cpu];
339
340 enum irql irql = spin_lock_high(&q->lock);
341 list_head_splice_back_init(dst: batch, src: &q->list);
342 spin_unlock(&q->lock, irql);
343 }
344}
345
346static bool rcu_callbacks_pending(void) {
347 for (cpu_id_t cpu = 0; cpu < global.core_count; cpu++) {
348 struct rcu_cpu *q = &rcu_global.cpus[cpu];
349
350 enum irql irql = spin_lock_high(&q->lock);
351 bool empty = list_head_empty(h: &q->list);
352 spin_unlock(&q->lock, irql);
353
354 if (!empty)
355 return true;
356 }
357 return false;
358}
359
360static bool rcu_work_pending(void) {
361 if (atomic_load_acq(&rcu_global.gp_requests) != 0)
362 return true;
363 return rcu_callbacks_pending();
364}
365
366/* This runs at PASSIVE so callbacks can do whatever */
367static void rcu_run_batch(struct list_head *batch, uint64_t seq) {
368 struct rcu_cb *cb, *tmp;
369 list_for_each_entry_safe(cb, tmp, batch, list) {
370 list_node_delete_init(n: &cb->list);
371
372#ifdef TEST_ENABLED
373 cb->gen_when_called = (size_t) seq;
374#endif
375
376 cb->fn(cb);
377 }
378}
379
380/* Bother everyone who's not quiesced
381 *
382 * CPUs that went to hlt after rcu_gp_start sampled them still need
383 * a QS. Instead of IPI'ing just for them to say they're idle,
384 * we check the percpu flag */
385static void rcu_kick_pending(uint64_t seq) TSA_NO_ANALYSIS {
386 /* Pairs with xchg in rcu_idle_exit */
387 cpu_memory_barrier();
388
389 for (size_t l = 0; l < rcu_global.leaf_count; l++) {
390 struct rcu_node *leaf = &rcu_global.leaves[l];
391
392 enum irql irql = spin_lock_high(&leaf->lock);
393
394 if (leaf->gp_seq != seq) {
395 spin_unlock(&leaf->lock, irql);
396 continue;
397 }
398
399 struct cpu_mask owing;
400 cpu_mask_copy(dst: &owing, src: &leaf->qs_cpus);
401
402 cpu_id_t cpu;
403 for_each_cpu(cpu, &owing) {
404 if (!atomic_load_acq(&rcu_global.cpus[cpu].idle))
405 continue;
406
407 cpu_mask_clear(m: &leaf->qs_cpus, cpu);
408 atomic_store_rlx(&rcu_global.cpus[cpu].reported_seq, seq);
409 }
410
411 struct cpu_mask pending;
412 cpu_mask_copy(dst: &pending, src: &leaf->qs_cpus);
413 rcu_propagate_done(node: leaf, seq, irql);
414
415 for_each_cpu(cpu, &pending) {
416 ipi_send(cpu: (uint32_t) cpu, IRQ_NOP);
417 }
418 }
419}
420
421/* TODO: nonfatal, we'll need to do something a bit smarter here with logging/
422 * errors for parseability and whatnot */
423static void rcu_report_stall(uint64_t seq, time_ms_t elapsed) {
424 struct rcu_stall_blocker blockers[RCU_STALL_MAX_REPORTED];
425 size_t nblockers = 0;
426
427 rcu_warn("grace period %lu stalled for %lu ms\n", seq, elapsed);
428
429 for (size_t l = 0; l < rcu_global.leaf_count; l++) {
430 struct rcu_node *leaf = &rcu_global.leaves[l];
431
432 struct cpu_mask pending = CPU_MASK_INIT;
433
434 enum irql irql = spin_lock_high(&leaf->lock);
435
436 if (leaf->gp_seq == seq)
437 cpu_mask_copy(dst: &pending, src: &leaf->qs_cpus);
438 uint32_t blocked = leaf->blocked_count;
439
440 struct thread *t;
441 list_for_each_entry(t, &leaf->blocked, rcu_list_node) {
442 if (t->rcu_blocked_seq != seq)
443 continue;
444
445 if (nblockers >= RCU_STALL_MAX_REPORTED)
446 break;
447
448 blockers[nblockers++] = (struct rcu_stall_blocker){
449 .id = t->id,
450 .name = t->name,
451 .state = (int) thread_get_state(t),
452 .read_seq = t->rcu_read_seq,
453 };
454 }
455
456 spin_unlock(&leaf->lock, irql);
457
458 if (!cpu_mask_empty(m: &pending) || blocked)
459 rcu_warn("leaf %zu: cpus %#lx pending, %u blocking reader(s)\n", l,
460
461 pending.bits[leaf->cpu_base / BITMAP_BITS_PER_WORD],
462 blocked);
463 }
464
465 for (size_t i = 0; i < nblockers; i++)
466 rcu_warn("tid %zu \"%s\" state %d read_seq %lu\n", blockers[i].id,
467 blockers[i].name ? blockers[i].name : "?", blockers[i].state,
468 blockers[i].read_seq);
469}
470
471static uint64_t rcu_gp_start(struct list_head *batch) TSA_NO_ANALYSIS {
472 enum irql outer = irql_raise(new_level: IRQL_DISPATCH_LEVEL);
473 rcu_detach_callbacks(batch);
474 atomic_store_rlx(&rcu_global.gp_requests, 0);
475
476 uint64_t prev = atomic_load_rlx(&rcu_global.gp_seq);
477 enforce(prev != UINT64_MAX, OVERFLOW, "RCU GP seq");
478 uint64_t seq = prev + 1;
479
480#ifdef TEST_ENABLED
481 struct rcu_cb *cb;
482 list_for_each_entry(cb, batch, list)
483 cb->target_gen = (size_t) seq;
484#endif
485
486 /* Root down, rcu.nodes is root then leaves */
487 for (size_t i = 0; i < rcu_global.node_count; i++) {
488 struct rcu_node *node = &rcu_global.nodes[i];
489
490 enum irql irql = spin_lock_high(&node->lock);
491 node->gp_seq = seq;
492 node->qs_children = node->full_children;
493
494 if (node->is_leaf) {
495 cpu_mask_copy(dst: &node->qs_cpus, src: &node->full_cpus);
496 node->blocked_count = 0;
497
498 /* Everyone here began before the GP */
499 struct thread *t;
500 list_for_each_entry(t, &node->blocked, rcu_list_node) {
501 uint64_t rseq = t->rcu_read_seq;
502 if (rseq < seq) {
503 t->rcu_blocked_seq = seq;
504 node->blocked_count++;
505 } else {
506 t->rcu_blocked_seq = 0;
507 }
508 }
509 }
510 spin_unlock(&node->lock, irql);
511 }
512
513 atomic_store_release(&rcu_global.gp_seq, seq);
514
515 /* Pairs with the xchg in rcu_idle_exit */
516 cpu_memory_barrier();
517
518 /* Retire quiescent and poke everyone else */
519 cpu_id_t self = smp_id(cond: TOPC_IRQL);
520
521 for (size_t l = 0; l < rcu_global.leaf_count; l++) {
522 struct rcu_node *leaf = &rcu_global.leaves[l];
523
524 enum irql irql = spin_lock_high(&leaf->lock);
525 if (leaf->gp_seq != seq) {
526 spin_unlock(&leaf->lock, irql);
527 continue;
528 }
529
530 /* Idle CPUs can't be in an RCU read section,
531 * and they'd clear the flag before they enter one */
532 cpu_id_t cpu;
533 for_each_cpu(cpu, &leaf->full_cpus) {
534 if (atomic_load_acq(&rcu_global.cpus[cpu].idle))
535 cpu_mask_clear(m: &leaf->qs_cpus, cpu);
536 }
537
538 if (leaf == rcu_leaf_for_cpu(cpu: self))
539 cpu_mask_clear(m: &leaf->qs_cpus, cpu: self);
540
541 struct cpu_mask pending;
542 cpu_mask_copy(dst: &pending, src: &leaf->qs_cpus);
543 rcu_propagate_done(node: leaf, seq, irql);
544
545 for_each_cpu(cpu, &pending)
546 ipi_send(cpu: (uint32_t) cpu, IRQ_NOP);
547 }
548
549 irql_lower(old_level: outer);
550 return seq;
551}
552
553static void rcu_gp_wait(uint64_t seq) {
554 time_ms_t started = time_get_ms();
555 time_ms_t last_kick = started;
556 time_ms_t last_stall = started;
557
558 while (atomic_load_acq(&rcu_global.gp_completed) < seq) {
559 scheduler_yield();
560
561 time_ms_t now = time_get_ms();
562
563 if (now - last_kick >= RCU_KICK_INTERVAL_MS) {
564 last_kick = now;
565 rcu_kick_pending(seq);
566 }
567
568 if (now - last_stall >= RCU_STALL_MS) {
569 last_stall = now;
570 rcu_report_stall(seq, elapsed: now - started);
571 }
572 }
573}
574
575static void rcu_gp_worker(void *unused_arg) {
576 cc_ignore(unused_arg);
577
578 while (true) {
579 semaphore_wait(s: &rcu_global.sem);
580
581 if (!rcu_work_pending())
582 continue;
583
584 struct list_head batch;
585 list_head_init(h: &batch);
586
587 uint64_t seq = rcu_gp_start(batch: &batch);
588 rcu_gp_wait(seq);
589 rcu_run_batch(batch: &batch, seq);
590 }
591}
592
593void rcu_synchronize(void) {
594 struct thread *t = thread_get_current();
595
596 enforce(!irq_in_interrupt() && !irq_in_nmi(), INVALID_CONTEXT);
597 enforce(irql_get() <= IRQL_APC_LEVEL, INVALID_CONTEXT);
598 enforce(!t || t->rcu_nesting == 0);
599 enforce(!rcu_global.ready || t != rcu_global.worker);
600
601 if (!rcu_global.ready)
602 return;
603
604 /* Caller's unpublish happened before this load, meaning
605 * grace periods that published a sequence greater than what's
606 * here started after the unpublish */
607 uint64_t target = atomic_load_acq(&rcu_global.gp_seq) + 1;
608
609 atomic_inc_release(&rcu_global.gp_requests);
610 semaphore_post(s: &rcu_global.sem);
611
612 while (atomic_load_acq(&rcu_global.gp_completed) < target)
613 scheduler_yield();
614}
615
616static void rcu_build_tree(void) {
617 size_t counts[RCU_MAX_LEVELS];
618 size_t levels = 0;
619
620 size_t n = global.core_count;
621 enforce(n > 0);
622
623 do {
624 n = DIV_ROUND_UP(n, RCU_FANOUT);
625 enforce(levels < RCU_MAX_LEVELS, CAPACITY_EXCEEDED,
626 "RCU tree deeper than RCU_MAX_LEVELS");
627 counts[levels++] = n;
628 } while (n > 1);
629
630 /* counts[0] is the leaf, counts[levels - 1] is the root */
631 size_t total = 0;
632 for (size_t i = 0; i < levels; i++)
633 total += counts[i];
634
635 rcu_global.nodes = nofail_vnews(rcu_global.nodes, total, ALLOC_ZERO);
636 rcu_global.node_count = total;
637
638 /* Root first so forward pass initializes parents before children */
639 size_t offsets[RCU_MAX_LEVELS];
640 offsets[levels - 1] = 0;
641 for (size_t i = levels - 1; i > 0; i--)
642 offsets[i - 1] = offsets[i] + counts[i];
643
644 for (size_t level = 0; level < levels; level++) {
645 for (size_t j = 0; j < counts[level]; j++) {
646 struct rcu_node *node = &rcu_global.nodes[offsets[level] + j];
647
648 spinlock_init(&node->lock);
649 node->gp_seq = 0;
650 node->completed_seq = 0;
651 node->qs_children = 0;
652
653 if (level == levels - 1) {
654 node->parent = NULL;
655 node->child_index = 0;
656 } else {
657 node->parent =
658 &rcu_global.nodes[offsets[level + 1] + (j / RCU_FANOUT)];
659 node->child_index = j % RCU_FANOUT;
660 }
661
662 size_t owned;
663 if (level == 0) {
664 node->is_leaf = true;
665 node->cpu_base = j * RCU_FANOUT;
666 list_head_init(h: &node->blocked);
667 owned = global.core_count - node->cpu_base;
668 } else {
669 node->is_leaf = false;
670 owned = counts[level - 1] - (j * RCU_FANOUT);
671 }
672
673 owned = MIN(owned, RCU_FANOUT);
674
675 if (node->is_leaf) {
676 cpu_mask_set_range(m: &node->full_cpus, start: node->cpu_base, len: owned);
677 } else {
678 bitmap_fill(map: &node->full_children, nbits: owned);
679 }
680 }
681 }
682
683 rcu_global.root = &rcu_global.nodes[0];
684 rcu_global.leaves = &rcu_global.nodes[offsets[0]];
685 rcu_global.leaf_count = counts[0];
686}
687
688void rcu_init(void) {
689 atomic_store_rlx(&rcu_global.gp_seq, 1);
690 atomic_store_rlx(&rcu_global.gp_completed, 1);
691 atomic_store_rlx(&rcu_global.gp_requests, 0);
692
693 semaphore_init(s: &rcu_global.sem, value: 0, SEMAPHORE_INIT_NORMAL);
694
695 rcu_build_tree();
696
697 rcu_global.cpus =
698 nofail_vnews(rcu_global.cpus, global.core_count, ALLOC_ZERO);
699 for (cpu_id_t cpu = 0; cpu < global.core_count; cpu++) {
700 spinlock_init(&rcu_global.cpus[cpu].lock);
701 list_head_init(h: &rcu_global.cpus[cpu].list);
702 atomic_store_rlx(&rcu_global.cpus[cpu].reported_seq, 0);
703 }
704
705 rcu_global.worker = thread_spawn("rcu_gp_worker", rcu_gp_worker);
706 rcu_global.ready = true;
707}
708