Markk116/smarm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
acf67fef066a3aeee402b09b9867738b67922a0c
smarm/tests/gen_server.rs
T

673 lines
23 KiB
Rust
Raw Blame History

//! 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, GenServerBuilder};
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;
type Info = ();
type Timer = ();
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 = ();
type Info = ();
type Timer = ();
fn init(&mut self, _ctx: &smarm::gen_server::GenServerCtx<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)));
}
// ---------------------------------------------------------------------------
// call_timeout
// ---------------------------------------------------------------------------
use smarm::gen_server::CallTimeoutError;
use std::time::{Duration, Instant};
/// Replies after sleeping `delay_ms` (parking the server actor, not the OS
/// thread), so callers can race a deadline against the reply.
struct Slow;
impl GenServer for Slow {
type Call = u64; // delay in ms
type Reply = u64;
type Cast = ();
type Info = ();
type Timer = ();
fn handle_call(&mut self, delay_ms: u64) -> u64 {
if delay_ms > 0 {
smarm::sleep(Duration::from_millis(delay_ms));
}
delay_ms
}
fn handle_cast(&mut self, _: ()) {}
}
#[test]
fn call_timeout_returns_reply_within_deadline() {
run(|| {
let srv = start(Slow);
assert_eq!(srv.call_timeout(0, Duration::from_secs(10)), Ok(0));
});
}
#[test]
fn call_timeout_times_out_on_slow_handler() {
run(|| {
let srv = start(Slow);
let start_t = Instant::now();
let r = srv.call_timeout(500, Duration::from_millis(50));
assert_eq!(r, Err(CallTimeoutError::Timeout));
let elapsed = start_t.elapsed();
// Gave up at the deadline, not at the reply.
assert!(elapsed >= Duration::from_millis(50));
assert!(elapsed < Duration::from_millis(500));
});
}
#[test]
fn server_survives_an_abandoned_call_and_late_reply_is_discarded() {
run(|| {
let srv = start(Slow);
assert_eq!(
srv.call_timeout(100, Duration::from_millis(20)),
Err(CallTimeoutError::Timeout)
);
// The timed-out request is still handled; its reply send fails
// harmlessly (receiver dropped). The server must keep serving, and
// the late reply must not leak into THIS call's reply channel.
assert_eq!(srv.call_timeout(0, Duration::from_secs(10)), Ok(0));
// Plain unbounded call still fine too.
assert_eq!(srv.call(0), Ok(0));
});
}
#[test]
fn call_timeout_to_dead_server_is_server_down_not_timeout() {
struct Bomb;
impl GenServer for Bomb {
type Call = ();
type Info = ();
type Timer = ();
type Reply = ();
type Cast = ();
fn handle_call(&mut self, _: ()) {
panic!("kaboom");
}
fn handle_cast(&mut self, _: ()) {}
}
run(|| {
let srv = start(Bomb);
// Dies mid-call: reply channel closes -> ServerDown (even though the
// generous deadline never fires).
assert_eq!(
srv.call_timeout((), Duration::from_secs(10)),
Err(CallTimeoutError::ServerDown)
);
// Already gone: inbox send fails -> ServerDown.
assert_eq!(
srv.call_timeout((), Duration::from_secs(10)),
Err(CallTimeoutError::ServerDown)
);
});
}
// ---------------------------------------------------------------------------
// 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;
type Timer = ();
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 = GenServerBuilder::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 = GenServerBuilder::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 = GenServerBuilder::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 = GenServerBuilder::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"]);
}
// ---------------------------------------------------------------------------
// handle_down: monitors handed to the loop via Watcher (v0.8)
// ---------------------------------------------------------------------------
use smarm::gen_server::Watcher;
use smarm::{monitor, spawn, DownReason, Pid};
/// The motivating pattern: a server that spawns workers from a handler,
/// watches them, and logs their deaths.
struct Pool {
watcher: Option<Watcher<Self>>,
log: Vec<DownReason>,
}
enum PoolCast {
SpawnDoomedWorker,
Watch(Pid),
}
impl GenServer for Pool {
type Call = ();
type Reply = Vec<DownReason>;
type Cast = PoolCast;
type Info = ();
type Timer = ();
fn init(&mut self, ctx: &smarm::gen_server::GenServerCtx<Self>) {
self.watcher = Some(ctx.watcher());
}
fn handle_call(&mut self, _: ()) -> Vec<DownReason> {
self.log.clone()
}
fn handle_cast(&mut self, cast: PoolCast) {
let watcher = self.watcher.as_ref().expect("init ran first");
match cast {
PoolCast::SpawnDoomedWorker => {
let h = spawn(|| panic!("worker died"));
watcher.watch(monitor(h.pid()));
}
PoolCast::Watch(pid) => watcher.watch(monitor(pid)),
}
}
fn handle_down(&mut self, down: smarm::Down) {
self.log.push(down.reason);
}
}
// A worker spawned and watched from inside a handler delivers its Down to
// handle_down. Down arms outrank the inbox, so the death is in the log by
// the time the follow-up call is answered.
#[test]
fn worker_pool_down_reaches_handle_down() {
let got = Arc::new(Mutex::new(Vec::new()));
let got2 = got.clone();
run(move || {
let server = start(Pool { watcher: None, log: Vec::new() });
server.cast(PoolCast::SpawnDoomedWorker).unwrap();
let _ = server.call(()).unwrap(); // sync point: cast handled, worker live
*got2.lock().unwrap() = server.call(()).unwrap();
});
assert_eq!(*got.lock().unwrap(), vec![DownReason::Panic]);
}
// Watching an already-dead pid yields an immediate NoProc Down, and the down
// arm outranks the inbox: the call cast *after* the watch still observes it.
#[test]
fn watch_dead_pid_is_noproc_down() {
let got = Arc::new(Mutex::new(Vec::new()));
let got2 = got.clone();
run(move || {
let h = spawn(|| {});
let dead = h.pid();
h.join().unwrap();
let server = start(Pool { watcher: None, log: Vec::new() });
server.cast(PoolCast::Watch(dead)).unwrap();
*got2.lock().unwrap() = server.call(()).unwrap();
});
assert_eq!(*got.lock().unwrap(), vec![DownReason::NoProc]);
}
// A state that never clones the Watcher closes the control arm; the loop
// falls back to the plain-inbox park and keeps serving. (Every pre-v0.8 test
// in this file also exercises this path.)
#[test]
fn unused_ctx_closes_control_arm_silently() {
let got = Arc::new(Mutex::new(0i64));
let got2 = got.clone();
run(move || {
let server = start(Counter { n: 0 });
server.cast(Op::Add(2)).unwrap();
server.cast(Op::Add(40)).unwrap();
*got2.lock().unwrap() = server.call(Req::Get).unwrap();
});
assert_eq!(*got.lock().unwrap(), 42);
}
// ---------------------------------------------------------------------------
// RFC 015 — gen_server timers. A server that arms one-shot timers from a cast
// and records each fire's payload, plus the cancel race signal.
// ---------------------------------------------------------------------------
use smarm::gen_server::{GenServerCtx, TimerHandle};
use smarm::TimerId;
enum TkCast {
Arm(Duration),
Tick(Duration),
CancelLast,
}
struct Timed {
timer: Option<TimerHandle<Self>>,
fired: Arc<Mutex<Vec<u32>>>,
cancel_won: Arc<Mutex<Option<bool>>>,
last: Option<TimerId>,
}
impl GenServer for Timed {
type Call = ();
type Reply = usize; // count of fires so far (a sync read point)
type Cast = TkCast;
type Info = ();
type Timer = u32;
fn init(&mut self, ctx: &GenServerCtx<Self>) {
self.timer = Some(ctx.timer());
}
fn handle_call(&mut self, _: ()) -> usize {
self.fired.lock().unwrap().len()
}
fn handle_cast(&mut self, c: TkCast) {
let t = self.timer.as_ref().expect("init ran first");
match c {
TkCast::Arm(d) => self.last = Some(t.arm_after(d, 7)),
TkCast::Tick(d) => self.last = Some(t.tick_every(d, 9)),
TkCast::CancelLast => {
let id = self.last.take().expect("a timer was armed");
*self.cancel_won.lock().unwrap() = Some(t.cancel(id));
}
}
}
fn handle_timer(&mut self, msg: u32) {
self.fired.lock().unwrap().push(msg);
}
}
fn timed(fired: Arc<Mutex<Vec<u32>>>, cancel_won: Arc<Mutex<Option<bool>>>) -> Timed {
Timed { timer: None, fired, cancel_won, last: None }
}
// A one-shot armed from a handler fires into handle_timer with its payload.
#[test]
fn arm_after_fires_into_handle_timer() {
let fired = Arc::new(Mutex::new(Vec::new()));
let f2 = fired.clone();
run(move || {
let cw = Arc::new(Mutex::new(None));
let server = start(timed(f2, cw));
server.cast(TkCast::Arm(Duration::from_millis(10))).unwrap();
let _ = server.call(()).unwrap(); // sync: arm done
smarm::sleep(Duration::from_millis(40)); // let the timer fire
let count = server.call(()).unwrap(); // timer arm outranks this inbox call
assert_eq!(count, 1, "the one-shot should have fired exactly once");
});
assert_eq!(*fired.lock().unwrap(), vec![7]);
}
// cancel before the deadline wins the race (returns true) and suppresses the
// fire entirely.
#[test]
fn cancel_before_fire_suppresses_it() {
let fired = Arc::new(Mutex::new(Vec::new()));
let cancel_won = Arc::new(Mutex::new(None));
let f2 = fired.clone();
let c2 = cancel_won.clone();
run(move || {
let server = start(timed(f2, c2));
server.cast(TkCast::Arm(Duration::from_millis(50))).unwrap();
server.cast(TkCast::CancelLast).unwrap();
let _ = server.call(()).unwrap(); // sync: arm + cancel both handled
smarm::sleep(Duration::from_millis(80)); // past the original deadline
let count = server.call(()).unwrap();
assert_eq!(count, 0, "cancelled timer must not fire");
});
assert_eq!(*cancel_won.lock().unwrap(), Some(true), "cancel beat the fire");
assert!(fired.lock().unwrap().is_empty());
}
// tick_every re-arms: a periodic fires repeatedly off one arm, each tick
// carrying a fresh payload. (Timer fires outrank the inbox by arm position, so
// a periodic cannot be starved by userspace traffic — the same property the
// info_outranks_inbox test pins for infos.)
#[test]
fn tick_every_rearms_repeatedly() {
let fired = Arc::new(Mutex::new(Vec::new()));
let f2 = fired.clone();
run(move || {
let cw = Arc::new(Mutex::new(None));
let server = start(timed(f2, cw));
server.cast(TkCast::Tick(Duration::from_millis(20))).unwrap();
let _ = server.call(()).unwrap(); // sync: periodic armed
smarm::sleep(Duration::from_millis(130)); // ~6 periods
let count = server.call(()).unwrap();
assert!(count >= 3, "periodic should have re-armed several times, got {count}");
});
// Every tick delivered the same payload.
assert!(fired.lock().unwrap().iter().all(|&v| v == 9));
}
// Cancelling a periodic stops the re-arm: no further ticks land after cancel.
#[test]
fn cancel_stops_a_periodic() {
let fired = Arc::new(Mutex::new(Vec::new()));
let cancel_won = Arc::new(Mutex::new(None));
let f2 = fired.clone();
let c2 = cancel_won.clone();
run(move || {
let server = start(timed(f2, c2));
server.cast(TkCast::Tick(Duration::from_millis(20))).unwrap();
let _ = server.call(()).unwrap();
smarm::sleep(Duration::from_millis(70)); // a few ticks
server.cast(TkCast::CancelLast).unwrap();
let after_cancel = server.call(()).unwrap(); // sync: cancel handled
smarm::sleep(Duration::from_millis(80)); // would be several more ticks
let later = server.call(()).unwrap();
assert_eq!(later, after_cancel, "no ticks may land after cancel");
});
// The periodic had fired at least once before being cancelled.
assert!(!fired.lock().unwrap().is_empty());
}
// ---------------------------------------------------------------------------
// RFC 015 §4.4 — idle / receive timeout. A server that sets an idle window in
// init and counts handle_idle fires; casts are traffic that resets the window.
// ---------------------------------------------------------------------------
struct Idler {
window: Duration,
idles: Arc<Mutex<u32>>,
}
impl GenServer for Idler {
type Call = ();
type Reply = u32; // idle fire count
type Cast = (); // a poke: traffic that resets the idle window
type Info = ();
type Timer = ();
fn init(&mut self, ctx: &GenServerCtx<Self>) {
ctx.idle_after(self.window);
}
fn handle_call(&mut self, _: ()) -> u32 {
*self.idles.lock().unwrap()
}
fn handle_cast(&mut self, _: ()) {}
fn handle_idle(&mut self) {
*self.idles.lock().unwrap() += 1;
}
}
// A quiet inbox fires handle_idle, and the window re-arms (steady detector):
// several fires across a quiet span.
#[test]
fn idle_fires_repeatedly_on_quiet() {
let idles = Arc::new(Mutex::new(0));
let i2 = idles.clone();
run(move || {
let server = start(Idler { window: Duration::from_millis(25), idles: i2 });
smarm::sleep(Duration::from_millis(130)); // quiet ⇒ ~5 windows
drop(server); // keep the server alive across the quiet span
});
assert!(*idles.lock().unwrap() >= 2, "idle should re-arm and fire several times");
}
// Traffic within the window keeps idle from firing; only once the inbox goes
// quiet does handle_idle fire.
#[test]
fn traffic_resets_the_idle_window() {
let idles = Arc::new(Mutex::new(0));
let i2 = idles.clone();
let before_quiet = Arc::new(Mutex::new(u32::MAX));
let bq = before_quiet.clone();
run(move || {
let server = start(Idler { window: Duration::from_millis(60), idles: i2 });
// Poke every 25ms (< 60ms window) for ~100ms: each cast resets the
// window before it can elapse.
for _ in 0..4 {
server.cast(()).unwrap();
smarm::sleep(Duration::from_millis(25));
}
*bq.lock().unwrap() = server.call(()).unwrap(); // count while traffic kept it quiet-free
smarm::sleep(Duration::from_millis(140)); // now genuinely quiet
drop(server);
});
assert_eq!(*before_quiet.lock().unwrap(), 0, "steady traffic must suppress idle");
assert!(*idles.lock().unwrap() >= 1, "idle fires once the inbox falls quiet");
}
// RFC 015 §4.7 — no armed timer survives loop exit. A server with a live
// periodic is dropped; the loop's drop guard drains and cancels it (a surviving
// re-arming timer would trip the in-Drop debug_assert), runs terminate, and no
// further tick lands after exit.
#[test]
fn no_timer_survives_exit() {
let fired = Arc::new(Mutex::new(Vec::new()));
let f_server = fired.clone();
let f_read = fired.clone();
run(move || {
let server = start(timed(f_server, Arc::new(Mutex::new(None))));
server.cast(TkCast::Tick(Duration::from_millis(15))).unwrap();
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.
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
assert_eq!(f_read.lock().unwrap().len(), at_exit, "no tick may fire after exit");
});
}
Reference in New Issue View Git Blame Copy Permalink