Resume interrupted futex waits, polls, and selects like Linux

litebox/src/platform/mock.rs

@@ -196,6 +196,13 @@ impl Instant for MockInstant {
             time: self.time.checked_add(duration_millis)?,
         })
     }
+
+    fn saturating_add(&self, duration: core::time::Duration) -> Self {
+        let duration_millis = u64::try_from(duration.as_millis()).unwrap_or(u64::MAX);
+        MockInstant {
+            time: self.time.saturating_add(duration_millis),
+        }
+    }
 }
 
 pub(crate) struct MockSystemTime {

litebox/src/sync/futex.rs

@@ -97,8 +97,8 @@ impl<Platform: RawSyncPrimitivesProvider + RawPointerProvider + TimeProvider>
     /// This function tests once if the futex word matches the expected value,
     /// returning immediately with
     /// [`FutexError::ImmediatelyWokenBecauseValueMismatch`] if it does not.
-    /// Otherwise, it waits until woken by a corresponding until
-    /// [`FutexManager::wake`] is called targeting the same futex word or until
+    /// Otherwise, it waits until woken by a corresponding call to
+    /// [`FutexManager::wake`] targeting the same futex word or until
     /// the wait times out or is interrupted.
     ///
     /// If `bitset` is `Some`, then the waiter is only woken if the wake call's
@@ -123,8 +123,8 @@ impl<Platform: RawSyncPrimitivesProvider + RawPointerProvider + TimeProvider>
             done: AtomicBool::new(false),
         },));
 
-        // Insert into the bucket's list. It will be removed when woken or the
-        // entry goes out of scope.
+        // Insert into the bucket's list. It will be removed when woken, when the
+        // wait fails, or when the entry goes out of scope.
         entry.as_mut().insert(bucket);
 
         // Check the value once. Do this only after inserting into the list so
@@ -135,8 +135,16 @@ impl<Platform: RawSyncPrimitivesProvider + RawPointerProvider + TimeProvider>
         // Only return when woken--don't reevaluate the futex word. This
         // ensures that the rate control mechanisms provided by the futex
         // interface are effective.
-        cx.wait_until(|| entry.get().done.load(Ordering::Acquire))
-            .map_err(FutexError::WaitError)
+        let result = cx.wait_until(|| entry.get().done.load(Ordering::Acquire));
+        if result.is_err() {
+            // A concurrent wake may occur while we're handling the timeout or interruption. In that case,
+            // the entry will indicate whether it was woken, and we should respect that.
+            entry.as_mut().remove();
+            if entry.get().done.load(Ordering::Acquire) {
+                return Ok(());
+            }
+        }
+        result.map_err(FutexError::WaitError)
     }
 
     /// Wakes waiters on the given futex word.
@@ -382,4 +390,46 @@ mod tests {
 
         assert!((1..=3).contains(&woken));
     }
+
+    #[test]
+    fn test_futex_wait_woken_as_interrupted_succeeds() {
+        use crate::event::wait::CheckForInterrupt;
+
+        /// Wakes the futex word, then interrupts the wait.
+        struct WakeThenInterrupt<'a> {
+            futex_manager: &'a FutexManager<MockPlatform>,
+            futex_addr: <MockPlatform as crate::platform::RawPointerProvider>::RawMutPointer<u32>,
+        }
+        impl CheckForInterrupt for WakeThenInterrupt<'_> {
+            fn check_for_interrupt(&self) -> bool {
+                let woken = self
+                    .futex_manager
+                    .wake(self.futex_addr, NonZeroU32::new(1).unwrap(), None)
+                    .unwrap();
+                assert_eq!(woken, 1);
+                true
+            }
+        }
+
+        let platform = MockPlatform::new();
+        let _litebox = LiteBox::new(platform);
+        let futex_manager = FutexManager::new();
+        let futex_word = AtomicU32::new(0);
+        let futex_addr =
+            <MockPlatform as crate::platform::RawPointerProvider>::RawMutPointer::from_usize(
+                futex_word.as_ptr() as usize,
+            );
+        let check = WakeThenInterrupt {
+            futex_manager: &futex_manager,
+            futex_addr,
+        };
+        let wait_state = WaitState::new(platform);
+        let result = futex_manager.wait(
+            &wait_state.context().with_check_for_interrupt(&check),
+            futex_addr,
+            0,
+            None,
+        );
+        assert!(result.is_ok(), "{result:?}");
+    }
 }

litebox/src/utilities/loan_list.rs

@@ -111,7 +111,6 @@ impl<'a, Platform: RawSyncPrimitivesProvider, T> LoanListEntry<'a, Platform, T>
     /// completes and the entry is fully returned.
     ///
     /// If the entry is not currently inserted, this method does nothing.
