diff --git a/src/io.rs b/src/io.rs index 6fa8d1b..2fd01dc 100644 --- a/src/io.rs +++ b/src/io.rs @@ -13,44 +13,68 @@ //! leaves the actor, no copying through an intermediary thread. Built on //! these are the conveniences `read(fd, &mut buf)` and `write(fd, &buf)`. //! -//! Architecture -//! ============ -//! Per `run()`, two OS threads: -//! - **epoll thread**: owns the epollfd. Loops in `epoll_wait`. On a -//! ready fd, pushes `Completion::FdReady { pid, fd, events }` to the -//! shared completion queue and writes the scheduler-wake pipe. On the -//! shutdown pipe (also registered in epollfd), exits. -//! - **pool thread**: blocks on the request mpsc. Runs the closure -//! inside `catch_unwind`, pushes `Completion::Blocking { pid, result }`, -//! writes the scheduler-wake pipe. +//! Architecture (RFC 018: driver-enqueues) +//! ======================================= +//! Per `run()`, two OS threads, each a *producer* behind the runtime's +//! two-call contract — make the actor runnable (`unpark_at`, whose enqueue +//! tail wakes a parked scheduler), nothing else: //! -//! Both threads share a single `completions: Arc>>` -//! and the same scheduler-wake pipe. +//! - **epoll thread**: owns `epoll_wait` on the epollfd. On a ready fd it +//! removes the parked waiter from the shared `waiters` map and DELs the +//! fd (both under the waiters lock — see below), then unparks the +//! actor directly. On the shutdown pipe (also registered in the +//! epollfd), exits. +//! - **pool thread**: blocks on the request mpsc. Runs the closure inside +//! `catch_unwind`, stashes the result in the actor's slot +//! (`pending_io_result`, under the cold lock, generation-checked), +//! decrements the runtime's `io_outstanding`, and unparks the actor. //! -//! `epoll_ctl` (register/unregister fd interest) is called by the -//! scheduler thread *directly* on the epollfd. That's well-defined per -//! `epoll_ctl(2)`: a thread may be calling `epoll_wait` on the epollfd -//! while another thread calls `epoll_ctl`. Avoids needing a second mpsc -//! and a second wake mechanism. +//! There is no shared completion queue and no wake pipe: each producer +//! routes its own completion, so the whole byte-vs-completion visibility +//! discipline of the drain era — and the stranded-completion hazards it +//! defended against — is unrepresentable. Producers reach the runtime +//! through a `Weak`: upgraded per completion (the path is +//! syscall-bound; the refcount op is noise) and avoiding an Arc cycle +//! through `RuntimeInner::io`. +//! +//! `epoll_ctl` (register fd interest) is called by the scheduler thread +//! directly on the epollfd. That's well-defined per `epoll_ctl(2)`: a +//! thread may be calling `epoll_wait` on the epollfd while another thread +//! calls `epoll_ctl`. //! //! Epoll mode //! ========== //! Level-triggered with EPOLLONESHOT. After a wakeup the kernel //! auto-disarms the fd, so we never get two wakeups for one -//! `wait_readable` call. The scheduler explicitly `EPOLL_CTL_DEL`s the fd -//! on completion to free the slot for re-registration. Net effect: each +//! `wait_readable` call. The epoll thread explicitly `EPOLL_CTL_DEL`s the +//! fd on readiness to free the slot for re-registration. Net effect: each //! `wait_readable(fd)` is one ADD, one wakeup, one DEL — symmetric and //! stateless between calls. //! +//! ## The waiters lock is the ADD/DEL serialization +//! +//! Registration (scheduler thread: check-vacant, defensive DEL, ADD, +//! insert) and readiness consumption (epoll thread: remove, DEL) each run +//! entirely under the `waiters` mutex. This is what makes the +//! oneshot-rearm race unrepresentable: a woken actor re-registering the +//! same fd cannot interleave with the epoll thread's DEL for the *previous* +//! registration — whichever takes the lock second sees a consistent +//! kernel-side state. Lock order: `io` (the runtime's outer mutex, held by +//! scheduler-side callers) → `waiters` → slot/queue leaves via `unpark_at`. +//! The epoll thread takes `waiters` without `io` — it must never take +//! `io`, both for lock-order hygiene and because teardown holds `io` while +//! joining it. +//! //! Fd hygiene //! ========== //! An actor stopped while waiting on an fd unwinds out of `wait_fd`'s park; //! a drop guard there (armed after a successful register, forgotten on a -//! normal wake) removes the `waiters` entry iff it is still that wait's -//! `(pid, epoch)` and only then `EPOLL_CTL_DEL`s the fd — an entry already -//! consumed by a racing `FdReady` means the fd may carry someone else's -//! fresh registration, which must be left alone. `epoll_register` keeps a -//! defensive bare DEL before ADD as belt-and-braces. +//! normal wake) calls [`IoThread::cancel_waiter`], which removes the +//! `waiters` entry iff it is still that wait's `(pid, epoch)` and only then +//! `EPOLL_CTL_DEL`s the fd — an entry already consumed by the epoll thread +//! means the fd may carry someone else's fresh registration, which must be +//! left alone. `epoll_register` keeps a defensive bare DEL before ADD as +//! belt-and-braces. //! //! Buffers used with `read`/`write` should be on fds opened with //! `O_NONBLOCK`. If they aren't, the syscall may block the scheduler @@ -68,13 +92,14 @@ //! they have no equivalent panic-propagation path. use crate::pid::Pid; +use crate::runtime::RuntimeInner; use std::any::Any; -use std::collections::{HashMap, VecDeque}; +use std::collections::HashMap; use std::io; use std::os::fd::RawFd; use std::panic; -use std::sync::mpsc; -use std::sync::{Arc, Mutex}; +use std::sync::atomic::Ordering; +use std::sync::{mpsc, Arc, Mutex, Weak}; use std::thread::JoinHandle as OsJoinHandle; // --------------------------------------------------------------------------- @@ -86,42 +111,29 @@ 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. + /// The submitter's park-epoch — the eventual wake is epoch-matched. epoch: u32, pid: Pid, /// The work to perform. Returns the wire-form result directly. work: Box IoResult + Send>, } -/// Completion message from either IO thread back to the scheduler. -pub enum Completion { - /// A `block_on_io` closure has finished (Ok = return value, Err = panic - /// payload). - 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 - /// thread doesn't have access to the `waiters` map; the scheduler - /// thread owns that. - FdReady { fd: RawFd, events: u32 }, -} +/// The parked-waiter map, shared between scheduler-side registration and +/// the epoll thread's readiness consumption. See the module docs on why +/// this single lock is the ADD/DEL serialization. +type Waiters = Arc>>; // --------------------------------------------------------------------------- -// IoThread — created per `run()`, owned by `SchedulerState`. +// IoThread — created per `run()`, owned by `RuntimeInner::io`. // --------------------------------------------------------------------------- pub struct IoThread { - // ----- Channels & queues ----- - /// Submission queue into the blocking-work pool. tx: mpsc::Sender, - /// Shared completion queue, fed by both the pool and the epoll thread. - completions: Arc>>, - /// Pipe the scheduler polls in its idle path. Both IO threads write to - /// `wake_write` after pushing a completion. - wake_read: RawFd, - wake_write: RawFd, + /// One parked actor per registered fd. Populated by `epoll_register`, + /// consumed by the epoll thread on readiness or `cancel_waiter` on an + /// unwound wait. + waiters: Waiters, // ----- Epoll machinery ----- @@ -133,39 +145,25 @@ pub struct IoThread { /// shutdown. shutdown_read: RawFd, shutdown_write: RawFd, - /// 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, // ----- Threads ----- pool_thread: Option>, epoll_thread: Option>, - - /// Number of `block_on_io` requests in-flight. Used by the scheduler's - /// idle path to decide whether to wait on the pipe or exit. Fd waits - /// are not counted here; they're counted by `waiters.len()`. - pub outstanding: u32, } impl IoThread { - pub fn start() -> io::Result { - // Scheduler-facing wake pipe. - let (wake_read, wake_write) = make_pipe()?; - // Pool submission channel + shared completion queue. + /// Start the pool and epoll threads. `rt` is the producers' route back + /// into the runtime (slot table + unpark protocol); a `Weak` so the + /// `RuntimeInner → IoThread → RuntimeInner` cycle never forms. + pub(crate) fn start(rt: Weak) -> io::Result { + // Pool submission channel. let (tx, rx) = mpsc::channel::(); - let completions: Arc>> = - Arc::new(Mutex::new(VecDeque::new())); + let waiters: Waiters = Arc::new(Mutex::new(HashMap::new())); // Epoll machinery. let epollfd = unsafe { libc::epoll_create1(libc::EPOLL_CLOEXEC) }; if epollfd < 0 { - // Best-effort fd cleanup before bailing. - unsafe { - libc::close(wake_read); - libc::close(wake_write); - } return Err(io::Error::last_os_error()); } @@ -174,8 +172,6 @@ impl IoThread { Err(e) => { unsafe { libc::close(epollfd); - libc::close(wake_read); - libc::close(wake_write); } return Err(e); } @@ -202,42 +198,37 @@ impl IoThread { libc::close(epollfd); libc::close(shutdown_read); libc::close(shutdown_write); - libc::close(wake_read); - libc::close(wake_write); } return Err(e); } // Spawn pool thread. - let pool_comps = completions.clone(); + let pool_rt = rt.clone(); let pool_thread = std::thread::Builder::new() .name("smarm-io-pool".into()) - .spawn(move || pool_loop(rx, pool_comps, wake_write))?; + .spawn(move || pool_loop(rx, pool_rt))?; // Spawn epoll thread. - let epoll_comps = completions.clone(); + let epoll_waiters = waiters.clone(); let epoll_thread = std::thread::Builder::new() .name("smarm-io-epoll".into()) - .spawn(move || epoll_loop(epollfd, epoll_comps, wake_write))?; + .spawn(move || epoll_loop(epollfd, epoll_waiters, rt))?; Ok(Self { tx, - completions, - wake_read, - wake_write, + waiters, epollfd, shutdown_read, shutdown_write, - waiters: HashMap::new(), pool_thread: Some(pool_thread), epoll_thread: Some(epoll_thread), - outstanding: 0, }) } - /// Hand a request to the pool. Increments `outstanding`. + /// Hand a request to the pool. The caller (scheduler.rs) increments + /// `io_outstanding` BEFORE calling — the pool decrements on completion, + /// and an increment that trailed the completion would underflow. 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. if self.tx.send(Request { pid, epoch, work }).is_err() { @@ -245,39 +236,13 @@ impl IoThread { } } - /// Drain every available completion. Caller (the scheduler) routes the - /// results and updates `outstanding` / `waiters` accordingly. - pub fn drain_completions(&mut self) -> Vec { - let mut q = match self.completions.lock() { - Ok(g) => g, - Err(e) => panic!("smarm: io completions lock poisoned (core corrupt): {e}"), - }; - let mut out = Vec::with_capacity(q.len()); - while let Some(c) = q.pop_front() { - out.push(c); - } - out - } - - pub fn wake_fd(&self) -> RawFd { - self.wake_read - } - - /// Write the wake pipe directly: rouse every scheduler thread blocked in - /// its idle `poll_wake`. Used by the terminal (AllDone) path — an idle - /// sibling may be blocked on a snapshot that nothing will ever refresh - /// (an orphaned timer deadline, or `io_outstanding` from a waiter that - /// was stop-cancelled and so never produces a completion). - pub fn wake(&self) { - wake_scheduler(self.wake_write); - } - /// Register interest in `fd` becoming readable/writable; record `pid` - /// as the parked waiter. The epoll thread will push a `FdReady` - /// completion when the kernel signals. + /// as the parked waiter. The epoll thread unparks it on readiness. + /// The caller increments `io_fd_waiters` BEFORE calling (mirror of + /// `submit`'s contract) and decrements it again if this errors. /// - /// EPOLLONESHOT: one wakeup per registration. The scheduler must - /// `epoll_del` on completion to free the slot for re-registration. + /// EPOLLONESHOT: one wakeup per registration; the epoll thread DELs on + /// readiness, `cancel_waiter` DELs on an unwound wait. pub fn epoll_register( &mut self, fd: RawFd, @@ -286,20 +251,24 @@ impl IoThread { readable: bool, writable: bool, ) -> io::Result<()> { + let mut waiters = match self.waiters.lock() { + Ok(g) => g, + Err(e) => panic!("smarm: io waiters lock poisoned (core corrupt): {e}"), + }; // Two actors waiting on the same fd would be a misuse: the kernel // delivers exactly one EPOLLONESHOT wakeup, so the second waiter // would hang. Reject up front. - if self.waiters.contains_key(&fd) { + if waiters.contains_key(&fd) { return Err(io::Error::new( io::ErrorKind::AlreadyExists, "fd already has a parked waiter", )); } - // Belt-and-braces: the unwind guard in `wait_fd` is responsible for - // cleaning up a stopped waiter's registration, but a bare DEL is - // harmless if the fd isn't registered (ENOENT) and removes any leak - // a path we haven't thought of might leave behind. + // Belt-and-braces: `cancel_waiter` is responsible for cleaning up a + // stopped waiter's registration, but a bare DEL is harmless if the + // fd isn't registered (ENOENT) and removes any leak a path we + // haven't thought of might leave behind. unsafe { libc::epoll_ctl(self.epollfd, libc::EPOLL_CTL_DEL, fd, std::ptr::null_mut()); } @@ -321,19 +290,29 @@ impl IoThread { if r < 0 { return Err(io::Error::last_os_error()); } - self.waiters.insert(fd, (pid, epoch)); + waiters.insert(fd, (pid, epoch)); Ok(()) } - /// Remove `fd` from the epollfd. Called by the scheduler after a - /// `FdReady` completion, so the next `wait_readable(fd)` can ADD again. - /// - /// Does NOT touch `waiters` — that's the scheduler's bookkeeping; this - /// is purely the kernel-side cleanup. - pub fn epoll_deregister(&mut self, fd: RawFd) { - // EPOLL_CTL_DEL of an already-removed fd returns ENOENT; ignore. - unsafe { - libc::epoll_ctl(self.epollfd, libc::EPOLL_CTL_DEL, fd, std::ptr::null_mut()); + /// Remove `fd`'s waiter iff it is still `(pid, epoch)`, DELing the fd + /// from the epollfd in the same critical section. Returns whether the + /// entry was removed (the caller then decrements `io_fd_waiters`). + /// `false` means the epoll thread consumed the registration first — + /// the fd may already carry someone else's fresh ADD; hands off. + pub fn cancel_waiter(&mut self, fd: RawFd, pid: Pid, epoch: u32) -> bool { + let mut waiters = match self.waiters.lock() { + Ok(g) => g, + Err(e) => panic!("smarm: io waiters lock poisoned (core corrupt): {e}"), + }; + if waiters.get(&fd) == Some(&(pid, epoch)) { + waiters.remove(&fd); + // EPOLL_CTL_DEL of an already-removed fd returns ENOENT; ignore. + unsafe { + libc::epoll_ctl(self.epollfd, libc::EPOLL_CTL_DEL, fd, std::ptr::null_mut()); + } + true + } else { + false } } } @@ -354,7 +333,10 @@ impl Drop for IoThread { let real_tx = std::mem::replace(&mut self.tx, dead_tx); drop(real_tx); - // 3. Join both threads. + // 3. Join both threads. Safe even while the caller holds the + // runtime's `io` mutex: neither thread ever takes it (they reach + // the runtime through a Weak they upgrade per completion, and + // the epoll thread's only lock is `waiters`). if let Some(h) = self.epoll_thread.take() { let _ = h.join(); } @@ -367,8 +349,6 @@ impl Drop for IoThread { libc::close(self.epollfd); libc::close(self.shutdown_read); libc::close(self.shutdown_write); - libc::close(self.wake_read); - libc::close(self.wake_write); } } } @@ -379,36 +359,38 @@ impl Drop for IoThread { const SHUTDOWN_EPOLL_TOKEN: u64 = u64::MAX; // --------------------------------------------------------------------------- -// Pool loop +// Pool loop (producer: Blocking completions) // --------------------------------------------------------------------------- -fn pool_loop( - rx: mpsc::Receiver, - completions: Arc>>, - wake_write: RawFd, -) { +fn pool_loop(rx: mpsc::Receiver, rt: Weak) { 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), }; - match completions.lock() { - Ok(mut g) => g.push_back(Completion::Blocking { pid, epoch, result }), - Err(e) => panic!("smarm: io completions lock poisoned (core corrupt): {e}"), + let Some(inner) = rt.upgrade() else { return }; + // Stash the result under the cold lock (generation-checked: an + // actor stopped with the op in flight discards it), decrement the + // in-flight count, then wake through the epoch-matched unpark. The + // unpark's enqueue tail wakes a parked scheduler; the actor stays + // `live` until it resumes and finalizes, so the decrement's + // ordering against the termination verdict is not load-bearing. + if let Some(slot) = inner.slot_at(pid) { + let mut cold = slot.cold.lock(); + if slot.generation() == pid.generation() { + cold.pending_io_result = Some(result); + } } - wake_scheduler(wake_write); + inner.io_outstanding.fetch_sub(1, Ordering::AcqRel); + inner.unpark_at(pid, epoch); } } // --------------------------------------------------------------------------- -// Epoll loop +// Epoll loop (producer: FdReady completions) // --------------------------------------------------------------------------- -fn epoll_loop( - epollfd: RawFd, - completions: Arc>>, - wake_write: RawFd, -) { +fn epoll_loop(epollfd: RawFd, waiters: Waiters, rt: Weak) { // Buffer for epoll_wait. 64 is plenty for our scale; if a real load // appears that needs more, this is a one-line change. const MAX_EVENTS: usize = 64; @@ -436,29 +418,41 @@ fn epoll_loop( } let mut shutdown_requested = false; - let mut pushed_any = false; - { - let mut q = match completions.lock() { - Ok(g) => g, - Err(e) => panic!("smarm: io completions lock poisoned (core corrupt): {e}"), + for ev in events.iter().take(n as usize) { + if ev.u64 == SHUTDOWN_EPOLL_TOKEN { + shutdown_requested = true; + continue; + } + let fd = ev.u64 as RawFd; + // Consume the registration: remove + DEL under the waiters + // lock (the ADD/DEL serialization — see module docs). A + // vanished entry means `cancel_waiter` beat us: the wake is + // already moot. + let entry = { + let mut w = match waiters.lock() { + Ok(g) => g, + Err(e) => { + panic!("smarm: io waiters lock poisoned (core corrupt): {e}") + } + }; + let entry = w.remove(&fd); + if entry.is_some() { + unsafe { + libc::epoll_ctl( + epollfd, + libc::EPOLL_CTL_DEL, + fd, + std::ptr::null_mut(), + ); + } + } + entry }; - for ev in events.iter().take(n as usize) { - if ev.u64 == SHUTDOWN_EPOLL_TOKEN { - shutdown_requested = true; - continue; - } - let fd = ev.u64 as RawFd; - let evs = ev.events; - q.push_back(Completion::FdReady { - fd, - events: evs, - }); - pushed_any = true; + if let Some((pid, epoch)) = entry { + let Some(inner) = rt.upgrade() else { return }; + inner.io_fd_waiters.fetch_sub(1, Ordering::AcqRel); + inner.unpark_at(pid, epoch); } - } - - if pushed_any { - wake_scheduler(wake_write); } if shutdown_requested { return; @@ -466,27 +460,8 @@ fn epoll_loop( } } -/// Write one byte to the scheduler's wake pipe. Retries on EINTR; ignores -/// EAGAIN (pipe full means there's already an outstanding wake we haven't -/// consumed yet, which is sufficient). -fn wake_scheduler(wake_write: RawFd) { - let buf: [u8; 1] = [0]; - unsafe { - loop { - let n = libc::write(wake_write, buf.as_ptr() as *const _, 1); - if n < 0 { - let e = *libc::__errno_location(); - if e == libc::EINTR { - continue; - } - } - break; - } - } -} - // --------------------------------------------------------------------------- -// Pipe helpers (unchanged from v0.2) +// Pipe helper // --------------------------------------------------------------------------- fn make_pipe() -> io::Result<(RawFd, RawFd)> { @@ -497,50 +472,3 @@ fn make_pipe() -> io::Result<(RawFd, RawFd)> { } Ok((fds[0], fds[1])) } - -/// Drain pending bytes from the wake pipe. Nonblocking (pipe is O_NONBLOCK). -/// -/// DISCIPLINE: called only by the phase-1 drain-lock winner, immediately -/// before `drain_completions`. Bytes are the notification channel for -/// completions; consuming one anywhere else can strand the completion it -/// announces (see the lost-wakeup note at the call site in `schedule_loop`). -pub fn drain_wake_pipe(fd: RawFd) { - let mut buf = [0u8; 64]; - loop { - let n = unsafe { libc::read(fd, buf.as_mut_ptr() as *mut _, buf.len()) }; - if n <= 0 { - break; - } - } -} - -/// Block on `fd` for up to `timeout`, returning when either there's data -/// to read or the timeout elapses. `None` for `timeout` means wait forever. -pub fn poll_wake(fd: RawFd, timeout: Option) { - let timeout_ms: libc::c_int = match timeout { - None => -1, - Some(d) => { - let ms = d.as_millis(); - if ms > i32::MAX as u128 { - i32::MAX - } else { - ms as i32 - } - } - }; - let mut pfd = libc::pollfd { - fd, - events: libc::POLLIN, - revents: 0, - }; - loop { - let r = unsafe { libc::poll(&mut pfd as *mut _, 1, timeout_ms) }; - if r < 0 { - let e = unsafe { *libc::__errno_location() }; - if e == libc::EINTR { - continue; - } - } - break; - } -} diff --git a/src/lib.rs b/src/lib.rs index a290d37..3ced4b4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -32,9 +32,6 @@ pub mod introspect; #[cfg(feature = "observer")] pub mod observer; pub mod runtime; -// TEMPORARY dead_code allow: park is standalone until the RFC 018 runtime -// swap (next commit) wires it into schedule_loop/enqueue; the allow dies there. -#[allow(dead_code)] pub(crate) mod park; pub(crate) mod raw_mutex; pub(crate) mod slot_state; diff --git a/src/park.rs b/src/park.rs index 9142474..f411fac 100644 --- a/src/park.rs +++ b/src/park.rs @@ -438,9 +438,10 @@ impl Coordinator { } } - /// Latest idle mask (RMW read — same handshake as `wake_one`; used by - /// the chain rule: "queue non-empty and mask non-zero ⇒ wake_one - /// again", where missing a just-parked sibling would strand its item). + /// Latest idle mask (RMW read — same handshake as `wake_one`). A + /// test-only observer: production expresses the chain rule through + /// `wake_one_if_idle` (fence + Relaxed load), not a mask read. + #[cfg(test)] pub(crate) fn idle_mask(&self) -> u64 { self.idle.fetch_or(0, Ordering::AcqRel) } @@ -503,7 +504,9 @@ impl Coordinator { } /// The armed-deadline snapshot (nanos since origin; `NO_DEADLINE` = - /// none). One Relaxed load. + /// none). Test-only introspection on the timekeeper's armed value; the + /// busy-path due-check reads `next_deadline`, not this. + #[cfg(test)] pub(crate) fn armed_deadline_nanos(&self) -> u64 { self.tk_armed.load(Ordering::Relaxed) } @@ -540,8 +543,10 @@ impl Coordinator { n != NO_DEADLINE && self.deadline_nanos(Instant::now()) >= n } - /// The earliest-deadline snapshot as an `Instant`, for the idle path's - /// park timeout (`None` = no timer pending). + /// The earliest-deadline snapshot as an `Instant` (`None` = no timer + /// pending). Test-only: the idle path arms the timekeeper from the + /// timer heap's own `peek_deadline` under the timers mutex. + #[cfg(test)] pub(crate) fn next_deadline_instant(&self) -> Option { let n = self.next_deadline.load(Ordering::Acquire); if n == NO_DEADLINE { @@ -875,9 +880,12 @@ mod loom_tests { .is_ok() { consumed.fetch_add(1, O::SeqCst); - // THE CHAIN RULE: surplus + idle sibling ⇒ wake. - if items.load(O::SeqCst) > 0 && c.idle_mask() != 0 { - c.wake_one(); + // THE CHAIN RULE, exactly as production expresses it + // (runtime.rs schedule_loop): surplus ⇒ the fenced + // fast-path wake. Models the Relaxed-load chain path, + // not just the RMW one. + if items.load(O::SeqCst) > 0 { + c.wake_one_if_idle(); } continue; } diff --git a/src/runtime.rs b/src/runtime.rs index f491d3c..bd5c025 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -86,21 +86,28 @@ //! # Termination (counter-based) //! //! The old all-clear scanned the slot table under the big lock. Now: -//! exit when `io_out == 0` (read *before* the queue lock, phase-1 ordering) -//! and, under the queue lock, the queue is empty and `live_actors == 0`. -//! `live_actors` is incremented in `spawn` before the enqueue and decremented -//! at the very END of `finalize_actor`, strictly after every wakeup that -//! finalize produces has been enqueued. The soundness crux: any enqueue -//! targets a live (not-yet-finalized) actor, so `live == 0` implies no wakeup -//! can still be in flight; combined with "spawner is itself live", observing -//! `(queue empty, live == 0)` under the queue lock means no work can ever -//! appear again. +//! exit when `io_outstanding + io_fd_waiters == 0` (two Relaxed/Acquire +//! atomic loads, read *before* the queue pop) and, under the queue lock, +//! the queue is empty and `live_actors == 0`. `live_actors` is incremented +//! in `spawn` before the enqueue and decremented at the very END of +//! `finalize_actor`, strictly after every wakeup that finalize produces has +//! been enqueued. The soundness crux: any enqueue targets a live +//! (not-yet-finalized) actor, so `live == 0` implies no wakeup can still be +//! in flight; combined with "spawner is itself live", observing +//! `(queue empty, live == 0)` means no work can ever appear again. //! -//! # Timer / IO drain (try-lock, one-winner) +//! # Scheduler park/wake (RFC 018) //! -//! Unchanged from phase 1: one winner per round drains due timers and IO -//! completions from their own mutexes; wakeups go through the unpark -//! protocol like everyone else's. +//! Schedulers sleep on per-thread futex parkers via the coordination layer +//! (`park.rs`), NOT on a shared wake pipe. IO backends are producers behind +//! a two-call contract — make the actor runnable (`unpark_at`), whose +//! `enqueue` tail wakes exactly one parked scheduler. The blocking pool and +//! epoll thread each route their own completions (driver-enqueues); there +//! is no shared completion queue, no drain lock, no one-winner drain phase. +//! Timers fire two ways: a busy-path due-check every loop iteration (one +//! Relaxed load of the earliest-deadline snapshot when no timer is armed), +//! and the timekeeper — at most one parked scheduler holds the timer +//! deadline, so an expiry wakes one scheduler, not a herd. use crate::actor::{ clear_current_pid, is_actor_done, reset_actor_done, set_current_actor_box, @@ -750,8 +757,21 @@ pub(crate) struct RuntimeInner { pub(crate) io: Mutex>, /// Monotonic `MonitorId` source. Never reused. pub(crate) next_monitor_id: AtomicU64, - /// Try-lock: exactly one scheduler thread drains timers/IO per iteration. - drain_lock: Mutex<()>, + /// RFC 018: the scheduler coordination layer — per-scheduler parkers, + /// idle mask, wake protocol, timekeeper role, earliest-deadline + /// snapshot. Arc'd because `Timers` shares it (insert-side deadline + /// notes run under the timers mutex). + pub(crate) coord: Arc, + /// `block_on_io` requests in flight. Incremented by the submitter + /// BEFORE submit (underflow-proof), decremented by the pool thread on + /// completion. Read lock-free by the idle path's termination verdict — + /// the per-pop `io.lock` of the drain era is gone. + pub(crate) io_outstanding: AtomicU32, + /// Parked fd waiters. Incremented by the registrar BEFORE + /// `epoll_register` (rolled back on error), decremented by whoever + /// consumes the registration (epoll thread on readiness, canceller on + /// an unwound wait). Same lock-free verdict read as `io_outstanding`. + pub(crate) io_fd_waiters: AtomicU32, /// Per-thread stats, indexed by scheduler thread slot (0..N). pub(crate) stats: Vec, /// Global counters for RFC 000 primitives. @@ -802,6 +822,12 @@ impl RuntimeInner { let slots: Box<[Slot]> = (0..max_actors).map(|_| Slot::vacant()).collect(); // Low indices on top of the stack so early spawns get low pids. let free: Vec = (0..max_actors as u32).rev().collect(); + // RFC 018: the coordination layer (asserts thread_count <= 64), and + // the timers' hook into it — every insert under the timers mutex + // notes its deadline (busy-path snapshot + timekeeper re-arm). + let coord = Arc::new(crate::park::Coordinator::new(thread_count)); + let mut timers = Timers::new(); + timers.attach_coordinator(coord.clone()); Arc::new(Self { run_queue: crate::run_queue::RunQueue::new(thread_count, max_actors), slots, @@ -810,10 +836,12 @@ impl RuntimeInner { root_bits: AtomicU64::new(u64::MAX), root_exited: AtomicBool::new(false), root_swept: AtomicBool::new(false), - timers: Mutex::new(Timers::new()), + timers: Mutex::new(timers), io: Mutex::new(None), next_monitor_id: AtomicU64::new(0), - drain_lock: Mutex::new(()), + coord, + io_outstanding: AtomicU32::new(0), + io_fd_waiters: AtomicU32::new(0), stats, io_parked: AtomicU32::new(0), sleeping: AtomicU32::new(0), @@ -874,6 +902,13 @@ impl RuntimeInner { ); self.run_queue.push(pid); crate::te!(crate::trace::Event::Enqueue(pid)); + // RFC 018 enqueue wake (fixes the silent enqueue): if a scheduler + // is parked, wake exactly one. The fast path when everyone is busy + // is a fence + one Relaxed load of an unmodified line — the + // pure-compute hot path pays (almost) nothing. Bias is over-wake: + // a spurious wake costs one futex round-trip and a failed pop; a + // missed wake would cost a stranded actor. + self.coord.wake_one_if_idle(); } /// Make `pid` runnable if it is parked; coalesce or defer otherwise. @@ -1076,7 +1111,16 @@ impl Runtime { self.inner.live_actors.load(Ordering::Acquire), 0, "run() called while previous run still active" ); - let io_thread = match IoThread::start() { + // RFC 018: the IO producers reach the runtime (slot table + unpark) + // through a Weak, so no RuntimeInner → IoThread → RuntimeInner cycle + // forms. Reset the in-flight counters BEFORE the threads can touch + // them (a prior run left them at 0 on a clean exit; the asserts pin + // that). + debug_assert_eq!(self.inner.io_outstanding.load(Ordering::Acquire), 0); + debug_assert_eq!(self.inner.io_fd_waiters.load(Ordering::Acquire), 0); + self.inner.io_outstanding.store(0, Ordering::Release); + self.inner.io_fd_waiters.store(0, Ordering::Release); + let io_thread = match IoThread::start(Arc::downgrade(&self.inner)) { Ok(io) => io, Err(e) => panic!("failed to start IO thread: {e}"), }; @@ -1179,6 +1223,8 @@ impl Runtime { } self.inner.io_parked.store(0, Ordering::Relaxed); self.inner.sleeping.store(0, Ordering::Relaxed); + self.inner.io_outstanding.store(0, Ordering::Relaxed); + self.inner.io_fd_waiters.store(0, Ordering::Relaxed); RUNTIME.with(|r| *r.borrow_mut() = None); @@ -1493,6 +1539,56 @@ fn stop_live_actors(inner: &Arc) { } } +// --------------------------------------------------------------------------- +// Timer firing — shared by the busy-path due-check and the timekeeper +// --------------------------------------------------------------------------- + +/// Pop and dispatch every due timer. `pop_due` re-anchors the +/// earliest-deadline snapshot under the timers mutex before returning, so +/// a caller that raced a concurrent insert simply comes back on the next +/// due-check. Dispatch runs with the timers lock released. +fn fire_due_timers(inner: &Arc, try_only: bool) { + let due = if try_only { + // Busy path: if another scheduler is already in the timers mutex + // (firing, inserting, or peeking) skip — the snapshot stays due + // until someone actually pops, so the check re-fires next loop. + match inner.timers.try_lock() { + Ok(mut t) => t.pop_due(std::time::Instant::now()), + Err(std::sync::TryLockError::WouldBlock) => return, + Err(std::sync::TryLockError::Poisoned(e)) => { + panic!("smarm: timers lock poisoned (core corrupt): {e}") + } + } + } else { + match inner.timers.lock() { + Ok(mut t) => t.pop_due(std::time::Instant::now()), + Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"), + } + }; + for entry in due { + match entry.reason { + // A sleep expiry is just an unpark: the protocol handles + // every interleaving — Parked (re-queue), Running (the + // actor is between `timers.insert_sleep` and + // `park_current`; RunningNotified makes the upcoming park + // re-queue), or gone (no-op). + 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); + } + // A `send_after` deadline: run the captured delivery thunk. + // It resolves the destination through the registry and + // sends now (a send can unpark a receiver) — same as any + // other in-loop unpark. The timers lock is already + // released; lock order Leaf -> Channel is preserved by the + // send itself. `pop_due` only returns still-armed Sends, so + // a cancelled one never reaches here. + crate::timer::Reason::Send { fire } => fire(), + } + } +} + // --------------------------------------------------------------------------- // schedule_loop — runs on each scheduler OS thread // --------------------------------------------------------------------------- @@ -1503,120 +1599,14 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { loop { // ---------------------------------------------------------------- - // 1. Try to win the drain lock (timers + IO). One winner per round; - // losers skip immediately and proceed to step 2. + // 1. Busy-path timer due-check (RFC 018 design point (a)): under + // saturation nobody parks, so no timekeeper exists — due timers + // must still fire. One Relaxed load + branch when no timer is + // armed; the clock is read only when one is. // ---------------------------------------------------------------- - if let Ok(_drain_guard) = inner.drain_lock.try_lock() { - // Timers and IO live behind their own mutexes (phase 1), so the - // pure-yield / pure-compute hot path never contends a global lock - // just to discover there is nothing to drain. The clock is read - // only when the timer heap is non-empty. - let due = { - let mut t = match inner.timers.lock() { - Ok(t) => t, - Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"), - }; - if t.is_empty() { - Vec::new() - } else { - t.pop_due(std::time::Instant::now()) - } - }; - let completions = match inner.io.lock() { - Ok(mut io) => io - .as_mut() - .map(|io| { - // Consume wake-pipe bytes ONLY here, under the drain - // lock and strictly before draining completions. - // Producers push their completion before writing the - // byte, so every byte consumed here has its completion - // visible to the drain below. Consuming bytes anywhere - // else — in particular after an idle poll, outside the - // lock — loses wakeups: a try_lock loser can eat the - // byte for a completion the winner never saw, leaving - // it stranded (and its EPOLLONESHOT fd disarmed) until - // an unrelated timer forces another drain pass. - crate::io::drain_wake_pipe(io.wake_fd()); - io.drain_completions() - }) - .unwrap_or_default(), - Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"), - }; - for entry in due { - match entry.reason { - // A sleep expiry is just an unpark: the protocol handles - // every interleaving — Parked (re-queue), Running (the - // actor is between `timers.insert_sleep` and - // `park_current`; RunningNotified makes the upcoming park - // re-queue), or gone (no-op). - 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); - } - // A `send_after` deadline: run the captured delivery thunk. - // It resolves the destination through the registry and - // sends now (a send can unpark a receiver) — same as any - // other in-loop unpark. The timers lock is already - // released; lock order Leaf -> Channel is preserved by the - // send itself. `pop_due` only returns still-armed Sends, so - // a cancelled one never reaches here. - crate::timer::Reason::Send { fire } => fire(), - } - } - - for completion in completions { - match completion { - crate::io::Completion::Blocking { pid, epoch, result } => { - match inner.io.lock() { - Ok(mut io) => { - if let Some(io) = io.as_mut() { - io.outstanding = io.outstanding.saturating_sub(1); - } - } - Err(e) => { - panic!("smarm: io lock poisoned (core corrupt): {e}") - } - } - // Stash the result under the cold lock, then unpark. - // The protocol also covers the submit→park window - // (RunningNotified), which the old code missed for - // Blocking completions — a latent lost wakeup. - if let Some(slot) = inner.slot_at(pid) { - { - let mut cold = slot.cold.lock(); - if slot.generation() == pid.generation() { - cold.pending_io_result = Some(result); - } else { - // Actor died (stopped) with the op in - // flight; discard the result. - } - } - 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 = match inner.io.lock() { - Ok(mut io) => io.as_mut().and_then(|io| { - let entry = io.waiters.remove(&fd); - io.epoll_deregister(fd); - entry - }), - Err(e) => { - panic!("smarm: io lock poisoned (core corrupt): {e}") - } - }; - if let Some((pid, epoch)) = parked { - inner.unpark_at(pid, epoch); - } - } - } - } - } // drain_guard drops here + if inner.coord.deadline_due() { + fire_due_timers(inner, true); + } // ---------------------------------------------------------------- // 2. Pop a runnable pid. Pop order (RFC 005): wake slot first, then @@ -1625,7 +1615,7 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { // ---------------------------------------------------------------- enum Pop { Got(Pid), - Idle { io_outstanding: u32, wake_fd: Option }, + Idle, AllDone, /// Root has exited and nothing is runnable: stop the parked-forever /// remainder, then re-pop. Fires at most once per run. @@ -1650,19 +1640,12 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { crate::te!(crate::trace::Event::SlotPop(pid)); pid } else { - // Read IO liveness BEFORE the queue lock (phase-1 ordering: a - // completion resurrects an actor only via the drain path, whose - // enqueue would be visible under the queue lock we take next). - let (io_out, io_fd) = { - let io = match inner.io.lock() { - Ok(io) => io, - Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"), - }; - match io.as_ref() { - Some(io) => (io.outstanding + io.waiters.len() as u32, Some(io.wake_fd())), - None => (0, None), - } - }; + // Read IO liveness BEFORE the queue pop — two atomic loads now + // (RFC 018), not a per-pop `io.lock`: a completion resurrects + // an actor via the producer's own unpark→enqueue, whose entry + // would be visible to the pop below. + let io_out = inner.io_outstanding.load(Ordering::Acquire) + + inner.io_fd_waiters.load(Ordering::Acquire); stats.run_queue_len.store(inner.run_queue.len(), Ordering::Relaxed); let pop = match inner.run_queue.pop() { @@ -1694,7 +1677,7 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { // the idle wait below on the next pass. Pop::RootDrain } else { - Pop::Idle { io_outstanding: io_out, wake_fd: io_fd } + Pop::Idle } } }; @@ -1709,22 +1692,15 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { Ok(mut timers) => timers.clear(), Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"), } - // Terminal wake: a sibling scheduler may be blocked in its - // idle wait on a snapshot that is now terminally stale — an - // orphaned long deadline (it would sleep it out in full) or - // a stale `io_outstanding > 0` from a stop-cancelled waiter - // (it would block in poll(-1) forever; cancellation produces - // no completion, so nothing else writes the wake pipe). - // One byte wakes every poller; each re-runs the verdict, - // reaches AllDone itself, and re-wakes — idempotent. - match inner.io.lock() { - Ok(io) => { - if let Some(io) = io.as_ref() { - io.wake(); - } - } - Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"), - } + // Terminal wake (replaces the wake-pipe byte): a sibling + // may be parked on a snapshot that is now terminally + // stale — an orphaned long deadline, or a stale + // `io_fd_waiters > 0` from a stop-cancelled waiter + // (cancellation produces no completion, so nothing else + // will ever wake it). `wake_all` permits every parker; + // each sibling re-runs the verdict, reaches AllDone + // itself, and re-wakes — idempotent. + inner.coord.wake_all(); return; } Pop::RootDrain => { @@ -1734,40 +1710,51 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { stop_live_actors(inner); continue; } - Pop::Idle { io_outstanding, wake_fd } => { - // Something is still in flight. Sleep on the appropriate - // source to avoid hammering the queue mutex; retry on wake. - let next_deadline = match inner.timers.lock() { - Ok(timers) => timers.peek_deadline(), - Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"), - }; - match (next_deadline, wake_fd) { - (Some(deadline), fd_opt) => { - let now = std::time::Instant::now(); - if deadline > now { - let timeout = deadline - now; - match fd_opt { - Some(fd) => { - // Wake only; the byte (if any) is - // consumed by the next drain-lock - // winner in phase 1. Level-triggered - // poll means an unconsumed byte makes - // this return immediately, so a loser - // spins briefly until the winner - // releases — never sleeps through it. - crate::io::poll_wake(fd, Some(timeout)); - } - None => thread::sleep(timeout), - } + Pop::Idle => { + // Something is still in flight. Park on our own futex + // until a producer wakes us (enqueue tail), a deadline + // passes, or the re-check finds the world changed. + // + // Timekeeper (RFC 018): at most one parked scheduler + // holds the timer deadline — the first idler to arm it + // parks with a timeout, the rest park indefinitely, so a + // timer expiry wakes one scheduler, not a herd. Peek and + // arm under the timers mutex (the serialization that + // makes the insert-side re-arm race-free). + let tk_deadline = { + let timers = match inner.timers.lock() { + Ok(t) => t, + Err(e) => { + panic!("smarm: timers lock poisoned (core corrupt): {e}") } - } - (None, Some(fd)) if io_outstanding > 0 => { - // See above: no byte consumption outside phase 1. - crate::io::poll_wake(fd, None); - } - _ => { - thread::sleep(std::time::Duration::from_micros(100)); - } + }; + timers + .peek_deadline() + .filter(|d| inner.coord.try_arm_timer(slot_idx, *d)) + }; + // The mandatory post-publish re-check: a producer that + // enqueued (or a verdict input that flipped) before it + // could see our idle bit has left us the evidence. + let _ = inner.coord.park(slot_idx, tk_deadline, || { + !inner.run_queue.is_empty() + || (inner.live_actors.load(Ordering::Acquire) == 0 + && inner.io_outstanding.load(Ordering::Acquire) == 0 + && inner.io_fd_waiters.load(Ordering::Acquire) == 0) + || (inner.root_exited.load(Ordering::Acquire) + && !inner.root_swept.load(Ordering::Acquire)) + || inner.coord.deadline_due() + }); + if tk_deadline.is_some() { + // Hand the role back BEFORE firing: pop_due can run + // `Send` thunks that insert new timers, and the + // insert-side re-arm check must see either no + // timekeeper (skip) or a real parked one — never us, + // awake and about to re-peek anyway. + inner.coord.disarm_timer(slot_idx); + // Woken for the deadline, for work, or to re-peek + // after an earlier insert — fire whatever is due; + // the next idle pass re-arms with the new minimum. + fire_due_timers(inner, false); } continue; } @@ -1780,6 +1767,21 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { // by the at-most-once-enqueued invariant nothing else can have // changed the state of a queued actor. // ---------------------------------------------------------------- + + // RFC 018 chain rule: we just took one runnable; if more remain and + // a sibling is parked, wake exactly one so the surplus runs in + // PARALLEL rather than serially behind us (without this the surplus + // is not stranded — we re-pop it after resuming — but it waits out + // our whole timeslice while an idle core sits available). Cheap: the + // queue-length check is queue-local, and `wake_one_if_idle` is a + // fence + one Relaxed mask load when nobody is parked. A Relaxed + // miss here is safe — the enqueue that created the surplus already + // issued its own wake (RFC 018 no-lost-wake); this only sharpens + // parallelism latency. + if !inner.run_queue.is_empty() { + inner.coord.wake_one_if_idle(); + } + let slot = match inner.slot_at(pid) { Some(s) => s, None => continue, // can't happen for real pids; defensive diff --git a/src/scheduler.rs b/src/scheduler.rs index dea289f..99eb666 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -72,6 +72,7 @@ use crate::runtime::{ self, RuntimeInner, YieldIntent, RUNTIME, }; use crate::supervisor::Signal; +use std::sync::atomic::Ordering; use std::sync::Arc; // --------------------------------------------------------------------------- @@ -753,7 +754,14 @@ where Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"), }; match io.as_mut() { - Some(io) => io.submit(me, epoch, work), + Some(io) => { + // RFC 018: count the op in flight BEFORE submit — the + // pool decrements on completion, and an increment that + // trailed the completion would underflow. Under the io + // lock, so ordered against the same-lock submit. + inner.io_outstanding.fetch_add(1, Ordering::AcqRel); + io.submit(me, epoch, work); + } None => panic!("io thread not started"), } }); @@ -813,7 +821,17 @@ fn wait_fd(fd: std::os::fd::RawFd, readable: bool, writable: bool) -> std::io::R Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"), }; match io.as_mut() { - Some(io) => io.epoll_register(fd, me, epoch, readable, writable), + Some(io) => { + // RFC 018: count the waiter BEFORE the ADD (mirror of + // submit); roll back if the registration fails so a + // rejected wait leaves the verdict counters clean. + inner.io_fd_waiters.fetch_add(1, Ordering::AcqRel); + let r = io.epoll_register(fd, me, epoch, readable, writable); + if r.is_err() { + inner.io_fd_waiters.fetch_sub(1, Ordering::AcqRel); + } + r + } None => panic!("io thread not started"), } })?; @@ -838,9 +856,12 @@ fn wait_fd(fd: std::os::fd::RawFd, readable: bool, writable: bool) -> std::io::R Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"), }; if let Some(io) = io.as_mut() { - if io.waiters.get(&self.fd) == Some(&(self.me, self.epoch)) { - io.waiters.remove(&self.fd); - io.epoll_deregister(self.fd); + // `cancel_waiter` removes + DELs iff still ours, all + // under the waiters lock (the ADD/DEL serialization); + // decrement only when we actually removed it — a + // FdReady that consumed it already did the decrement. + if io.cancel_waiter(self.fd, self.me, self.epoch) { + inner.io_fd_waiters.fetch_sub(1, Ordering::AcqRel); } } }); @@ -908,7 +929,14 @@ impl crate::channel::Selectable for FdArm { }; match io.as_mut() { Some(io) => { - io.epoll_register(self.fd, pid, epoch, self.readable, self.writable) + inner.io_fd_waiters.fetch_add(1, Ordering::AcqRel); + let r = io.epoll_register( + self.fd, pid, epoch, self.readable, self.writable, + ); + if r.is_err() { + inner.io_fd_waiters.fetch_sub(1, Ordering::AcqRel); + } + r } None => panic!("io thread not started"), } @@ -936,9 +964,8 @@ impl crate::channel::Selectable for FdArm { Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"), }; if let Some(io) = io.as_mut() { - if io.waiters.get(&self.fd) == Some(&(pid, epoch)) { - io.waiters.remove(&self.fd); - io.epoll_deregister(self.fd); + if io.cancel_waiter(self.fd, pid, epoch) { + inner.io_fd_waiters.fetch_sub(1, Ordering::AcqRel); } } }); diff --git a/src/timer.rs b/src/timer.rs index 143cae0..e16c431 100644 --- a/src/timer.rs +++ b/src/timer.rs @@ -141,6 +141,14 @@ impl PartialOrd for Entry { #[derive(Default)] pub struct Timers { + /// RFC 018: the scheduler coordination layer. Attached once at + /// `RuntimeInner::new`; every insert notes its deadline (min-maintained + /// snapshot for the busy-path due-check + the timekeeper re-arm wake) + /// and every pop/clear re-anchors the snapshot to the heap minimum. + /// All calls happen under the timers mutex — the serialization the + /// coordinator's timer protocol mandates. `None` only in unit tests + /// that construct a bare `Timers`. + coord: Option>, /// Reverse-wrapped so the smallest deadline is at the top. heap: BinaryHeap>, /// Monotonic counter for the tiebreaker `seq` field (and the `TimerId` of a @@ -157,7 +165,18 @@ pub struct Timers { impl Timers { pub fn new() -> Self { - Self { heap: BinaryHeap::new(), next_seq: 0, armed: std::collections::HashSet::new() } + Self { + coord: None, + heap: BinaryHeap::new(), + next_seq: 0, + armed: std::collections::HashSet::new(), + } + } + + /// Attach the scheduler coordination layer (RFC 018). Called once, at + /// runtime construction, before any scheduler thread exists. + pub(crate) fn attach_coordinator(&mut self, c: std::sync::Arc) { + self.coord = Some(c); } /// Insert a `Sleep` timer. Convenience for the common case. @@ -242,6 +261,13 @@ impl Timers { #[cfg(feature = "smarm-causal")] wall, })); + // RFC 018: publish the (possibly new-minimum) deadline to the + // busy-path snapshot and wake the timekeeper if it is parked + // toward a later one. We hold the timers mutex — the mandated + // serialization for both. + if let Some(c) = &self.coord { + c.note_deadline(deadline); + } seq } @@ -255,6 +281,9 @@ impl Timers { pub fn clear(&mut self) { self.heap.clear(); self.armed.clear(); + if let Some(c) = &self.coord { + c.refresh_deadline(None); + } } /// Soonest pending deadline, or `None` if the heap is empty. @@ -324,6 +353,13 @@ impl Timers { } out.push(entry); } + // RFC 018: re-anchor the busy-path snapshot to the new heap minimum + // (still under the timers mutex). A causal-shift re-queue above went + // through `heap.push` directly, so this peek is the one place the + // snapshot is guaranteed to catch up. + if let Some(c) = &self.coord { + c.refresh_deadline(self.peek_deadline()); + } out } } diff --git a/tests/park_wake.rs b/tests/park_wake.rs new file mode 100644 index 0000000..2224a94 --- /dev/null +++ b/tests/park_wake.rs @@ -0,0 +1,69 @@ +//! RFC 018 scheduler park/wake — observable-behavior guards. +//! +//! These pin the two timer-latency properties the park/wake swap must +//! preserve or introduce: +//! +//! - `sleep_fires_under_saturation`: due timers fire even when every +//! scheduler is busy (nobody parked ⇒ no timekeeper) — the busy-path +//! due-check, ratified design point (a). The old drain phase gave this +//! for free (timers drained every loop iteration); the new design must +//! not lose it. +//! - `submillisecond_sleep_is_prompt`: a sub-ms sleep completes promptly. +//! Under the old wake pipe, `poll_wake`'s `as_millis` truncation turned +//! sub-ms deadlines into 0ms busy-polls (correct wall time, pathological +//! CPU); under park/wake the futex timespec carries full nanosecond +//! precision. + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +#[test] +fn sleep_fires_under_saturation() { + let rt = smarm::runtime::init(smarm::runtime::Config::exact(4)); + rt.run(|| { + let stop = Arc::new(AtomicBool::new(false)); + let mut spinners = Vec::new(); + // 8 spinners over 4 schedulers: the run queue never empties, so no + // scheduler ever parks and no timekeeper exists. Only the busy-path + // due-check can fire the sleeper's timer before the spinners quit. + for _ in 0..8 { + let stop = stop.clone(); + spinners.push(smarm::spawn(move || { + let t0 = Instant::now(); + while !stop.load(Ordering::Relaxed) && t0.elapsed() < Duration::from_secs(5) { + smarm::yield_now(); + } + })); + } + let t0 = Instant::now(); + smarm::sleep(Duration::from_millis(10)); + let dt = t0.elapsed(); + stop.store(true, Ordering::Relaxed); + for s in spinners { + let _ = s.join(); + } + assert!( + dt < Duration::from_millis(500), + "10ms sleep took {dt:?} under scheduler saturation — busy-path \ + timer firing is broken (timekeeper-only firing stalls under load)" + ); + }); +} + +#[test] +fn submillisecond_sleep_is_prompt() { + let rt = smarm::runtime::init(smarm::runtime::Config::exact(2)); + rt.run(|| { + // Warm one iteration, then measure. + smarm::sleep(Duration::from_micros(500)); + let t0 = Instant::now(); + smarm::sleep(Duration::from_micros(500)); + let dt = t0.elapsed(); + assert!(dt >= Duration::from_micros(400), "woke early: {dt:?}"); + assert!( + dt < Duration::from_millis(100), + "500µs sleep took {dt:?} — sub-ms deadline handling is broken" + ); + }); +}