diff --git a/src/gen_server.rs b/src/gen_server.rs index f7485ad..aec901f 100644 --- a/src/gen_server.rs +++ b/src/gen_server.rs @@ -49,16 +49,17 @@ //! Revisit against a real consumer. (Monitor `Down` forwarding is dynamic — //! see `handle_down` — because monitors are inherently created at runtime.) -use crate::channel::{channel, select, Receiver, Selectable, Sender}; +use crate::channel::{channel, select, select_timeout, Receiver, RecvTimeoutError, 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::{cancel_timer, request_stop, send_after_to, spawn, spawn_under}; use crate::timer::TimerId; +use std::cell::Cell; use std::collections::HashMap; use std::marker::PhantomData; use std::sync::{Arc, Mutex}; -use std::time::Duration; +use std::time::{Duration, Instant}; /// Behaviour for a gen_server: a state value plus call/cast handlers. /// @@ -279,6 +280,11 @@ enum Sys { pub struct ServerCtx { sys_tx: Sender>, reg: Arc>>, + /// The idle/receive-timeout window, set once via [`idle_after`](Self::idle_after) + /// during `init` and read by the loop after `init` returns. Interior-mutable + /// because `init` holds only `&ctx`; not `Send`, but `ServerCtx` is only ever + /// borrowed on the actor's own stack during `init`, never sent. + idle: Cell>, } impl ServerCtx { @@ -301,6 +307,20 @@ impl ServerCtx { pub fn timer(&self) -> TimerHandle { TimerHandle { sys_tx: self.sys_tx.clone(), reg: self.reg.clone() } } + + /// Set the idle / receive-timeout window: if no message of any kind is + /// dispatched for `after`, the loop fires [`GenServer::handle_idle`] + /// (RFC 015 §4.4). Set once, in `init`; the window is loop-owned and fixed + /// (dynamic per-message reconfiguration is a non-goal). The window resets on + /// every dispatched message and re-arms after `handle_idle` (a steady idle + /// detector); a server wanting one-shot idle-shutdown requests its own stop + /// in `handle_idle`. + /// + /// Distinct from [`ServerRef::call_timeout`], which is a *client-side* call + /// deadline — same word, different axis (§7). + pub fn idle_after(&self, after: Duration) { + self.idle.set(Some(after)); + } } /// Per-server timer bookkeeping, shared between the loop and every @@ -717,30 +737,63 @@ fn server_loop( // loop falls back to the plain-inbox park. let (sys_tx, sys_rx) = channel::>(); // 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. + // on fire, and 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() }); + // Bind the ctx so the idle window set during init can be read back, then + // drop it — that drops the loop's own Sys sender, so a state that cloned no + // Watcher/TimerHandle lets the arm auto-close (the unused-ctx behaviour). + let ctx = ServerCtx { sys_tx, reg: reg.clone(), idle: Cell::new(None) }; + guard.0.init(&ctx); + let idle = ctx.idle.get(); + drop(ctx); let mut monitors: Vec = Vec::new(); let mut sys_open = true; + // The receive-timeout deadline, live only when an idle window is set. Reset + // to `now + idle` after any dispatched message and after handle_idle fires + // (steady detector); `None` disables the timeout entirely. + let mut idle_deadline: Option = idle.map(|d| Instant::now() + d); + let reset_idle = |dl: &mut Option| { + if let Some(d) = idle { + *dl = Some(Instant::now() + d); + } + }; loop { 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() { - Ok(env) => dispatch(&mut guard.0, env), - // All ServerRefs dropped → inbox closed → graceful shutdown. - Err(_) => break, + // Nothing to select over: park on the inbox alone (the pre-v0.8 + // loop), with the idle window as the recv timeout when one is set. + match idle_deadline { + Some(dl) => { + let wait = dl.saturating_duration_since(Instant::now()); + match rx.recv_timeout(wait) { + Ok(env) => { + dispatch(&mut guard.0, env); + reset_idle(&mut idle_deadline); + } + // Quiet for the whole window → the idle event. + Err(RecvTimeoutError::Timeout) => { + guard.0.handle_idle(); + reset_idle(&mut idle_deadline); + } + // All ServerRefs dropped → inbox closed → shutdown. + Err(RecvTimeoutError::Disconnected) => break, + } + } + None => match rx.recv() { + Ok(env) => dispatch(&mut guard.0, env), + Err(_) => break, + }, } } else { // 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. + // The whole wait carries the idle window as its timeout. let nd = monitors.len(); let nw = sys_open as usize; - let i = { + let sel = { let mut arms: Vec<&dyn Selectable> = Vec::with_capacity(nd + nw + infos.len() + 1); for m in &monitors { @@ -753,7 +806,22 @@ fn server_loop( arms.push(r); } arms.push(&rx); - select(&arms) + match idle_deadline { + Some(dl) => { + select_timeout(&arms, dl.saturating_duration_since(Instant::now())) + } + None => Some(select(&arms)), + } + }; + let i = match sel { + Some(i) => i, + // No arm ready for the whole window → idle, then re-arm steady. + None => { + guard.0.handle_idle(); + reset_idle(&mut idle_deadline); + crate::check!(); + continue; + } }; if i < nd { // A Down retires its arm either way: delivered (one-shot) or @@ -761,9 +829,11 @@ fn server_loop( let m = monitors.remove(i); if let Ok(Some(down)) = m.rx.try_recv() { guard.0.handle_down(down); + reset_idle(&mut idle_deadline); } } else if i < nd + nw { match sys_rx.try_recv() { + // Control intake, not a dispatched message: no idle reset. Ok(Some(Sys::Watch(m))) => monitors.push(m), Ok(Some(Sys::Timer(id, msg))) => { // The one-shot fired: retire its registry entry so the @@ -771,6 +841,7 @@ fn server_loop( // dispatch. reg.lock().unwrap().oneshots.remove(&id); guard.0.handle_timer(msg); + reset_idle(&mut idle_deadline); } Ok(Some(Sys::Tick(id))) => { // A periodic fired. Under the lock: if it is still live @@ -798,6 +869,7 @@ fn server_loop( }; if let Some(msg) = msg { guard.0.handle_timer(msg); + reset_idle(&mut idle_deadline); } } // Single-receiver: nothing can drain the arm between @@ -809,7 +881,10 @@ fn server_loop( } else if i < nd + nw + infos.len() { let j = i - nd - nw; match infos[j].try_recv() { - Ok(Some(info)) => guard.0.handle_info(info), + Ok(Some(info)) => { + guard.0.handle_info(info); + reset_idle(&mut idle_deadline); + } Ok(None) => debug_assert!(false, "ready info arm was empty"), // Senders all gone: drop the arm, keep serving. `remove` // (not swap_remove) — order is priority. @@ -819,7 +894,10 @@ fn server_loop( } } else { match rx.try_recv() { - Ok(Some(env)) => dispatch(&mut guard.0, env), + Ok(Some(env)) => { + dispatch(&mut guard.0, env); + reset_idle(&mut idle_deadline); + } Ok(None) => debug_assert!(false, "ready inbox was empty"), Err(_) => break, } diff --git a/tests/gen_server.rs b/tests/gen_server.rs index 61c467c..169466b 100644 --- a/tests/gen_server.rs +++ b/tests/gen_server.rs @@ -580,3 +580,69 @@ fn cancel_stops_a_periodic() { // The periodic had fired at least once before being cancelled. assert!(!fired.lock().unwrap().is_empty()); } + +// --------------------------------------------------------------------------- +// RFC 015 §4.4 — idle / receive timeout. A server that sets an idle window in +// init and counts handle_idle fires; casts are traffic that resets the window. +// --------------------------------------------------------------------------- +struct Idler { + window: Duration, + idles: Arc>, +} + +impl GenServer for Idler { + type Call = (); + type Reply = u32; // idle fire count + type Cast = (); // a poke: traffic that resets the idle window + type Info = (); + type Timer = (); + + fn init(&mut self, ctx: &ServerCtx) { + ctx.idle_after(self.window); + } + fn handle_call(&mut self, _: ()) -> u32 { + *self.idles.lock().unwrap() + } + fn handle_cast(&mut self, _: ()) {} + fn handle_idle(&mut self) { + *self.idles.lock().unwrap() += 1; + } +} + +// A quiet inbox fires handle_idle, and the window re-arms (steady detector): +// several fires across a quiet span. +#[test] +fn idle_fires_repeatedly_on_quiet() { + let idles = Arc::new(Mutex::new(0)); + let i2 = idles.clone(); + run(move || { + let server = start(Idler { window: Duration::from_millis(25), idles: i2 }); + smarm::sleep(Duration::from_millis(130)); // quiet ⇒ ~5 windows + drop(server); // keep the server alive across the quiet span + }); + assert!(*idles.lock().unwrap() >= 2, "idle should re-arm and fire several times"); +} + +// Traffic within the window keeps idle from firing; only once the inbox goes +// quiet does handle_idle fire. +#[test] +fn traffic_resets_the_idle_window() { + let idles = Arc::new(Mutex::new(0)); + let i2 = idles.clone(); + let before_quiet = Arc::new(Mutex::new(u32::MAX)); + let bq = before_quiet.clone(); + run(move || { + let server = start(Idler { window: Duration::from_millis(60), idles: i2 }); + // Poke every 25ms (< 60ms window) for ~100ms: each cast resets the + // window before it can elapse. + for _ in 0..4 { + server.cast(()).unwrap(); + smarm::sleep(Duration::from_millis(25)); + } + *bq.lock().unwrap() = server.call(()).unwrap(); // count while traffic kept it quiet-free + smarm::sleep(Duration::from_millis(140)); // now genuinely quiet + drop(server); + }); + assert_eq!(*before_quiet.lock().unwrap(), 0, "steady traffic must suppress idle"); + assert!(*idles.lock().unwrap() >= 1, "idle fires once the inbox falls quiet"); +}