From e5d1b3b54b8e814f544fcfe4c6c662565b8b74f2 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 10 Jun 2026 14:56:51 +0000 Subject: [PATCH] =?UTF-8?q?feat(gen=5Fserver):=20handle=5Finfo=20=E2=80=94?= =?UTF-8?q?=20static=20info=20arms=20selected=20ahead=20of=20the=20inbox?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit type Info + handle_info (no-op default) on GenServer; ServerBuilder (with_info/under/start) so start variants don't multiply — start/start_under stay as wrappers. The loop selects [infos.., inbox] in priority order when info arms exist, and keeps the plain recv() park when none do. Closed info arms are silently removed (closed-arm-is-ready-forever); a closed inbox still means graceful shutdown. --- src/gen_server.rs | 169 ++++++++++++++++++++++++++++++++++++-------- src/lib.rs | 2 +- tests/gen_server.rs | 112 ++++++++++++++++++++++++++++- 3 files changed, 250 insertions(+), 33 deletions(-) diff --git a/src/gen_server.rs b/src/gen_server.rs index 75446f1..d8e09de 100644 --- a/src/gen_server.rs +++ b/src/gen_server.rs @@ -6,14 +6,23 @@ //! [`call`](ServerRef::call) (synchronous, returns a reply) or //! [`cast`](ServerRef::cast) (fire-and-forget). //! -//! ## Why a single inbox +//! ## One inbox, many arms //! -//! smarm has no `select` and no unified per-process mailbox (see the runtime -//! gotchas in `task.md`), so a server cannot wait on a "call channel" and a -//! "cast channel" simultaneously. Instead every message — call or cast — -//! travels the *same* inbox channel as an [`Envelope`], and the server loop -//! dispatches by variant. A `call` carries a freshly made one-shot reply -//! channel; the server sends the reply straight back down it. +//! Every call and cast travels the *same* inbox channel as an [`Envelope`] +//! (calls and casts are ordered relative to each other, like Erlang); the +//! server loop dispatches by variant. A `call` carries a freshly made +//! one-shot reply channel; the server sends the reply straight back down it. +//! +//! Out-of-band messages ride *separate* channels composed at the wait via +//! [`select`](crate::channel::select): info channels handed over at start +//! ([`ServerBuilder::with_info`]) are dispatched to +//! [`handle_info`](GenServer::handle_info). Arm priority is +//! **infos before inbox**, in declaration order — a hot inbox cannot starve +//! an out-of-band message; conversely a hot info channel CAN starve the +//! inbox, deliberately (system-message semantics). An info channel whose +//! senders are all gone is silently dropped from the arm set (per the +//! closed-arm-is-ready-forever rule on `select`); a closed *inbox* still +//! means graceful shutdown. //! //! ## Server death //! @@ -36,11 +45,11 @@ //! //! ## Not here (yet) //! -//! No `handle_info`: it needs the cross-channel mailbox merge that selective -//! receive deliberately left unmade. (Call timeouts landed as -//! [`ServerRef::call_timeout`], on top of `Receiver::recv_timeout`.) +//! No dynamic info subscription: the info-channel set is fixed at start. +//! Revisit against a real consumer. (Monitor `Down` forwarding is dynamic — +//! see `handle_down` — because monitors are inherently created at runtime.) -use crate::channel::{channel, Receiver, Sender}; +use crate::channel::{channel, select, Receiver, Selectable, Sender}; use crate::pid::Pid; use crate::scheduler::{spawn, spawn_under}; @@ -56,6 +65,11 @@ pub trait GenServer: Send + 'static { type Reply: Send + 'static; /// Asynchronous request type (carried by [`ServerRef::cast`]). type Cast: Send + 'static; + /// Out-of-band message type, delivered to [`handle_info`](Self::handle_info) + /// from the info channels registered at start + /// ([`ServerBuilder::with_info`]). Servers with several out-of-band + /// sources enum them up into one `Info`. Use `()` if unused. + type Info: Send + 'static; /// Runs once inside the server actor before any message is handled. fn init(&mut self) {} @@ -66,6 +80,10 @@ pub trait GenServer: Send + 'static { /// Handle a fire-and-forget cast. fn handle_cast(&mut self, request: Self::Cast); + /// Handle an out-of-band message from one of the info channels. Default: + /// drop it. + fn handle_info(&mut self, _info: Self::Info) {} + /// Runs as the server actor exits, on any exit path (see module docs). fn terminate(&mut self) {} } @@ -172,24 +190,72 @@ impl ServerRef { } } +/// Configure-then-start construction for servers that need more than a bare +/// [`start`]: info channels, a supervisor. Consumed by [`start`](Self::start). +/// +/// ```ignore +/// let server = ServerBuilder::new(state) +/// .with_info(events_rx) +/// .under(supervisor_pid) +/// .start(); +/// ``` +pub struct ServerBuilder { + state: G, + infos: Vec>, + supervisor: Option, +} + +impl ServerBuilder { + pub fn new(state: G) -> Self { + ServerBuilder { state, infos: Vec::new(), supervisor: None } + } + + /// Add an out-of-band channel; messages arriving on it are dispatched to + /// [`GenServer::handle_info`]. Repeatable; arm priority is call order + /// (earlier = higher), and every info channel outranks the inbox. + pub fn with_info(mut self, rx: Receiver) -> Self { + self.infos.push(rx); + self + } + + /// Spawn the server under an explicit supervisor pid (via [`spawn_under`]) + /// so it slots into the supervision tree. + pub fn under(mut self, supervisor: Pid) -> Self { + self.supervisor = Some(supervisor); + self + } + + /// Spawn the server actor and hand back its [`ServerRef`]. The server's + /// lifetime is governed by its refs, not by joining, so the backing join + /// handle is dropped. + pub fn start(self) -> ServerRef { + let (tx, rx) = channel::>(); + let ServerBuilder { state, infos, supervisor } = self; + let handle = match supervisor { + Some(sup) => spawn_under(sup, move || server_loop::(rx, state, infos)), + None => spawn(move || server_loop::(rx, state, infos)), + }; + ServerRef { tx, pid: handle.pid() } + } +} + /// Spawn `state` as a server under the current actor (via [`spawn`]). Returns a -/// [`ServerRef`]; the server's lifetime is governed by its refs, not by -/// joining, so the backing join handle is dropped. +/// [`ServerRef`]. Shorthand for `ServerBuilder::new(state).start()`. pub fn start(state: G) -> ServerRef { - let (tx, rx) = channel::>(); - let handle = spawn(move || server_loop::(rx, state)); - ServerRef { tx, pid: handle.pid() } + ServerBuilder::new(state).start() } /// Like [`start`], but spawns the server under an explicit supervisor pid (via /// [`spawn_under`]) so it slots into the supervision tree. pub fn start_under(supervisor: Pid, state: G) -> ServerRef { - let (tx, rx) = channel::>(); - let handle = spawn_under(supervisor, move || server_loop::(rx, state)); - ServerRef { tx, pid: handle.pid() } + ServerBuilder::new(state).under(supervisor).start() } -fn server_loop(rx: Receiver>, state: G) { +fn server_loop( + rx: Receiver>, + state: G, + mut infos: Vec>, +) { // terminate() must run on every exit path (clean close, panic, stop), so it // lives in this guard's Drop rather than after the loop. struct Terminate(G); @@ -199,22 +265,63 @@ fn server_loop(rx: Receiver>, state: G) { } } - let mut guard = Terminate(state); - guard.0.init(); - - loop { - match rx.recv() { - Ok(Envelope::Call(request, reply_tx)) => { - let reply = guard.0.handle_call(request); + fn dispatch(state: &mut G, env: Envelope) { + match env { + Envelope::Call(request, reply_tx) => { + let reply = state.handle_call(request); // The caller may have gone away (e.g. cancelled while parked); // a failed reply send is not the server's problem. let _ = reply_tx.send(reply); } - Ok(Envelope::Cast(request)) => guard.0.handle_cast(request), - // All ServerRefs dropped → inbox closed → graceful shutdown. - Err(_) => break, + Envelope::Cast(request) => state.handle_cast(request), } - // Observation point so a server whose inbox is never empty stays + } + + let mut guard = Terminate(state); + guard.0.init(); + + loop { + if 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, + } + } else { + // Arm priority: infos (declaration order), then the inbox. The + // arm slice is rebuilt per iteration because `infos` shrinks as + // arms close. + let i = { + let mut arms: Vec<&dyn Selectable> = Vec::with_capacity(infos.len() + 1); + for r in &infos { + arms.push(r); + } + arms.push(&rx); + select(&arms) + }; + if i < infos.len() { + match infos[i].try_recv() { + Ok(Some(info)) => guard.0.handle_info(info), + // Single-receiver: nothing can drain the arm between + // select's ready and our try_recv. + 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. + Err(_) => { + infos.remove(i); + } + } + } else { + match rx.try_recv() { + Ok(Some(env)) => dispatch(&mut guard.0, env), + Ok(None) => debug_assert!(false, "ready inbox was empty"), + Err(_) => break, + } + } + } + // Observation point so a server whose arms are never empty stays // preemptible and cancellable. crate::check!(); } diff --git a/src/lib.rs b/src/lib.rs index be57621..b78407c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -46,7 +46,7 @@ static ALLOCATOR: preempt::PreemptingAllocator = preempt::PreemptingAllocator; // --------------------------------------------------------------------------- pub use channel::{channel, select, select_timeout, Receiver, RecvError, RecvTimeoutError, Selectable, Sender}; -pub use gen_server::{CallError, CallTimeoutError, CastError, GenServer, ServerRef}; +pub use gen_server::{CallError, CallTimeoutError, CastError, GenServer, ServerBuilder, ServerRef}; pub use link::{link, trap_exit, unlink, ExitSignal}; pub use monitor::{demonitor, monitor, Down, DownReason, Monitor, MonitorId}; pub use mutex::{LockTimeout, Mutex, MutexGuard}; diff --git a/tests/gen_server.rs b/tests/gen_server.rs index 6479ab9..35f7bba 100644 --- a/tests/gen_server.rs +++ b/tests/gen_server.rs @@ -1,7 +1,7 @@ //! gen_server tests: call round-trip, cast, lifecycle callbacks, and the two //! server-down detection paths (reply-channel close vs. inbox-send failure). -use smarm::gen_server::{start, CallError, GenServer}; +use smarm::gen_server::{start, CallError, GenServer, ServerBuilder}; use smarm::run; use std::sync::{Arc, Mutex}; @@ -26,6 +26,7 @@ impl GenServer for Counter { type Call = Req; type Reply = i64; type Cast = Op; + type Info = (); fn handle_call(&mut self, req: Req) -> i64 { match req { @@ -68,6 +69,7 @@ impl GenServer for Lifecycle { type Call = (); type Reply = (); type Cast = (); + type Info = (); fn init(&mut self) { self.log.lock().unwrap().push("init"); @@ -151,6 +153,7 @@ impl GenServer for Slow { type Call = u64; // delay in ms type Reply = u64; type Cast = (); + type Info = (); fn handle_call(&mut self, delay_ms: u64) -> u64 { if delay_ms > 0 { @@ -206,6 +209,7 @@ fn call_timeout_to_dead_server_is_server_down_not_timeout() { struct Bomb; impl GenServer for Bomb { type Call = (); + type Info = (); type Reply = (); type Cast = (); fn handle_call(&mut self, _: ()) { @@ -229,3 +233,109 @@ fn call_timeout_to_dead_server_is_server_down_not_timeout() { ); }); } + +// --------------------------------------------------------------------------- +// handle_info: out-of-band channels selected alongside the inbox (v0.8) +// --------------------------------------------------------------------------- + +/// Logs every message it handles, in order; a call reads the log back. +struct Logger { + log: Vec<&'static str>, +} + +impl GenServer for Logger { + type Call = (); + type Reply = Vec<&'static str>; + type Cast = (); + type Info = &'static str; + + fn handle_call(&mut self, _: ()) -> Vec<&'static str> { + self.log.clone() + } + + fn handle_cast(&mut self, _: ()) { + self.log.push("cast"); + } + + fn handle_info(&mut self, info: &'static str) { + self.log.push(info); + } +} + +// An info message is dispatched to handle_info, interleaved with normal +// service. +#[test] +fn info_is_dispatched() { + let got = Arc::new(Mutex::new(Vec::new())); + let got2 = got.clone(); + run(move || { + let (info_tx, info_rx) = smarm::channel::<&'static str>(); + let server = ServerBuilder::new(Logger { log: Vec::new() }) + .with_info(info_rx) + .start(); + info_tx.send("info").unwrap(); + *got2.lock().unwrap() = server.call(()).unwrap(); + }); + assert_eq!(*got.lock().unwrap(), vec!["info"]); +} + +// Arm priority: with a cast AND an info both queued before the server first +// runs, the info is handled first — info arms outrank the inbox. Relies on +// run()'s deterministic single-thread ordering. +#[test] +fn info_outranks_inbox() { + let got = Arc::new(Mutex::new(Vec::new())); + let got2 = got.clone(); + run(move || { + let (info_tx, info_rx) = smarm::channel::<&'static str>(); + let server = ServerBuilder::new(Logger { log: Vec::new() }) + .with_info(info_rx) + .start(); + // The server actor hasn't run yet: both messages are queued before + // its first select. Inbox first in *send* order, info first in *arm* + // order — arm order must win. + server.cast(()).unwrap(); + info_tx.send("info").unwrap(); + *got2.lock().unwrap() = server.call(()).unwrap(); + }); + assert_eq!(*got.lock().unwrap(), vec!["info", "cast"]); +} + +// Two info channels: declaration order is priority order. +#[test] +fn info_arms_keep_declaration_priority() { + let got = Arc::new(Mutex::new(Vec::new())); + let got2 = got.clone(); + run(move || { + let (hi_tx, hi_rx) = smarm::channel::<&'static str>(); + let (lo_tx, lo_rx) = smarm::channel::<&'static str>(); + let server = ServerBuilder::new(Logger { log: Vec::new() }) + .with_info(hi_rx) + .with_info(lo_rx) + .start(); + // Sent low-priority first; handled high-priority first. + lo_tx.send("lo").unwrap(); + hi_tx.send("hi").unwrap(); + *got2.lock().unwrap() = server.call(()).unwrap(); + }); + assert_eq!(*got.lock().unwrap(), vec!["hi", "lo"]); +} + +// A closed info arm is silently dropped and the server keeps serving; the +// closure does NOT reach handle_info and does NOT starve the inbox (the +// closed-arm-is-ready-forever gotcha). +#[test] +fn closed_info_arm_is_dropped_silently() { + let got = Arc::new(Mutex::new(Vec::new())); + let got2 = got.clone(); + run(move || { + let (info_tx, info_rx) = smarm::channel::<&'static str>(); + let server = ServerBuilder::new(Logger { log: Vec::new() }) + .with_info(info_rx) + .start(); + drop(info_tx); // closed before the server's first select + server.cast(()).unwrap(); + *got2.lock().unwrap() = server.call(()).unwrap(); + }); + assert_eq!(*got.lock().unwrap(), vec!["cast"]); +}