diff --git a/src/gen_server.rs b/src/gen_server.rs new file mode 100644 index 0000000..42ccf4f --- /dev/null +++ b/src/gen_server.rs @@ -0,0 +1,183 @@ +//! gen_server — synchronous call / asynchronous cast over a single actor. +//! +//! A thin request-reply layer on top of [`channel`](crate::channel), modelled +//! on Erlang's `gen_server`. A *server* is an actor owning a state value that +//! implements [`GenServer`]; clients hold a clonable [`ServerRef`] and issue +//! [`call`](ServerRef::call) (synchronous, returns a reply) or +//! [`cast`](ServerRef::cast) (fire-and-forget). +//! +//! ## Why a single inbox +//! +//! 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. +//! +//! ## Server death +//! +//! Detection falls out of channel closure, so no monitor is required: +//! - if the server is already gone, its inbox is closed and the `send` in +//! `call`/`cast` fails → [`CallError::ServerDown`] / [`CastError::ServerDown`]; +//! - if the server dies *after* a call is enqueued but before it replies +//! (a handler panic, or a cooperative `request_stop`), the reply sender is +//! dropped as the server's stack unwinds, closing the reply channel; the +//! parked caller wakes and its `recv` returns `Err` → `ServerDown`. +//! +//! ## Lifecycle / callbacks +//! +//! [`GenServer::init`] runs once inside the server actor before the first +//! message; [`GenServer::terminate`] runs on the way out. terminate is wired +//! through a drop guard, so it fires on *every* exit path — graceful inbox +//! close, a handler panic, or a cooperative `request_stop` — not only the clean +//! one. Keep it cheap and non-blocking: it may run mid-unwind, and a panic +//! inside it during an unwind aborts the process (a double panic). +//! +//! ## Not here (yet) +//! +//! No `handle_info` and no call timeout. Both need machinery smarm currently +//! defers: a call timeout needs a per-`recv` deadline (cf. the deferred +//! `Signal::Timeout`), and `handle_info` needs the cross-channel mailbox merge +//! that selective receive deliberately left unmade. They are slated to land +//! together with timeouts. + +use crate::channel::{channel, Receiver, Sender}; +use crate::pid::Pid; +use crate::scheduler::{spawn, spawn_under}; + +/// Behaviour for a gen_server: a state value plus call/cast handlers. +/// +/// `handle_call` and `handle_cast` are required; [`init`](Self::init) and +/// [`terminate`](Self::terminate) are optional lifecycle hooks with no-op +/// defaults. +pub trait GenServer: Send + 'static { + /// Synchronous request type (carried by [`ServerRef::call`]). + type Call: Send + 'static; + /// Reply type returned for a `Call`. + type Reply: Send + 'static; + /// Asynchronous request type (carried by [`ServerRef::cast`]). + type Cast: Send + 'static; + + /// Runs once inside the server actor before any message is handled. + fn init(&mut self) {} + + /// Handle a synchronous call and produce the reply sent back to the caller. + fn handle_call(&mut self, request: Self::Call) -> Self::Reply; + + /// Handle a fire-and-forget cast. + fn handle_cast(&mut self, request: Self::Cast); + + /// Runs as the server actor exits, on any exit path (see module docs). + fn terminate(&mut self) {} +} + +/// What travels the server's single inbox channel: a synchronous call (with a +/// reply sender) or an asynchronous cast. +enum Envelope { + Call(G::Call, Sender), + Cast(G::Cast), +} + +/// A clonable handle to a running server. Cloning yields another sender to the +/// same inbox; the server lives until the last `ServerRef` is dropped, at which +/// point its inbox closes and the loop exits normally. +pub struct ServerRef { + tx: Sender>, + pid: Pid, +} + +impl Clone for ServerRef { + fn clone(&self) -> Self { + ServerRef { tx: self.tx.clone(), pid: self.pid } + } +} + +/// Returned by [`ServerRef::call`] when the server is no longer reachable. +#[derive(Debug, PartialEq, Eq, Clone, Copy)] +pub enum CallError { + /// The server was already gone, or died before replying. + ServerDown, +} + +/// Returned by [`ServerRef::cast`] when the server is no longer reachable. +#[derive(Debug, PartialEq, Eq, Clone, Copy)] +pub enum CastError { + /// The server inbox was closed (the server is gone). + ServerDown, +} + +impl ServerRef { + /// The server actor's pid — usable with `monitor`, `request_stop`, `link`. + pub fn pid(&self) -> Pid { + self.pid + } + + /// Synchronous request-reply. Blocks (parking the calling actor) until the + /// server replies, or returns [`CallError::ServerDown`] if the server is or + /// becomes unreachable before a reply arrives. + pub fn call(&self, request: G::Call) -> Result { + let (reply_tx, reply_rx) = channel::(); + self.tx + .send(Envelope::Call(request, reply_tx)) + .map_err(|_| CallError::ServerDown)?; + reply_rx.recv().map_err(|_| CallError::ServerDown) + } + + /// Fire-and-forget request. Returns once the message is enqueued; does not + /// wait for the server to handle it. [`CastError::ServerDown`] if the inbox + /// is already closed. + pub fn cast(&self, request: G::Cast) -> Result<(), CastError> { + self.tx + .send(Envelope::Cast(request)) + .map_err(|_| CastError::ServerDown) + } +} + +/// 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. +pub fn start(state: G) -> ServerRef { + let (tx, rx) = channel::>(); + let handle = spawn(move || server_loop::(rx, state)); + ServerRef { tx, pid: handle.pid() } +} + +/// 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() } +} + +fn server_loop(rx: Receiver>, state: G) { + // 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); + impl Drop for Terminate { + fn drop(&mut self) { + self.0.terminate(); + } + } + + 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); + // 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, + } + // Observation point so a server whose inbox is never empty stays + // preemptible and cancellable. + crate::check!(); + } +} diff --git a/src/lib.rs b/src/lib.rs index 65067d4..839ca40 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -24,6 +24,7 @@ pub mod io; pub mod mutex; pub mod monitor; pub mod link; +pub mod gen_server; pub mod runtime; pub mod trace; @@ -39,6 +40,7 @@ static ALLOCATOR: preempt::PreemptingAllocator = preempt::PreemptingAllocator; // --------------------------------------------------------------------------- pub use channel::{channel, Receiver, RecvError, Sender}; +pub use gen_server::{CallError, CastError, GenServer, ServerRef}; pub use link::{link, trap_exit, unlink, ExitSignal}; pub use monitor::{monitor, Down, DownReason}; pub use mutex::{LockTimeout, Mutex, MutexGuard}; diff --git a/tests/gen_server.rs b/tests/gen_server.rs new file mode 100644 index 0000000..92de18d --- /dev/null +++ b/tests/gen_server.rs @@ -0,0 +1,137 @@ +//! 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::run; +use std::sync::{Arc, Mutex}; + +// --------------------------------------------------------------------------- +// A trivial counter server: casts mutate, calls read (or blow up). +// --------------------------------------------------------------------------- + +struct Counter { + n: i64, +} + +enum Req { + Get, + Boom, +} + +enum Op { + Add(i64), +} + +impl GenServer for Counter { + type Call = Req; + type Reply = i64; + type Cast = Op; + + fn handle_call(&mut self, req: Req) -> i64 { + match req { + Req::Get => self.n, + Req::Boom => panic!("boom"), + } + } + + fn handle_cast(&mut self, op: Op) { + match op { + Op::Add(x) => self.n += x, + } + } +} + +// Casts are applied in order and a later call observes the accumulated state. +#[test] +fn cast_then_call_roundtrip() { + let got = Arc::new(Mutex::new(0i64)); + let got2 = got.clone(); + run(move || { + let server = start(Counter { n: 0 }); + server.cast(Op::Add(5)).unwrap(); + server.cast(Op::Add(3)).unwrap(); + let n = server.call(Req::Get).unwrap(); + *got2.lock().unwrap() = n; + }); + assert_eq!(*got.lock().unwrap(), 8); +} + +// --------------------------------------------------------------------------- +// Lifecycle: init runs before the first message, terminate on graceful exit. +// --------------------------------------------------------------------------- + +struct Lifecycle { + log: Arc>>, +} + +impl GenServer for Lifecycle { + type Call = (); + type Reply = (); + type Cast = (); + + fn init(&mut self) { + self.log.lock().unwrap().push("init"); + } + + fn handle_call(&mut self, _req: ()) { + self.log.lock().unwrap().push("call"); + } + + fn handle_cast(&mut self, _req: ()) {} + + fn terminate(&mut self) { + self.log.lock().unwrap().push("terminate"); + } +} + +// init -> handle_call -> (drop last ref closes inbox) -> terminate. +#[test] +fn init_and_terminate_run() { + let log = Arc::new(Mutex::new(Vec::new())); + let log2 = log.clone(); + run(move || { + let server = start(Lifecycle { log: log2 }); + server.call(()).unwrap(); + // Dropping the only ref closes the inbox; the server breaks out of its + // recv loop and runs terminate. run() will not return until it has. + drop(server); + }); + assert_eq!(*log.lock().unwrap(), vec!["init", "call", "terminate"]); +} + +// --------------------------------------------------------------------------- +// Server-down detection. +// --------------------------------------------------------------------------- + +// The server dies *while* a call is in flight (handler panics): the reply +// sender drops on unwind, closing the reply channel, so the parked caller's +// recv returns Err -> ServerDown. +#[test] +fn call_to_panicking_handler_is_server_down() { + let got = Arc::new(Mutex::new(None)); + let got2 = got.clone(); + run(move || { + let server = start(Counter { n: 0 }); + let r = server.call(Req::Boom); + *got2.lock().unwrap() = Some(r); + }); + assert_eq!(*got.lock().unwrap(), Some(Err(CallError::ServerDown))); +} + +// A call issued *after* the server is already gone: the inbox is closed, so the +// send itself fails -> ServerDown (the other detection path). +#[test] +fn call_after_server_gone_is_server_down() { + let got = Arc::new(Mutex::new(None)); + let got2 = got.clone(); + run(move || { + let server = start(Counter { n: 0 }); + let server2 = server.clone(); + // This kills the server (and is itself ServerDown). + assert_eq!(server.call(Req::Boom), Err(CallError::ServerDown)); + // Inbox now closed; a fresh call can't even be enqueued. + let r = server2.call(Req::Get); + *got2.lock().unwrap() = Some(r); + }); + assert_eq!(*got.lock().unwrap(), Some(Err(CallError::ServerDown))); +}