Skip to main content

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}