fuchsia_rcu/state_machine.rs
1// Copyright 2025 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use crate::atomic_stack::{AtomicListIterator, AtomicStack};
6use crate::rcu_callback::RcuCallback;
7use crate::rcu_droppable::RcuDroppable;
8use fuchsia_sync::{Completion, Mutex};
9use std::marker::PhantomData;
10use std::sync::atomic::{AtomicPtr, AtomicU8, AtomicUsize, Ordering};
11use std::thread_local;
12use std::time::Duration;
13
14#[cfg(feature = "rseq_backend")]
15use crate::read_counters::RcuReadCounters;
16#[cfg(feature = "rseq_backend")]
17use std::cell::Cell;
18
19struct RcuControlBlock {
20 /// The generation counter.
21 ///
22 /// The generation counter is incremented whenever the state machine leaves the `Idle` state.
23 generation: AtomicUsize,
24
25 /// The read counters.
26 ///
27 /// Readers increment the counter for the generation that they are reading from. For example,
28 /// if the `generation` is even, then readers increment the counter for the `read_counters[0]`.
29 /// If the `generation` is odd, then readers increment the counter for the `read_counters[1]`.
30 #[cfg(not(feature = "rseq_backend"))]
31 read_counters: [AtomicUsize; 2],
32
33 #[cfg(feature = "rseq_backend")]
34 read_counters: RcuReadCounters,
35
36 /// The chain of callbacks that are waiting to be run.
37 ///
38 /// Writers add callbacks to this chain after writing to the object. The callbacks are run when
39 /// all currently in-flight read operations have completed.
40 callback_chain: AtomicStack<RcuCallback>,
41
42 /// The futex used to put the background advancer thread to sleep when there are no callbacks.
43 advancer_thread_state: zx::Futex,
44
45 /// Callbacks that are ready to run after the next grace period.
46 waiting_callbacks: Mutex<AtomicListIterator<RcuCallback>>,
47}
48
49const ADVANCER_THREAD_SLEEPING: i32 = 0;
50const ADVANCER_THREAD_ACTIVE: i32 = 1;
51
52/// The number of times to spin checking for active readers before yielding.
53const ADVANCER_SPIN_LIMIT: u32 = 64;
54/// The number of spin/yield iterations before falling back to sleeping.
55const ADVANCER_YIELD_LIMIT: u32 = 66;
56/// Initial sleep duration when active readers are still present after spinning and yielding.
57const ADVANCER_INITIAL_SLEEP: Duration = Duration::from_micros(50);
58/// Maximum sleep duration for exponential backoff during long stalls.
59const ADVANCER_MAX_SLEEP: Duration = Duration::from_millis(1);
60
61impl RcuControlBlock {
62 /// Create a new control block for the RCU state machine.
63 const fn new() -> Self {
64 #[cfg(feature = "rseq_backend")]
65 let read_counters = RcuReadCounters::new();
66
67 #[cfg(not(feature = "rseq_backend"))]
68 let read_counters = [AtomicUsize::new(0), AtomicUsize::new(0)];
69
70 Self {
71 generation: AtomicUsize::new(0),
72 read_counters,
73 callback_chain: AtomicStack::new(),
74 advancer_thread_state: zx::Futex::new(ADVANCER_THREAD_SLEEPING),
75 waiting_callbacks: Mutex::new(AtomicListIterator::empty()),
76 }
77 }
78}
79
80/// The control block for the RCU state machine.
81static RCU_CONTROL_BLOCK: RcuControlBlock = RcuControlBlock::new();
82
83struct RcuThreadBlock {
84 /// The number of times the thread has nested into a read lock.
85 nesting_level: AtomicUsize,
86
87 /// The index of the read counter that the thread incremented when it entered its outermost read
88 /// lock.
89 counter_index: AtomicU8,
90
91 /// The restartable sequence for the current thread.
92 #[cfg(feature = "rseq_backend")]
93 rseq: Cell<Option<fuchsia_rseq::Rseq>>,
94}
95
96impl RcuThreadBlock {
97 /// Creates a new `RcuThreadBlock`.
98 const fn new() -> Self {
99 Self {
100 nesting_level: AtomicUsize::new(0),
101 counter_index: AtomicU8::new(0),
102 #[cfg(feature = "rseq_backend")]
103 rseq: Cell::new(None),
104 }
105 }
106
107 /// Returns true if the thread is holding a read lock.
108 fn holding_read_lock(&self) -> bool {
109 self.nesting_level.load(Ordering::Relaxed) > 0
110 }
111}
112thread_local! {
113 /// Thread-specific data for the RCU state machine.
114 ///
115 /// This data is used to track the nesting level of read locks and the index of the read counter
116 /// that the thread incremented when it entered its outermost read lock.
117 static RCU_THREAD_BLOCK: RcuThreadBlock = const { RcuThreadBlock::new() };
118}
119
120/// An RAII guard that keeps the current thread registered with RCU and RSEQ.
121///
122/// Unregisters the thread when dropped.
123#[derive(Debug)]
124#[must_use = "the thread is unregistered when the guard is dropped"]
125pub struct RcuThreadRegistration {
126 _marker: PhantomData<*const ()>,
127}
128
129impl RcuThreadRegistration {
130 /// Leaks the registration, keeping the current thread registered indefinitely.
131 #[inline]
132 pub fn leak(self) {
133 std::mem::forget(self);
134 }
135}
136
137impl Drop for RcuThreadRegistration {
138 fn drop(&mut self) {
139 unregister_thread();
140 }
141}
142
143/// Registers the current thread for RCU and RSEQ.
144///
145/// With RSEQ enabled, RCU requires registration of its RSEQ critical section. By default, it
146/// registers and unregisters on each `rcu_read_lock`. To avoid this overhead, if a thread knows the only
147/// RSEQ it needs is for RCU, it may create an [`RcuThreadRegistration`] which will keep the
148/// registration alive for as long as the [`RcuThreadRegistration`] exists. This avoids the overhead of
149/// registration on each read lock, but no other RSEQs may be used by the thread until dropped.
150pub fn register_thread() -> RcuThreadRegistration {
151 #[cfg(feature = "rseq_backend")]
152 {
153 RCU_THREAD_BLOCK.with(|block| {
154 assert!(
155 !block.holding_read_lock(),
156 "Cannot register thread while holding an RCU read lock"
157 );
158 assert!(block.rseq.get().is_none(), "thread already registered");
159 let rseq = fuchsia_rseq::rseq_register_thread_with_cs(
160 crate::read_counters::rcu_critical_section(),
161 );
162 block.rseq.set(Some(rseq));
163 });
164 }
165
166 RcuThreadRegistration { _marker: PhantomData }
167}
168
169/// Unregisters the current thread from RCU and RSEQ.
170pub(crate) fn unregister_thread() {
171 #[cfg(feature = "rseq_backend")]
172 {
173 RCU_THREAD_BLOCK.with(|block| {
174 assert!(
175 !block.holding_read_lock(),
176 "Cannot unregister thread while holding an RCU read lock"
177 );
178 assert!(block.rseq.take().is_some(), "thread not registered");
179 });
180 // SAFETY: We've verified no read lock is held and cleared the cached `rseq`
181 // handle from `RCU_THREAD_BLOCK`, so no further access will be made using that handle.
182 unsafe {
183 fuchsia_rseq::rseq_unregister_thread();
184 }
185 }
186}
187
188/// Exposes the thread-local counters for RCU stall detection.
189pub fn with_thread_block_counters<F>(f: F)
190where
191 F: FnOnce(*const AtomicUsize, *const AtomicU8),
192{
193 RCU_THREAD_BLOCK.with(|thread_block| {
194 f(&thread_block.nesting_level as *const _, &thread_block.counter_index as *const _);
195 });
196}
197
198/// Acquire a read lock.
199///
200/// This function is used to acquire a read lock on the RCU state machine. The RCU state machine
201/// defers calling callbacks until all currently in-flight read operations have completed.
202///
203/// Must be balanced by a call to `rcu_read_unlock` on the same thread.
204#[inline]
205pub(crate) fn rcu_read_lock() {
206 RCU_THREAD_BLOCK.with(|thread_block| {
207 let nesting_level = thread_block.nesting_level.load(Ordering::Relaxed);
208 if nesting_level > 0 {
209 // If this thread already has a read lock, increment the nesting level instead of the
210 // incrementing the read counter. This approach is a performance optimization to reduce
211 // the number of atomic operations that need to be performed.
212 thread_block.nesting_level.store(nesting_level + 1, Ordering::Relaxed);
213 } else {
214 // This is the outermost read lock. Increment the read counter.
215 let control_block = &RCU_CONTROL_BLOCK;
216
217 // There's a race here where we capture `index` and then go on to increment the read
218 // counter. The choice of `index` here isn't actually important for correctness because
219 // we always wait at least two grace periods before calling the callbacks, so it doesn't
220 // matter which counter we increment. It does mean that a thread waiting for the read
221 // counter to drop to zero, could actually find that the read counter increases before
222 // it eventually reaches zero, which should be fine.
223 let index = control_block.generation.load(Ordering::Relaxed) & 1;
224
225 #[cfg(feature = "rseq_backend")]
226 {
227 match thread_block.rseq.get() {
228 // SAFETY: `register_thread()` initialized `rseq` with `rcu_critical_section()`.
229 Some(rseq) => unsafe { control_block.read_counters.begin(&rseq, index) },
230 None => control_block.read_counters.begin_slow(index),
231 }
232 std::sync::atomic::compiler_fence(Ordering::SeqCst);
233 }
234
235 #[cfg(not(feature = "rseq_backend"))]
236 {
237 // Synchronization point [A] (see design.md)
238 control_block.read_counters[index].fetch_add(1, Ordering::SeqCst);
239 }
240
241 thread_block.counter_index.store(index as u8, Ordering::Relaxed);
242 thread_block.nesting_level.store(1, Ordering::Relaxed);
243 }
244 });
245}
246
247/// Release a read lock.
248///
249/// This function is used to release a read lock on the RCU state machine. See `rcu_read_lock` for
250/// more details.
251#[inline]
252pub(crate) fn rcu_read_unlock() {
253 RCU_THREAD_BLOCK.with(|thread_block| {
254 let nesting_level = thread_block.nesting_level.load(Ordering::Relaxed);
255 if nesting_level > 1 {
256 // If the nesting level is greater than 1, this is not the outermost read lock.
257 // Decrement the nesting level instead of the read counter.
258 thread_block.nesting_level.store(nesting_level - 1, Ordering::Relaxed);
259 } else {
260 // This is the outermost read lock. Decrement the read counter.
261 let index = thread_block.counter_index.load(Ordering::Relaxed) as usize;
262 let control_block = &RCU_CONTROL_BLOCK;
263
264 #[cfg(feature = "rseq_backend")]
265 {
266 std::sync::atomic::compiler_fence(Ordering::SeqCst);
267 match thread_block.rseq.get() {
268 // SAFETY: `register_thread()` initialized `rseq` with `rcu_critical_section()`.
269 Some(rseq) => unsafe { control_block.read_counters.end(&rseq, index) },
270 None => control_block.read_counters.end_slow(index),
271 }
272 }
273
274 #[cfg(not(feature = "rseq_backend"))]
275 {
276 // Synchronization point [B] (see design.md)
277 control_block.read_counters[index].fetch_sub(1, Ordering::SeqCst);
278 }
279
280 thread_block.nesting_level.store(0, Ordering::Relaxed);
281 thread_block.counter_index.store(u8::MAX, Ordering::Relaxed);
282 }
283 });
284}
285
286/// Read the value of an RCU pointer.
287///
288/// This function cannot be called unless the current thread is holding a read lock. The returned
289/// pointer is valid until the read lock is released.
290pub(crate) fn rcu_read_pointer<T>(ptr: &AtomicPtr<T>) -> *const T {
291 // Synchronization point [D] (see design.md)
292 ptr.load(Ordering::Acquire)
293}
294
295/// Assign a new value to an RCU pointer.
296///
297/// Concurrent readers may continue to reference the old value of the pointer until the RCU state
298/// machine has made sufficient progress. To clean up the old value of the pointer, use `rcu_call`
299/// or `rcu_drop`, which defer processing until all in-flight read operations have completed.
300pub(crate) fn rcu_assign_pointer<T>(ptr: &AtomicPtr<T>, new_ptr: *mut T) {
301 // Synchronization point [E] (see design.md)
302 ptr.store(new_ptr, Ordering::Release);
303}
304
305/// Replace the value of an RCU pointer.
306///
307/// Concurrent readers may continue to reference the old value of the pointer until the RCU state
308/// machine has made sufficient progress. To clean up the old value of the pointer, use `rcu_call`
309/// or `rcu_drop`, which defer processing until all in-flight read operations have completed.
310pub(crate) fn rcu_replace_pointer<T>(ptr: &AtomicPtr<T>, new_ptr: *mut T) -> *mut T {
311 // Synchronization point [F] (see design.md)
312 ptr.swap(new_ptr, Ordering::AcqRel)
313}
314
315/// Call a callback to run after all in-flight read operations have completed.
316///
317/// To wait until the callback is ready to run, call `rcu_synchronize()`. Note that
318/// `rcu_synchronize()` requires an advancer thread or a thread calling `rcu_run_callbacks()` to
319/// make progress and invoke callbacks. The callback might be called from an arbitrary thread.
320///
321/// NOTE: The order in which callbacks are called is not guaranteed since they can be called
322/// concurrently from multiple threads.
323pub(crate) fn rcu_call(callback: impl FnOnce() + Send + Sync + 'static) {
324 #[cfg(not(feature = "rseq_backend"))]
325 {
326 // We need to synchronize with rcu_read_lock. We need to ensure that all prior stores are
327 // visible to threads that have called rcu_read_lock. We must synchronize with both read
328 // counters using a store operation. We don't need to change the value.
329 std::sync::atomic::fence(Ordering::Release);
330
331 RCU_CONTROL_BLOCK.read_counters[0].fetch_add(0, Ordering::Relaxed);
332 RCU_CONTROL_BLOCK.read_counters[1].fetch_add(0, Ordering::Relaxed);
333 }
334
335 // Synchronization point [G] (see design.md)
336 RCU_CONTROL_BLOCK.callback_chain.push_front(RcuCallback::new(callback));
337
338 // Wake the rcu advancer thread if it is sleeping on the futex.
339 let thread_state = &RCU_CONTROL_BLOCK.advancer_thread_state;
340
341 // This write is required to be SeqCst to ensure total ordering with additions to the
342 // callback_chain. See the comment in rcu_advancer_wait_for_work for details.
343 if thread_state.swap(ADVANCER_THREAD_ACTIVE, Ordering::SeqCst) == ADVANCER_THREAD_SLEEPING {
344 thread_state.wake(1);
345 }
346}
347
348/// Schedule the object to be dropped after all in-flight read operations have completed.
349///
350/// To wait until the object is dropped, call `rcu_synchronize()`. Note that `rcu_synchronize()`
351/// requires an advancer thread or a thread calling `rcu_run_callbacks()` to make progress and drop
352/// the object.
353///
354/// To be safely passed to [rcu_drop] either directly, or indirectly by an rcu container, the type
355/// must implement the marker trait [RcuDroppable] to indicate:
356/// - Dropping T must not take locks or otherwise block.
357/// - It is safe to drop T from an arbitrary thread.
358/// - There is no guarantee as to _when_ T will actually be dropped unless `rcu_synchronize()` or
359/// `rcu_run_callbacks()` is called.
360pub fn rcu_drop<T: RcuDroppable + Sync>(value: T) {
361 rcu_call(move || {
362 std::mem::drop(value);
363 });
364}
365
366/// Check if there are any active readers for the given generation.
367fn has_active_readers(generation: usize) -> bool {
368 let index = generation & 1;
369
370 #[cfg(feature = "rseq_backend")]
371 {
372 return RCU_CONTROL_BLOCK.read_counters.has_active(index);
373 }
374
375 #[cfg(not(feature = "rseq_backend"))]
376 {
377 // Synchronization point [C] (see design.md)
378 RCU_CONTROL_BLOCK.read_counters[index].load(Ordering::SeqCst) > 0
379 }
380}
381
382/// Wake the rcu advancer thread if it's sleeping.
383pub fn rcu_advancer_wake() {
384 // Scheduling a no-op callback is a convenient way to both wake the rcu advancer and also have
385 // it consider the wakeup to be non spurious so it doesn't immediately go back to sleep.
386 rcu_call(|| {});
387}
388
389/// Blocks the current thread until all in-flight read operations have completed for the given
390/// generation.
391///
392/// Postcondition: The number of active readers for the given generation is zero.
393fn rcu_advancer_wait_for_readers(generation: usize) {
394 let mut spins = 0u32;
395 let mut sleep_duration = ADVANCER_INITIAL_SLEEP;
396 while has_active_readers(generation) {
397 // In practice, we tend to see a bimodel distribution of read locks that either release
398 // within a few hundred ns, or around a few µs.
399 //
400 // Then, we see long tail of cases where the release can take > 1ms because a thread got
401 // context switched out while holding a read lock.
402 //
403 // We attempt to model this behavior by first spinning, then slowly backing off.
404 if spins < ADVANCER_SPIN_LIMIT {
405 std::hint::spin_loop();
406 spins += 1;
407 } else if spins < ADVANCER_YIELD_LIMIT {
408 std::thread::yield_now();
409 spins += 1;
410 } else {
411 std::thread::sleep(sleep_duration);
412 sleep_duration = std::cmp::min(sleep_duration * 2, ADVANCER_MAX_SLEEP);
413 }
414 }
415}
416
417/// Advance the RCU state machine.
418///
419/// This function blocks until all in-flight read operations have completed for the current
420/// generation and all callbacks have been run.
421fn rcu_grace_period() {
422 let callbacks = {
423 let mut waiting_callbacks = RCU_CONTROL_BLOCK.waiting_callbacks.lock();
424
425 // We are in the *Idle* state.
426
427 // Swap out the callbacks that we can run when this grace period has passed with the
428 // callbacks that can run after the next period.
429 // Synchronization point [H] (see design.md)
430 let callbacks =
431 std::mem::replace(&mut *waiting_callbacks, RCU_CONTROL_BLOCK.callback_chain.take());
432
433 // Issue an IPI to all CPUs to force them to serialize their execution.
434 // This ensures that all prior stores by all writers are visible to
435 // any thread that subsequently enters an RCU read-side critical section.
436 #[cfg(feature = "rseq_backend")]
437 unsafe {
438 zx::sys::zx_membarrier_sync_process_data()
439 };
440
441 let generation = RCU_CONTROL_BLOCK.generation.fetch_add(1, Ordering::Relaxed);
442
443 // Enter the *Waiting* state
444 rcu_advancer_wait_for_readers(generation);
445
446 // Return to the *Idle* state.
447 callbacks
448 };
449
450 for callback in callbacks {
451 callback.invoke();
452 }
453}
454
455/// Block until all in-flight read operations have completed for callbacks registered prior to this
456/// call.
457///
458/// Note: This function does not advance the RCU state machine itself; it registers a callback and
459/// blocks until that callback is run. If no thread is calling `rcu_run_callbacks()` (for example,
460/// via a dedicated advancer thread), this function will block indefinitely.
461pub fn rcu_synchronize() {
462 RCU_THREAD_BLOCK.with(|block| {
463 assert!(!block.holding_read_lock());
464 });
465
466 let completion = std::sync::Arc::new(Completion::new());
467 let c = completion.clone();
468 rcu_call(move || {
469 c.signal();
470 });
471
472 // If callbacks have run, then we know all read operations must have completed for the
473 // generation we registered the callback on.
474 completion.wait();
475}
476
477/// Check if there is any work waiting to be processed by the RCU advancer.
478fn has_pending_work() -> bool {
479 !RCU_CONTROL_BLOCK.callback_chain.is_empty()
480 || !RCU_CONTROL_BLOCK.waiting_callbacks.lock().is_empty()
481}
482
483/// Blocks the calling advancer thread until RCU callbacks are scheduled.
484pub fn rcu_advancer_wait_for_work() {
485 let thread_state = &RCU_CONTROL_BLOCK.advancer_thread_state;
486 while !has_pending_work() {
487 // This needs to be SeqCst to make the has_pending_work() call synchronize properly, as
488 // with the write in rcu_call. Without a SeqCst, it's possible that a waker sees our write,
489 // and thus doesn't wake us, and simultaneously, we don't see the pending work, and thus go
490 // to sleep, causing a lost wakeup.
491 //
492 // With the SeqCst total ordering, we're guaranteed that either a thread scheduling
493 // callbacks doesn't observe our write, and thus tries to wake us, or that we observe the
494 // added callback, and thus don't sleep.
495 thread_state.store(ADVANCER_THREAD_SLEEPING, Ordering::SeqCst);
496
497 // Double-check after storing SLEEPING to prevent race with rcu_call, rcu_synchronize, or
498 // rcu_advancer_wake.
499 if has_pending_work() {
500 break;
501 }
502
503 // In the case of a spurious wakeup, we recheck has_pending_work and if there is no work to
504 // be done, attempt to return to sleep.
505 let _ = thread_state.wait(ADVANCER_THREAD_SLEEPING, None, zx::MonotonicInstant::INFINITE);
506 }
507 thread_state.store(ADVANCER_THREAD_ACTIVE, Ordering::Relaxed);
508}
509
510/// Advances the RCU state machine if work is pending and runs ready callbacks.
511///
512/// If callbacks are pending, this runs two grace periods and invokes any ready callbacks.
513///
514/// Returns `true` if callbacks were processed, or `false` if no work was pending.
515pub fn rcu_run_callbacks() -> bool {
516 RCU_THREAD_BLOCK.with(|block| {
517 assert!(!block.holding_read_lock());
518 });
519
520 let thread_state = &RCU_CONTROL_BLOCK.advancer_thread_state;
521 thread_state.store(ADVANCER_THREAD_ACTIVE, Ordering::Relaxed);
522
523 if has_pending_work() {
524 rcu_grace_period();
525 rcu_grace_period();
526 true
527 } else {
528 false
529 }
530}
531
532#[cfg(test)]
533mod tests {
534 use super::*;
535 use std::sync::Arc;
536 use std::sync::atomic::{AtomicBool, Ordering};
537
538 static TEST_MUTEX: std::sync::Mutex<()> = std::sync::Mutex::new(());
539
540 #[test]
541 fn test_rcu_delay_regression() {
542 let _lock = TEST_MUTEX.lock().unwrap();
543 let flag = Arc::new(AtomicBool::new(false));
544 let moved_flag = flag.clone();
545
546 rcu_call(move || {
547 moved_flag.store(true, Ordering::SeqCst);
548 });
549
550 rcu_grace_period();
551
552 assert!(
553 !flag.load(Ordering::SeqCst),
554 "Callback executed too early! RCU requires 2 grace periods delay."
555 );
556
557 rcu_grace_period();
558 assert!(flag.load(Ordering::SeqCst), "Callback should have executed after 2 grace periods");
559 }
560
561 #[test]
562 fn test_rcu_synchronize() {
563 let _lock = TEST_MUTEX.lock().unwrap();
564 let stop = Arc::new(AtomicBool::new(false));
565 let stop_clone = stop.clone();
566
567 let handle = std::thread::spawn(move || {
568 while !stop_clone.load(Ordering::Relaxed) {
569 rcu_advancer_wait_for_work();
570 rcu_run_callbacks();
571 }
572 rcu_run_callbacks();
573 });
574
575 let completion = Arc::new(Completion::new());
576 let c = completion.clone();
577
578 rcu_call(move || {
579 c.signal();
580 });
581
582 rcu_synchronize();
583
584 // Callbacks within a batch are not guaranteed to execute in a specific order,
585 // so the callback may complete slightly before or after rcu_synchronize() returns.
586 completion.wait();
587
588 stop.store(true, Ordering::Relaxed);
589 rcu_advancer_wake();
590 handle.join().unwrap();
591 RCU_CONTROL_BLOCK.advancer_thread_state.store(ADVANCER_THREAD_SLEEPING, Ordering::SeqCst);
592 }
593
594 #[test]
595 fn test_rcu_run_callbacks() {
596 let _lock = TEST_MUTEX.lock().unwrap();
597 let flag = Arc::new(AtomicBool::new(false));
598 let moved_flag = flag.clone();
599
600 rcu_call(move || {
601 moved_flag.store(true, Ordering::SeqCst);
602 });
603
604 assert!(rcu_run_callbacks());
605 assert!(
606 flag.load(Ordering::SeqCst),
607 "Callback should have executed after rcu_run_callbacks()"
608 );
609 // Second step has no pending work.
610 assert!(!rcu_run_callbacks());
611 RCU_CONTROL_BLOCK.advancer_thread_state.store(ADVANCER_THREAD_SLEEPING, Ordering::SeqCst);
612 }
613
614 #[test]
615 fn test_rcu_advancer_thread_wake() {
616 let _lock = TEST_MUTEX.lock().unwrap();
617 let flag = Arc::new(AtomicBool::new(false));
618 let flag_clone = flag.clone();
619
620 let stop = Arc::new(AtomicBool::new(false));
621 let stop_clone = stop.clone();
622
623 let handle = std::thread::spawn(move || {
624 while !stop_clone.load(Ordering::Relaxed) {
625 rcu_advancer_wait_for_work();
626 rcu_run_callbacks();
627 }
628 rcu_run_callbacks();
629 });
630
631 rcu_call(move || {
632 flag_clone.store(true, Ordering::SeqCst);
633 });
634
635 // Wait for advancer thread to wake up and process callback.
636 let start = std::time::Instant::now();
637 while !flag.load(Ordering::SeqCst) {
638 assert!(
639 start.elapsed() < std::time::Duration::from_secs(5),
640 "Timed out waiting for advancer thread to process callback"
641 );
642 std::thread::sleep(std::time::Duration::from_millis(10));
643 }
644
645 stop.store(true, Ordering::Relaxed);
646 // Wake the thread if it went back to sleep so it can terminate.
647 rcu_advancer_wake();
648 handle.join().unwrap();
649 RCU_CONTROL_BLOCK.advancer_thread_state.store(ADVANCER_THREAD_SLEEPING, Ordering::SeqCst);
650 }
651
652 #[test]
653 fn test_rcu_synchronize_no_callbacks() {
654 let _lock = TEST_MUTEX.lock().unwrap();
655 let stop = Arc::new(AtomicBool::new(false));
656 let stop_clone = stop.clone();
657
658 let handle = std::thread::spawn(move || {
659 while !stop_clone.load(Ordering::Relaxed) {
660 rcu_advancer_wait_for_work();
661 rcu_run_callbacks();
662 }
663 rcu_run_callbacks();
664 });
665
666 // Calling rcu_synchronize() with no prior callbacks should wake the advancer,
667 // wait for the grace periods to complete, and return successfully.
668 rcu_synchronize();
669
670 stop.store(true, Ordering::Relaxed);
671 rcu_advancer_wake();
672 handle.join().unwrap();
673 RCU_CONTROL_BLOCK.advancer_thread_state.store(ADVANCER_THREAD_SLEEPING, Ordering::SeqCst);
674 }
675}