diff --git a/ROADMAP.md b/ROADMAP.md index c36505a..47bf681 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -95,10 +95,17 @@ and the full 013/014 history in git. Now unblocked (RFC 013/014 shipped: a timer's `Dest` resolves to a `Pid` / `Name` *on fire*, not a bespoke channel/closure). Two layers: -**Substrate — `send_after` / `cancel_timer`.** Message-delivery timer on the -existing min-heap (`timer.rs`): deliver a value to an address at a deadline, -cancellable. Small, and the thing the gen_server idioms below have no clean -expression without today. +**Substrate — `send_after` / `cancel_timer`.** ✅ SHIPPED — message-delivery +timer on the existing `timer.rs` min-heap: deliver a value to an address +(`Pid` via `send_to`, `Name` via `send`), resolved *on fire* so a dead +target / restarted name is observed at fire time; failed resolve dropped +(Erlang `erlang:send_after`). A new `Reason::Send { fire }` thunk carries +delivery type-erased; cancellation is an `armed` set keyed on the entry `seq` +(only `Send` timers use it — `Sleep`/`WaitTimeout` keep inert-stale), exposed +as an opaque `TimerId`. `cancel` is unscoped (any holder of the id) and returns +whether it landed before the fire (the race signal). `peek_deadline`'s contract +relaxed to "≤ true next deadline" so a future hierarchical timing wheel can back +`Timers` without touching `send_after` or the scheduler idle path. **The gen_server patterns themselves** (the headline). The OTP time vocabulary, expressed against the v0.8 server loop and its `select` arm priority: @@ -111,10 +118,10 @@ expressed against the v0.8 server loop and its `select` arm priority: exponential re-arm on failure. Call deadlines already exist in part (`CallTimeoutError`); fold them in rather -than reinvent. Needs an RFC: the timer-arming surface on `ServerCtx`, the -timeout-directive shape, and how a fired timer interacts with `handle_info` / -arm priority. Land the `send_after` substrate first; `gen_statem` state timeouts -(Low priority) then sit on the same mechanism. +than reinvent. The `send_after` substrate is now landed (above); `gen_statem` +state timeouts (Low priority) sit on the same mechanism. Still needs an RFC: the +timer-arming surface on `ServerCtx`, the timeout-directive shape, and how a +fired timer interacts with `handle_info` / arm priority. --- diff --git a/src/lib.rs b/src/lib.rs index 809c70e..2bd46f1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -65,11 +65,12 @@ pub use registry::{ }; pub use runtime::{init, Config, Runtime}; pub use scheduler::{ - block_on_io, request_stop, run, self_pid, sleep, spawn, spawn_addr, spawn_under, wait_readable, - wait_readable_timeout, wait_writable, wait_writable_timeout, yield_now, FdArm, JoinError, - JoinHandle, + block_on_io, cancel_timer, request_stop, run, self_pid, send_after, send_after_named, sleep, + spawn, spawn_addr, spawn_under, wait_readable, wait_readable_timeout, wait_writable, + wait_writable_timeout, yield_now, FdArm, JoinError, JoinHandle, }; pub use supervisor::{ChildSpec, OneForOne, Restart, Signal, Strategy}; +pub use timer::TimerId; // --------------------------------------------------------------------------- // check!() diff --git a/src/runtime.rs b/src/runtime.rs index acdbf41..6a95e3d 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1232,6 +1232,14 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { // 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(), } } diff --git a/src/scheduler.rs b/src/scheduler.rs index c1a9131..9fddb08 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -9,7 +9,7 @@ use crate::actor::current_pid; use crate::channel::Sender; -use crate::pid::Pid; +use crate::pid::{Name, Pid}; use crate::runtime::{ self, RuntimeInner, YieldIntent, RUNTIME, }; @@ -388,6 +388,57 @@ pub fn insert_wait_timer( }); } +// --------------------------------------------------------------------------- +// send_after / cancel_timer — message-delivery timers (Erlang send_after). +// +// Arm a timer that delivers `msg` to an address after `after`, returning a +// `TimerId`. The destination is resolved *on fire*, not at arm time: a +// `Pid` that has since died yields `SendError::Dead`, a `Name` resolves +// to whoever currently holds it (so a restarted server is reached). Either way +// a failed resolve / closed inbox is dropped, matching `erlang:send_after`. +// `cancel_timer` prevents an as-yet-unfired delivery; it returns whether the +// timer was still armed. +// --------------------------------------------------------------------------- + +/// Deliver `msg` to the exact actor named by `dest` (identity-bound, no +/// redirect — see [`send_to`](crate::registry::send_to)) after `after`. +/// Returns a [`TimerId`](crate::timer::TimerId) for [`cancel_timer`]. +pub fn send_after( + after: std::time::Duration, + dest: Pid, + msg: A::Msg, +) -> crate::timer::TimerId { + let deadline = crate::timer::deadline_from_now(after); + let fire = Box::new(move || { + let _ = crate::registry::send_to(dest, msg); + }); + with_runtime(|inner| inner.timers.lock().unwrap().insert_send(deadline, dest.erase(), fire)) +} + +/// Deliver `msg` to whichever actor holds the name `dest` at fire time +/// (re-resolving [`send`](crate::registry::send) semantics) after `after`. +/// Returns a [`TimerId`](crate::timer::TimerId) for [`cancel_timer`]. +pub fn send_after_named( + after: std::time::Duration, + dest: Name, + msg: M, +) -> crate::timer::TimerId { + let deadline = crate::timer::deadline_from_now(after); + // Informational only (who armed it); not used for delivery. + let armed_by = current_pid().unwrap_or(Pid::new(0, 0)); + let fire = Box::new(move || { + let _ = crate::registry::send(dest, msg); + }); + with_runtime(|inner| inner.timers.lock().unwrap().insert_send(deadline, armed_by, fire)) +} + +/// Cancel a timer armed by [`send_after`] / [`send_after_named`]. Returns +/// `true` if it was still pending (delivery now prevented), `false` if it had +/// already fired or been cancelled. +pub fn cancel_timer(id: crate::timer::TimerId) -> bool { + with_runtime(|inner| inner.timers.lock().unwrap().cancel(id)) +} + // --------------------------------------------------------------------------- // block_on_io / wait_readable / wait_writable / read / write // --------------------------------------------------------------------------- diff --git a/src/timer.rs b/src/timer.rs index 10cf402..118d1f5 100644 --- a/src/timer.rs +++ b/src/timer.rs @@ -15,12 +15,14 @@ //! `BinaryHeap` is a max-heap; entries are wrapped in `Reverse` to get //! min-heap behaviour. //! -//! 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 / -//! 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". +//! Cancellation is selective. A `Sleep` / `WaitTimeout` entry is left in the +//! heap on a non-timer wakeup (lock granted before timeout): it is popped +//! eventually and no-ops because a stale unpark fails its epoch CAS — cheap +//! (~32 bytes per stale entry plus a few cycles on pop), bounded by one entry +//! per parked actor. A `Send` entry is different: running its thunk delivers a +//! real message, so a stale one is *not* inert. `send_after` therefore carries +//! true cancellation via the `armed` set keyed on the entry's `seq`; `pop_due` +//! fires a `Send` only while it is still armed, and `cancel` removes the arm. //! //! Stale pids (slot reused since the timer was inserted) are filtered on //! pop by the scheduler — same convention as the run queue. @@ -51,8 +53,28 @@ pub enum Reason { target: Arc, epoch: u32, }, + /// `send_after`: deliver a message to an address at the deadline, + /// cancellable. The destination (a `Pid` / `Name`) and the message + /// are captured inside `fire`, which resolves the address through the + /// registry and sends *when run* — so a target that died or, for a name, + /// was restarted is observed at fire time, not arm time. A failed resolve + /// or send is dropped (Erlang `erlang:send_after` semantics). + /// + /// Unlike `Sleep` / `WaitTimeout`, a stale `Send` is **not** inert — running + /// the thunk delivers a real message — so these are the only timers that + /// carry true cancellation (the `armed` set on [`Timers`], keyed by the + /// entry's `seq`). `pop_due` fires the thunk only for an entry still armed. + Send { fire: Box }, } +/// Opaque handle to an armed `send_after` timer, returned by +/// [`Timers::insert_send`] and consumed by [`Timers::cancel`]. The inner value +/// is the entry's insertion `seq`; callers must treat it as opaque so the +/// backing structure can change (e.g. a future hierarchical timing wheel) with +/// no API churn. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct TimerId(u64); + /// Callback the scheduler invokes when a `WaitTimeout` entry pops. /// /// Implementors: do not touch `SchedulerState` other than via the public @@ -97,13 +119,21 @@ impl PartialOrd for Entry { pub struct Timers { /// Reverse-wrapped so the smallest deadline is at the top. heap: BinaryHeap>, - /// Monotonic counter for the tiebreaker `seq` field. + /// Monotonic counter for the tiebreaker `seq` field (and the `TimerId` of a + /// `Send` timer — the two are the same value). next_seq: u64, + /// Presence set of *live* `Send` timers, keyed by `seq`. Populated on + /// `insert_send`, removed on fire (in `pop_due`) and on `cancel`. A `Send` + /// entry fires only while present, so a `cancel` that lands before the + /// entry pops prevents delivery; a `cancel` after it has fired finds + /// nothing (the race signal). Bounded by armed-but-not-yet-resolved timers + /// and self-collecting — no sweep. `Sleep` / `WaitTimeout` never touch it. + armed: std::collections::HashSet, } impl Timers { pub fn new() -> Self { - Self { heap: BinaryHeap::new(), next_seq: 0 } + Self { heap: BinaryHeap::new(), next_seq: 0, armed: std::collections::HashSet::new() } } /// Insert a `Sleep` timer. Convenience for the common case. @@ -111,6 +141,33 @@ impl Timers { self.insert(deadline, pid, Reason::Sleep { epoch }); } + /// Arm a cancellable `send_after` timer: run `fire` at `deadline` unless + /// [`cancel`](Self::cancel)led first. `pid` is informational only (the + /// destination, or who armed it — useful for introspection); it is *not* + /// used to wake anyone, the delivery lives entirely inside `fire`. Returns + /// a [`TimerId`] for cancellation. + pub fn insert_send( + &mut self, + deadline: Instant, + pid: Pid, + fire: Box, + ) -> TimerId { + let seq = self.next_seq; + self.next_seq = self.next_seq.wrapping_add(1); + self.armed.insert(seq); + self.heap.push(Reverse(Entry { deadline, seq, pid, reason: Reason::Send { fire } })); + TimerId(seq) + } + + /// Cancel an armed `send_after` timer. Returns `true` if the timer was + /// still armed (delivery is now prevented), `false` if it had already + /// fired or been cancelled. The heap entry, if still pending, is left to be + /// discarded when its deadline passes — `pop_due` drops any `Send` entry + /// whose `seq` is no longer armed. + pub fn cancel(&mut self, id: TimerId) -> bool { + self.armed.remove(&id.0) + } + /// Insert an arbitrary timer entry. pub fn insert(&mut self, deadline: Instant, pid: Pid, reason: Reason) { let seq = self.next_seq; @@ -127,6 +184,7 @@ impl Timers { /// discarded so it can't keep the runtime alive. pub fn clear(&mut self) { self.heap.clear(); + self.armed.clear(); } /// Soonest pending deadline, or `None` if the heap is empty. @@ -136,11 +194,21 @@ impl Timers { /// Pop every entry whose deadline is ≤ `now`, in deadline order. /// The scheduler dispatches each entry by inspecting `entry.reason`. + /// + /// A due `Send` entry is returned only if it is still armed; a cancelled + /// one is silently dropped here (its `seq` was already removed from + /// `armed` by [`cancel`](Self::cancel)). Returning it removes it from + /// `armed`, so a later `cancel` of a fired timer reports `false`. pub fn pop_due(&mut self, now: Instant) -> Vec { let mut out = Vec::new(); while let Some(r) = self.heap.peek() { if r.0.deadline <= now { - out.push(self.heap.pop().unwrap().0); + let entry = self.heap.pop().unwrap().0; + if matches!(entry.reason, Reason::Send { .. }) && !self.armed.remove(&entry.seq) { + // Cancelled before it came due: discard, do not deliver. + continue; + } + out.push(entry); } else { break; } diff --git a/tests/timer.rs b/tests/timer.rs index fdeb942..00d0c01 100644 --- a/tests/timer.rs +++ b/tests/timer.rs @@ -205,3 +205,203 @@ fn same_deadline_entries_pop_in_insertion_order() { let pids: Vec = due.iter().map(|e| e.pid.index()).collect(); assert_eq!(pids, vec![0, 1, 2]); } + +// --------------------------------------------------------------------------- +// send_after / cancel_timer — the message-delivery timer substrate. +// +// Unit tests drive `Timers` directly with a flag-flipping fire thunk (no +// runtime needed, mirroring RecordingTarget above). Integration tests drive +// the public scheduler API and assert real registry-resolved delivery. +// --------------------------------------------------------------------------- + +use std::sync::atomic::{AtomicBool, Ordering}; + +// Pull the Send fire thunk out of a popped entry and run it. +fn run_fire(entry: smarm::timer::Entry) { + match entry.reason { + Reason::Send { fire } => fire(), + _ => panic!("expected a Send entry"), + } +} + +#[test] +fn armed_send_timer_is_returned_and_fires() { + let mut t = Timers::new(); + let now = Instant::now(); + let fired = Arc::new(AtomicBool::new(false)); + let f = fired.clone(); + let _id = t.insert_send( + now + Duration::from_millis(10), + Pid::new(0, 0), + Box::new(move || f.store(true, Ordering::SeqCst)), + ); + + let mut due = t.pop_due(now + Duration::from_millis(20)); + assert_eq!(due.len(), 1, "an armed send timer should pop when due"); + assert!(!fired.load(Ordering::SeqCst), "pop must not fire on its own"); + run_fire(due.pop().unwrap()); + assert!(fired.load(Ordering::SeqCst), "running the thunk delivers"); + assert!(t.is_empty()); +} + +#[test] +fn cancelled_send_timer_is_discarded_not_returned() { + let mut t = Timers::new(); + let now = Instant::now(); + let fired = Arc::new(AtomicBool::new(false)); + let f = fired.clone(); + let id = t.insert_send( + now + Duration::from_millis(10), + Pid::new(0, 0), + Box::new(move || f.store(true, Ordering::SeqCst)), + ); + + assert!(t.cancel(id), "cancel before fire returns true"); + let due = t.pop_due(now + Duration::from_millis(20)); + assert!(due.is_empty(), "a cancelled send timer must not pop"); + assert!(!fired.load(Ordering::SeqCst)); +} + +#[test] +fn cancel_after_fire_returns_false() { + // The race signal Mark wanted: cancelling a timer that already fired tells + // you it was too late. + let mut t = Timers::new(); + let now = Instant::now(); + let id = t.insert_send( + now + Duration::from_millis(5), + Pid::new(0, 0), + Box::new(|| {}), + ); + let due = t.pop_due(now + Duration::from_millis(10)); + assert_eq!(due.len(), 1); + assert!(!t.cancel(id), "cancel after the timer fired returns false"); +} + +#[test] +fn cancel_unknown_id_returns_false() { + let mut t = Timers::new(); + let now = Instant::now(); + let id = t.insert_send(now + Duration::from_millis(5), Pid::new(0, 0), Box::new(|| {})); + assert!(t.cancel(id)); + // Second cancel of the same id: already gone. + assert!(!t.cancel(id)); +} + +#[test] +fn send_timers_interleave_with_sleep_in_deadline_order() { + let mut t = Timers::new(); + let now = Instant::now(); + t.insert_sleep(now + Duration::from_millis(30), Pid::new(0, 0), 1); + let _id = t.insert_send(now + Duration::from_millis(10), Pid::new(1, 0), Box::new(|| {})); + t.insert_sleep(now + Duration::from_millis(20), Pid::new(2, 0), 1); + + let due = t.pop_due(now + Duration::from_millis(50)); + assert_eq!(due.len(), 3); + // 10ms Send, then 20ms Sleep, then 30ms Sleep. + assert!(matches!(due[0].reason, Reason::Send { .. })); + assert_eq!(due[1].pid.index(), 2); + assert_eq!(due[2].pid.index(), 0); +} + +#[test] +fn clear_drops_armed_send_timers() { + let mut t = Timers::new(); + let now = Instant::now(); + let id = t.insert_send(now + Duration::from_millis(10), Pid::new(0, 0), Box::new(|| {})); + t.clear(); + assert!(t.is_empty()); + // The arm record is gone too: cancelling reports nothing to cancel. + assert!(!t.cancel(id)); +} + +// --- Integration: real delivery through the scheduler + registry. --- + +use smarm::{cancel_timer, channel, register, send_after_named, Name}; + +#[test] +fn send_after_named_delivers_after_the_delay() { + const PING: Name = Name::new("send_after_ping"); + run(|| { + let (tx, rx) = channel::(); + register(PING, tx).unwrap(); + let t0 = Instant::now(); + let _id = send_after_named(Duration::from_millis(30), PING, 99); + assert_eq!(rx.recv().unwrap(), 99); + assert!( + t0.elapsed() >= Duration::from_millis(25), + "delivered too early: {:?}", + t0.elapsed() + ); + }); +} + +#[test] +fn cancel_timer_prevents_delivery() { + const C: Name = Name::new("send_after_cancel"); + run(|| { + let (tx, rx) = channel::(); + register(C, tx).unwrap(); + let id = send_after_named(Duration::from_millis(50), C, 7); + assert!(cancel_timer(id), "cancel before fire returns true"); + sleep(Duration::from_millis(90)); + assert_eq!(rx.try_recv(), Ok(None), "cancelled timer delivered anyway"); + }); +} + +#[test] +fn send_after_to_unresolved_name_is_silent() { + const NOPE: Name = Name::new("send_after_nobody_home"); + run(|| { + // Nobody registered NOPE; firing resolves to nothing and is dropped. + let _id = send_after_named(Duration::from_millis(10), NOPE, 1); + sleep(Duration::from_millis(40)); // let it fire and no-op + // Reaching here without a panic is the assertion. + }); +} + +// --- Integration: typed Pid delivery (exercises send_to on fire). --- + +use smarm::{send_after, send_to, spawn_addr, Addressable, Receiver}; + +struct Sink; +impl Addressable for Sink { + type Msg = u64; +} + +#[test] +fn send_after_delivers_to_typed_pid() { + run(|| { + // A reply channel so the test actor learns what Sink received. + let (report_tx, report_rx) = channel::(); + let sink: Pid = spawn_addr::(move |rx: Receiver| { + if let Ok(v) = rx.recv() { + let _ = report_tx.send(v); + } + }); + let _id = send_after(Duration::from_millis(25), sink, 1234); + assert_eq!(report_rx.recv().unwrap(), 1234); + }); +} + +#[test] +fn send_after_to_dead_typed_pid_is_silent() { + run(|| { + // Sink exits immediately after handling one message; arm a second + // delivery for after it's gone. The fire-time send_to returns Dead and + // is dropped — no panic. + let (report_tx, report_rx) = channel::(); + let sink: Pid = spawn_addr::(move |rx: Receiver| { + if let Ok(v) = rx.recv() { + let _ = report_tx.send(v); + } + // body returns -> actor exits + }); + send_to(sink, 1).unwrap(); + assert_eq!(report_rx.recv().unwrap(), 1); // sink has now exited + let _id = send_after(Duration::from_millis(15), sink, 2); + sleep(Duration::from_millis(45)); // let it fire against the dead pid + // No panic, and nothing further delivered. + assert_eq!(report_rx.try_recv(), Ok(None)); + }); +}