Allow parking threads and wake correctly

src/lib.rs

@@ -30,6 +30,7 @@ pub mod uapi;
 pub use hal_cortex_m::*;
 // Add new hals here. No cfg needed.
 
+pub use crate::sync::waiter::ParkedWaiter;
 pub use hal::Machinelike;
 pub use hal_api::error::*;
 pub use proc_macros::app_main;

src/sched.rs

@@ -296,10 +296,15 @@ impl<const N: usize> Scheduler<N> {
         Ok(())
     }
 
-    /// `kick` lookup by raw `UId::as_usize()`. Synthetic `tid` is a placeholder.
+    /// Lookup `UId` from a raw `UId::as_usize()`. Synthetic `tid` is a
+    /// placeholder; `UId` equality is by raw uid only.
+    fn synthetic_uid(uid: usize) -> thread::UId {
+        thread::UId::new(uid, thread::Id::new(0, crate::sched::task::UId::new(0)))
+    }
+
+    /// `kick` lookup by raw `UId::as_usize()`.
     pub fn kick_by_uid(&mut self, uid: usize) -> Result<()> {
-        let lookup_uid = thread::UId::new(uid, thread::Id::new(0, crate::sched::task::UId::new(0)));
-        self.kick(lookup_uid)
+        self.kick(Self::synthetic_uid(uid))
     }
 
     pub fn current_uid(&self) -> Option<usize> {
@@ -315,6 +320,10 @@ impl<const N: usize> Scheduler<N> {
 
         if let Some(thread) = self.threads.get_mut(uid) {
             thread.resume();
+            if res.is_err() {
+                // Not sleeping yet: latch the wake for the next park.
+                thread.set_pending_wake();
+            }
         } else {
             return Err(kerr!(EINVAL)); // Thread does not exist.
         }
@@ -325,6 +334,13 @@ impl<const N: usize> Scheduler<N> {
         Ok(())
     }
 
+    /// Take-and-clear the latched wake for `uid`.
+    pub fn take_pending_wake_by_uid(&mut self, uid: usize) -> bool {
+        self.threads
+            .get_mut(Self::synthetic_uid(uid))
+            .map_or(false, |t| t.take_pending_wake())
+    }
+
     /// This will make the thread not runnable, but it will not remove it from other lists.
     /// If the thread is currently running, reschedule will be triggered.
     ///

src/sched/thread.rs

@@ -289,6 +289,9 @@ pub struct Thread {
 
     waiter: Option<Waiter>,
 
+    /// Wake latched while not sleeping; consumed by the next park.
+    pending_wake: bool,
+
     #[list(tag = RRList, idx = UId)]
     rr_links: list::Links<RRList, UId>,
 
@@ -311,6 +314,7 @@ impl Thread {
             uid,
             rt_server: server,
             waiter: None,
+            pending_wake: false,
             rr_links: list::Links::new(),
             thread_links: list::Links::new(),
         }
@@ -328,6 +332,15 @@ impl Thread {
         self.waiter.is_some()
     }
 
+    pub fn set_pending_wake(&mut self) {
+        self.pending_wake = true;
+    }
+
+    /// Returns and clears the latched wake.
+    pub fn take_pending_wake(&mut self) -> bool {
+        core::mem::replace(&mut self.pending_wake, false)
+    }
+
     pub fn save_ctx(&mut self, ctx: *mut c_void) -> Result<()> {
         let sp = self.state.stack.create_sp(ctx)?;
         self.state.stack.set_sp(sp);

src/sync/waiter.rs

@@ -82,4 +82,92 @@ impl ParkedWaiter {
             crate::sched::kick_thread(uid as u32);
         }
     }
+
+    /// Userspace wait: re-`probe`s, parking via the pending-wake
+    /// syscall between tries, until `Some` or `timeout_ticks` elapse.
+    /// `u64::MAX` waits forever. Single consumer: returns `probe()`
+    /// immediately if another waiter holds the slot (`EBUSY`).
+    /// Userspace-only; kernel code uses [`park`](Self::park).
+    pub fn wait_while<T>(
+        &self,
+        uid: usize,
+        timeout_ticks: u64,
+        mut probe: impl FnMut() -> Option<T>,
+    ) -> Option<T> {
+        if let Some(v) = probe() {
+            return Some(v);
+        }
+        if self.arm(uid).is_err() {
+            return probe();
+        }
+        let forever = timeout_ticks == u64::MAX;
+        let deadline = crate::uapi::time::tick().saturating_add(timeout_ticks);
+        let result = loop {
+            if let Some(v) = probe() {
+                break Some(v);
+            }
+            let now = crate::uapi::time::tick();
+            if !forever && now >= deadline {
+                break None;
+            }
+            let remaining = if forever { u64::MAX } else { deadline - now };
+            let _ = crate::uapi::sched::park_pending(remaining);
+        };
+        self.disarm();
+        result
+    }
+}
+
+#[cfg(test)]
+impl ParkedWaiter {
+    pub(crate) fn armed_uid(&self) -> usize {
+        self.uid.load(Ordering::Acquire)
+    }
+}
+
+// Scheduler-free contract tests. The arm↔kick / park_pending
+// interleaving is verified on QEMU/HW.
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn arm_is_single_consumer_without_overwrite() {
+        let w = ParkedWaiter::new();
+        assert!(w.arm(1).is_ok());
+        assert!(w.arm(2).is_err());
+        assert_eq!(w.armed_uid(), 1);
+        w.disarm();
+        assert!(w.arm(2).is_ok());
+    }
+
+    #[test]
+    fn arm_rejects_idle_uid() {
+        let w = ParkedWaiter::new();
+        assert!(matches!(
+            w.arm(UNARMED).unwrap_err().kind,
+            crate::error::PosixError::EINVAL
+        ));
+    }
+
+    #[test]
+    fn wake_unarmed_is_noop() {
+        ParkedWaiter::new().wake();
+    }
+
+    #[test]
+    fn wait_while_ready_returns_without_arming() {
+        let w = ParkedWaiter::new();
+        assert_eq!(w.wait_while(5, 0, || Some(42)), Some(42));
+        assert_eq!(w.armed_uid(), UNARMED);
+    }
+
+    #[test]
+    fn wait_while_degrades_when_slot_busy() {
+        let w = ParkedWaiter::new();
+        w.arm(1).unwrap();
+        assert_eq!(w.wait_while(2, 0, || -> Option<()> { None }), None);
+        // First consumer's uid is left intact.
+        assert_eq!(w.armed_uid(), 1);
+    }
 }

src/syscalls/sched.rs

@@ -87,6 +87,29 @@ fn kick_thread(_uid: usize) -> c_int {
     0
 }
 
+/// Sleep until `duration` ticks elapse, unless a wake was latched for
+/// this thread meanwhile (then return at once). `u64::MAX` = forever.
+#[syscall_handler(num = 7)]
+fn park_pending(duration_hi: u32, duration_lo: u32) -> c_int {
+    let duration = ((duration_hi as u64) << 32) | (duration_lo as u64);
+    sched::with(|sched| {
+        let Some(uid) = sched.current_uid() else {
+            bug!("no current thread set.");
+        };
+        if sched.take_pending_wake_by_uid(uid) {
+            return 0;
+        }
+        let now = time::tick();
+        if sched
+            .sleep_until(None, now.saturating_add(duration), now)
+            .is_err()
+        {
+            bug!("no current thread set.");
+        }
+        0
+    })
+}
+
 #[syscall_handler(num = 6)]
 fn current_id() -> c_int {
     sched::with(|sched| match sched.current_uid() {

src/uapi/sched.rs

@@ -11,6 +11,12 @@ pub fn sleep_for(_duration: u64) -> isize {
     hal::asm::syscall!(2, (_duration >> 32) as u32, _duration as u32)
 }
 
+/// Sleep until `timeout_ticks` elapse, unless a wake was latched for
+/// this thread meanwhile. `u64::MAX` waits forever.
+pub fn park_pending(timeout_ticks: u64) -> isize {
+    hal::asm::syscall!(7, (timeout_ticks >> 32) as u32, timeout_ticks as u32)
+}
+
 pub fn yield_thread() -> isize {
     let _until = u64::MAX;
     hal::asm::syscall!(1, (_until >> 32) as u32, _until as u32)