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 <acpi/lapic.h>
25#include <console/printf.h>
26#include <global.h>
27#include <irq/irq.h>
28#include <kassert.h>
29#include <log.h>
30#include <mem/alloc.h>
31#include <mem/alloc_or_die.h>
32#include <sch/sched.h>
33#include <smp/core.h>
34#include <stdatomic.h>
35#include <structures/list.h>
36#include <sync/rcu.h>
37#include <sync/semaphore.h>
38#include <sync/spinlock.h>
39#include <thread/thread.h>
40#include <time/spin_sleep.h>
41#include <time/time.h>
42
43#include "rcu_internal.h"
44
45LOG_SITE_DECLARE_PRINT(rcu);
46LOG_HANDLE_DECLARE_PRINT(rcu);
47static struct rcu_state rcu;
48
49static inline struct rcu_node *rcu_leaf_for_cpu(cpu_id_t cpu) {
50 return &rcu.leaves[cpu / RCU_FANOUT];
51}
52
53static inline bool rcu_node_pending(const struct rcu_node *node) {
54 if (node->is_leaf)
55 return !cpu_mask_empty(m: &node->qs_cpus) || node->blocked_count != 0;
56
57 return node->qs_children != 0;
58}
59
60/* Send completed node up
61 *
62 * A node completes when it's empty, with no child mask bits,
63 * no readers for this GP, and completed_seq keeps this idempotent
64 */
65static void rcu_propagate_done(struct rcu_node *node, uint64_t seq,
66 enum irql irql) {
67 SPINLOCK_ASSERT_HELD(&node->lock);
68 while (true) {
69 if (node->gp_seq != seq || node->completed_seq == seq ||
70 rcu_node_pending(node)) {
71 spin_unlock(&node->lock, irql);
72 return;
73 }
74
75 kassert(seq > node->completed_seq);
76 node->completed_seq = seq;
77
78 struct rcu_node *parent = node->parent;
79 size_t child_index = node->child_index;
80
81 spin_unlock(&node->lock, irql);
82
83 if (!parent) {
84 /* Root is done, GP over */
85 atomic_store_explicit(&rcu.gp_completed, seq, memory_order_release);
86 return;
87 }
88
89 irql = spin_lock_irq_disable(&parent->lock);
90 if (parent->gp_seq != seq) {
91 spin_unlock(&parent->lock, irql);
92 return;
93 }
94
95 bitmap_clear(map: &parent->qs_children, bit: child_index);
96 node = parent;
97 }
98}
99
100/* cpu has passed through a quiescent state */
101static void rcu_report_cpu_locked(struct rcu_node *leaf, cpu_id_t cpu,
102 uint64_t gp_seq_seen, enum irql irql) {
103 SPINLOCK_ASSERT_HELD(&leaf->lock);
104 if (leaf->gp_seq != gp_seq_seen || leaf->completed_seq == gp_seq_seen) {
105 spin_unlock(&leaf->lock, irql);
106 return;
107 }
108
109 atomic_store_explicit(&rcu.cpus[cpu].reported_seq, gp_seq_seen,
110 memory_order_relaxed);
111
112 cpu_mask_clear(m: &leaf->qs_cpus, cpu);
113 rcu_propagate_done(node: leaf, seq: gp_seq_seen, irql);
114}
115
116void rcu_read_lock(void) {
117 struct thread *t = thread_get_current();
118 if (unlikely(!t))
119 return;
120
121 kassert_debug(!irq_in_nmi(), "RCU read section in NMI context");
122
123 if (atomic_load_explicit(&t->rcu_nesting, memory_order_relaxed) == 0) {
124 uint64_t seq = atomic_load_explicit(&rcu.gp_seq, memory_order_acquire);
125 atomic_store_explicit(&t->rcu_read_seq, seq, memory_order_relaxed);
126 }
127
128 /* TODO: There should be a smart way that avoids seq_cst ordering...
129 *
130 * But we do this so that this can't come after the first
131 * rcu_dereference(), and because I will optimize later */
132 uint32_t old =
133 atomic_fetch_add_explicit(&t->rcu_nesting, 1, memory_order_seq_cst);
134
135 kassert(old != UINT32_MAX, "RCU nesting overflow");
136 crash_unwind_enter_rcu();
137}
138
139/* Remove a reader off the leaf, and this can run on whatever CPU
140 * the reader happens to wake on */
141static void rcu_unregister_reader(struct thread *t) {
142 enum irql outer = irql_raise(new_level: IRQL_HIGH_LEVEL);
143
144 struct rcu_node *leaf = t->rcu_leaf;
145 if (leaf) {
146 enum irql irql = spin_lock_irq_disable(&leaf->lock);
147
148 list_del_init(entry: &t->rcu_list_node);
149 t->rcu_leaf = NULL;
150
151 uint64_t bseq = t->rcu_blocked_seq;
152 t->rcu_blocked_seq = 0;
153
154 if (bseq != 0 && bseq == leaf->gp_seq) {
155 kassert(leaf->blocked_count != 0, "RCU blocked reader underflow");
156 leaf->blocked_count--;
157 rcu_propagate_done(node: leaf, seq: bseq, irql);
158 } else {
159 spin_unlock(&leaf->lock, irql);
160 }
161 }
162
163 irql_lower(old_level: outer);
164}
165
166void rcu_read_unlock(void) {
167 struct thread *t = thread_get_current();
168 if (unlikely(!t))
169 return;
170
171 uint32_t old =
172 atomic_fetch_sub_explicit(&t->rcu_nesting, 1, memory_order_seq_cst);
173 kassert(old != 0, "RCU nesting underflow");
174
175 /* Nesting is zero, we're registered, unregister
176 *
177 * We do this AFTER decrement so if we get preempted in this window,
178 * the scheduler just sees an idle reader and moves on */
179 if (unlikely(old == 1 && t->rcu_leaf))
180 rcu_unregister_reader(t); /* NOTE: This can potentially result in cross
181 * node traffic and cache unhappiness, but
182 * this is also the "very slow path"....
183 *
184 * Perhaps something can be done (?)
185 */
186
187 crash_unwind_exit_rcu();
188}
189
190/* We might want to change next_is_idle to a thread pointer...
191 *
192 * Lock ordering here is scheduler -> leaf -> parents */
193void rcu_note_context_switch(struct thread *outgoing, bool next_is_idle) {
194 if (!unlikely(rcu.ready))
195 return;
196
197 enum irql outer = irql_raise(new_level: IRQL_HIGH_LEVEL);
198
199 cpu_id_t cpu = smp_id(cond: TOPC_IRQL);
200 struct rcu_node *leaf = rcu_leaf_for_cpu(cpu);
201 uint64_t gp_seq_seen =
202 atomic_load_explicit(&rcu.gp_seq, memory_order_acquire);
203
204 enum irql irql = spin_lock_irq_disable(&leaf->lock);
205
206 if (next_is_idle)
207 cpu_mask_set(m: &leaf->idle_cpus, cpu);
208 else
209 cpu_mask_clear(m: &leaf->idle_cpus, cpu);
210
211 /* Registration uses the leaf's tracked seq, NOT gp_seq_seen
212 *
213 * Grace period starts initialize leaves before publishing the
214 * sequence, and readers that get switched out in the window
215 * come before the new GP, although this CPU hasn't seen the publish */
216 uint64_t lseq = leaf->gp_seq;
217 bool leaf_active = leaf->completed_seq != lseq;
218
219 if (outgoing &&
220 atomic_load_explicit(&outgoing->rcu_nesting, memory_order_relaxed) !=
221 0 &&
222 !outgoing->rcu_leaf) {
223 outgoing->rcu_leaf = leaf;
224 outgoing->rcu_blocked_seq = 0;
225 list_add_tail(new: &outgoing->rcu_list_node, head: &leaf->blocked);
226
227 uint64_t rseq =
228 atomic_load_explicit(&outgoing->rcu_read_seq, memory_order_relaxed);
229 if (leaf_active && rseq < lseq) {
230 outgoing->rcu_blocked_seq = lseq;
231 leaf->blocked_count++;
232 }
233 }
234
235 rcu_report_cpu_locked(leaf, cpu, gp_seq_seen, irql);
236
237 irql_lower(old_level: outer);
238}
239
240/* This just lets the CPU answer IRQ_NOP */
241void rcu_note_irq_exit(void) {
242 if (!rcu.ready)
243 return;
244
245 kassert(irq_in_interrupt());
246 struct thread *t = thread_get_current();
247
248 /* Just report when the interrupted context isn't holding an unregistered
249 * read side critical section, as readers that are accounted in a leaf
250 * don't need to be handled over here */
251 if (t && atomic_load_explicit(&t->rcu_nesting, memory_order_relaxed) != 0 &&
252 !t->rcu_leaf)
253 return;
254
255 uint64_t gp_seq_seen =
256 atomic_load_explicit(&rcu.gp_seq, memory_order_acquire);
257 if (gp_seq_seen ==
258 atomic_load_explicit(&rcu.gp_completed, memory_order_relaxed))
259 return;
260
261 cpu_id_t cpu = smp_id(cond: TOPC_IRQ);
262 if (atomic_load_explicit(&rcu.cpus[cpu].reported_seq,
263 memory_order_relaxed) == gp_seq_seen)
264 return;
265
266 /* Technically the IRQL stuff here is a no-op, but we'll
267 * preserve it in case we move the irq.c callsite +
268 * gives us a bit of semantic/invariant clarity */
269 enum irql outer = irql_raise(new_level: IRQL_HIGH_LEVEL);
270
271 struct rcu_node *leaf = rcu_leaf_for_cpu(cpu);
272
273 enum irql irql = spin_lock_irq_disable(&leaf->lock);
274 rcu_report_cpu_locked(leaf, cpu, gp_seq_seen, irql);
275
276 irql_lower(old_level: outer);
277}
278
279void rcu_defer(struct rcu_cb *cb, rcu_fn func, void *arg) {
280 kassert(cb && func);
281
282 cb->fn = func;
283 cb->arg = arg;
284 cb->target_gen = 0;
285 cb->gen_when_called = 0;
286 INIT_LIST_HEAD(list: &cb->list);
287
288 /* Pin + disable interrupts for queue selection */
289 enum irql outer = irql_raise(new_level: IRQL_HIGH_LEVEL);
290
291 cb->enqueued_waiting_on_gen =
292 (size_t) atomic_load_explicit(&rcu.gp_seq, memory_order_relaxed);
293
294 if (unlikely(!rcu.ready)) {
295 /* Before rcu_init(), we have no tree, no worker, and the boot CPU
296 * is the only thing running RCU operations, so we can just do this */
297 irql_lower(old_level: outer);
298 cb->fn(cb, cb->arg);
299 return;
300 }
301
302 struct rcu_cpu *q = &rcu.cpus[smp_id(cond: TOPC_IRQL)];
303
304 enum irql irql = spin_lock_irq_disable(&q->lock);
305 list_add_tail(new: &cb->list, head: &q->list);
306 spin_unlock(&q->lock, irql);
307
308 irql_lower(old_level: outer);
309
310 semaphore_post(s: &rcu.sem);
311}
312
313static void rcu_detach_callbacks(struct list_head *batch) {
314 for (cpu_id_t cpu = 0; cpu < global.core_count; cpu++) {
315 struct rcu_cpu *q = &rcu.cpus[cpu];
316
317 enum irql irql = spin_lock_irq_disable(&q->lock);
318 list_splice_tail_init(list: &q->list, head: batch);
319 spin_unlock(&q->lock, irql);
320 }
321}
322
323static bool rcu_callbacks_pending(void) {
324 for (cpu_id_t cpu = 0; cpu < global.core_count; cpu++) {
325 struct rcu_cpu *q = &rcu.cpus[cpu];
326
327 enum irql irql = spin_lock_irq_disable(&q->lock);
328 bool empty = list_empty(head: &q->list);
329 spin_unlock(&q->lock, irql);
330
331 if (!empty)
332 return true;
333 }
334 return false;
335}
336
337static bool rcu_work_pending(void) {
338 if (atomic_load_explicit(&rcu.gp_requests, memory_order_acquire) != 0)
339 return true;
340 return rcu_callbacks_pending();
341}
342
343/* This runs at PASSIVE so callbacks can do whatever */
344static void rcu_run_batch(struct list_head *batch, uint64_t seq) {
345 struct rcu_cb *cb, *tmp;
346 list_for_each_entry_safe(cb, tmp, batch, list) {
347 list_del_init(entry: &cb->list);
348 cb->gen_when_called = (size_t) seq;
349 cb->fn(cb, cb->arg);
350 }
351}
352
353/* Bother everyone who's not quiesced, skip idlers */
354static void rcu_kick_pending(uint64_t seq) {
355 for (size_t l = 0; l < rcu.leaf_count; l++) {
356 struct rcu_node *leaf = &rcu.leaves[l];
357
358 struct cpu_mask pending = CPU_MASK_INIT;
359
360 enum irql irql = spin_lock_irq_disable(&leaf->lock);
361 if (leaf->gp_seq == seq)
362 cpu_mask_copy(dst: &pending, src: &leaf->qs_cpus);
363 spin_unlock(&leaf->lock, irql);
364
365 cpu_id_t cpu;
366 for_each_cpu(cpu, &pending) {
367 ipi_send(apic_id: (uint32_t) cpu, IRQ_NOP);
368 }
369 }
370}
371
372/* TODO: nonfatal, we'll need to do something a bit smarter here with logging/
373 * errors for parseability and whatnot */
374static void rcu_report_stall(uint64_t seq, time_ms_t elapsed) {
375 struct rcu_stall_blocker blockers[RCU_STALL_MAX_REPORTED];
376 size_t nblockers = 0;
377
378 rcu_warn("grace period %llu stalled for %llu ms\n",
379 (unsigned long long) seq, (unsigned long long) elapsed);
380
381 for (size_t l = 0; l < rcu.leaf_count; l++) {
382 struct rcu_node *leaf = &rcu.leaves[l];
383
384 struct cpu_mask pending = CPU_MASK_INIT;
385
386 enum irql irql = spin_lock_irq_disable(&leaf->lock);
387
388 if (leaf->gp_seq == seq)
389 cpu_mask_copy(dst: &pending, src: &leaf->qs_cpus);
390 uint32_t blocked = leaf->blocked_count;
391
392 struct thread *t;
393 list_for_each_entry(t, &leaf->blocked, rcu_list_node) {
394 if (t->rcu_blocked_seq != seq)
395 continue;
396 if (nblockers >= RCU_STALL_MAX_REPORTED)
397 break;
398
399 blockers[nblockers++] = (struct rcu_stall_blocker){
400 .id = t->id,
401 .name = t->name,
402 .state = (int) thread_get_state(t),
403 .read_seq = atomic_load_explicit(&t->rcu_read_seq,
404 memory_order_relaxed),
405 };
406 }
407
408 spin_unlock(&leaf->lock, irql);
409
410 if (!cpu_mask_empty(m: &pending) || blocked)
411 rcu_warn("leaf %zu: cpus %#llx pending, %u blocking reader(s)\n", l,
412 (unsigned long long)
413 pending.bits[leaf->cpu_base / BITMAP_BITS_PER_WORD],
414 blocked);
415 }
416
417 for (size_t i = 0; i < nblockers; i++)
418 rcu_warn("tid %llu \"%s\" state %d read_seq %llu\n",
419 (unsigned long long) blockers[i].id,
420 blockers[i].name ? blockers[i].name : "?", blockers[i].state,
421 (unsigned long long) blockers[i].read_seq);
422}
423
424static uint64_t rcu_gp_start(struct list_head *batch) {
425 enum irql outer = irql_raise(new_level: IRQL_DISPATCH_LEVEL);
426 rcu_detach_callbacks(batch);
427 atomic_store_explicit(&rcu.gp_requests, 0, memory_order_relaxed);
428
429 uint64_t prev = atomic_load_explicit(&rcu.gp_seq, memory_order_relaxed);
430 kassert(prev != UINT64_MAX, "RCU GP seq wrap");
431 uint64_t seq = prev + 1;
432
433 struct rcu_cb *cb;
434 list_for_each_entry(cb, batch, list) cb->target_gen = (size_t) seq;
435
436 /* Root down, rcu.nodes is root then leaves */
437 for (size_t i = 0; i < rcu.node_count; i++) {
438 struct rcu_node *node = &rcu.nodes[i];
439
440 enum irql irql = spin_lock_irq_disable(&node->lock);
441 node->gp_seq = seq;
442 node->qs_children = node->full_children;
443
444 if (node->is_leaf) {
445 cpu_mask_copy(dst: &node->qs_cpus, src: &node->full_cpus);
446 node->blocked_count = 0;
447
448 /* Everyone here began before the GP */
449 struct thread *t;
450 list_for_each_entry(t, &node->blocked, rcu_list_node) {
451 uint64_t rseq = atomic_load_explicit(&t->rcu_read_seq,
452 memory_order_relaxed);
453 if (rseq < seq) {
454 t->rcu_blocked_seq = seq;
455 node->blocked_count++;
456 } else {
457 t->rcu_blocked_seq = 0;
458 }
459 }
460 }
461 spin_unlock(&node->lock, irql);
462 }
463
464 atomic_store_explicit(&rcu.gp_seq, seq, memory_order_release);
465
466 /* Retire quiescent and poke everyone else */
467 cpu_id_t self = smp_id(cond: TOPC_IRQL);
468
469 for (size_t l = 0; l < rcu.leaf_count; l++) {
470 struct rcu_node *leaf = &rcu.leaves[l];
471
472 enum irql irql = spin_lock_irq_disable(&leaf->lock);
473 if (leaf->gp_seq != seq) {
474 spin_unlock(&leaf->lock, irql);
475 continue;
476 }
477
478 cpu_mask_andnot(dst: &leaf->qs_cpus, a: &leaf->qs_cpus, b: &leaf->idle_cpus);
479 if (leaf == rcu_leaf_for_cpu(cpu: self))
480 cpu_mask_clear(m: &leaf->qs_cpus, cpu: self);
481
482 struct cpu_mask pending;
483 cpu_mask_copy(dst: &pending, src: &leaf->qs_cpus);
484 rcu_propagate_done(node: leaf, seq, irql);
485
486 cpu_id_t cpu;
487 for_each_cpu(cpu, &pending) ipi_send(apic_id: (uint32_t) cpu, IRQ_NOP);
488 }
489
490 irql_lower(old_level: outer);
491 return seq;
492}
493
494static void rcu_gp_wait(uint64_t seq) {
495 time_ms_t started = time_get_ms();
496 time_ms_t last_kick = started;
497 time_ms_t last_stall = started;
498
499 while (atomic_load_explicit(&rcu.gp_completed, memory_order_acquire) <
500 seq) {
501 scheduler_yield();
502
503 time_ms_t now = time_get_ms();
504
505 if (now - last_kick >= RCU_KICK_INTERVAL_MS) {
506 last_kick = now;
507 rcu_kick_pending(seq);
508 }
509
510 if (now - last_stall >= RCU_STALL_MS) {
511 last_stall = now;
512 rcu_report_stall(seq, elapsed: now - started);
513 }
514 }
515}
516
517static void rcu_gp_worker(void *unused_arg) {
518 unused(unused_arg);
519
520 while (true) {
521 semaphore_wait(s: &rcu.sem);
522
523 if (!rcu_work_pending())
524 continue;
525
526 struct list_head batch;
527 INIT_LIST_HEAD(list: &batch);
528
529 uint64_t seq = rcu_gp_start(batch: &batch);
530 rcu_gp_wait(seq);
531 rcu_run_batch(batch: &batch, seq);
532 }
533}
534
535void rcu_synchronize(void) {
536 struct thread *t = thread_get_current();
537
538 kassert(!irq_in_interrupt() && !irq_in_nmi(),
539 "rcu_synchronize() from interrupt context");
540 kassert(irql_get() <= IRQL_APC_LEVEL, "rcu_synchronize() above APC level");
541 kassert(!t || atomic_load_explicit(&t->rcu_nesting, memory_order_relaxed) ==
542 0,
543 "rcu_synchronize() inside an RCU read section");
544 kassert(!rcu.ready || t != rcu.worker,
545 "rcu_synchronize() from the RCU worker deadlocks");
546
547 if (!rcu.ready)
548 return;
549
550 /* Caller's unpublish happened before this load, meaning
551 * grace periods that published a sequence greater than what's
552 * here started after the unpublish */
553 uint64_t target =
554 atomic_load_explicit(&rcu.gp_seq, memory_order_acquire) + 1;
555
556 atomic_fetch_add_explicit(&rcu.gp_requests, 1, memory_order_release);
557 semaphore_post(s: &rcu.sem);
558
559 while (atomic_load_explicit(&rcu.gp_completed, memory_order_acquire) <
560 target)
561 scheduler_yield();
562}
563
564static void rcu_build_tree(void) {
565 size_t counts[RCU_MAX_LEVELS];
566 size_t levels = 0;
567
568 size_t n = global.core_count;
569 kassert(n > 0);
570
571 do {
572 n = (n + RCU_FANOUT - 1) / RCU_FANOUT;
573 kassert(levels < RCU_MAX_LEVELS, "RCU tree deeper than RCU_MAX_LEVELS");
574 counts[levels++] = n;
575 } while (n > 1);
576
577 /* counts[0] is the leaf, counts[levels - 1] is the root */
578 size_t total = 0;
579 for (size_t i = 0; i < levels; i++)
580 total += counts[i];
581
582 rcu.nodes =
583 kmalloc_or_die(total * sizeof(struct rcu_node), ALLOC_FLAGS_ZERO);
584 rcu.node_count = total;
585
586 /* Root first so forward pass initializes parents before children */
587 size_t offsets[RCU_MAX_LEVELS];
588 offsets[levels - 1] = 0;
589 for (size_t i = levels - 1; i > 0; i--)
590 offsets[i - 1] = offsets[i] + counts[i];
591
592 for (size_t level = 0; level < levels; level++) {
593 for (size_t j = 0; j < counts[level]; j++) {
594 struct rcu_node *node = &rcu.nodes[offsets[level] + j];
595
596 spinlock_init(&node->lock);
597 node->gp_seq = 0;
598 node->completed_seq = 0;
599 node->qs_children = 0;
600
601 if (level == levels - 1) {
602 node->parent = NULL;
603 node->child_index = 0;
604 } else {
605 node->parent =
606 &rcu.nodes[offsets[level + 1] + (j / RCU_FANOUT)];
607 node->child_index = j % RCU_FANOUT;
608 }
609
610 size_t owned;
611 if (level == 0) {
612 node->is_leaf = true;
613 node->cpu_base = j * RCU_FANOUT;
614 INIT_LIST_HEAD(list: &node->blocked);
615 owned = global.core_count - node->cpu_base;
616 } else {
617 node->is_leaf = false;
618 owned = counts[level - 1] - (j * RCU_FANOUT);
619 }
620
621 if (owned > RCU_FANOUT)
622 owned = RCU_FANOUT;
623
624 if (node->is_leaf)
625 cpu_mask_set_range(m: &node->full_cpus, start: node->cpu_base, len: owned);
626 else
627 bitmap_fill(map: &node->full_children, nbits: owned);
628 }
629 }
630
631 rcu.root = &rcu.nodes[0];
632 rcu.leaves = &rcu.nodes[offsets[0]];
633 rcu.leaf_count = counts[0];
634}
635
636void rcu_init(void) {
637 atomic_store(&rcu.gp_seq, 1);
638 atomic_store(&rcu.gp_completed, 1);
639 atomic_store(&rcu.gp_requests, 0);
640
641 semaphore_init(s: &rcu.sem, value: 0, SEMAPHORE_INIT_NORMAL);
642
643 rcu_build_tree();
644
645 rcu.cpus = kmalloc_or_die(global.core_count * sizeof(struct rcu_cpu),
646 ALLOC_FLAGS_ZERO);
647 for (cpu_id_t cpu = 0; cpu < global.core_count; cpu++) {
648 spinlock_init(&rcu.cpus[cpu].lock);
649 INIT_LIST_HEAD(list: &rcu.cpus[cpu].list);
650 atomic_store(&rcu.cpus[cpu].reported_seq, 0);
651 }
652
653 rcu.worker = thread_spawn(name: "rcu_gp_worker", entry: rcu_gp_worker, NULL);
654 rcu.ready = true;
655}
656