diff --git a/src/channel.rs b/src/channel.rs index c461a13..43b07be 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -39,27 +39,22 @@ pub fn channel() -> (Sender, Receiver) { parked_receiver: None, senders: 1, receiver_alive: true, - cur_wait: None, - next_wait_seq: 0, - timed_out: false, })); (Sender { inner: inner.clone() }, Receiver { inner }) } struct Inner { queue: VecDeque, - parked_receiver: Option, + /// The parked receiver's `(pid, park-epoch)`. The epoch is the slot + /// word's runtime-wide wait identity (see slot_state.rs): wakers call + /// `unpark_at(pid, epoch)`, so an entry left over from an already-woken + /// wait — a `select` loser arm, a satisfied `recv_timeout`'s timer — is + /// inert: the wake fails the word's epoch CAS and no-ops. This replaces + /// the old per-channel `cur_wait`/`next_wait_seq`/`timed_out` trio: wait + /// identity now exists exactly once, in the slot word. + parked_receiver: Option<(Pid, u32)>, senders: usize, receiver_alive: bool, - /// The seq of the receiver's current *bounded* wait (`recv_timeout`), or - /// `None` outside one. Lets `on_timeout` tell "this wait" apart from "a - /// later wait on the same channel" — same convention as `Mutex`. - cur_wait: Option, - /// Monotonic source for `cur_wait` seqs. - next_wait_seq: u64, - /// Set by `on_timeout` when it cancels the current bounded wait; consumed - /// by the woken receiver to distinguish a timeout from a message wake. - timed_out: bool, } pub struct Sender { @@ -129,8 +124,8 @@ impl Drop for Sender { None } }; - if let Some(pid) = unpark { - crate::scheduler::unpark(pid); + if let Some((pid, epoch)) = unpark { + crate::scheduler::unpark_at(pid, epoch); } } } @@ -151,9 +146,9 @@ impl Sender { g.queue.push_back(value); g.parked_receiver.take() }; - if let Some(pid) = unpark { + if let Some((pid, epoch)) = unpark { crate::te!(crate::trace::Event::Send { sender: crate::actor::current_pid().unwrap_or(crate::pid::Pid::new(u32::MAX, u32::MAX)), receiver: Some(pid) }); - crate::scheduler::unpark(pid); + crate::scheduler::unpark_at(pid, epoch); } else { crate::te!(crate::trace::Event::Send { sender: crate::actor::current_pid().unwrap_or(crate::pid::Pid::new(u32::MAX, u32::MAX)), receiver: None }); } @@ -178,7 +173,10 @@ impl Receiver { g.parked_receiver.is_none(), "channel has more than one receiver" ); - g.parked_receiver = Some(me); + // begin_wait is lock-free — legal under the Channel lock; + // registering in the same critical section makes the epoch + // atomic with the senders' view of the registration. + g.parked_receiver = Some((me, crate::scheduler::begin_wait())); crate::te!(crate::trace::Event::RecvPark(me)); } // Release the lock before parking — the unparker will need it. @@ -192,14 +190,21 @@ impl Receiver { /// `timeout` has elapsed, returning [`RecvTimeoutError::Timeout`]. /// /// Built on the same timer machinery as `Mutex::lock_timeout`: the wait - /// registers a `WaitTimeout` entry tagged with a per-wait seq; on expiry - /// the channel (as the [`TimerTarget`](crate::timer::TimerTarget)) checks - /// whether *this* wait is still parked and, only then, cancels it. A wake - /// that races the deadline resolves message-first: if a message is - /// available when the receiver runs, it is delivered even if the timer - /// had already fired. A satisfied or abandoned wait leaves its timer - /// entry to expire as a no-op (seq mismatch), per the no-cancellation - /// convention in `timer.rs`. + /// registers a `WaitTimeout` entry stamped with the wait's park-epoch; + /// on expiry the channel (as the + /// [`TimerTarget`](crate::timer::TimerTarget)) checks whether *this* + /// wait is still parked and, only then, cancels it. A wake that races + /// the deadline resolves message-first: if a message is available when + /// the receiver runs, it is delivered even if the timer had already + /// fired. A satisfied or abandoned wait leaves its timer entry to expire + /// as a no-op (registration gone; epoch consumed), per the + /// no-cancellation convention in `timer.rs`. + /// + /// The wake is classified from state alone — wakes are precise (the only + /// stamped wakers of this wait are a send, the last-sender drop, and the + /// timer; a stop wake unwinds out of `park_current` and never reaches + /// the classification), so: message queued → `Ok`; `senders == 0` → + /// `Disconnected`; neither → it was the timer → `Timeout`. /// /// `Duration::ZERO` is a valid timeout: it parks until the immediately- /// due timer is drained, then reports `Timeout` unless a message was @@ -212,7 +217,7 @@ impl Receiver { .expect("recv_timeout() called outside an actor"); // Fast path + wait registration, one critical section. - let seq; + let epoch; { let mut g = self.inner.lock(); if let Some(v) = g.queue.pop_front() { @@ -225,11 +230,8 @@ impl Receiver { g.parked_receiver.is_none(), "channel has more than one receiver" ); - seq = g.next_wait_seq; - g.next_wait_seq = g.next_wait_seq.wrapping_add(1); - g.cur_wait = Some(seq); - g.timed_out = false; - g.parked_receiver = Some(me); + epoch = crate::scheduler::begin_wait(); + g.parked_receiver = Some((me, epoch)); crate::te!(crate::trace::Event::RecvPark(me)); } @@ -239,34 +241,18 @@ impl Receiver { // protocol makes the park below return immediately in that case. let deadline = crate::timer::deadline_from_now(timeout); let target: std::sync::Arc = self.inner.clone(); - crate::scheduler::insert_wait_timer(deadline, me, target, seq); + crate::scheduler::insert_wait_timer(deadline, me, target, epoch); - loop { - crate::scheduler::park_current(); - crate::te!(crate::trace::Event::RecvWake(crate::actor::current_pid().unwrap())); - let mut g = self.inner.lock(); - if let Some(v) = g.queue.pop_front() { - g.cur_wait = None; - g.timed_out = false; - return Ok(v); - } - if g.senders == 0 { - g.cur_wait = None; - g.timed_out = false; - return Err(RecvTimeoutError::Disconnected); - } - if g.timed_out { - g.cur_wait = None; - g.timed_out = false; - return Err(RecvTimeoutError::Timeout); - } - // Defensive: no current wake source reaches here (a send leaves a - // message, closure zeroes senders, the timer sets timed_out, and - // a stop unwinds out of park_current) — but if one ever does, - // re-register and keep waiting; cur_wait == Some(seq) keeps the - // already-armed timer valid for this wait. - g.parked_receiver = Some(me); + crate::scheduler::park_current(); + crate::te!(crate::trace::Event::RecvWake(crate::actor::current_pid().unwrap())); + let mut g = self.inner.lock(); + if let Some(v) = g.queue.pop_front() { + return Ok(v); } + if g.senders == 0 { + return Err(RecvTimeoutError::Disconnected); + } + Err(RecvTimeoutError::Timeout) } /// Selective receive: remove and return the first queued message for which @@ -300,7 +286,7 @@ impl Receiver { g.parked_receiver.is_none(), "channel has more than one receiver" ); - g.parked_receiver = Some(me); + g.parked_receiver = Some((me, crate::scheduler::begin_wait())); crate::te!(crate::trace::Event::RecvPark(me)); } // Release the lock before parking — the unparker will need it. @@ -346,16 +332,18 @@ impl Receiver { // --------------------------------------------------------------------------- impl crate::timer::TimerTarget for RawMutex> { - fn on_timeout(&self, pid: Pid, wait_seq: u64) { - // Cancel the wait only if THIS wait (seq match) is still parked. If a - // sender already took `parked_receiver`, the receiver is waking with - // a message — message wins, the timer no-ops. If `cur_wait` moved on, - // this entry is stale (the wait was satisfied or abandoned) — no-op. + fn on_timeout(&self, pid: Pid, epoch: u32) { + // Cancel the wait only if THIS wait (epoch match) is still + // registered. If a sender already took `parked_receiver`, the + // receiver is waking with a message — message wins, the timer + // no-ops. If a later wait by the same receiver is registered, the + // epoch mismatches — stale entry, no-op. (The unpark_at would fail + // its word CAS in either case anyway; checking under the lock keeps + // the registration bookkeeping exact.) let unpark = { let mut g = self.lock(); - if g.cur_wait == Some(wait_seq) && g.parked_receiver == Some(pid) { + if g.parked_receiver == Some((pid, epoch)) { g.parked_receiver = None; - g.timed_out = true; true } else { false @@ -364,7 +352,7 @@ impl crate::timer::TimerTarget for RawMutex> { // Unpark outside the channel lock — it may take the run-queue lock; // legal under a Channel lock, but pointless to nest. if unpark { - crate::scheduler::unpark(pid); + crate::scheduler::unpark_at(pid, epoch); } } } diff --git a/src/io.rs b/src/io.rs index acf70d5..f7e50ef 100644 --- a/src/io.rs +++ b/src/io.rs @@ -84,6 +84,9 @@ use std::thread::JoinHandle as OsJoinHandle; pub type IoResult = Result, Box>; struct Request { + /// The submitter's park-epoch — carried through to the `Blocking` + /// completion so the wake is epoch-matched. + epoch: u32, pid: Pid, /// The work to perform. Returns the wire-form result directly. work: Box IoResult + Send>, @@ -93,7 +96,7 @@ struct Request { pub enum Completion { /// A `block_on_io` closure has finished (Ok = return value, Err = panic /// payload). - Blocking { pid: Pid, result: IoResult }, + Blocking { pid: Pid, epoch: u32, result: IoResult }, /// An fd registered via `wait_readable`/`wait_writable` is ready. The /// scheduler looks up the parked pid in `waiters`, unparks it, and /// removes the entry. `pid` isn't in this variant because the epoll @@ -131,7 +134,7 @@ pub struct IoThread { /// One parked actor per registered fd. Populated by `wait_readable` / /// `wait_writable` and drained by the scheduler when a `FdReady` /// completion is processed. - pub waiters: HashMap, + pub waiters: HashMap, // ----- Threads ----- @@ -231,12 +234,12 @@ impl IoThread { } /// Hand a request to the pool. Increments `outstanding`. - pub fn submit(&mut self, pid: Pid, work: Box IoResult + Send>) { + pub fn submit(&mut self, pid: Pid, epoch: u32, work: Box IoResult + Send>) { self.outstanding += 1; // Send can only fail if the pool has hung up, which only happens // on shutdown. submit during shutdown is a bug. self.tx - .send(Request { pid, work }) + .send(Request { pid, epoch, work }) .expect("io pool hung up unexpectedly"); } @@ -265,6 +268,7 @@ impl IoThread { &mut self, fd: RawFd, pid: Pid, + epoch: u32, readable: bool, writable: bool, ) -> io::Result<()> { @@ -303,7 +307,7 @@ impl IoThread { if r < 0 { return Err(io::Error::last_os_error()); } - self.waiters.insert(fd, pid); + self.waiters.insert(fd, (pid, epoch)); Ok(()) } @@ -369,7 +373,7 @@ fn pool_loop( completions: Arc>>, wake_write: RawFd, ) { - while let Ok(Request { pid, work }) = rx.recv() { + while let Ok(Request { pid, epoch, work }) = rx.recv() { let result: IoResult = match panic::catch_unwind(panic::AssertUnwindSafe(work)) { Ok(r) => r, Err(payload) => Err(payload), @@ -377,7 +381,7 @@ fn pool_loop( completions .lock() .unwrap() - .push_back(Completion::Blocking { pid, result }); + .push_back(Completion::Blocking { pid, epoch, result }); wake_scheduler(wake_write); } } diff --git a/src/mutex.rs b/src/mutex.rs index 5bfc239..e5784fa 100644 --- a/src/mutex.rs +++ b/src/mutex.rs @@ -33,13 +33,15 @@ impl std::error::Error for LockTimeout {} struct Wait { pid: Pid, - seq: u64, + /// The wait's park-epoch (slot-word wait identity, see slot_state.rs). + /// Grants and timeouts wake via `unpark_at(pid, epoch)`; a stale entry + /// can neither be granted by mistake nor wake the wrong wait. + epoch: u32, } struct MutexState { holder: Option, waiters: VecDeque, - next_seq: u64, default_timeout: Duration, } @@ -53,7 +55,6 @@ impl MutexCore { state: StdMutex::new(MutexState { holder: None, waiters: VecDeque::new(), - next_seq: 0, default_timeout, }), } @@ -61,17 +62,17 @@ impl MutexCore { } impl TimerTarget for MutexCore { - fn on_timeout(&self, pid: Pid, wait_seq: u64) { + fn on_timeout(&self, pid: Pid, epoch: u32) { let unpark = { let mut st = self.state.lock().unwrap(); - // Remove from waiters only if still there with matching seq. + // Remove from waiters only if still there with matching epoch. // If the lock was already granted (holder == Some(pid)), the // timer fired after the grant — treat as no-op; the actor // will see `is_holder == true` and return Ok. if st.holder == Some(pid) { return; } - let pos = st.waiters.iter().position(|w| w.pid == pid && w.seq == wait_seq); + let pos = st.waiters.iter().position(|w| w.pid == pid && w.epoch == epoch); if pos.is_some() { st.waiters.remove(pos.unwrap()); true @@ -80,7 +81,7 @@ impl TimerTarget for MutexCore { } }; if unpark { - scheduler::unpark(pid); + scheduler::unpark_at(pid, epoch); } } } @@ -133,20 +134,25 @@ impl Mutex { // Slow path: register as a waiter, set timeout, park. let _np = scheduler::NoPreempt::enter(); - let seq = { + let epoch = { let mut st = self.core.state.lock().unwrap(); - let seq = st.next_seq; - st.next_seq = st.next_seq.wrapping_add(1); - st.waiters.push_back(Wait { pid: me, seq }); - seq + // begin_wait is lock-free — legal under the state lock; this + // makes the epoch atomic with the registration's visibility to + // grants and timeouts. + let epoch = scheduler::begin_wait(); + st.waiters.push_back(Wait { pid: me, epoch }); + epoch }; let target: Arc = self.core.clone(); let deadline = timer::deadline_from_now(timeout); - scheduler::insert_wait_timer(deadline, me, target, seq); + scheduler::insert_wait_timer(deadline, me, target, epoch); scheduler::park_current(); - // Resumed. Are we the holder? + // Resumed — precisely: only our grant or our timer can wake this + // wait (both epoch-stamped; a stop wake unwinds out of + // park_current). The one-shot interpretation below is therefore + // exhaustive. Are we the holder? let is_holder = self.core.state.lock().unwrap().holder == Some(me); if is_holder { let value = self.value.lock().unwrap().take() @@ -228,12 +234,12 @@ impl Drop for MutexGuard<'_, T> { let v = self.value.take().expect("MutexGuard: double drop"); *self.mutex.value.lock().unwrap() = Some(v); - let next_pid = { + let next = { let mut st = self.mutex.core.state.lock().unwrap(); match st.waiters.pop_front() { Some(w) => { st.holder = Some(w.pid); - Some(w.pid) + Some((w.pid, w.epoch)) } None => { st.holder = None; @@ -241,8 +247,8 @@ impl Drop for MutexGuard<'_, T> { } } }; - if let Some(pid) = next_pid { - scheduler::unpark(pid); + if let Some((pid, epoch)) = next { + scheduler::unpark_at(pid, epoch); } } } diff --git a/src/runtime.rs b/src/runtime.rs index 9494f8c..7807bb8 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -17,7 +17,7 @@ //! } //! //! Slot { -//! word: AtomicU64 ← (generation << 32) | state — THE state machine +//! word: AtomicU64 ← (gen << 32) | (epoch << 8) | state — THE state machine //! sp: AtomicUsize ← saved stack pointer //! stop_ptr:AtomicPtr<...> ← into the actor's Arc //! closure: AtomicPtr<...> ← first-resume closure, swap-to-take @@ -320,7 +320,9 @@ pub(crate) type Closure = Box; /// link / finalize), never on the yield/park/unpark hot path. pub(crate) struct SlotCold { pub(crate) actor: Option, - pub(crate) waiters: Vec, + /// Parked joiners as `(pid, park-epoch)`; finalize wakes each via the + /// epoch-matched unpark. + pub(crate) waiters: Vec<(Pid, u32)>, pub(crate) outcome: Option, pub(crate) supervisor_channel: Option>, /// Watchers registered via `monitor()`, each tagged with its @@ -543,6 +545,14 @@ impl RuntimeInner { self.unpark_inner(pid, Some(epoch)); } + /// Open a new wait for `pid` (the calling actor itself): bump its + /// park-epoch and return it. See slot_state.rs for the rules. + #[must_use] + pub(crate) fn begin_wait(&self, pid: Pid) -> u32 { + let slot = self.slot_at(pid).expect("begin_wait: own slot vanished"); + slot.word.begin_wait(pid.generation()) + } + fn unpark_inner(&self, pid: Pid, want: Option) { if let Some(slot) = self.slot_at(pid) { match slot.word.unpark(pid.generation(), want) { @@ -937,9 +947,9 @@ fn finalize_actor(inner: &Arc, pid: Pid, outcome: Outcome) { } } - // Unpark joiners. - for joiner in waiters { - inner.unpark(joiner); + // Unpark joiners (epoch-matched: each registered under the cold lock). + for (joiner, epoch) in waiters { + inner.unpark_at(joiner, epoch); } // Reclaim if no outstanding handles (re-verified inside). @@ -990,17 +1000,19 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { // actor is between `timers.insert_sleep` and // `park_current`; RunningNotified makes the upcoming park // re-queue), or gone (no-op). - crate::timer::Reason::Sleep => inner.unpark(entry.pid), - crate::timer::Reason::WaitTimeout { target, wait_seq } => { - // The callback may call unpark itself. - target.on_timeout(entry.pid, wait_seq); + crate::timer::Reason::Sleep { epoch } => { + inner.unpark_at(entry.pid, epoch) + } + crate::timer::Reason::WaitTimeout { target, epoch } => { + // The callback may call unpark_at itself. + target.on_timeout(entry.pid, epoch); } } } for completion in completions { match completion { - crate::io::Completion::Blocking { pid, result } => { + crate::io::Completion::Blocking { pid, epoch, result } => { if let Some(io) = inner.io.lock().unwrap().as_mut() { io.outstanding = io.outstanding.saturating_sub(1); } @@ -1018,19 +1030,19 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { // flight; discard the result. } } - inner.unpark(pid); + inner.unpark_at(pid, epoch); } } crate::io::Completion::FdReady { fd, events: _ } => { // Resolve the parked pid under the io lock, then wake // through the protocol. Lock order: io before all. - let parked_pid = inner.io.lock().unwrap().as_mut().and_then(|io| { - let pid = io.waiters.remove(&fd); + let parked = inner.io.lock().unwrap().as_mut().and_then(|io| { + let entry = io.waiters.remove(&fd); io.epoll_deregister(fd); - pid + entry }); - if let Some(pid) = parked_pid { - inner.unpark(pid); + if let Some((pid, epoch)) = parked { + inner.unpark_at(pid, epoch); } } } diff --git a/src/scheduler.rs b/src/scheduler.rs index f0d3bcd..5c4295b 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -96,7 +96,10 @@ impl JoinHandle { Some(cold.outcome.take().expect("Done slot must have outcome")) } crate::slot_state::Status::Live => { - cold.waiters.push(me); + // begin_wait is lock-free, legal under the cold lock; + // registering under it makes the epoch atomic with + // the check-Done-or-register linearization point. + cold.waiters.push((me, begin_wait())); None } } @@ -219,11 +222,30 @@ pub fn park_current() { } pub fn unpark(pid: Pid) { - // The whole protocol lives on the slot's packed word (gen + state in one - // CAS) — see Slot::unpark_action. No runtime lock unless we enqueue. + // The whole protocol lives on the slot's packed word (gen + epoch + + // state in one CAS) — see slot_state.rs. No runtime lock unless we + // enqueue. WILDCARD form: reserved for terminal wakes (request_stop); + // every registration-based waker must use `unpark_at`. let _ = try_with_runtime(|inner| inner.unpark(pid)); } +/// Epoch-matched unpark: wake `pid` only if its current wait is still the +/// one this waker registered for. The form every registration-based waker +/// (channel senders, mutex grants, wait-timers, io completions, joiner +/// wakes) must use — see slot_state.rs for the consuming-wake rules. +pub(crate) fn unpark_at(pid: Pid, epoch: u32) { + let _ = try_with_runtime(|inner| inner.unpark_at(pid, epoch)); +} + +/// Open a new wait for the CURRENT actor: bump its park-epoch and return +/// it. Call once per wait, before registering `(pid, epoch)` with any +/// waker. Lock-free (one CAS on the own slot word), so it is legal under +/// any lock, including a Channel-class lock. +pub(crate) fn begin_wait() -> u32 { + let me = current_pid().expect("begin_wait() called outside an actor"); + with_runtime(|inner| inner.begin_wait(me)) +} + /// Request cooperative cancellation of `pid`. /// /// Sets the actor's stop flag and wakes it so it observes the flag promptly: @@ -281,8 +303,9 @@ impl Drop for NoPreempt { pub fn sleep(duration: std::time::Duration) { let me = current_pid().expect("sleep() called outside an actor"); let _np = NoPreempt::enter(); + let epoch = begin_wait(); let deadline = crate::timer::deadline_from_now(duration); - with_runtime(|inner| inner.timers.lock().unwrap().insert_sleep(deadline, me)); + with_runtime(|inner| inner.timers.lock().unwrap().insert_sleep(deadline, me, epoch)); park_current(); } @@ -290,13 +313,13 @@ pub fn insert_wait_timer( deadline: std::time::Instant, pid: Pid, target: std::sync::Arc, - wait_seq: u64, + epoch: u32, ) { with_runtime(|inner| { inner.timers.lock().unwrap().insert( deadline, pid, - crate::timer::Reason::WaitTimeout { target, wait_seq }, + crate::timer::Reason::WaitTimeout { target, epoch }, ); }); } @@ -317,9 +340,10 @@ where }); { let _np = NoPreempt::enter(); + let epoch = begin_wait(); with_runtime(|inner| { let mut io = inner.io.lock().unwrap(); - io.as_mut().expect("io thread not started").submit(me, work); + io.as_mut().expect("io thread not started").submit(me, epoch, work); }); park_current(); } @@ -351,9 +375,10 @@ pub fn wait_writable(fd: std::os::fd::RawFd) -> std::io::Result<()> { fn wait_fd(fd: std::os::fd::RawFd, readable: bool, writable: bool) -> std::io::Result<()> { let me = current_pid().expect("wait_*() called outside an actor"); let _np = NoPreempt::enter(); + let epoch = begin_wait(); with_runtime(|inner| { let mut io = inner.io.lock().unwrap(); - io.as_mut().expect("io thread not started").epoll_register(fd, me, readable, writable) + io.as_mut().expect("io thread not started").epoll_register(fd, me, epoch, readable, writable) })?; park_current(); Ok(()) diff --git a/src/timer.rs b/src/timer.rs index 6bbfdbd..10cf402 100644 --- a/src/timer.rs +++ b/src/timer.rs @@ -18,9 +18,9 @@ //! No cancellation. When a non-timer wakeup happens (e.g. lock granted //! before timeout), the timer entry is left in the heap. It will be popped //! eventually and the dispatch will observe "actor is no longer parked / -//! wait_seq is stale" and no-op. Cost is ~32 bytes per stale entry plus a -//! few cycles on pop; acceptable given the upper bound is "one entry per -//! parked actor". +//! the wait's epoch was consumed" and no-op. Cost is ~32 bytes per stale +//! entry plus a few cycles on pop; acceptable given the upper bound is "one +//! entry per parked actor". //! //! Stale pids (slot reused since the timer was inserted) are filtered on //! pop by the scheduler — same convention as the run queue. @@ -35,18 +35,21 @@ use std::time::{Duration, Instant}; /// /// Held inside `Entry`, dispatched by the scheduler in `pop_due`. pub enum Reason { - /// `loom::sleep(d)`. Unpark `pid` unconditionally (modulo the usual - /// "still parked?" check the scheduler applies). - Sleep, - /// A bounded wait — currently only `Mutex::lock_timeout`. On expiry the - /// scheduler calls `target.on_timeout(pid, wait_seq)`. The target then - /// decides whether `pid` was actually still waiting, and if so unparks - /// it with whatever error the wait was bounded for. `wait_seq` lets the - /// target tell apart "this wait" from "a later wait by the same actor - /// on the same target". + /// `sleep(d)`. Wake `pid` via the epoch-matched unpark: if anything + /// else (necessarily a terminal wake) already consumed the wait, the + /// entry is stale and no-ops at the CAS. + Sleep { epoch: u32 }, + /// A bounded wait (`Mutex::lock_timeout`, `Receiver::recv_timeout`, + /// `select_timeout`). On expiry the scheduler calls + /// `target.on_timeout(pid, epoch)`. The target then decides whether + /// `pid` was actually still waiting (registration still present under + /// its lock), and if so takes the registration and unparks via + /// `unpark_at`. The epoch is the slot-word park-epoch — the runtime-wide + /// wait identity — so a stale entry is doubly inert: the registration + /// check misses, and even a racing unpark fails the word's epoch CAS. WaitTimeout { target: Arc, - wait_seq: u64, + epoch: u32, }, } @@ -55,7 +58,7 @@ pub enum Reason { /// Implementors: do not touch `SchedulerState` other than via the public /// `unpark` / channel APIs. The scheduler is mid-iteration when this fires. pub trait TimerTarget: Send + Sync { - fn on_timeout(&self, pid: Pid, wait_seq: u64); + fn on_timeout(&self, pid: Pid, epoch: u32); } pub struct Entry { @@ -104,8 +107,8 @@ impl Timers { } /// Insert a `Sleep` timer. Convenience for the common case. - pub fn insert_sleep(&mut self, deadline: Instant, pid: Pid) { - self.insert(deadline, pid, Reason::Sleep); + pub fn insert_sleep(&mut self, deadline: Instant, pid: Pid, epoch: u32) { + self.insert(deadline, pid, Reason::Sleep { epoch }); } /// Insert an arbitrary timer entry. diff --git a/tests/timer.rs b/tests/timer.rs index 2f94dd7..fdeb942 100644 --- a/tests/timer.rs +++ b/tests/timer.rs @@ -124,11 +124,11 @@ use smarm::pid::Pid; use smarm::timer::{Reason, TimerTarget, Timers}; struct RecordingTarget { - calls: Mutex>, + calls: Mutex>, } impl TimerTarget for RecordingTarget { - fn on_timeout(&self, pid: Pid, seq: u64) { - self.calls.lock().unwrap().push((pid, seq)); + fn on_timeout(&self, pid: Pid, epoch: u32) { + self.calls.lock().unwrap().push((pid, epoch)); } } @@ -137,9 +137,9 @@ fn timers_pop_due_returns_entries_in_deadline_order() { let mut t = Timers::new(); let now = Instant::now(); // Insert out of order; pop_due should hand them back sorted by deadline. - t.insert_sleep(now + Duration::from_millis(30), Pid::new(0, 0)); - t.insert_sleep(now + Duration::from_millis(10), Pid::new(1, 0)); - t.insert_sleep(now + Duration::from_millis(20), Pid::new(2, 0)); + t.insert_sleep(now + Duration::from_millis(30), Pid::new(0, 0), 1); + t.insert_sleep(now + Duration::from_millis(10), Pid::new(1, 0), 1); + t.insert_sleep(now + Duration::from_millis(20), Pid::new(2, 0), 1); // Advance past all of them. let due = t.pop_due(now + Duration::from_millis(50)); @@ -152,8 +152,8 @@ fn timers_pop_due_returns_entries_in_deadline_order() { fn timers_only_pop_entries_whose_deadline_has_passed() { let mut t = Timers::new(); let now = Instant::now(); - t.insert_sleep(now + Duration::from_millis(5), Pid::new(0, 0)); - t.insert_sleep(now + Duration::from_millis(100), Pid::new(1, 0)); + t.insert_sleep(now + Duration::from_millis(5), Pid::new(0, 0), 1); + t.insert_sleep(now + Duration::from_millis(100), Pid::new(1, 0), 1); let due = t.pop_due(now + Duration::from_millis(20)); assert_eq!(due.len(), 1); @@ -169,11 +169,11 @@ fn timers_mix_sleep_and_wait_timeout_reasons() { let target = Arc::new(RecordingTarget { calls: Mutex::new(Vec::new()) }); let now = Instant::now(); - t.insert_sleep(now + Duration::from_millis(5), Pid::new(0, 0)); + t.insert_sleep(now + Duration::from_millis(5), Pid::new(0, 0), 1); t.insert( now + Duration::from_millis(10), Pid::new(1, 0), - Reason::WaitTimeout { target: target.clone(), wait_seq: 42 }, + Reason::WaitTimeout { target: target.clone(), epoch: 42 }, ); let due = t.pop_due(now + Duration::from_millis(20)); @@ -181,11 +181,11 @@ fn timers_mix_sleep_and_wait_timeout_reasons() { // Order: Sleep (5ms) first, WaitTimeout (10ms) second. match &due[0].reason { - Reason::Sleep => {} + Reason::Sleep { .. } => {} _ => panic!("first entry should be a Sleep"), } match &due[1].reason { - Reason::WaitTimeout { wait_seq, .. } => assert_eq!(*wait_seq, 42), + Reason::WaitTimeout { epoch, .. } => assert_eq!(*epoch, 42), _ => panic!("second entry should be a WaitTimeout"), } } @@ -197,9 +197,9 @@ fn same_deadline_entries_pop_in_insertion_order() { let mut t = Timers::new(); let now = Instant::now(); let d = now + Duration::from_millis(10); - t.insert_sleep(d, Pid::new(0, 0)); - t.insert_sleep(d, Pid::new(1, 0)); - t.insert_sleep(d, Pid::new(2, 0)); + t.insert_sleep(d, Pid::new(0, 0), 1); + t.insert_sleep(d, Pid::new(1, 0), 1); + t.insert_sleep(d, Pid::new(2, 0), 1); let due = t.pop_due(now + Duration::from_millis(20)); let pids: Vec = due.iter().map(|e| e.pid.index()).collect();