-    #[cfg_attr(not(test), expect(dead_code))]
     pub fn remove(self: Pin<&mut Self>) {
         if let Some(list) = self.list {
             list.remove_node(&self.node);

litebox_broker_platform_linux_userland/Cargo.toml

@@ -9,7 +9,7 @@ litebox_broker_protocol = { path = "../litebox_broker_protocol", version = "0.1.
 litebox_platform = { path = "../litebox_platform", version = "0.1.0" }
 # getpgrp, whose result rustix cannot represent in a foreign PID namespace.
 libc = { version = "0.2.177", default-features = false }
-rustix = { version = "1.1.2", default-features = false, features = ["event", "fs", "net", "std", "termios", "thread"] }
+rustix = { version = "1.1.2", default-features = false, features = ["event", "fs", "net", "std", "termios", "thread", "time"] }
 
 [features]
 lock_tracing = ["litebox_platform/lock_tracing"]

litebox_broker_platform_linux_userland/src/time.rs

@@ -18,7 +18,9 @@ impl TimeProvider for LinuxTimeProvider {
     type SystemTime = LinuxSystemTime;
 
     fn now(&self) -> Self::Instant {
-        LinuxInstant(std::time::Instant::now())
+        let now = rustix::time::clock_gettime(rustix::time::ClockId::Monotonic);
+        // `CLOCK_MONOTONIC` is never negative.
+        LinuxInstant(Duration::try_from(now).unwrap_or(Duration::ZERO))
     }
 
     fn current_time(&self) -> Self::SystemTime {
@@ -28,16 +30,20 @@ impl TimeProvider for LinuxTimeProvider {
 
 /// Monotonic clock value for a Linux-userland broker.
 #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
-pub struct LinuxInstant(std::time::Instant);
+pub struct LinuxInstant(Duration);
 
 impl InstantTrait for LinuxInstant {
     fn checked_duration_since(&self, earlier: &Self) -> Option<Duration> {
-        self.0.checked_duration_since(earlier.0)
+        self.0.checked_sub(earlier.0)
     }
 
     fn checked_add(&self, duration: Duration) -> Option<Self> {
         self.0.checked_add(duration).map(Self)
     }
+
+    fn saturating_add(&self, duration: Duration) -> Self {
+        Self(self.0.saturating_add(duration))
+    }
 }
 
 /// Wall-clock value for a Linux-userland broker.

litebox_common_linux/src/lib.rs

@@ -2340,7 +2340,7 @@ pub enum FutexOperation {
 }
 
 bitflags::bitflags! {
-    #[derive(Debug)]
+    #[derive(Clone, Copy, Debug)]
     pub struct FutexFlags: i32 {
         const PRIVATE = 0x80; // FUTEX_PRIVATE_FLAG
         const CLOCK_REALTIME = 0x100; // FUTEX_CLOCK_REALTIME
@@ -2979,6 +2979,11 @@ pub enum SyscallRequest {
         size: i32,
         flags: EpollCreateFlags,
     },
+    Poll {
+        fds: UserPtrMut<Pollfd>,
+        nfds: usize,
+        timeout: TimeParam,
+    },
     Ppoll {
         fds: UserPtrMut<Pollfd>,
         nfds: usize,
@@ -3578,9 +3583,7 @@ impl SyscallRequest {
                 sys_req!(Ppoll { fds:*, nfds, timeout: { =*> TimeParam::timespec_old }, sigmask:*, sigsetsize })
             }
             #[cfg(target_arch = "x86_64")]
-            Sysno::poll => {
-                sys_req!(Ppoll { fds:*, nfds, timeout: { => TimeParam::Milliseconds }, sigmask: { None }, sigsetsize: { 0 } })
-            }
+            Sysno::poll => sys_req!(Poll { fds:*, nfds, timeout: { => TimeParam::Milliseconds } }),
             #[cfg(target_arch = "x86_64")]
             Sysno::select => {
                 sys_req!(Pselect {

litebox_platform/src/time.rs

@@ -32,6 +32,11 @@ pub trait Instant: Copy + Clone + PartialEq + Eq + PartialOrd + Ord + Send + Syn
 
     /// Returns a new instant advanced by `duration`, or `None` on overflow.
     fn checked_add(&self, duration: Duration) -> Option<Self>;
+
+    /// Returns a new instant advanced by `duration`, or the latest instant the platform can
+    /// represent on overflow.
+    #[must_use]
+    fn saturating_add(&self, duration: Duration) -> Self;
 }
 
 /// A measurement of the system wall clock.

litebox_platform_linux_kernel/src/lib.rs

@@ -259,6 +259,13 @@ impl InstantTrait for Instant {
             duration_micros.checked_mul(CPU_MHZ.load(core::sync::atomic::Ordering::Relaxed))?,
         )?))
     }
+
+    fn saturating_add(&self, duration: core::time::Duration) -> Self {
+        let duration_micros = u64::try_from(duration.as_micros()).unwrap_or(u64::MAX);
+        Instant(self.0.saturating_add(
+            duration_micros.saturating_mul(CPU_MHZ.load(core::sync::atomic::Ordering::Relaxed)),
+        ))
+    }
 }
 
 impl Instant {

litebox_platform_linux_userland/src/lib.rs

@@ -1055,6 +1055,11 @@ impl InstantTrait for Instant {
             inner: self.inner.checked_add(duration)?,
         })
     }
+    fn saturating_add(&self, duration: Duration) -> Self {
+        Self {
+            inner: self.inner.saturating_add(duration),
+        }
+    }
 }
 
 pub struct SystemTime {

litebox_platform_lvbs/src/lib.rs

@@ -935,6 +935,14 @@ impl InstantTrait for Instant {
         let ticks = nanos / crate::arch::timer::REF_COUNTER_TICK_NANOS;
         Some(Instant(self.0.checked_add(ticks)?))
     }
+
+    fn saturating_add(&self, duration: core::time::Duration) -> Self {
+        let ticks = duration.as_nanos() / u128::from(crate::arch::timer::REF_COUNTER_TICK_NANOS);
+        Instant(
+            self.0
+                .saturating_add(u64::try_from(ticks).unwrap_or(u64::MAX)),
+        )
+    }
 }
 
 impl Instant {

litebox_platform_macos_userland/src/lib.rs

@@ -298,6 +298,11 @@ impl InstantTrait for Instant {
             inner: self.inner.checked_add(duration)?,
         })
     }
+    fn saturating_add(&self, duration: Duration) -> Self {
+        Self {
+            inner: self.inner.saturating_add(duration),
+        }
+    }
 }
 
 pub struct SystemTime {

litebox_platform_vm_kernel/src/lib.rs

@@ -900,6 +900,11 @@ impl InstantTrait for Instant {
         let nanos: u64 = duration.as_nanos().try_into().ok()?;
         Some(Instant(self.0.checked_add(nanos)?))
     }
+
+    fn saturating_add(&self, duration: core::time::Duration) -> Self {
+        let nanos = u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX);
+        Instant(self.0.saturating_add(nanos))
+    }
 }
 
 impl SystemTimeTrait for SystemTime {

litebox_platform_vm_userland/src/lib.rs

@@ -417,6 +417,11 @@ impl InstantTrait for Instant {
         let nanos: u64 = duration.as_nanos().try_into().ok()?;
         Some(Instant(self.0.checked_add(nanos)?))
     }
+
+    fn saturating_add(&self, duration: core::time::Duration) -> Self {
+        let nanos = u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX);
+        Instant(self.0.saturating_add(nanos))
+    }
 }
 
 impl SystemTimeTrait for SystemTime {

litebox_platform_windows_userland/src/lib.rs

@@ -1869,6 +1869,11 @@ impl InstantTrait for Instant {
         let new = self.0.checked_add(duration_100ns)?;
         Some(Instant(new))
     }
+
+    fn saturating_add(&self, duration: core::time::Duration) -> Self {
+        let duration_100ns = u64::try_from(duration.as_nanos() / 100).unwrap_or(u64::MAX);
+        Instant(self.0.saturating_add(duration_100ns))
+    }
 }
 
 pub struct SystemTime {

litebox_runner_linux_userland/tests/helpers.h

@@ -195,6 +195,32 @@ static inline void expect_poll_lacks(int fd, short events_mask, short forbidden,
     }
 }
 
+// A signal handler that does nothing, so that the signal interrupts what it comes during.
+static inline void handle_signal(int signal) { (void)signal; }
+
+// Sets the action of `signal` to run `handler` with `flags`.
+static inline void set_handler(int signal, void (*handler)(int), int flags) {
+    struct sigaction action;
+    memset(&action, 0, sizeof(action));
+    action.sa_handler = handler;
+    action.sa_flags = flags;
+    sigemptyset(&action.sa_mask);
+    TEST_ASSERT(sigaction(signal, &action, NULL) == 0, "sigaction");
+}
+
+// Sends `signal` to the process every millisecond until `stop(arg)` holds, for at most `max_ms`.
+// Returns whether `stop(arg)` held.
+static inline int signal_until(int signal, int (*stop)(void *), void *arg, long max_ms) {
+    for (long ms = 0; ms < max_ms; ms++) {
+        if (stop(arg)) {
+            return 1;
+        }
+        TEST_ASSERT(kill(getpid(), signal) == 0, "kill");
+        sleep_ms(1);
+    }
+    return stop(arg);
+}
+
 #if defined(__x86_64__) || defined(__i386__)
 #define SPIN_HINT() __asm__ __volatile__("pause")
 #elif defined(__aarch64__)