From 401a1465d51798ba4466dde79983deddf80b345f Mon Sep 17 00:00:00 2001 From: smarm-agent Date: Thu, 18 Jun 2026 18:58:00 +0000 Subject: [PATCH] =?UTF-8?q?gen=5Fserver:=20TimerHandle=20arm=5Fafter/cance?= =?UTF-8?q?l=20+=20Sys::Timer=20dispatch=20(RFC=20015=20=C2=A74.3,=20?= =?UTF-8?q?=C2=A75)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/gen_server.rs | 118 +++++++++++++++++++++++++++++++++++++++----- src/lib.rs | 2 +- src/scheduler.rs | 3 -- src/timer.rs | 11 +++++ tests/gen_server.rs | 92 ++++++++++++++++++++++++++++++++++ 5 files changed, 211 insertions(+), 15 deletions(-) diff --git a/src/gen_server.rs b/src/gen_server.rs index a6eed0c..b18bac3 100644 --- a/src/gen_server.rs +++ b/src/gen_server.rs @@ -53,8 +53,12 @@ use crate::channel::{channel, select, Receiver, Selectable, Sender}; use crate::monitor::{demonitor, monitor, Down, Monitor}; use crate::pid::Pid; use crate::registry::{register_with, resolve_named_sender, RegisterError}; -use crate::scheduler::{request_stop, spawn, spawn_under}; +use crate::scheduler::{cancel_timer, request_stop, send_after_to, spawn, spawn_under}; +use crate::timer::TimerId; +use std::collections::HashMap; use std::marker::PhantomData; +use std::sync::{Arc, Mutex}; +use std::time::Duration; /// Behaviour for a gen_server: a state value plus call/cast handlers. /// @@ -261,28 +265,110 @@ enum Sys { /// A fired one-shot timer carrying its loop-local id (so the loop can retire /// it from the live set) and the server's payload, dispatched to /// [`GenServer::handle_timer`]. - // Constructed by `TimerHandle::arm_after` in the next RFC 015 chunk; until - // then only the `Watch` variant is produced. - #[allow(dead_code)] Timer(crate::timer::TimerId, G::Timer), } -/// The server loop's runtime hook, passed to [`GenServer::init`]. Currently -/// carries only the [`Watcher`]; opaque so fields can grow without breaking. +/// The server loop's runtime hook, passed to [`GenServer::init`]. Hands out the +/// loop's two clonable intake handles — the [`Watcher`] (monitors) and the +/// [`TimerHandle`] (timers) — plus the one-shot idle-window setter. Opaque, so +/// fields can grow without breaking. pub struct ServerCtx { - watcher: Watcher, + sys_tx: Sender>, + reg: Arc>, } impl ServerCtx { /// A clonable handle to the loop's monitor intake. Store it in the state /// during `init` to watch monitors from later handlers. pub fn watcher(&self) -> Watcher { - self.watcher.clone() + Watcher { tx: self.sys_tx.clone() } } /// Shorthand for `ctx.watcher().watch(m)` when watching during `init`. pub fn watch(&self, m: Monitor) { - self.watcher.watch(m) + self.watcher().watch(m) + } + + /// A clonable handle to the loop's timer intake (the [`Watcher`] pattern, + /// for time). Store it on the state during `init` to + /// [`arm_after`](TimerHandle::arm_after) / + /// [`tick_every`](TimerHandle::tick_every) / + /// [`cancel`](TimerHandle::cancel) from any later handler. + pub fn timer(&self) -> TimerHandle { + TimerHandle { sys_tx: self.sys_tx.clone(), reg: self.reg.clone() } + } +} + +/// Per-server timer bookkeeping, shared between the loop and every +/// [`TimerHandle`] clone. A gen_server actor is single-threaded — handle calls +/// (made from inside handlers) and the loop never run concurrently — so this +/// `Mutex` is always uncontended; it is `Arc` rather than `Rc` +/// only because the handle is stored on `self` and `GenServer: Send` forces the +/// handle (hence its shared state) to be `Send`. +#[derive(Default)] +struct TimerReg { + /// Monotonic minter for loop-local [`TimerId`]s — the ids handed to users, + /// kept distinct from the substrate `seq` (which changes on every periodic + /// re-arm). A local id is resolved only through this registry, never passed + /// to [`cancel_timer`], so the two id roles never mix. + next_local: u64, + /// Live one-shot timers: local id → current substrate id. Present from + /// [`arm_after`](TimerHandle::arm_after) until the loop retires it on fire + /// or [`cancel`](TimerHandle::cancel) removes it. + oneshots: HashMap, +} + +impl TimerReg { + /// Mint the next loop-local id. + fn mint(&mut self) -> TimerId { + let id = TimerId::from_raw(self.next_local); + self.next_local = self.next_local.wrapping_add(1); + id + } +} + +/// Arms, cancels, and (chunk 4) periodically ticks server timers, the time-side +/// twin of [`Watcher`] (RFC 015 §4.3). Handed out by [`ServerCtx::timer`] in +/// `init` and stored on the state; later handlers arm timers without any change +/// to their signatures. Clonable; the arm stays open while any clone (or an +/// in-flight fire) lives. +pub struct TimerHandle { + sys_tx: Sender>, + reg: Arc>, +} + +// Manual Clone for the same reason as `Watcher`: no `G: Clone` needed. +impl Clone for TimerHandle { + fn clone(&self) -> Self { + TimerHandle { sys_tx: self.sys_tx.clone(), reg: self.reg.clone() } + } +} + +impl TimerHandle { + /// Arm a one-shot timer: deliver `msg` to [`GenServer::handle_timer`] after + /// `after`, unless [`cancel`](Self::cancel)led first. Returns a loop-local + /// [`TimerId`] for cancellation. A thin wrapper over the `send_after` + /// substrate that lands the fire on the loop's own system arm (above infos + /// and the inbox), not in the inbox. + pub fn arm_after(&self, after: Duration, msg: G::Timer) -> TimerId { + let mut reg = self.reg.lock().unwrap(); + let local = reg.mint(); + // The fire thunk carries the local id so the loop can retire the entry + // on dispatch; `msg` is moved in (no `Clone` needed for one-shots). + let sub = send_after_to(after, self.sys_tx.clone(), Sys::Timer(local, msg)); + reg.oneshots.insert(local, sub); + local + } + + /// Cancel an armed timer. Returns the substrate's race signal — `true` if + /// the cancel beat the fire (delivery is now prevented), `false` if the + /// timer had already fired or been cancelled / is unknown. + pub fn cancel(&self, id: TimerId) -> bool { + let mut reg = self.reg.lock().unwrap(); + if let Some(sub) = reg.oneshots.remove(&id) { + return cancel_timer(sub); + } + false } } @@ -549,7 +635,11 @@ fn server_loop( // Watcher closes the arm, the first select observes the closure, and the // loop falls back to the plain-inbox park. let (sys_tx, sys_rx) = channel::>(); - guard.0.init(&ServerCtx { watcher: Watcher { tx: sys_tx } }); + // Shared timer bookkeeping: the loop keeps one clone (to retire one-shots + // on fire, and — later — to drain on exit); every TimerHandle the state + // stores holds another. + let reg = Arc::new(Mutex::new(TimerReg::default())); + guard.0.init(&ServerCtx { sys_tx, reg: reg.clone() }); let mut monitors: Vec = Vec::new(); let mut sys_open = true; @@ -594,7 +684,13 @@ fn server_loop( } else if i < nd + nw { match sys_rx.try_recv() { Ok(Some(Sys::Watch(m))) => monitors.push(m), - Ok(Some(Sys::Timer(_id, msg))) => guard.0.handle_timer(msg), + Ok(Some(Sys::Timer(id, msg))) => { + // The one-shot fired: retire its registry entry so the + // live set tracks only still-pending timers, then + // dispatch. + reg.lock().unwrap().oneshots.remove(&id); + guard.0.handle_timer(msg); + } // Single-receiver: nothing can drain the arm between // select's ready and our try_recv. Ok(None) => debug_assert!(false, "ready system arm was empty"), diff --git a/src/lib.rs b/src/lib.rs index 2bd46f1..1793ec8 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -52,7 +52,7 @@ pub use channel::{ }; pub use gen_server::{ call, cast, shutdown, whereis_server, CallError, CallTimeoutError, CastError, GenServer, - NamedServerBuilder, ServerBuilder, ServerCtx, ServerName, ServerRef, Watcher, + NamedServerBuilder, ServerBuilder, ServerCtx, ServerName, ServerRef, TimerHandle, Watcher, }; pub use link::{link, trap_exit, unlink, ExitSignal}; pub use monitor::{demonitor, monitor, Down, DownReason, Monitor, MonitorId}; diff --git a/src/scheduler.rs b/src/scheduler.rs index 215680b..861ffc9 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -443,9 +443,6 @@ pub fn send_after_named( /// send onto a channel whose receiver is gone is dropped, exactly as a failed /// address resolve is (Erlang `send_after` semantics). `pid` is informational /// (who armed it); delivery lives entirely in the thunk. -// Wired into `TimerHandle::arm_after` in the RFC 015 timer chunk; until then it -// is referenced only from tests, so the non-test build would flag it dead. -#[allow(dead_code)] pub(crate) fn send_after_to( after: std::time::Duration, tx: Sender, diff --git a/src/timer.rs b/src/timer.rs index 118d1f5..1d1c47d 100644 --- a/src/timer.rs +++ b/src/timer.rs @@ -75,6 +75,17 @@ pub enum Reason { #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub struct TimerId(u64); +impl TimerId { + /// Wrap a raw value. Crate-internal: the gen_server timer layer mints its + /// own loop-local `TimerId`s (the public ids it hands out, decoupled from + /// the per-re-arm substrate `seq`) and maps them to live substrate ids. + /// These local ids are only ever resolved through that layer's registry — + /// never passed back to [`Timers::cancel`] — so the two id roles do not mix. + pub(crate) fn from_raw(v: u64) -> Self { + TimerId(v) + } +} + /// Callback the scheduler invokes when a `WaitTimeout` entry pops. /// /// Implementors: do not touch `SchedulerState` other than via the public diff --git a/tests/gen_server.rs b/tests/gen_server.rs index 2349173..164eb74 100644 --- a/tests/gen_server.rs +++ b/tests/gen_server.rs @@ -443,3 +443,95 @@ fn unused_ctx_closes_control_arm_silently() { }); assert_eq!(*got.lock().unwrap(), 42); } + +// --------------------------------------------------------------------------- +// RFC 015 — gen_server timers. A server that arms one-shot timers from a cast +// and records each fire's payload, plus the cancel race signal. +// --------------------------------------------------------------------------- +use smarm::gen_server::{ServerCtx, TimerHandle}; +use smarm::TimerId; + +enum TkCast { + Arm(Duration), + CancelLast, +} + +struct Timed { + timer: Option>, + fired: Arc>>, + cancel_won: Arc>>, + last: Option, +} + +impl GenServer for Timed { + type Call = (); + type Reply = usize; // count of fires so far (a sync read point) + type Cast = TkCast; + type Info = (); + type Timer = u32; + + fn init(&mut self, ctx: &ServerCtx) { + self.timer = Some(ctx.timer()); + } + + fn handle_call(&mut self, _: ()) -> usize { + self.fired.lock().unwrap().len() + } + + fn handle_cast(&mut self, c: TkCast) { + let t = self.timer.as_ref().expect("init ran first"); + match c { + TkCast::Arm(d) => self.last = Some(t.arm_after(d, 7)), + TkCast::CancelLast => { + let id = self.last.take().expect("a timer was armed"); + *self.cancel_won.lock().unwrap() = Some(t.cancel(id)); + } + } + } + + fn handle_timer(&mut self, msg: u32) { + self.fired.lock().unwrap().push(msg); + } +} + +fn timed(fired: Arc>>, cancel_won: Arc>>) -> Timed { + Timed { timer: None, fired, cancel_won, last: None } +} + +// A one-shot armed from a handler fires into handle_timer with its payload. +#[test] +fn arm_after_fires_into_handle_timer() { + let fired = Arc::new(Mutex::new(Vec::new())); + let f2 = fired.clone(); + run(move || { + let cw = Arc::new(Mutex::new(None)); + let server = start(timed(f2, cw)); + server.cast(TkCast::Arm(Duration::from_millis(10))).unwrap(); + let _ = server.call(()).unwrap(); // sync: arm done + smarm::sleep(Duration::from_millis(40)); // let the timer fire + let count = server.call(()).unwrap(); // timer arm outranks this inbox call + assert_eq!(count, 1, "the one-shot should have fired exactly once"); + }); + assert_eq!(*fired.lock().unwrap(), vec![7]); +} + +// cancel before the deadline wins the race (returns true) and suppresses the +// fire entirely. +#[test] +fn cancel_before_fire_suppresses_it() { + let fired = Arc::new(Mutex::new(Vec::new())); + let cancel_won = Arc::new(Mutex::new(None)); + let f2 = fired.clone(); + let c2 = cancel_won.clone(); + run(move || { + let server = start(timed(f2, c2)); + server.cast(TkCast::Arm(Duration::from_millis(50))).unwrap(); + server.cast(TkCast::CancelLast).unwrap(); + let _ = server.call(()).unwrap(); // sync: arm + cancel both handled + smarm::sleep(Duration::from_millis(80)); // past the original deadline + let count = server.call(()).unwrap(); + assert_eq!(count, 0, "cancelled timer must not fire"); + }); + assert_eq!(*cancel_won.lock().unwrap(), Some(true), "cancel beat the fire"); + assert!(fired.lock().unwrap().is_empty()); +}