//! A stateful actor with typed request-reply messaging. //! //! ## What is a gen_server? //! //! Whenever you need mutable state shared between threads, the obvious tool is //! `Arc>`. That works, but it scatters lock guards and error //! handling everywhere, and offers no natural place to put the logic that //! operates on the state. //! //! A gen_server is a cleaner alternative: a dedicated thread that *owns* the //! state and handles one message at a time. Other threads talk to it by sending //! typed messages and optionally waiting for a typed reply. Because messages are //! serialised through a single inbox. The server is the only thing that ever //! touches the state, so there are no lock guards at all. //! //! You write the logic by implementing [`GenServer`]. smarm runs the receive //! loop, the reply plumbing, and the lifecycle. //! //! ## A first server //! //! Let's build a counter that can be incremented from any thread and queried for //! its current value. //! //! ```ignore //! use smarm::gen_server::{self, GenServer, GenServerRef}; //! //! // State //! //! struct Counter { //! count: u64, //! } //! //! // Message types //! //! // Calls expect a reply; casts do not. //! enum CounterCall { Get } //! enum CounterCast { Increment } //! //! // Behaviour //! //! impl GenServer for Counter { //! type Call = CounterCall; //! type Reply = u64; //! type Cast = CounterCast; //! type Info = (); // no out-of-band messages //! type Timer = (); // no timers //! //! fn handle_call(&mut self, request: CounterCall) -> u64 { //! match request { //! CounterCall::Get => self.count, //! } //! } //! //! fn handle_cast(&mut self, request: CounterCast) { //! match request { //! CounterCast::Increment => self.count += 1, //! } //! } //! } //! ``` //! //! Start the server and talk to it: //! //! ```ignore //! let server: GenServerRef = gen_server::start(Counter { count: 0 }); //! //! server.cast(CounterCast::Increment).unwrap(); //! server.cast(CounterCast::Increment).unwrap(); //! //! let n = server.call(CounterCall::Get).unwrap(); //! // n == 2 //! ``` //! //! [`GenServerRef`] is cheaply clonable; hand copies to as many threads as you //! like. They all share the same inbox; the server handles messages one at a //! time, in arrival order. //! //! ## Client / server APIs //! //! In practice, callers should not have to know about `CounterCall` or //! `CounterCast`. The recommended pattern is to wrap the message types in plain //! functions on the same module: //! //! ```ignore //! // Client API //! //! impl Counter { //! pub fn start() -> GenServerRef { //! gen_server::start(Counter { count: 0 }) //! } //! //! pub fn increment(server: &GenServerRef) { //! server.cast(CounterCast::Increment).unwrap(); //! } //! //! pub fn get(server: &GenServerRef) -> u64 { //! server.call(CounterCall::Get).unwrap() //! } //! } //! //! // Callers now just do: //! let counter = Counter::start(); //! Counter::increment(&counter); //! Counter::increment(&counter); //! let n = Counter::get(&counter); // n == 2 //! ``` //! //! ## call vs cast //! //! `call` sends a request and blocks the calling thread until the server //! replies. Use it when you need a return value, or when you need to know that //! the server has processed the message before continuing. //! //! `cast` enqueues a message and returns immediately, without waiting for the //! server to handle it. Use it for fire-and-forget updates where you don't need //! confirmation. //! //! Both return `Err(ServerDown)` if the inbox is closed (the server is gone). //! If you need to cap how long you wait for a reply, use //! [`GenServerRef::call_timeout`], which additionally returns `Err(Timeout)` if //! the deadline passes before a reply arrives. //! //! ## Lifecycle //! //! Two optional callbacks bracket the server's life: //! //! - [`GenServer::init`] runs once before the first message. Use it to start //! timers or set up monitors; see the [`GenServerCtx`] it receives. //! - [`GenServer::terminate`] runs when the server is about to exit. It fires //! on every exit path (all `GenServerRef`s dropped, a handler panic, or an //! explicit [`GenServerRef::shutdown`]), not only on clean shutdown. Keep it //! short and non-blocking: if `terminate` panics while the server is already //! unwinding from a handler panic, the process aborts. //! //! ## When the server stops //! //! The server runs as long as at least one [`GenServerRef`] exists. When the last //! one is dropped, the inbox closes and the loop exits gracefully. To stop a //! server explicitly and wait for it to finish, call [`GenServerRef::shutdown`]. //! //! If the server panics inside a handler, the panic unwinds the server thread. //! Any caller currently waiting in `call` sees `Err(ServerDown)`: the reply //! channel closes as the server unwinds, which wakes the caller. //! //! ## Going further //! //! Out-of-band messages: a server can receive messages from sources other //! than `call`/`cast`, for example notifications from a background task. Pass //! extra [`Receiver`] channels to [`GenServerBuilder::with_info`]; they are //! dispatched to [`GenServer::handle_info`], always before inbox messages. //! //! Monitors: to watch another actor and be notified when it exits, clone a //! [`Watcher`] from [`GenServerCtx::watcher`] during `init` and call //! [`Watcher::watch`] with a [`Monitor`] from any handler. The loop dispatches //! the resulting [`Down`] to [`GenServer::handle_down`]. //! //! Timers: clone a [`TimerHandle`] from [`GenServerCtx::timer`] during `init`. //! Use [`TimerHandle::arm_after`] for a one-shot and [`TimerHandle::tick_every`] //! for a periodic; both fire into [`GenServer::handle_timer`]. //! //! Idle detection: call [`GenServerCtx::idle_after`] during `init` to set a //! quiet window: if no message of any kind is dispatched for that duration, //! the loop calls [`GenServer::handle_idle`] and resets the window. This differs //! from the client-side `call_timeout`. Idle measures silence on the *server*, //! while `call_timeout` caps how long *one caller* waits. //! //! Names and registration: a server can be given a static name so other //! actors can reach it without holding a `GenServerRef`. Use //! [`GenServerBuilder::named`] to register on start, and the free functions //! [`call`], [`cast`], and [`whereis_server`] to address it by name. Registered //! servers are a natural fit for supervision; see `supervisor` for how to //! build a tree that restarts servers on failure. //! //! ## Limitations //! //! The info-channel set is fixed at start; channels cannot be added or removed //! while the server is running. Monitors are dynamic (they can be registered //! from any handler via [`Watcher::watch`]) because monitors are inherently //! created at runtime. The idle window is set once, in `init`. 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, Instant}; /// The behaviour you implement to make a type into a gen_server. /// /// Implement this on your state struct. `handle_call` and `handle_cast` are /// required; all other methods have no-op defaults and can be added as needed. pub trait GenServer: Send + 'static { /// The request type sent by [`GenServerRef::call`]. Must produce a [`Reply`](Self::Reply). type Call: Send + 'static; /// The value returned to the caller by [`handle_call`](Self::handle_call). type Reply: Send + 'static; /// The request type sent by [`GenServerRef::cast`]. No reply is produced. type Cast: Send + 'static; /// Out-of-band message type delivered to [`handle_info`](Self::handle_info) /// from channels registered via [`GenServerBuilder::with_info`]. If you don't /// use info channels, set this to `()`. type Info: Send + 'static; /// Payload type for timers armed via [`TimerHandle`], delivered to /// [`handle_timer`](Self::handle_timer). Kept separate from [`Info`](Self::Info) /// so that timer fires (which your server schedules itself) stay distinct /// from external messages (which arrive from outside). Use `()` if unused. type Timer: Send + 'static; /// Runs once inside the server actor before any message is handled. The /// [`GenServerCtx`] is the loop's one runtime hook: clone whichever of /// [`ctx.watcher()`](GenServerCtx::watcher) (monitors) and /// [`ctx.timer()`](GenServerCtx::timer) (timers) you will need from later /// handlers into the state here. If you need neither, the unused ctx drops /// and the loop's system arm auto-closes (see module docs). fn init(&mut self, _ctx: &GenServerCtx) where Self: Sized, { } /// 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); /// Handle an out-of-band message from one of the info channels. Default: /// drop it. fn handle_info(&mut self, _info: Self::Info) {} /// Handle a [`Down`] from a monitor handed to the loop via /// [`Watcher::watch`]. Default: drop it. fn handle_down(&mut self, _down: Down) {} /// Handle a fired timer armed through [`TimerHandle`] /// ([`arm_after`](TimerHandle::arm_after) / /// [`tick_every`](TimerHandle::tick_every)). Timer fires are delivered /// before info and inbox messages, so a heartbeat cannot be starved by /// application traffic. Default: drop the message. fn handle_timer(&mut self, _msg: Self::Timer) {} /// Handle a quiet-period notification: called when the server has gone the /// full idle window (set via [`GenServerCtx::idle_after`] in `init`) without /// dispatching any message. The window resets automatically after this /// fires, so it acts as a steady idle detector. To shut down after one idle /// period, call [`request_stop`](crate::scheduler::request_stop) here. /// Default: no-op. fn handle_idle(&mut self) {} /// 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. Private — callers use [`GenServerRef`]. 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 `GenServerRef` is dropped, at which /// point its inbox closes and the loop exits normally. pub struct GenServerRef { tx: Sender>, pid: Pid, } impl Clone for GenServerRef { fn clone(&self) -> Self { GenServerRef { tx: self.tx.clone(), pid: self.pid } } } /// Returned by [`GenServerRef::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 [`GenServerRef::call_timeout`]. #[derive(Debug, PartialEq, Eq, Clone, Copy)] pub enum CallTimeoutError { /// The server was already gone, or died before replying. ServerDown, /// The deadline passed before a reply arrived. The request is still in the /// server's inbox and will be handled — the reply is simply discarded /// because the reply channel was dropped when the timeout fired. Design /// calls that may time out to be idempotent, so a late reply causes no harm. Timeout, } /// Returned by [`GenServerRef::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 GenServerRef { /// 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) } /// Like [`call`](Self::call), but gives up after `timeout` and returns /// [`CallTimeoutError::Timeout`] if no reply arrives in time. /// /// Note that "timeout" means two different things here and in /// [`GenServerCtx::idle_after`]: this one is a *client-side call deadline* — /// how long this particular caller waits. The idle timeout measures silence /// on the *server's* inbox. They are completely independent. /// /// On timeout, the server still handles the request; only the reply is /// discarded (the reply channel is dropped, so the server's send fails /// silently). Design timed-out calls to be idempotent. pub fn call_timeout( &self, request: G::Call, timeout: std::time::Duration, ) -> Result { let (reply_tx, reply_rx) = channel::(); self.tx .send(Envelope::Call(request, reply_tx)) .map_err(|_| CallTimeoutError::ServerDown)?; reply_rx.recv_timeout(timeout).map_err(|e| match e { crate::channel::RecvTimeoutError::Disconnected => CallTimeoutError::ServerDown, crate::channel::RecvTimeoutError::Timeout => CallTimeoutError::Timeout, }) } /// 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) } /// Stop the server and block until it has fully exited. /// /// Sends a cooperative stop signal to the server actor and waits for it to /// exit, so [`GenServer::terminate`] has run by the time this returns. /// Returns immediately if the server is already gone. /// /// This is the right teardown for a server kept alive by a registered /// [`GenServerName`], where dropping every external `GenServerRef` is not enough /// to close the inbox. Like all cooperative cancellation, it is best-effort: /// a server wedged in a tight loop with no observation point cannot be /// stopped this way. Panics if called outside `Runtime::run()`. pub fn shutdown(&self) { let mon = monitor(self.pid); request_stop(self.pid); // The Down lands when the server finalizes; an already-dead target makes // `monitor` deliver NoProc immediately, so this never blocks forever. let _ = mon.rx.recv(); demonitor(&mon); } } /// Internal channel carrying control messages to the server loop. Monitors /// and timer fires both need to land above the inbox in priority, so they /// share one channel dispatched by variant. The arm stays open while any /// [`Watcher`] or [`TimerHandle`] clone (or an in-flight timer fire) holds a sender. enum Sys { /// A monitor handed in via [`Watcher::watch`]; pushed onto the loop's /// monitor set. Watch(Monitor), /// A fired one-shot timer: loop-local id (so the loop can retire the /// registry entry) plus the server's payload. Timer(crate::timer::TimerId, G::Timer), /// A fired periodic tick carrying its stable local id. The loop looks up /// the payload factory, dispatches it to [`GenServer::handle_timer`], and /// re-arms the next tick before returning. Tick(crate::timer::TimerId), } /// The server loop's runtime hook, passed to [`GenServer::init`]. Hands out the /// loop's two clonable intake handles — the [`Watcher`] (monitors) and the /// [`TimerHandle`] (timers) — plus the one-shot idle-window setter. All fields /// are private so the struct can gain new hooks in future without breaking /// existing implementations. pub struct GenServerCtx { 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 `GenServerCtx` is only ever /// borrowed on the actor's own stack during `init`, never sent. idle: Cell>, } impl GenServerCtx { /// A clonable handle to the loop's monitor intake. Store it in the state /// during `init` to watch monitors from later handlers. pub fn watcher(&self) -> Watcher { Watcher { tx: self.sys_tx.clone() } } /// Shorthand for `ctx.watcher().watch(m)` when watching during `init`. pub fn watch(&self, m: Monitor) { self.watcher().watch(m) } /// A clonable handle to the loop's timer intake (the [`Watcher`] pattern, /// for time). Store it on the state during `init` to /// [`arm_after`](TimerHandle::arm_after) / /// [`tick_every`](TimerHandle::tick_every) / /// [`cancel`](TimerHandle::cancel) from any later handler. pub fn timer(&self) -> TimerHandle { TimerHandle { sys_tx: self.sys_tx.clone(), reg: self.reg.clone() } } /// Set a quiet-period window: if the loop goes `after` without dispatching /// any message, it calls [`GenServer::handle_idle`] and resets the window. /// Call this once during `init`; the window is fixed for the server's /// lifetime. /// /// This is distinct from [`GenServerRef::call_timeout`], which caps how long a /// single caller waits for a reply. This one measures silence on the whole /// inbox. pub fn idle_after(&self, after: Duration) { self.idle.set(Some(after)); } } /// Per-server timer bookkeeping, shared between the loop and every /// [`TimerHandle`] clone. A gen_server actor is single-threaded — handlers /// and the loop never run concurrently — so this `Mutex` is always /// uncontended at runtime; `Arc` (rather than `Rc`) is used /// only because `GenServer: Send` forces the handle (hence this shared state) /// to be `Send`. struct TimerReg { /// Monotonic minter for loop-local [`TimerId`]s — the ids handed to users, /// kept distinct from the substrate `seq` (which changes on every periodic /// re-arm). A local id is resolved only through this registry, never passed /// to [`cancel_timer`], so the two id roles never mix. next_local: u64, /// Live one-shot timers: local id → current substrate id. Present from /// [`arm_after`](TimerHandle::arm_after) until the loop retires it on fire /// or [`cancel`](TimerHandle::cancel) removes it. oneshots: HashMap, /// Live periodic timers: stable local id → its bookkeeping. The stable id /// is minted once by [`tick_every`](TimerHandle::tick_every) and survives /// every re-arm; the entry lives until [`cancel`](TimerHandle::cancel) /// removes it (steady re-arm never self-removes). periodics: HashMap>, /// A `Sys` sender held *only while ≥1 periodic is live*, used by the loop to /// re-arm the next tick. `Some` exactly when `periodics` is non-empty: this /// keeps the system arm open for a server whose only system use is a /// periodic, while leaving a non-timer server's arm to auto-close (the /// `unused_ctx` behaviour) since this stays `None` for it. rearm_tx: Option>>, } /// One periodic timer's loop-side bookkeeping. struct Periodic { /// Re-arm interval; each fire schedules the next at `now + every`. every: Duration, /// The currently-armed underlying timer for this periodic (changes on every /// re-arm); cancelled by [`cancel`](TimerHandle::cancel). live: crate::timer::TimerId, /// Produces a fresh payload for each tick. Built in `tick_every` as /// `move || msg.clone()` — so the `Clone` bound lives on that one method /// and never appears on `type Timer` or anywhere else in the loop. make: Box G::Timer + Send>, } // Manual Default: deriving would demand `G: Default`, but a fresh registry is // independent of the server type. impl Default for TimerReg { fn default() -> Self { TimerReg { next_local: 0, oneshots: HashMap::new(), periodics: HashMap::new(), rearm_tx: None, } } } impl TimerReg { /// Mint the next loop-local id. fn mint(&mut self) -> TimerId { let id = TimerId::from_raw(self.next_local); self.next_local = self.next_local.wrapping_add(1); id } } /// Arms and cancels timers for a server loop, the time-side twin of [`Watcher`]. /// Handed out by [`GenServerCtx::timer`] in `init`; store it on the state so /// handlers can arm timers without changing their signatures. Clonable; the /// loop's timer arm stays open while any clone (or an in-flight fire) lives. pub struct TimerHandle { sys_tx: Sender>, reg: Arc>>, } // Manual Clone for the same reason as `Watcher`: no `G: Clone` needed. impl Clone for TimerHandle { fn clone(&self) -> Self { TimerHandle { sys_tx: self.sys_tx.clone(), reg: self.reg.clone() } } } impl TimerHandle { /// Arm a one-shot timer: deliver `msg` to [`GenServer::handle_timer`] once, /// after `after`, unless [`cancel`](Self::cancel)led first. Returns a /// [`TimerId`] you can pass to `cancel`. Timer fires arrive before info and /// inbox messages. pub fn arm_after(&self, after: Duration, msg: G::Timer) -> TimerId { let mut reg = match self.reg.lock() { Ok(g) => g, Err(e) => panic!("smarm: gen_server reg lock poisoned (core corrupt): {e}"), }; let local = reg.mint(); // The fire thunk carries the local id so the loop can retire the entry // on dispatch; `msg` is moved in (no `Clone` needed for one-shots). let sub = send_after_to(after, self.sys_tx.clone(), Sys::Timer(local, msg)); reg.oneshots.insert(local, sub); local } /// Arm a periodic timer: deliver a fresh `msg` to /// [`GenServer::handle_timer`] every `every`, until /// [`cancel`](Self::cancel)led. The underlying timer is one-shot; the loop /// re-arms it after each fire to produce the repeating cadence, and exposes /// a single stable [`TimerId`] so one `cancel` stops both the pending /// instance and all future ones. /// /// Requires `G::Timer: Clone` because the same logical message is cloned /// fresh for each tick. This bound is on this method only; one-shot /// [`arm_after`](Self::arm_after) and the `type Timer` declaration are /// unconstrained. pub fn tick_every(&self, every: Duration, msg: G::Timer) -> TimerId where G::Timer: Clone, { let mut reg = match self.reg.lock() { Ok(g) => g, Err(e) => panic!("smarm: gen_server reg lock poisoned (core corrupt): {e}"), }; let local = reg.mint(); // A live periodic must keep the system arm open so the next tick can be // re-armed; the first periodic installs the loop's re-arm sender. if reg.periodics.is_empty() { reg.rearm_tx = Some(self.sys_tx.clone()); } let make: Box G::Timer + Send> = Box::new(move || msg.clone()); // First instance fires after `every`; the payload is produced loop-side // from `make` on fire, so the tick carries only the stable id. let sub = send_after_to(every, self.sys_tx.clone(), Sys::Tick(local)); reg.periodics.insert(local, Periodic { every, live: sub, make }); debug_assert!(reg.rearm_tx.is_some(), "rearm_tx must be Some while periodics is non-empty"); local } /// Cancel an armed timer (one-shot or periodic). Returns the substrate's /// race signal — `true` if the cancel beat the fire, `false` if the timer /// had already fired, been cancelled, or is otherwise unknown. Most callers /// can ignore the return value: a fired-but-not-yet-dispatched one-shot is /// discarded by the loop on delivery (see the `Tick` arm in `server_loop`), /// and a cancelled periodic will not re-arm. For a periodic, `cancel` also /// stops re-arming: the entry is removed so the loop will not schedule /// another tick even if the pending one already escaped onto the channel. pub fn cancel(&self, id: TimerId) -> bool { let mut reg = match self.reg.lock() { Ok(g) => g, Err(e) => panic!("smarm: gen_server reg lock poisoned (core corrupt): {e}"), }; if let Some(sub) = reg.oneshots.remove(&id) { return cancel_timer(sub); } if let Some(p) = reg.periodics.remove(&id) { let beat = cancel_timer(p.live); if reg.periodics.is_empty() { reg.rearm_tx = None; } debug_assert!( reg.rearm_tx.is_some() != reg.periodics.is_empty(), "rearm_tx must be Some iff periodics is non-empty" ); return beat; } false } } /// Hands [`Monitor`]s to a server loop; their [`Down`]s are dispatched to /// [`GenServer::handle_down`]. Down arms outrank every other arm (a death /// notice cannot be starved), and a delivered `Down` retires its arm — /// monitors are one-shot. pub struct Watcher { tx: Sender>, } // Manual Clone: deriving would demand `G: Clone`, but the handle is clonable // regardless of the server type (it clones only the inner sender). impl Clone for Watcher { fn clone(&self) -> Self { Watcher { tx: self.tx.clone() } } } impl Watcher { /// Transfer `m` to the server loop. If the server is already gone the /// monitor is silently dropped (its queued `Down`, if any, with it) — /// there is no loop left to care. pub fn watch(&self, m: Monitor) { let _ = self.tx.send(Sys::Watch(m)); } } /// 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 = GenServerBuilder::new(state) /// .with_info(events_rx) /// .under(supervisor_pid) /// .start(); /// ``` pub struct GenServerBuilder { state: G, infos: Vec>, supervisor: Option, } impl GenServerBuilder { pub fn new(state: G) -> Self { GenServerBuilder { 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 [`GenServerRef`]. The server's /// lifetime is governed by its refs, not by joining, so the backing join /// handle is dropped. pub fn start(self) -> GenServerRef { self.spawn_server() } /// Bind the server to a durable [`GenServerName`] as it starts. Switches to the /// fallible [`NamedGenServerBuilder::start`] (the name may already be held by a /// live server). Consumes the builder, carrying its `with_info` / `under` /// configuration through. pub fn named(self, name: GenServerName) -> NamedGenServerBuilder { NamedGenServerBuilder { builder: self, name: name.as_str() } } /// Private shared body behind [`start`](Self::start) and /// [`NamedGenServerBuilder::start`]: allocate the inbox, spawn the loop, /// return the ref. The named path additionally publishes the inbox sender /// under the name before returning. fn spawn_server(self) -> GenServerRef { let (tx, rx) = channel::>(); let GenServerBuilder { 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)), }; GenServerRef { tx, pid: handle.pid() } } } /// A typed, static name for a gen_server, used to address it through the /// registry without holding a [`GenServerRef`]. Because a gen_server handles /// multiple message types, the name is typed by the whole server (`G`) rather /// than by a single message type. Declare it as a constant: /// /// ```ignore /// const COUNTER: GenServerName = GenServerName::new("counter"); /// ``` /// /// Use [`GenServerBuilder::named`] to bind the name on start, and the free /// functions [`call`], [`cast`], and [`whereis_server`] to address the server /// by name. Under the hood the server's inbox sender is stored in the registry /// keyed by name and message `TypeId`; `Envelope` is private, so only /// `GenServerName` can open that slot. pub struct GenServerName { name: &'static str, _marker: PhantomData G>, } impl GenServerName { /// Bind a static string as a server name. `const`, so names live as /// associated constants at call sites. #[inline] pub const fn new(name: &'static str) -> Self { Self { name, _marker: PhantomData } } /// The underlying registry key. #[inline] pub const fn as_str(self) -> &'static str { self.name } } impl Copy for GenServerName {} impl Clone for GenServerName { fn clone(&self) -> Self { *self } } /// A [`GenServerBuilder`] that will bind a [`GenServerName`] as it starts. Reached via /// [`GenServerBuilder::named`]; its [`start`](Self::start) is fallible because the /// name may already be held by a live server. `with_info` / `under` stay /// available so configuration can come before or after `named`. pub struct NamedGenServerBuilder { builder: GenServerBuilder, name: &'static str, } impl NamedGenServerBuilder { /// Add an out-of-band info channel (see [`GenServerBuilder::with_info`]). pub fn with_info(mut self, rx: Receiver) -> Self { self.builder = self.builder.with_info(rx); self } /// Spawn under an explicit supervisor (see [`GenServerBuilder::under`]). pub fn under(mut self, supervisor: Pid) -> Self { self.builder = self.builder.under(supervisor); self } /// Spawn the server and bind its name in one step. Fallible: returns /// [`RegisterError::NameTaken`] if the name is already held by a different /// live server. /// /// The inbox sender is published under the name **from the parent side, /// before this returns**, so a by-name `call` / `cast` resolves the instant /// `start()` returns — no race with the server body. On a name clash the /// just-spawned server is wound down (its only ref is dropped, closing the /// inbox), so a failed bind leaks no actor. pub fn start(self) -> Result, RegisterError> { let NamedGenServerBuilder { builder, name } = self; let server = builder.spawn_server(); match register_with::>(server.pid, name, server.tx.clone()) { Ok(()) => Ok(server), Err(e) => { drop(server); // inbox closes → loop exits gracefully Err(e) } } } } /// Resolve a [`GenServerName`] to a [`GenServerRef`] when you want a handle to hold or /// pass on rather than resolve per call. Rebuilds the ref from the registry's /// stored inbox sender; `None` if no live server holds the name. /// /// Panics if called outside `Runtime::run()`. pub fn whereis_server(name: GenServerName) -> Option> { resolve_named_sender::>(name.as_str()).map(|(pid, tx)| GenServerRef { tx, pid }) } /// Synchronous request-reply to the server currently registered under `name`, /// resolving through the registry on every call (so a server restarted under /// the same name is reached transparently). [`CallError::ServerDown`] if no live /// server holds the name, or if it dies before replying. /// /// Panics if called outside `Runtime::run()`. pub fn call(name: GenServerName, request: G::Call) -> Result { match whereis_server(name) { Some(server) => server.call(request), None => Err(CallError::ServerDown), } } /// Fire-and-forget to the server registered under `name`, resolving per cast. /// [`CastError::ServerDown`] if no live server holds the name. /// /// Panics if called outside `Runtime::run()`. pub fn cast(name: GenServerName, request: G::Cast) -> Result<(), CastError> { match whereis_server(name) { Some(server) => server.cast(request), None => Err(CastError::ServerDown), } } /// Terminate the server registered under `name` and block until it is down (see /// [`GenServerRef::shutdown`]). A no-op if no live server holds the name. /// /// Panics if called outside `Runtime::run()`. pub fn shutdown(name: GenServerName) { if let Some(server) = whereis_server(name) { server.shutdown(); } } /// Spawn `state` as a server under the current actor (via [`spawn`]). Returns a /// [`GenServerRef`]. Shorthand for `GenServerBuilder::new(state).start()`. pub fn start(state: G) -> GenServerRef { GenServerBuilder::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) -> GenServerRef { GenServerBuilder::new(state).under(supervisor).start() } fn server_loop( rx: Receiver>, state: G, mut infos: Vec>, ) { // Drop guard — owns the server state and the timer registry. // // Why a guard rather than code after the loop: // - `terminate()` must fire on *every* exit path: clean inbox close, // a handler panic, and cooperative `request_stop`. A guard's `Drop` // covers all three; code after the loop only covers the clean path. // - The timer drain must run *before* `terminate()`: terminate may // inspect state but must not arm new timers into a dead loop. By // owning the registry here the drain and the terminate call are // sequenced correctly and can never be reordered. // - A debug_assert pins the post-drain invariant cheaply. struct Terminate(G, Arc>>); impl Drop for Terminate { fn drop(&mut self) { { let mut reg = match self.1.lock() { Ok(g) => g, Err(e) => panic!("smarm: gen_server reg lock poisoned (core corrupt): {e}"), }; for (_, sub) in reg.oneshots.drain() { cancel_timer(sub); } for (_, p) in reg.periodics.drain() { cancel_timer(p.live); } reg.rearm_tx = None; // A fired-but-gone send already no-ops on the closed Sys channel; // this guards the armed-but-unfired re-arming case specifically. debug_assert!( reg.oneshots.is_empty() && reg.periodics.is_empty(), "an armed timer survived gen_server loop exit" ); } self.0.terminate(); } } // Unwrap an envelope and hand it to the right handler. For calls, a failed // reply send means the caller went away (e.g. cancelled while parked) — // that is the caller's problem, not the server's. 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); } Envelope::Cast(request) => state.handle_cast(request), } } // Shared timer bookkeeping: the loop keeps one clone (to retire one-shots on // fire, re-arm periodics, and read the idle window); the guard holds another // to drain on exit; every TimerHandle the state stores holds more. let reg = Arc::new(Mutex::new(TimerReg::default())); let mut guard = Terminate(state, reg.clone()); // The system intake arm: Watchers feed Monitors and armed timers feed fires // to the loop through it. The ctx (and with it the loop's own sender) drops // right after init — a state that cloned no Watcher/TimerHandle closes the // arm, the first select observes the closure, and the loop falls back to the // plain-inbox park. let (sys_tx, sys_rx) = channel::>(); // 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 = GenServerCtx { 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() { // Fast path: no extra arms, no select overhead — park directly on // the inbox. Mirrors the inbox arm of the select path below; any // change there must be applied here too. 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 { // Slow path: one or more extra arms live — build the arm slice and // select. Arm order encodes priority: downs → system → infos → // inbox. The slice is rebuilt each iteration because the monitor // and info sets shrink/grow. Mirrors the fast-path inbox park // above; keep them in sync. let nd = monitors.len(); // monitor band: [0, nd) let nw = sys_open as usize; // system arm: [nd, nd+nw) // info band: [nd+nw, nd+nw+ni) // inbox arm: [nd+nw+ni] let sel = { let mut arms: Vec<&dyn Selectable> = Vec::with_capacity(nd + nw + infos.len() + 1); for m in &monitors { arms.push(&m.rx); } if sys_open { arms.push(&sys_rx); } for r in &infos { arms.push(r); } arms.push(&rx); 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 { // Monitor band: a Down retires its arm either way (one-shot) // or closes without delivering (defensive; shouldn't happen). 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 // live set tracks only still-pending timers, then // dispatch. match reg.lock() { Ok(mut g) => { g.oneshots.remove(&id); } Err(e) => panic!("smarm: gen_server reg lock poisoned (core corrupt): {e}"), } 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 // (cancel didn't beat us here), produce its payload and // re-arm the next tick at now + every *before* dispatch, // so the period is measured from fire-handling and a // handler that cancels stops the just-armed instance. let msg = { let mut g = match reg.lock() { Ok(g) => g, Err(e) => panic!("smarm: gen_server reg lock poisoned (core corrupt): {e}"), }; let r = &mut *g; if let Some(p) = r.periodics.get_mut(&id) { let every = p.every; let msg = (p.make)(); let tx = match r.rearm_tx.as_ref() { Some(tx) => tx.clone(), None => panic!( "smarm: live periodic without rearm_tx (logic bug)" ), }; p.live = send_after_to(every, tx, Sys::Tick(id)); Some(msg) } else { // Cancelled between fire and dispatch: discard. None } }; if let Some(msg) = msg { guard.0.handle_timer(msg); reset_idle(&mut idle_deadline); } } // Single-receiver: nothing can drain the arm between // select's ready and our try_recv. Ok(None) => debug_assert!(false, "ready system arm was empty"), // Every Watcher/TimerHandle gone: stop selecting on the arm. Err(_) => sys_open = false, } } else if i < nd + nw + infos.len() { // Info band. let j = i - nd - nw; match infos[j].try_recv() { 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. Err(_) => { infos.remove(j); } } } else { // Inbox arm (mirrors the fast-path park above). match rx.try_recv() { Ok(Some(env)) => { dispatch(&mut guard.0, env); reset_idle(&mut idle_deadline); } 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!(); } }