From 57eadb5c6cfddddcfddfbed8dbe83afc236a8e88 Mon Sep 17 00:00:00 2001 From: smarm-agent Date: Thu, 18 Jun 2026 18:54:24 +0000 Subject: [PATCH] =?UTF-8?q?gen=5Fserver:=20type=20Timer=20+=20handle=5Ftim?= =?UTF-8?q?er/handle=5Fidle;=20fold=20control=20into=20Sys=20channel=20(RF?= =?UTF-8?q?C=20015=20=C2=A74.5,=20=C2=A76)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- examples/named_genserver.rs | 1 + src/gen_server.rs | 113 +++++++++++++++++++++++++++--------- tests/gen_server.rs | 12 +++- 3 files changed, 94 insertions(+), 32 deletions(-) diff --git a/examples/named_genserver.rs b/examples/named_genserver.rs index 379987e..d99ec01 100644 --- a/examples/named_genserver.rs +++ b/examples/named_genserver.rs @@ -24,6 +24,7 @@ impl GenServer for Counter { type Reply = u64; type Cast = Update; type Info = (); + type Timer = (); fn handle_call(&mut self, q: Query) -> u64 { match q { diff --git a/src/gen_server.rs b/src/gen_server.rs index b079425..a6eed0c 100644 --- a/src/gen_server.rs +++ b/src/gen_server.rs @@ -73,12 +73,25 @@ pub trait GenServer: Send + 'static { /// ([`ServerBuilder::with_info`]). Servers with several out-of-band /// sources enum them up into one `Info`. Use `()` if unused. type Info: Send + 'static; + /// The server's own scheduled-timer payload, delivered to + /// [`handle_timer`](Self::handle_timer) when a timer armed via the loop's + /// [`TimerHandle`] fires (RFC 015). Kept distinct from + /// [`Info`](Self::Info) — `Info` is *external* out-of-band traffic, `Timer` + /// is the server's *own* fires — so the system/userspace split stays + /// legible and a tagged timer round-trips + /// (`arm_after(d, Tk::Retry(n))` → `handle_timer(Tk::Retry(n))`). Use `()` + /// if unused, exactly as `Info` does. + type Timer: Send + 'static; /// Runs once inside the server actor before any message is handled. The /// [`ServerCtx`] is the loop's one runtime hook: clone its [`Watcher`] /// into the state here to be able to [`watch`](Watcher::watch) monitors /// from any later handler. - fn init(&mut self, _ctx: &ServerCtx) {} + fn init(&mut self, _ctx: &ServerCtx) + where + Self: Sized, + { + } /// Handle a synchronous call and produce the reply sent back to the caller. fn handle_call(&mut self, request: Self::Call) -> Self::Reply; @@ -94,6 +107,21 @@ pub trait GenServer: Send + 'static { /// [`Watcher::watch`]. Default: drop it. fn handle_down(&mut self, _down: Down) {} + /// Handle a fired timer armed through the loop's [`TimerHandle`] + /// ([`arm_after`](TimerHandle::arm_after) / + /// [`tick_every`](TimerHandle::tick_every)). Timer fires re-enter at system + /// priority — above infos and the inbox — so a heartbeat cannot be starved + /// by userspace traffic. Default: drop it (RFC 015 §6). + fn handle_timer(&mut self, _msg: Self::Timer) {} + + /// Handle a receive/idle timeout: fired when the loop has waited a full + /// idle window (set once via [`ServerCtx::idle_after`]) with no message of + /// any kind dispatched. The window resets on every dispatched message and + /// re-arms after this fires (a steady idle detector); a server wanting + /// one-shot idle-shutdown simply requests its own stop here. Default: no-op + /// (RFC 015 §4.4, §6). + fn handle_idle(&mut self) {} + /// Runs as the server actor exits, on any exit path (see module docs). fn terminate(&mut self) {} } @@ -221,16 +249,34 @@ impl ServerRef { } } -/// The server loop's runtime hook, passed to [`GenServer::init`]. Currently -/// carries only the [`Watcher`]; opaque so fields can grow without breaking. -pub struct ServerCtx { - watcher: Watcher, +/// The server loop's internal *system intake* channel (RFC 015 §4.5). Control +/// (monitor handoff) and armed-timer fires are structurally the same — both are +/// loop-internal sources selected above the inbox — so they fold into one +/// channel, dispatched by variant. The arm stays open while any [`Watcher`] or +/// [`TimerHandle`] clone (or an in-flight fire thunk) still holds a sender. +enum Sys { + /// A monitor handed in via [`Watcher::watch`]; pushed onto the loop's + /// monitor set. + Watch(Monitor), + /// 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), } -impl ServerCtx { +/// The server loop's runtime hook, passed to [`GenServer::init`]. Currently +/// carries only the [`Watcher`]; opaque so fields can grow without breaking. +pub struct ServerCtx { + watcher: Watcher, +} + +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 { + pub fn watcher(&self) -> Watcher { self.watcher.clone() } @@ -244,17 +290,24 @@ impl ServerCtx { /// [`GenServer::handle_down`]. Down arms outrank every other arm (a death /// notice cannot be starved), and a delivered `Down` retires its arm — /// monitors are one-shot. -#[derive(Clone)] -pub struct Watcher { - tx: Sender, +pub struct Watcher { + tx: Sender>, } -impl Watcher { +// Manual Clone: deriving would demand `G: Clone`, but the handle is clonable +// regardless of the server type (it clones only the inner sender). +impl Clone for Watcher { + fn clone(&self) -> Self { + Watcher { tx: self.tx.clone() } + } +} + +impl Watcher { /// Transfer `m` to the server loop. If the server is already gone the /// monitor is silently dropped (its queued `Down`, if any, with it) — /// there is no loop left to care. pub fn watch(&self, m: Monitor) { - let _ = self.tx.send(m); + let _ = self.tx.send(Sys::Watch(m)); } } @@ -490,18 +543,19 @@ fn server_loop( let mut guard = Terminate(state); - // The control arm: Watchers feed Monitors to the loop through it. The - // ctx (and with it the loop's own sender) drops right after init — a - // state that didn't clone the Watcher closes the arm, the first select - // observes the closure, and the loop falls back to the plain-inbox park. - let (watch_tx, watch_rx) = channel::(); - guard.0.init(&ServerCtx { watcher: Watcher { tx: watch_tx } }); + // The system intake arm: Watchers feed Monitors and (next chunk) armed + // timers feed fires to the loop through it. The ctx (and with it the loop's + // own sender) drops right after init — a state that didn't clone the + // 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 } }); let mut monitors: Vec = Vec::new(); - let mut watch_open = true; + let mut sys_open = true; loop { - if monitors.is_empty() && !watch_open && infos.is_empty() { + if monitors.is_empty() && !sys_open && infos.is_empty() { // Nothing to select over: park on the inbox alone, exactly the // pre-v0.8 loop. match rx.recv() { @@ -510,19 +564,19 @@ fn server_loop( Err(_) => break, } } else { - // Arm priority: downs, then the control arm, then infos (each in + // Arm priority: downs, then the system arm, then infos (each in // declaration order), then the inbox. The arm slice is rebuilt // per iteration because every set but the inbox shrinks or grows. let nd = monitors.len(); - let nw = watch_open as usize; + let nw = sys_open as usize; let i = { let mut arms: Vec<&dyn Selectable> = Vec::with_capacity(nd + nw + infos.len() + 1); for m in &monitors { arms.push(&m.rx); } - if watch_open { - arms.push(&watch_rx); + if sys_open { + arms.push(&sys_rx); } for r in &infos { arms.push(r); @@ -538,13 +592,14 @@ fn server_loop( guard.0.handle_down(down); } } else if i < nd + nw { - match watch_rx.try_recv() { - Ok(Some(m)) => monitors.push(m), + match sys_rx.try_recv() { + Ok(Some(Sys::Watch(m))) => monitors.push(m), + Ok(Some(Sys::Timer(_id, msg))) => 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 control arm was empty"), - // Every Watcher gone: stop selecting on the arm. - Err(_) => watch_open = false, + Ok(None) => debug_assert!(false, "ready system arm was empty"), + // Every Watcher/TimerHandle gone: stop selecting on the arm. + Err(_) => sys_open = false, } } else if i < nd + nw + infos.len() { let j = i - nd - nw; diff --git a/tests/gen_server.rs b/tests/gen_server.rs index cc042ab..2349173 100644 --- a/tests/gen_server.rs +++ b/tests/gen_server.rs @@ -27,6 +27,7 @@ impl GenServer for Counter { type Reply = i64; type Cast = Op; type Info = (); + type Timer = (); fn handle_call(&mut self, req: Req) -> i64 { match req { @@ -70,8 +71,9 @@ impl GenServer for Lifecycle { type Reply = (); type Cast = (); type Info = (); + type Timer = (); - fn init(&mut self, _ctx: &smarm::gen_server::ServerCtx) { + fn init(&mut self, _ctx: &smarm::gen_server::ServerCtx) { self.log.lock().unwrap().push("init"); } @@ -154,6 +156,7 @@ impl GenServer for Slow { type Reply = u64; type Cast = (); type Info = (); + type Timer = (); fn handle_call(&mut self, delay_ms: u64) -> u64 { if delay_ms > 0 { @@ -210,6 +213,7 @@ fn call_timeout_to_dead_server_is_server_down_not_timeout() { impl GenServer for Bomb { type Call = (); type Info = (); + type Timer = (); type Reply = (); type Cast = (); fn handle_call(&mut self, _: ()) { @@ -248,6 +252,7 @@ impl GenServer for Logger { type Reply = Vec<&'static str>; type Cast = (); type Info = &'static str; + type Timer = (); fn handle_call(&mut self, _: ()) -> Vec<&'static str> { self.log.clone() @@ -350,7 +355,7 @@ use smarm::{monitor, spawn, DownReason, Pid}; /// The motivating pattern: a server that spawns workers from a handler, /// watches them, and logs their deaths. struct Pool { - watcher: Option, + watcher: Option>, log: Vec, } @@ -364,8 +369,9 @@ impl GenServer for Pool { type Reply = Vec; type Cast = PoolCast; type Info = (); + type Timer = (); - fn init(&mut self, ctx: &smarm::gen_server::ServerCtx) { + fn init(&mut self, ctx: &smarm::gen_server::ServerCtx) { self.watcher = Some(ctx.watcher()); }