feat(gen_server,gen_statem): lifetime is the actor's — refs are addresses; inline named run
Root cause behind the "pin the endpoint" gotcha and the trapping-wrapper
pattern: a gen_server had two lifetime authorities — its refs (last one
dropped → inbox closes → exit) and, when supervised, its supervisor. OTP has
one: a process lives until it stops, is shut down, or is killed; a pid is an
address. Root exit now shutting down every forest root removes the reason the
ref-governed idiom existed (a forgotten server no longer hangs the run), so
adopt the one rule:
- The server/machine loop holds one inbox sender for its life; the inbox
never closes. GenServerRef / GenStatemRef are addresses. Explicit close is
`shutdown()`; a forgotten one is swept at root exit.
- `NamedGenServerBuilder::run()` / `gen_statem::run_named(name, m)` run the
loop inline as the current actor: a server is a direct ChildSpec child,
gets the supervisor's shutdown as handle_shutdown / a shutdown row, re-binds
its name on restart, and is addressed by name. The wrapper in
examples/graceful_shutdown.rs is gone.
- gen_statem gains GenStatemName + whereis_machine/send/call/shutdown by name
(parity with gen_server); the macro gets `Sm::new`.
- Root-exit sweep records `Event::RootSweep { target, trapping }` under
smarm-trace ("root_sweep shutdown|stopped"): unsupervised leftovers are
visible rather than silently owned-by-refs.
- Named start() name-clash path stops the spawned actor instead of relying
on ref drop.
Tests: tests/gen_server_lifetime.rs, tests/gen_statem_lifetime.rs,
tests/root_sweep_trace.rs (feature-gated); three existing tests that used
drop-closes-inbox now use shutdown(). Docs/README/ROADMAP/Deep Dive updated.
This commit is contained in:
Claude (sandbox)
+6
-6
@@ -88,7 +88,7 @@ impl GenServer for Lifecycle {
|
||||
}
|
||||
}
|
||||
|
||||
// init -> handle_call -> (drop last ref closes inbox) -> terminate.
|
||||
// init -> handle_call -> shutdown -> terminate.
|
||||
#[test]
|
||||
fn init_and_terminate_run() {
|
||||
let log = Arc::new(Mutex::new(Vec::new()));
|
||||
@@ -96,9 +96,9 @@ fn init_and_terminate_run() {
|
||||
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);
|
||||
// Refs are addresses: dropping one does not end the server. The
|
||||
// explicit close does, and waits for terminate.
|
||||
server.shutdown();
|
||||
});
|
||||
assert_eq!(*log.lock().unwrap(), vec!["init", "call", "terminate"]);
|
||||
}
|
||||
@@ -702,8 +702,8 @@ fn no_timer_survives_exit() {
|
||||
let _ = server.call(()).unwrap(); // sync: periodic armed
|
||||
smarm::sleep(Duration::from_millis(45)); // a couple of ticks
|
||||
let mon = smarm::monitor(server.pid());
|
||||
drop(server); // inbox closes → loop exits → guard drains timers
|
||||
// Clean Down ⇒ the loop returned without the no-leak assert aborting.
|
||||
server.shutdown(); // loop exits → guard drains timers
|
||||
// Clean Down ⇒ the loop returned without the no-leak assert aborting.
|
||||
assert!(mon.rx.recv().is_ok());
|
||||
let at_exit = f_read.lock().unwrap().len();
|
||||
smarm::sleep(Duration::from_millis(90)); // would be several more ticks
|
||||
|
||||
@@ -0,0 +1,243 @@
|
||||
//! gen_server lifetime is the actor's, not its refs' (OTP: a pid is an
|
||||
//! address, a process lives until it stops, is shut down, or is killed).
|
||||
//!
|
||||
//! - Dropping the last `GenServerRef` does NOT end the server. It ends via
|
||||
//! `StopHandle::stop`, `request_shutdown` / `GenServerRef::shutdown`,
|
||||
//! `request_stop`, or a handler panic.
|
||||
//! - `GenServerBuilder::named(N).run()` runs the loop inline as the *current*
|
||||
//! actor, so a server is a direct `ChildSpec` child: the supervisor's
|
||||
//! shutdown reaches it as `handle_shutdown`, a restart re-binds the name,
|
||||
//! and by-name `call`/`cast` reach whichever incarnation is live.
|
||||
|
||||
use smarm::gen_server::{
|
||||
self, GenServer, GenServerBuilder, GenServerCtx, GenServerName, ShutdownAction, StopHandle,
|
||||
};
|
||||
use smarm::registry::RegisterError;
|
||||
use smarm::supervisor::{ChildSpec, OneForOne, Restart, Shutdown};
|
||||
use smarm::{monitor, request_shutdown, run, sleep, spawn, DownReason};
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
#[derive(Default, Clone)]
|
||||
struct Log(Arc<Mutex<Vec<String>>>);
|
||||
impl Log {
|
||||
fn push(&self, s: impl Into<String>) {
|
||||
self.0.lock().unwrap().push(s.into());
|
||||
}
|
||||
fn get(&self) -> Vec<String> {
|
||||
self.0.lock().unwrap().clone()
|
||||
}
|
||||
}
|
||||
|
||||
struct Counter {
|
||||
log: Log,
|
||||
n: u64,
|
||||
trap: bool,
|
||||
stop: Option<StopHandle<Counter>>,
|
||||
}
|
||||
|
||||
enum Call {
|
||||
Get,
|
||||
}
|
||||
enum Cast {
|
||||
Inc,
|
||||
Stop,
|
||||
}
|
||||
|
||||
impl GenServer for Counter {
|
||||
type Call = Call;
|
||||
type Reply = u64;
|
||||
type Cast = Cast;
|
||||
type Info = ();
|
||||
type Timer = ();
|
||||
|
||||
fn init(&mut self, ctx: &GenServerCtx<Self>) {
|
||||
if self.trap {
|
||||
ctx.trap_exit();
|
||||
}
|
||||
self.stop = Some(ctx.stop_handle());
|
||||
self.log.push("init");
|
||||
}
|
||||
fn handle_call(&mut self, Call::Get: Call) -> u64 {
|
||||
self.n
|
||||
}
|
||||
fn handle_cast(&mut self, c: Cast) {
|
||||
match c {
|
||||
Cast::Inc => self.n += 1,
|
||||
Cast::Stop => self.stop.as_ref().unwrap().stop(),
|
||||
}
|
||||
}
|
||||
fn handle_shutdown(&mut self) -> ShutdownAction {
|
||||
self.log.push("handle_shutdown");
|
||||
ShutdownAction::Exit
|
||||
}
|
||||
fn terminate(&mut self) {
|
||||
self.log.push("terminate");
|
||||
}
|
||||
}
|
||||
|
||||
fn counter(log: &Log, trap: bool) -> Counter {
|
||||
Counter {
|
||||
log: log.clone(),
|
||||
n: 0,
|
||||
trap,
|
||||
stop: None,
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Refs are addresses: dropping the last one does not end the server.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[test]
|
||||
fn dropping_last_ref_does_not_end_server() {
|
||||
let log = Log::default();
|
||||
let l = log.clone();
|
||||
run(move || {
|
||||
let srv = gen_server::start(counter(&l, true));
|
||||
let pid = srv.pid();
|
||||
srv.cast(Cast::Inc).unwrap();
|
||||
assert_eq!(srv.call(Call::Get).unwrap(), 1);
|
||||
let mon = monitor(pid);
|
||||
drop(srv);
|
||||
sleep(Duration::from_millis(30));
|
||||
assert!(
|
||||
mon.rx.try_recv().unwrap().is_none(),
|
||||
"server must outlive its last ref"
|
||||
);
|
||||
assert_eq!(l.get(), vec!["init"], "terminate must not have run");
|
||||
// Explicit teardown still works, and is what ends it.
|
||||
request_shutdown(pid);
|
||||
let down = mon.rx.recv().unwrap();
|
||||
assert_eq!(down.reason, DownReason::Exit);
|
||||
});
|
||||
assert_eq!(log.get(), vec!["init", "handle_shutdown", "terminate"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ref_shutdown_is_the_explicit_close() {
|
||||
let log = Log::default();
|
||||
let l = log.clone();
|
||||
run(move || {
|
||||
let srv = gen_server::start(counter(&l, true));
|
||||
srv.call(Call::Get).unwrap(); // sync: init (and trap_exit) has run
|
||||
srv.shutdown(); // graceful, waits
|
||||
assert_eq!(l.get(), vec!["init", "handle_shutdown", "terminate"]);
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn forgotten_server_is_shut_down_at_root_exit() {
|
||||
// A ref-less server is not a hung run: root exit shuts it down.
|
||||
let log = Log::default();
|
||||
let l = log.clone();
|
||||
run(move || {
|
||||
let srv = gen_server::start(counter(&l, false));
|
||||
drop(srv);
|
||||
});
|
||||
assert_eq!(log.get(), vec!["init", "terminate"]);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Inline run: a gen_server as a direct ChildSpec child.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const COUNTER: GenServerName<Counter> = GenServerName::new("lifetime-counter");
|
||||
|
||||
#[test]
|
||||
fn named_run_is_a_direct_supervised_child_and_gets_shutdown() {
|
||||
let log = Log::default();
|
||||
let l = log.clone();
|
||||
run(move || {
|
||||
let l2 = l.clone();
|
||||
let sup = spawn(move || {
|
||||
let l3 = l2.clone();
|
||||
OneForOne::new()
|
||||
.child(
|
||||
ChildSpec::new(Restart::Permanent, move || {
|
||||
GenServerBuilder::new(counter(&l3, true))
|
||||
.named(COUNTER)
|
||||
.run()
|
||||
.expect("name free");
|
||||
})
|
||||
.shutdown(Shutdown::Infinity),
|
||||
)
|
||||
.run();
|
||||
});
|
||||
sleep(Duration::from_millis(10));
|
||||
gen_server::cast(COUNTER, Cast::Inc).unwrap();
|
||||
assert_eq!(gen_server::call(COUNTER, Call::Get).unwrap(), 1);
|
||||
request_shutdown(sup.pid());
|
||||
sup.join()
|
||||
.expect("ordered shutdown, supervisor returns normally");
|
||||
assert_eq!(l.get(), vec!["init", "handle_shutdown", "terminate"]);
|
||||
assert!(gen_server::whereis_server(COUNTER).is_none());
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn named_run_child_restarts_and_rebinds_name() {
|
||||
let log = Log::default();
|
||||
let inits = Arc::new(AtomicUsize::new(0));
|
||||
let l = log.clone();
|
||||
let i = inits.clone();
|
||||
run(move || {
|
||||
let l2 = l.clone();
|
||||
let i2 = i.clone();
|
||||
let sup = spawn(move || {
|
||||
let l3 = l2.clone();
|
||||
let i3 = i2.clone();
|
||||
OneForOne::new()
|
||||
.child(ChildSpec::new(Restart::Permanent, move || {
|
||||
i3.fetch_add(1, Ordering::SeqCst);
|
||||
GenServerBuilder::new(counter(&l3, false))
|
||||
.named(COUNTER)
|
||||
.run()
|
||||
.expect("name free on (re)start");
|
||||
}))
|
||||
.run();
|
||||
});
|
||||
sleep(Duration::from_millis(10));
|
||||
gen_server::cast(COUNTER, Cast::Inc).unwrap();
|
||||
assert_eq!(gen_server::call(COUNTER, Call::Get).unwrap(), 1);
|
||||
// Normal self-exit → Permanent restarts it, fresh state, same name.
|
||||
gen_server::cast(COUNTER, Cast::Stop).unwrap();
|
||||
sleep(Duration::from_millis(30));
|
||||
assert_eq!(i.load(Ordering::SeqCst), 2, "restarted once");
|
||||
assert_eq!(gen_server::call(COUNTER, Call::Get).unwrap(), 0);
|
||||
request_shutdown(sup.pid());
|
||||
sup.join().unwrap();
|
||||
});
|
||||
assert_eq!(log.get(), vec!["init", "terminate", "init", "terminate"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn named_run_name_clash_fails_before_init() {
|
||||
let log = Log::default();
|
||||
let l = log.clone();
|
||||
run(move || {
|
||||
let first = GenServerBuilder::new(counter(&l, false))
|
||||
.named(COUNTER)
|
||||
.start()
|
||||
.unwrap();
|
||||
let l2 = l.clone();
|
||||
let res = Arc::new(Mutex::new(None));
|
||||
let r2 = res.clone();
|
||||
let first_pid = first.pid();
|
||||
spawn(move || {
|
||||
let r = GenServerBuilder::new(counter(&l2, false))
|
||||
.named(COUNTER)
|
||||
.run();
|
||||
*r2.lock().unwrap() = Some(r);
|
||||
})
|
||||
.join()
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
*res.lock().unwrap(),
|
||||
Some(Err(RegisterError::NameTaken { holder: first_pid }))
|
||||
);
|
||||
assert_eq!(l.get(), vec!["init"], "clashing server never ran init");
|
||||
first.shutdown();
|
||||
});
|
||||
}
|
||||
@@ -196,12 +196,18 @@ fn linked_peer_death_reaches_handle_exit() {
|
||||
r.cast(Cast::Note("still-serving")).unwrap();
|
||||
sleep(Duration::from_millis(20));
|
||||
a.store(true, Ordering::SeqCst);
|
||||
let mon = monitor(r.pid());
|
||||
drop(r); // inbox closes → clean exit
|
||||
let _ = mon.rx.recv(); // don't let the root-exit sweep race terminate
|
||||
r.shutdown(); // graceful; waits for terminate
|
||||
});
|
||||
assert!(alive.load(Ordering::SeqCst));
|
||||
assert_eq!(log.get(), vec!["handle_exit", "still-serving", "terminate"]);
|
||||
assert_eq!(
|
||||
log.get(),
|
||||
vec![
|
||||
"handle_exit",
|
||||
"still-serving",
|
||||
"handle_shutdown",
|
||||
"terminate"
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -0,0 +1,128 @@
|
||||
//! gen_statem lifetime parity with gen_server: a machine lives until it
|
||||
//! stops, is shut down, or is killed — its refs are addresses. And
|
||||
//! `gen_statem::run_named` runs a machine inline as the current actor, so it
|
||||
//! is a direct `ChildSpec` child addressed by name.
|
||||
|
||||
use smarm::gen_statem::{self, GenStatemName, Reply};
|
||||
use smarm::supervisor::{ChildSpec, OneForOne, Restart, Shutdown};
|
||||
use smarm::{monitor, request_shutdown, run, sleep, spawn, DownReason};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
#[derive(Default, Clone)]
|
||||
struct Log(Arc<Mutex<Vec<&'static str>>>);
|
||||
impl Log {
|
||||
fn push(&self, e: &'static str) {
|
||||
self.0.lock().unwrap().push(e);
|
||||
}
|
||||
fn get(&self) -> Vec<&'static str> {
|
||||
self.0.lock().unwrap().clone()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
|
||||
enum S {
|
||||
On,
|
||||
}
|
||||
|
||||
struct D {
|
||||
log: Log,
|
||||
trap: bool,
|
||||
n: u64,
|
||||
}
|
||||
|
||||
enum Cast {
|
||||
Inc,
|
||||
StopNow,
|
||||
}
|
||||
enum Call {
|
||||
Get(Reply<u64>),
|
||||
}
|
||||
|
||||
smarm::gen_statem! {
|
||||
machine: Sm { state: S, data: D };
|
||||
event: Ev { cast: Cast, call: Call, info: () };
|
||||
context(data, prev, cx);
|
||||
|
||||
enter {
|
||||
S::On => { data.log.push("enter"); if data.trap { cx.trap_exit() } },
|
||||
}
|
||||
|
||||
on S::On => {
|
||||
cast Cast::Inc => { data.n += 1; prev },
|
||||
cast Cast::StopNow => stop,
|
||||
call Call::Get(r) => { r.reply(data.n); prev },
|
||||
shutdown => { data.log.push("shutdown"); cx.stop(); prev },
|
||||
state_timeout => unhandled,
|
||||
timeout _ => unhandled,
|
||||
}
|
||||
|
||||
terminate { data.log.push("terminate"); }
|
||||
}
|
||||
|
||||
fn d(log: &Log, trap: bool) -> D {
|
||||
D {
|
||||
log: log.clone(),
|
||||
trap,
|
||||
n: 0,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn dropping_last_ref_does_not_end_machine() {
|
||||
let log = Log::default();
|
||||
let l = log.clone();
|
||||
run(move || {
|
||||
let m = Sm::start(S::On, d(&l, true));
|
||||
let pid = m.pid();
|
||||
m.send(Ev::Cast(Cast::Inc)).unwrap();
|
||||
assert_eq!(m.call(|r| Ev::Call(Call::Get(r))).unwrap(), 1);
|
||||
let mon = monitor(pid);
|
||||
drop(m);
|
||||
sleep(Duration::from_millis(30));
|
||||
assert!(
|
||||
mon.rx.try_recv().unwrap().is_none(),
|
||||
"machine must outlive its refs"
|
||||
);
|
||||
assert_eq!(l.get(), vec!["enter"]);
|
||||
request_shutdown(pid);
|
||||
assert_eq!(mon.rx.recv().unwrap().reason, DownReason::Exit);
|
||||
});
|
||||
assert_eq!(log.get(), vec!["enter", "shutdown", "terminate"]);
|
||||
}
|
||||
|
||||
const SM: GenStatemName<Sm> = GenStatemName::new("lifetime-sm");
|
||||
|
||||
#[test]
|
||||
fn run_named_is_a_direct_supervised_child() {
|
||||
let log = Log::default();
|
||||
let l = log.clone();
|
||||
run(move || {
|
||||
let l2 = l.clone();
|
||||
let sup = spawn(move || {
|
||||
let l3 = l2.clone();
|
||||
OneForOne::new()
|
||||
.child(
|
||||
ChildSpec::new(Restart::Permanent, move || {
|
||||
gen_statem::run_named(SM, Sm::new(S::On, d(&l3, true))).expect("name free");
|
||||
})
|
||||
.shutdown(Shutdown::Infinity),
|
||||
)
|
||||
.run();
|
||||
});
|
||||
sleep(Duration::from_millis(10));
|
||||
gen_statem::send(SM, Ev::Cast(Cast::Inc)).unwrap();
|
||||
assert_eq!(gen_statem::call(SM, |r| Ev::Call(Call::Get(r))).unwrap(), 1);
|
||||
// Normal self-exit → Permanent restart → fresh data, same name.
|
||||
gen_statem::send(SM, Ev::Cast(Cast::StopNow)).unwrap();
|
||||
sleep(Duration::from_millis(30));
|
||||
assert_eq!(gen_statem::call(SM, |r| Ev::Call(Call::Get(r))).unwrap(), 0);
|
||||
request_shutdown(sup.pid());
|
||||
sup.join().unwrap();
|
||||
assert!(gen_statem::whereis_machine(SM).is_none());
|
||||
});
|
||||
assert_eq!(
|
||||
log.get(),
|
||||
vec!["enter", "terminate", "enter", "shutdown", "terminate"]
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
//! Under `smarm-trace`, every actor the root-exit sweep reaches is recorded
|
||||
//! as a `root_sweep` event — the way to *see* unsupervised leftovers. One test
|
||||
//! per binary: the trace file is process-global.
|
||||
#![cfg(feature = "smarm-trace")]
|
||||
|
||||
use smarm::gen_server::{self, GenServer, GenServerCtx};
|
||||
use smarm::run;
|
||||
|
||||
struct Quiet;
|
||||
impl GenServer for Quiet {
|
||||
type Call = ();
|
||||
type Reply = ();
|
||||
type Cast = ();
|
||||
type Info = ();
|
||||
type Timer = ();
|
||||
fn init(&mut self, _: &GenServerCtx<Self>) {}
|
||||
fn handle_call(&mut self, _: ()) {}
|
||||
fn handle_cast(&mut self, _: ()) {}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn forgotten_server_shows_up_as_root_sweep() {
|
||||
let path = std::env::temp_dir().join(format!("smarm_root_sweep_{}.json", std::process::id()));
|
||||
std::env::set_var("SMARM_TRACE_FILE", &path);
|
||||
run(|| {
|
||||
let srv = gen_server::start(Quiet);
|
||||
srv.call(()).unwrap();
|
||||
drop(srv); // forgotten: nobody supervises it, nobody holds it
|
||||
});
|
||||
let trace = std::fs::read_to_string(&path).expect("trace file written");
|
||||
let _ = std::fs::remove_file(&path);
|
||||
assert!(
|
||||
trace.contains("root_sweep stopped"),
|
||||
"expected a root_sweep line for the non-trapping leftover; got:\n{trace}"
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user