feat(gen_server): synchronous call / async cast over a single actor (roadmap #5)
Thin request-reply layer on channels, no runtime change. A server is an
actor owning a GenServer state; clients hold a clonable ServerRef and issue
call (sync, parks for the reply) or cast (fire-and-forget).
- Single inbox carrying an internal Envelope { Call(req, reply_tx) | Cast },
forced by the no-select / no-unified-mailbox invariant; call makes a
one-shot reply channel and parks on it.
- Server-down detection is pure channel closure (no monitor): send fails if
the inbox is gone; the reply sender drops on the server's unwind so a
parked caller wakes to Err. Both collapse to CallError/CastError::ServerDown.
- init/terminate are optional trait hooks; terminate runs via a drop guard so
it fires on every exit path (clean close, panic, request_stop). Must stay
non-blocking — may run mid-unwind.
- ServerRef carries pid() for monitor/request_stop/link; start + start_under.
Tests: cast-then-call roundtrip, init/terminate ordering, both server-down
paths (reply-channel close on handler panic; inbox-send failure when gone).
This commit is contained in:
@@ -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<G: GenServer> {
|
||||
Call(G::Call, Sender<G::Reply>),
|
||||
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<G: GenServer> {
|
||||
tx: Sender<Envelope<G>>,
|
||||
pid: Pid,
|
||||
}
|
||||
|
||||
impl<G: GenServer> Clone for ServerRef<G> {
|
||||
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<G: GenServer> ServerRef<G> {
|
||||
/// 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<G::Reply, CallError> {
|
||||
let (reply_tx, reply_rx) = channel::<G::Reply>();
|
||||
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<G: GenServer>(state: G) -> ServerRef<G> {
|
||||
let (tx, rx) = channel::<Envelope<G>>();
|
||||
let handle = spawn(move || server_loop::<G>(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<G: GenServer>(supervisor: Pid, state: G) -> ServerRef<G> {
|
||||
let (tx, rx) = channel::<Envelope<G>>();
|
||||
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) {
|
||||
// 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: GenServer>(G);
|
||||
impl<G: GenServer> Drop for Terminate<G> {
|
||||
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!();
|
||||
}
|
||||
}
|
||||
@@ -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};
|
||||
|
||||
@@ -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<Mutex<Vec<&'static str>>>,
|
||||
}
|
||||
|
||||
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)));
|
||||
}
|
||||
Reference in New Issue
Block a user