feat(gen_server): handle_info — static info arms selected ahead of the inbox

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.
This commit is contained in:
Claude
2026-06-10 14:56:51 +00:00
parent e545f818fa
commit e5d1b3b54b
3 changed files with 250 additions and 33 deletions
+136 -29
View File
@@ -6,14 +6,23 @@
//! [`call`](ServerRef::call) (synchronous, returns a reply) or //! [`call`](ServerRef::call) (synchronous, returns a reply) or
//! [`cast`](ServerRef::cast) (fire-and-forget). //! [`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 //! Every call and cast travels the *same* inbox channel as an [`Envelope`]
//! gotchas in `task.md`), so a server cannot wait on a "call channel" and a //! (calls and casts are ordered relative to each other, like Erlang); the
//! "cast channel" simultaneously. Instead every message — call or cast — //! server loop dispatches by variant. A `call` carries a freshly made
//! travels the *same* inbox channel as an [`Envelope`], and the server loop //! one-shot reply channel; the server sends the reply straight back down it.
//! 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 //! ## Server death
//! //!
@@ -36,11 +45,11 @@
//! //!
//! ## Not here (yet) //! ## Not here (yet)
//! //!
//! No `handle_info`: it needs the cross-channel mailbox merge that selective //! No dynamic info subscription: the info-channel set is fixed at start.
//! receive deliberately left unmade. (Call timeouts landed as //! Revisit against a real consumer. (Monitor `Down` forwarding is dynamic —
//! [`ServerRef::call_timeout`], on top of `Receiver::recv_timeout`.) //! 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::pid::Pid;
use crate::scheduler::{spawn, spawn_under}; use crate::scheduler::{spawn, spawn_under};
@@ -56,6 +65,11 @@ pub trait GenServer: Send + 'static {
type Reply: Send + 'static; type Reply: Send + 'static;
/// Asynchronous request type (carried by [`ServerRef::cast`]). /// Asynchronous request type (carried by [`ServerRef::cast`]).
type Cast: Send + 'static; 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. /// Runs once inside the server actor before any message is handled.
fn init(&mut self) {} fn init(&mut self) {}
@@ -66,6 +80,10 @@ pub trait GenServer: Send + 'static {
/// Handle a fire-and-forget cast. /// Handle a fire-and-forget cast.
fn handle_cast(&mut self, request: Self::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). /// Runs as the server actor exits, on any exit path (see module docs).
fn terminate(&mut self) {} fn terminate(&mut self) {}
} }
@@ -172,24 +190,72 @@ impl<G: GenServer> ServerRef<G> {
} }
} }
/// Spawn `state` as a server under the current actor (via [`spawn`]). Returns a /// Configure-then-start construction for servers that need more than a bare
/// [`ServerRef`]; the server's lifetime is governed by its refs, not by /// [`start`]: info channels, a supervisor. Consumed by [`start`](Self::start).
/// joining, so the backing join handle is dropped. ///
pub fn start<G: GenServer>(state: G) -> ServerRef<G> { /// ```ignore
/// let server = ServerBuilder::new(state)
/// .with_info(events_rx)
/// .under(supervisor_pid)
/// .start();
/// ```
pub struct ServerBuilder<G: GenServer> {
state: G,
infos: Vec<Receiver<G::Info>>,
supervisor: Option<Pid>,
}
impl<G: GenServer> ServerBuilder<G> {
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<G::Info>) -> 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<G> {
let (tx, rx) = channel::<Envelope<G>>(); let (tx, rx) = channel::<Envelope<G>>();
let handle = spawn(move || server_loop::<G>(rx, state)); let ServerBuilder { state, infos, supervisor } = self;
let handle = match supervisor {
Some(sup) => spawn_under(sup, move || server_loop::<G>(rx, state, infos)),
None => spawn(move || server_loop::<G>(rx, state, infos)),
};
ServerRef { tx, pid: handle.pid() } ServerRef { tx, pid: handle.pid() }
}
}
/// Spawn `state` as a server under the current actor (via [`spawn`]). Returns a
/// [`ServerRef`]. Shorthand for `ServerBuilder::new(state).start()`.
pub fn start<G: GenServer>(state: G) -> ServerRef<G> {
ServerBuilder::new(state).start()
} }
/// Like [`start`], but spawns the server under an explicit supervisor pid (via /// Like [`start`], but spawns the server under an explicit supervisor pid (via
/// [`spawn_under`]) so it slots into the supervision tree. /// [`spawn_under`]) so it slots into the supervision tree.
pub fn start_under<G: GenServer>(supervisor: Pid, state: G) -> ServerRef<G> { pub fn start_under<G: GenServer>(supervisor: Pid, state: G) -> ServerRef<G> {
let (tx, rx) = channel::<Envelope<G>>(); ServerBuilder::new(state).under(supervisor).start()
let handle = spawn_under(supervisor, move || server_loop::<G>(rx, state));
ServerRef { tx, pid: handle.pid() }
} }
fn server_loop<G: GenServer>(rx: Receiver<Envelope<G>>, state: G) { fn server_loop<G: GenServer>(
rx: Receiver<Envelope<G>>,
state: G,
mut infos: Vec<Receiver<G::Info>>,
) {
// terminate() must run on every exit path (clean close, panic, stop), so it // terminate() must run on every exit path (clean close, panic, stop), so it
// lives in this guard's Drop rather than after the loop. // lives in this guard's Drop rather than after the loop.
struct Terminate<G: GenServer>(G); struct Terminate<G: GenServer>(G);
@@ -199,22 +265,63 @@ fn server_loop<G: GenServer>(rx: Receiver<Envelope<G>>, state: G) {
} }
} }
let mut guard = Terminate(state); fn dispatch<G: GenServer>(state: &mut G, env: Envelope<G>) {
guard.0.init(); match env {
Envelope::Call(request, reply_tx) => {
loop { let reply = state.handle_call(request);
match rx.recv() {
Ok(Envelope::Call(request, reply_tx)) => {
let reply = guard.0.handle_call(request);
// The caller may have gone away (e.g. cancelled while parked); // The caller may have gone away (e.g. cancelled while parked);
// a failed reply send is not the server's problem. // a failed reply send is not the server's problem.
let _ = reply_tx.send(reply); let _ = reply_tx.send(reply);
} }
Ok(Envelope::Cast(request)) => guard.0.handle_cast(request), Envelope::Cast(request) => state.handle_cast(request),
}
}
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. // All ServerRefs dropped → inbox closed → graceful shutdown.
Err(_) => break, Err(_) => break,
} }
// Observation point so a server whose inbox is never empty stays } 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. // preemptible and cancellable.
crate::check!(); crate::check!();
} }
+1 -1
View File
@@ -46,7 +46,7 @@ static ALLOCATOR: preempt::PreemptingAllocator = preempt::PreemptingAllocator;
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
pub use channel::{channel, select, select_timeout, Receiver, RecvError, RecvTimeoutError, Selectable, Sender}; 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 link::{link, trap_exit, unlink, ExitSignal};
pub use monitor::{demonitor, monitor, Down, DownReason, Monitor, MonitorId}; pub use monitor::{demonitor, monitor, Down, DownReason, Monitor, MonitorId};
pub use mutex::{LockTimeout, Mutex, MutexGuard}; pub use mutex::{LockTimeout, Mutex, MutexGuard};
+111 -1
View File
@@ -1,7 +1,7 @@
//! gen_server tests: call round-trip, cast, lifecycle callbacks, and the two //! gen_server tests: call round-trip, cast, lifecycle callbacks, and the two
//! server-down detection paths (reply-channel close vs. inbox-send failure). //! 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 smarm::run;
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
@@ -26,6 +26,7 @@ impl GenServer for Counter {
type Call = Req; type Call = Req;
type Reply = i64; type Reply = i64;
type Cast = Op; type Cast = Op;
type Info = ();
fn handle_call(&mut self, req: Req) -> i64 { fn handle_call(&mut self, req: Req) -> i64 {
match req { match req {
@@ -68,6 +69,7 @@ impl GenServer for Lifecycle {
type Call = (); type Call = ();
type Reply = (); type Reply = ();
type Cast = (); type Cast = ();
type Info = ();
fn init(&mut self) { fn init(&mut self) {
self.log.lock().unwrap().push("init"); self.log.lock().unwrap().push("init");
@@ -151,6 +153,7 @@ impl GenServer for Slow {
type Call = u64; // delay in ms type Call = u64; // delay in ms
type Reply = u64; type Reply = u64;
type Cast = (); type Cast = ();
type Info = ();
fn handle_call(&mut self, delay_ms: u64) -> u64 { fn handle_call(&mut self, delay_ms: u64) -> u64 {
if delay_ms > 0 { if delay_ms > 0 {
@@ -206,6 +209,7 @@ fn call_timeout_to_dead_server_is_server_down_not_timeout() {
struct Bomb; struct Bomb;
impl GenServer for Bomb { impl GenServer for Bomb {
type Call = (); type Call = ();
type Info = ();
type Reply = (); type Reply = ();
type Cast = (); type Cast = ();
fn handle_call(&mut self, _: ()) { 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"]);
}