Markk116/smarm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

gen_server: idle/receive timeout via select_timeout/recv_timeout + handle_idle (RFC 015 §4.4)

Browse Source
This commit is contained in:
smarm-agent
2026-06-18 19:17:20 +00:00
parent c0cfa01f37
commit 8c8af55928
2 changed files with 159 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/gen_server.rs
+93 -15
View File
@@ -49,16 +49,17 @@
//! Revisit against a real consumer. (Monitor `Down` forwarding is dynamic —
//! see `handle_down` — because monitors are inherently created at runtime.)
use crate::channel::{channel, select, Receiver, Selectable, Sender};
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;
use std::time::{Duration, Instant};
/// Behaviour for a gen_server: a state value plus call/cast handlers.
///
@@ -279,6 +280,11 @@ enum Sys<G: GenServer> {
pub struct ServerCtx<G: GenServer> {
sys_tx: Sender<Sys<G>>,
reg: Arc<Mutex<TimerReg<G>>>,
/// 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 `ServerCtx` is only ever
/// borrowed on the actor's own stack during `init`, never sent.
idle: Cell<Option<Duration>>,
}
impl<G: GenServer> ServerCtx<G> {
@@ -301,6 +307,20 @@ impl<G: GenServer> ServerCtx<G> {
pub fn timer(&self) -> TimerHandle<G> {
TimerHandle { sys_tx: self.sys_tx.clone(), reg: self.reg.clone() }
}
/// Set the idle / receive-timeout window: if no message of any kind is
/// dispatched for `after`, the loop fires [`GenServer::handle_idle`]
/// (RFC 015 §4.4). Set once, in `init`; the window is loop-owned and fixed
/// (dynamic per-message reconfiguration is a non-goal). The window resets on
/// every dispatched message and re-arms after `handle_idle` (a steady idle
/// detector); a server wanting one-shot idle-shutdown requests its own stop
/// in `handle_idle`.
///
/// Distinct from [`ServerRef::call_timeout`], which is a *client-side* call
/// deadline — same word, different axis (§7).
pub fn idle_after(&self, after: Duration) {
self.idle.set(Some(after));
}
}
/// Per-server timer bookkeeping, shared between the loop and every
@@ -717,30 +737,63 @@ fn server_loop<G: GenServer>(
// loop falls back to the plain-inbox park.
let (sys_tx, sys_rx) = channel::<Sys<G>>();
// Shared timer bookkeeping: the loop keeps one clone (to retire one-shots
// on fire, and — later — to drain on exit); every TimerHandle the state
// stores holds another.
// on fire, and to drain on exit); every TimerHandle the state stores holds
// another.
let reg = Arc::new(Mutex::new(TimerReg::default()));
guard.0.init(&ServerCtx { sys_tx, reg: reg.clone() });
// 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 = ServerCtx { sys_tx, reg: reg.clone(), idle: Cell::new(None) };
guard.0.init(&ctx);
let idle = ctx.idle.get();
drop(ctx);
let mut monitors: Vec<Monitor> = 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<Instant> = idle.map(|d| Instant::now() + d);
let reset_idle = |dl: &mut Option<Instant>| {
if let Some(d) = idle {
*dl = Some(Instant::now() + d);
}
};
loop {
if monitors.is_empty() && !sys_open && infos.is_empty() {
// Nothing to select over: park on the inbox alone, exactly the
// pre-v0.8 loop.
match rx.recv() {
Ok(env) => dispatch(&mut guard.0, env),
// All ServerRefs dropped → inbox closed → graceful shutdown.
Err(_) => break,
// Nothing to select over: park on the inbox alone (the pre-v0.8
// loop), with the idle window as the recv timeout when one is set.
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 {
// Arm priority: downs, then the system arm, then infos (each in
// declaration order), then the inbox. The arm slice is rebuilt
// per iteration because every set but the inbox shrinks or grows.
// The whole wait carries the idle window as its timeout.
let nd = monitors.len();
let nw = sys_open as usize;
let i = {
let sel = {
let mut arms: Vec<&dyn Selectable> =
Vec::with_capacity(nd + nw + infos.len() + 1);
for m in &monitors {
@@ -753,7 +806,22 @@ fn server_loop<G: GenServer>(
arms.push(r);
}
arms.push(&rx);
select(&arms)
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 {
// A Down retires its arm either way: delivered (one-shot) or
@@ -761,9 +829,11 @@ fn server_loop<G: GenServer>(
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
@@ -771,6 +841,7 @@ fn server_loop<G: GenServer>(
// dispatch.
reg.lock().unwrap().oneshots.remove(&id);
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
@@ -798,6 +869,7 @@ fn server_loop<G: GenServer>(
};
if let Some(msg) = msg {
guard.0.handle_timer(msg);
reset_idle(&mut idle_deadline);
}
}
// Single-receiver: nothing can drain the arm between
@@ -809,7 +881,10 @@ fn server_loop<G: GenServer>(
} else if i < nd + nw + infos.len() {
let j = i - nd - nw;
match infos[j].try_recv() {
Ok(Some(info)) => guard.0.handle_info(info),
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.
@@ -819,7 +894,10 @@ fn server_loop<G: GenServer>(
}
} else {
match rx.try_recv() {
Ok(Some(env)) => dispatch(&mut guard.0, env),
Ok(Some(env)) => {
dispatch(&mut guard.0, env);
reset_idle(&mut idle_deadline);
}
Ok(None) => debug_assert!(false, "ready inbox was empty"),
Err(_) => break,
}