Add futex wake-path store-buffering reproduction

litebox/Cargo.toml

@@ -34,6 +34,10 @@ windows-sys = { version = "0.60.2", features = [
 lock_tracing = ["dep:arrayvec", "spin/mutex"]
 panic_on_unclosed_fd_drop = []
 enforce_singleton_litebox_instance = []
+# Exposes the wake-path ordering stress instrumentation (see `src/ordering_stress.rs`)
+# to dependent crates so they can drive the reproductions over a real platform.
+# Test-only; never enable in production builds.
+ordering_stress = []
 
 [lints]
 workspace = true

litebox/src/event/polling.rs

@@ -69,12 +69,19 @@ impl<Platform: RawSyncPrimitivesProvider + TimeProvider> WaitContext<'_, Platfor
             events | Events::ALWAYS_POLLED,
         )
         .map_err(TryOpError::Other)?;
+        #[cfg(any(test, feature = "ordering_stress"))]
+        crate::ordering_stress::waiter_registered();
         loop {
             match try_op() {
                 Err(TryOpError::TryAgain) => {}
                 ret => return ret,
             }
-            match self.wait_until(|| observer.is_ready()) {
+            match self.wait_until(|| {
+                let ready = observer.is_ready();
+                #[cfg(any(test, feature = "ordering_stress"))]
+                crate::ordering_stress::record_waiter_done(ready);
+                ready
+            }) {
                 Ok(()) => {}
                 Err(err) => return Err(TryOpError::WaitError(err)),
             }
@@ -169,8 +176,131 @@ impl<Platform: RawSyncPrimitivesProvider> PolleeObserver<Platform> {
 
 impl<Platform: RawSyncPrimitivesProvider> Observer<Events> for PolleeObserver<Platform> {
     fn on_events(&self, _events: &Events) {
+        #[cfg(any(test, feature = "ordering_stress"))]
+        crate::ordering_stress::waker_rendezvous();
         self.ready
             .store(true, core::sync::atomic::Ordering::Release);
         self.waker.wake();
     }
 }
+
+#[cfg(test)]
+mod tests {
+    extern crate std;
+
+    use super::{Pollee, TryOpError};
+    use crate::LiteBox;
+    use crate::event::Events;
+    use crate::event::wait::{WaitError, WaitState};
+    use crate::platform::mock::MockPlatform;
+    use alloc::sync::Arc;
+    use core::convert::Infallible;
+    use core::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::Barrier;
+    use std::thread;
+    use std::time::Duration;
+
+    /// Reproduces the store-buffering hazard in the polling notify path, the sibling of
+    /// the futex one in [`crate::sync::futex`]: `PolleeObserver::on_events` publishes
+    /// `ready` with a `Release` store (a plain `mov` on x86) and then calls `wake`, whose
+    /// `fetch_update` reads the thread state, while the waiter stores `WAITING` and then
+    /// loads `ready`. Both sides can read the other's stale value, so the notifier skips
+    /// the wake on `RUNNING_IN_HOST` while the waiter blocks on a stale `ready == false`.
+    /// Ignored because it is probabilistic; run with `LITEBOX_POLL_STRESS_ITERS` to
+    /// control the iteration count.
+    #[test]
+    #[ignore = "probabilistic weak-memory stress test"]
+    fn stress_registered_poller_does_not_miss_notification() {
+        let platform = MockPlatform::new();
+        let _litebox = LiteBox::new(platform);
+        let pollee = Arc::new(Pollee::<MockPlatform>::new());
+        let iterations = std::env::var("LITEBOX_POLL_STRESS_ITERS")
+            .ok()
+            .and_then(|value| value.parse().ok())
+            .unwrap_or(1_000_000);
+        crate::ordering_stress::activate();
+        let iteration_start = Arc::new(Barrier::new(3));
+        let iteration_finish = Arc::new(Barrier::new(3));
+        let waiter_timed_out = Arc::new(AtomicBool::new(false));
+
+        let waiter = {
+            let pollee = Arc::clone(&pollee);
+            let iteration_start = Arc::clone(&iteration_start);
+            let iteration_finish = Arc::clone(&iteration_finish);
+            let waiter_timed_out = Arc::clone(&waiter_timed_out);
+            thread::spawn(move || {
+                for _ in 0..iterations {
+                    iteration_start.wait();
+                    // The pre-registration probe and the first in-loop probe must report
+                    // `TryAgain` so the waiter always reaches the racing `wait_until`; a
+                    // later probe reports success so a delivered notification ends the wait.
+                    let mut probes = 0u32;
+                    let result: Result<(), TryOpError<Infallible>> = pollee.wait(
+                        &WaitState::new(platform)
+                            .context()
+                            .with_timeout(Duration::from_millis(20)),
+                        false,
+                        Events::IN,
+                        || {
+                            probes += 1;
+                            if probes <= 2 {
+                                Err(TryOpError::TryAgain)
+                            } else {
+                                Ok(())
+                            }
+                        },
+                    );
+                    waiter_timed_out.store(
+                        matches!(result, Err(TryOpError::WaitError(WaitError::TimedOut))),
+                        Ordering::Relaxed,
+                    );
+                    iteration_finish.wait();
+                }
+            })
+        };
+
+        let notifier = {
+            let pollee = Arc::clone(&pollee);
+            let iteration_start = Arc::clone(&iteration_start);
+            let iteration_finish = Arc::clone(&iteration_finish);
+            thread::spawn(move || {
+                for _ in 0..iterations {
+                    iteration_start.wait();
+                    while !crate::ordering_stress::waiter_is_registered() {
+                        core::hint::spin_loop();
+                    }
+                    pollee.notify_observers(Events::IN);
+                    iteration_finish.wait();
+                }
+            })
+        };
+
+        let mut both_old = 0;
+        let mut lost_wakeups = 0;
+
+        for _ in 0..iterations {
+            crate::ordering_stress::begin_round();
+            waiter_timed_out.store(false, Ordering::Relaxed);
+            iteration_start.wait();
+            crate::ordering_stress::wait_until_parked();
+            crate::ordering_stress::release();
+            iteration_finish.wait();
+            if crate::ordering_stress::observed_both_old() {
+                both_old += 1;
+            }
+            if waiter_timed_out.load(Ordering::Relaxed) {
+                lost_wakeups += 1;
+            }
+        }
+
+        waiter.join().unwrap();
+        notifier.join().unwrap();
+        crate::ordering_stress::deactivate();
+        std::eprintln!("iterations={iterations} both_old={both_old} lost_wakeups={lost_wakeups}");
+        assert_eq!(both_old, 0, "the polling notify path saw both old values");
+        assert_eq!(
+            lost_wakeups, 0,
+            "a registered poller missed its notification"
+        );
+    }
+}

litebox/src/event/wait.rs

@@ -68,6 +68,9 @@ impl<Platform: RawSyncPrimitivesProvider> Waker<Platform> {
     /// Causes the thread blocked in [`WaitContext::wait_until`] to wake up and
     /// reevaluate its wait condition.
     ///
+    /// Callers must publish whatever condition the waiter tests *before* calling this;
+    /// the fence inside pairs with the one in `start_wait` to make that store visible.
+    ///
     /// Note that this does not interrupt guest execution; to interrupt guest
     /// execution, use [`ThreadHandle::interrupt`].
     pub fn wake(&self) {
@@ -161,6 +164,12 @@ impl<Platform: RawSyncPrimitivesProvider> WaitState<Platform> {
 impl<Platform: RawSyncPrimitivesProvider> WaitStateInner<Platform> {
     /// Wakes up the thread if it is waiting (but not if it is running in the guest).
     fn wake(&self) {
+        // Callers publish their condition flag immediately before this call. Without a
+        // fence on *both* sides, that store and the state read below are a store-buffering
+        // pair: each thread can read the other's stale value, so the waiter blocks on an
+        // unset condition while this skips the wake on a stale `RUNNING_IN_HOST`.
+        // Acquire/Release cannot express this -- only a store-load barrier can.
+        core::sync::atomic::fence(Ordering::SeqCst);
         let condvar = &self.condvar;
         let v = condvar.underlying_atomic().fetch_update(
             Ordering::Release,
@@ -174,6 +183,8 @@ impl<Platform: RawSyncPrimitivesProvider> WaitStateInner<Platform> {
                 state => unreachable!("{state:?}"),
             },
         );
+        #[cfg(any(test, feature = "ordering_stress"))]
+        crate::ordering_stress::record_waker_result(v);
         match v.map(ThreadState) {
             Ok(ThreadState::WAITING) => {
                 condvar.wake_one();
@@ -211,6 +222,10 @@ impl<Platform: RawSyncPrimitivesProvider + ThreadProvider> ThreadHandle<Platform
     /// condition and interrupt condition. If it is running guest code, the
     /// platform will interrupt the thread and re-enter the shim.
     pub fn interrupt(&self) {
+        // See the fence in `WaitStateInner::wake`: callers publish their interrupt
+        // condition (e.g. `is_exiting`) immediately before this call, and
+        // `check_for_interrupt` reads it after the waiter's `WAITING` store.
+        core::sync::atomic::fence(Ordering::SeqCst);
         let condvar = &self.waker.0.condvar;
         let v = condvar.underlying_atomic().fetch_update(
             Ordering::Release,
@@ -224,6 +239,8 @@ impl<Platform: RawSyncPrimitivesProvider + ThreadProvider> ThreadHandle<Platform
                 state => unreachable!("{state:?}"),
             },
         );
+        #[cfg(any(test, feature = "ordering_stress"))]
+        crate::ordering_stress::record_waker_result(v);
         match v.map(ThreadState) {
             Ok(ThreadState::WAITING) => {
                 condvar.wake_one();
@@ -375,9 +392,14 @@ impl<'a, Platform: RawSyncPrimitivesProvider + TimeProvider> WaitContext<'a, Pla
     /// missed.
     fn start_wait(&self) {
         self.waker.0.platform.update_waker(Some(self.waker.clone()));
+        #[cfg(any(test, feature = "ordering_stress"))]
+        crate::ordering_stress::waiter_rendezvous();
         self.waker
             .0
             .set_state(ThreadState::WAITING, Ordering::SeqCst);
+        // Pairs with the fences in `WaitStateInner::wake` and `ThreadHandle::interrupt`;
+        // both sides need one to forbid the store-buffering outcome.
+        core::sync::atomic::fence(Ordering::SeqCst);
     }
 
     /// Returns the thread to the running state after a wait.
@@ -480,3 +502,140 @@ pub enum WaitError {
     #[error("wait timed out")]
     TimedOut,
 }
+
+#[cfg(test)]
+mod tests {
+    extern crate std;
+
+    use super::{CheckForInterrupt, WaitError, WaitState};
+    use crate::LiteBox;
+    use crate::platform::mock::MockPlatform;
+    use alloc::sync::Arc;
+    use core::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::Barrier;
+    use std::thread;
+
+    /// Stands in for the shim's `is_exiting`: published with a plain relaxed store by the
+    /// interrupting thread and read by the waiter inside `check_for_interrupt`.
+    struct ExitFlag(AtomicBool);
+
+    impl CheckForInterrupt for ExitFlag {
+        fn check_for_interrupt(&self) -> bool {
+            let exiting = self.0.load(Ordering::Relaxed);
+            crate::ordering_stress::record_waiter_done(exiting);
+            exiting
+        }
+    }
+
+    /// The third instance of the wake-path store-buffering hazard, alongside the futex and
+    /// polling ones: an interrupter publishes its flag and then calls
+    /// [`ThreadHandle::interrupt`], whose `fetch_update` reads the thread state, while the
+    /// waiter stores `WAITING` and then reads that flag via `check_for_interrupt`. Both
+    /// sides can read the other's stale value, so the interrupt is skipped on
+    /// `RUNNING_IN_HOST` while the waiter blocks on a stale flag.
+    ///
+    /// The wait deliberately has **no deadline**, for two reasons: it is the case that
+    /// actually hangs in production (`kill_other_threads` during `execve` waits forever for
+    /// a thread that never got its interrupt), and a deadline makes `commit_wait` call
+    /// `remaining_timeout()` between the `WAITING` store and the flag read, which widens the
+    /// window enough to hide the race entirely. To keep the suite from hanging, the
+    /// interrupter issues a second, rescue interrupt; by then the `WAITING` store has
+    /// drained, so it always lands. `record_waker_result` is first-write-wins, so the rescue
+    /// cannot overwrite the racing result.
+    ///
+    /// Ignored because it is probabilistic; run with `LITEBOX_INTERRUPT_STRESS_ITERS` to
+    /// control the iteration count.
+    #[test]
+    #[ignore = "probabilistic weak-memory stress test"]
+    fn stress_interrupted_waiter_is_not_missed() {
+        let platform = MockPlatform::new();
+        let _litebox = LiteBox::new(platform);
+        let iterations = std::env::var("LITEBOX_INTERRUPT_STRESS_ITERS")
+            .ok()
+            .and_then(|value| value.parse().ok())
+            .unwrap_or(200_000);
+        crate::ordering_stress::activate();
+        let iteration_start = Arc::new(Barrier::new(3));
+        let iteration_finish = Arc::new(Barrier::new(3));
+        let exit_flag = Arc::new(ExitFlag(AtomicBool::new(false)));
+        let waiter_interrupted = Arc::new(AtomicBool::new(false));
+        let (handle_tx, handle_rx) = std::sync::mpsc::channel();
+
+        let waiter = {
+            let iteration_start = Arc::clone(&iteration_start);
+            let iteration_finish = Arc::clone(&iteration_finish);
+            let exit_flag = Arc::clone(&exit_flag);
+            let waiter_interrupted = Arc::clone(&waiter_interrupted);
+            thread::spawn(move || {
+                // `WaitState` is `!Sync`, so it must live on the waiting thread; only its
+                // `ThreadHandle` crosses over to the interrupter.
+                let wait_state = WaitState::new(platform);
+                handle_tx.send(wait_state.thread_handle()).unwrap();
+                for _ in 0..iterations {
+                    iteration_start.wait();
+                    crate::ordering_stress::waiter_registered();
+                    let err = wait_state
+                        .context()
+                        .with_check_for_interrupt(&*exit_flag)
+                        .sleep();
+                    waiter_interrupted
+                        .store(matches!(err, WaitError::Interrupted), Ordering::Relaxed);
+                    iteration_finish.wait();
+                }
+            })
+        };
+
+        let handle = handle_rx.recv().unwrap();
+        let interrupter = {
+            let iteration_start = Arc::clone(&iteration_start);
+            let iteration_finish = Arc::clone(&iteration_finish);
+            let exit_flag = Arc::clone(&exit_flag);
+            thread::spawn(move || {
+                for _ in 0..iterations {
+                    iteration_start.wait();
+                    while !crate::ordering_stress::waiter_is_registered() {
+                        core::hint::spin_loop();
+                    }
+                    crate::ordering_stress::waker_rendezvous();
+                    exit_flag.0.store(true, Ordering::Relaxed);
+                    handle.interrupt();
+                    // Liveness rescue: if the first interrupt was dropped on a stale
+                    // `RUNNING_IN_HOST`, production would hang here forever.
+                    handle.interrupt();
+                    iteration_finish.wait();
+                }
+            })
+        };
+
+        let mut both_old = 0;
+        let mut not_interrupted = 0;
+
+        for _ in 0..iterations {
+            crate::ordering_stress::begin_round();
+            exit_flag.0.store(false, Ordering::Relaxed);
+            waiter_interrupted.store(false, Ordering::Relaxed);
+            iteration_start.wait();
+            crate::ordering_stress::wait_until_parked();
+            crate::ordering_stress::release();
+            iteration_finish.wait();
+            if crate::ordering_stress::observed_both_old() {
+                both_old += 1;
+            }
+            if !waiter_interrupted.load(Ordering::Relaxed) {
+                not_interrupted += 1;
+            }
+        }
+
+        waiter.join().unwrap();
+        interrupter.join().unwrap();
+        crate::ordering_stress::deactivate();
+        std::eprintln!(
+            "iterations={iterations} both_old={both_old} not_interrupted={not_interrupted}"
+        );
+        assert_eq!(not_interrupted, 0, "a wait ended without being interrupted");
+        assert_eq!(
+            both_old, 0,
+            "the interrupt path saw both old values (each one hangs without the rescue)"
+        );
+    }
+}

litebox/src/lib.rs

@@ -28,6 +28,13 @@ pub mod shim;
 pub mod sync;
 pub mod tls;
 
+// Test-only instrumentation for the wake-path store-buffering stress tests.
+// Compiled under `cfg(test)` for the in-crate mock tests, or the `ordering_stress`
+// feature so a dependent crate can drive the same reproductions over a real
+// platform. Never enable in production builds.
+#[cfg(any(test, feature = "ordering_stress"))]
+pub mod ordering_stress;
+
 // The core [`LiteBox`] object itself, re-exported here publicly, just to keep management of the
 // code cleaner.
 mod litebox;