gen_server: type Timer + handle_timer/handle_idle; fold control into Sys channel (RFC 015 §4.5, §6)
This commit is contained in:
@@ -24,6 +24,7 @@ impl GenServer for Counter {
|
|||||||
type Reply = u64;
|
type Reply = u64;
|
||||||
type Cast = Update;
|
type Cast = Update;
|
||||||
type Info = ();
|
type Info = ();
|
||||||
|
type Timer = ();
|
||||||
|
|
||||||
fn handle_call(&mut self, q: Query) -> u64 {
|
fn handle_call(&mut self, q: Query) -> u64 {
|
||||||
match q {
|
match q {
|
||||||
|
|||||||
+84
-29
@@ -73,12 +73,25 @@ pub trait GenServer: Send + 'static {
|
|||||||
/// ([`ServerBuilder::with_info`]). Servers with several out-of-band
|
/// ([`ServerBuilder::with_info`]). Servers with several out-of-band
|
||||||
/// sources enum them up into one `Info`. Use `()` if unused.
|
/// sources enum them up into one `Info`. Use `()` if unused.
|
||||||
type Info: Send + 'static;
|
type Info: Send + 'static;
|
||||||
|
/// The server's own scheduled-timer payload, delivered to
|
||||||
|
/// [`handle_timer`](Self::handle_timer) when a timer armed via the loop's
|
||||||
|
/// [`TimerHandle`] fires (RFC 015). Kept distinct from
|
||||||
|
/// [`Info`](Self::Info) — `Info` is *external* out-of-band traffic, `Timer`
|
||||||
|
/// is the server's *own* fires — so the system/userspace split stays
|
||||||
|
/// legible and a tagged timer round-trips
|
||||||
|
/// (`arm_after(d, Tk::Retry(n))` → `handle_timer(Tk::Retry(n))`). Use `()`
|
||||||
|
/// if unused, exactly as `Info` does.
|
||||||
|
type Timer: Send + 'static;
|
||||||
|
|
||||||
/// Runs once inside the server actor before any message is handled. The
|
/// Runs once inside the server actor before any message is handled. The
|
||||||
/// [`ServerCtx`] is the loop's one runtime hook: clone its [`Watcher`]
|
/// [`ServerCtx`] is the loop's one runtime hook: clone its [`Watcher`]
|
||||||
/// into the state here to be able to [`watch`](Watcher::watch) monitors
|
/// into the state here to be able to [`watch`](Watcher::watch) monitors
|
||||||
/// from any later handler.
|
/// from any later handler.
|
||||||
fn init(&mut self, _ctx: &ServerCtx) {}
|
fn init(&mut self, _ctx: &ServerCtx<Self>)
|
||||||
|
where
|
||||||
|
Self: Sized,
|
||||||
|
{
|
||||||
|
}
|
||||||
|
|
||||||
/// Handle a synchronous call and produce the reply sent back to the caller.
|
/// Handle a synchronous call and produce the reply sent back to the caller.
|
||||||
fn handle_call(&mut self, request: Self::Call) -> Self::Reply;
|
fn handle_call(&mut self, request: Self::Call) -> Self::Reply;
|
||||||
@@ -94,6 +107,21 @@ pub trait GenServer: Send + 'static {
|
|||||||
/// [`Watcher::watch`]. Default: drop it.
|
/// [`Watcher::watch`]. Default: drop it.
|
||||||
fn handle_down(&mut self, _down: Down) {}
|
fn handle_down(&mut self, _down: Down) {}
|
||||||
|
|
||||||
|
/// Handle a fired timer armed through the loop's [`TimerHandle`]
|
||||||
|
/// ([`arm_after`](TimerHandle::arm_after) /
|
||||||
|
/// [`tick_every`](TimerHandle::tick_every)). Timer fires re-enter at system
|
||||||
|
/// priority — above infos and the inbox — so a heartbeat cannot be starved
|
||||||
|
/// by userspace traffic. Default: drop it (RFC 015 §6).
|
||||||
|
fn handle_timer(&mut self, _msg: Self::Timer) {}
|
||||||
|
|
||||||
|
/// Handle a receive/idle timeout: fired when the loop has waited a full
|
||||||
|
/// idle window (set once via [`ServerCtx::idle_after`]) with no message of
|
||||||
|
/// any kind dispatched. The window resets on every dispatched message and
|
||||||
|
/// re-arms after this fires (a steady idle detector); a server wanting
|
||||||
|
/// one-shot idle-shutdown simply requests its own stop here. Default: no-op
|
||||||
|
/// (RFC 015 §4.4, §6).
|
||||||
|
fn handle_idle(&mut self) {}
|
||||||
|
|
||||||
/// Runs as the server actor exits, on any exit path (see module docs).
|
/// Runs as the server actor exits, on any exit path (see module docs).
|
||||||
fn terminate(&mut self) {}
|
fn terminate(&mut self) {}
|
||||||
}
|
}
|
||||||
@@ -221,16 +249,34 @@ impl<G: GenServer> ServerRef<G> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// The server loop's runtime hook, passed to [`GenServer::init`]. Currently
|
/// The server loop's internal *system intake* channel (RFC 015 §4.5). Control
|
||||||
/// carries only the [`Watcher`]; opaque so fields can grow without breaking.
|
/// (monitor handoff) and armed-timer fires are structurally the same — both are
|
||||||
pub struct ServerCtx {
|
/// loop-internal sources selected above the inbox — so they fold into one
|
||||||
watcher: Watcher,
|
/// channel, dispatched by variant. The arm stays open while any [`Watcher`] or
|
||||||
|
/// [`TimerHandle`] clone (or an in-flight fire thunk) still holds a sender.
|
||||||
|
enum Sys<G: GenServer> {
|
||||||
|
/// A monitor handed in via [`Watcher::watch`]; pushed onto the loop's
|
||||||
|
/// monitor set.
|
||||||
|
Watch(Monitor),
|
||||||
|
/// A fired one-shot timer carrying its loop-local id (so the loop can retire
|
||||||
|
/// it from the live set) and the server's payload, dispatched to
|
||||||
|
/// [`GenServer::handle_timer`].
|
||||||
|
// Constructed by `TimerHandle::arm_after` in the next RFC 015 chunk; until
|
||||||
|
// then only the `Watch` variant is produced.
|
||||||
|
#[allow(dead_code)]
|
||||||
|
Timer(crate::timer::TimerId, G::Timer),
|
||||||
}
|
}
|
||||||
|
|
||||||
impl ServerCtx {
|
/// The server loop's runtime hook, passed to [`GenServer::init`]. Currently
|
||||||
|
/// carries only the [`Watcher`]; opaque so fields can grow without breaking.
|
||||||
|
pub struct ServerCtx<G: GenServer> {
|
||||||
|
watcher: Watcher<G>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<G: GenServer> ServerCtx<G> {
|
||||||
/// A clonable handle to the loop's monitor intake. Store it in the state
|
/// A clonable handle to the loop's monitor intake. Store it in the state
|
||||||
/// during `init` to watch monitors from later handlers.
|
/// during `init` to watch monitors from later handlers.
|
||||||
pub fn watcher(&self) -> Watcher {
|
pub fn watcher(&self) -> Watcher<G> {
|
||||||
self.watcher.clone()
|
self.watcher.clone()
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -244,17 +290,24 @@ impl ServerCtx {
|
|||||||
/// [`GenServer::handle_down`]. Down arms outrank every other arm (a death
|
/// [`GenServer::handle_down`]. Down arms outrank every other arm (a death
|
||||||
/// notice cannot be starved), and a delivered `Down` retires its arm —
|
/// notice cannot be starved), and a delivered `Down` retires its arm —
|
||||||
/// monitors are one-shot.
|
/// monitors are one-shot.
|
||||||
#[derive(Clone)]
|
pub struct Watcher<G: GenServer> {
|
||||||
pub struct Watcher {
|
tx: Sender<Sys<G>>,
|
||||||
tx: Sender<Monitor>,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Watcher {
|
// Manual Clone: deriving would demand `G: Clone`, but the handle is clonable
|
||||||
|
// regardless of the server type (it clones only the inner sender).
|
||||||
|
impl<G: GenServer> Clone for Watcher<G> {
|
||||||
|
fn clone(&self) -> Self {
|
||||||
|
Watcher { tx: self.tx.clone() }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<G: GenServer> Watcher<G> {
|
||||||
/// Transfer `m` to the server loop. If the server is already gone the
|
/// Transfer `m` to the server loop. If the server is already gone the
|
||||||
/// monitor is silently dropped (its queued `Down`, if any, with it) —
|
/// monitor is silently dropped (its queued `Down`, if any, with it) —
|
||||||
/// there is no loop left to care.
|
/// there is no loop left to care.
|
||||||
pub fn watch(&self, m: Monitor) {
|
pub fn watch(&self, m: Monitor) {
|
||||||
let _ = self.tx.send(m);
|
let _ = self.tx.send(Sys::Watch(m));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -490,18 +543,19 @@ fn server_loop<G: GenServer>(
|
|||||||
|
|
||||||
let mut guard = Terminate(state);
|
let mut guard = Terminate(state);
|
||||||
|
|
||||||
// The control arm: Watchers feed Monitors to the loop through it. The
|
// The system intake arm: Watchers feed Monitors and (next chunk) armed
|
||||||
// ctx (and with it the loop's own sender) drops right after init — a
|
// timers feed fires to the loop through it. The ctx (and with it the loop's
|
||||||
// state that didn't clone the Watcher closes the arm, the first select
|
// own sender) drops right after init — a state that didn't clone the
|
||||||
// observes the closure, and the loop falls back to the plain-inbox park.
|
// Watcher closes the arm, the first select observes the closure, and the
|
||||||
let (watch_tx, watch_rx) = channel::<Monitor>();
|
// loop falls back to the plain-inbox park.
|
||||||
guard.0.init(&ServerCtx { watcher: Watcher { tx: watch_tx } });
|
let (sys_tx, sys_rx) = channel::<Sys<G>>();
|
||||||
|
guard.0.init(&ServerCtx { watcher: Watcher { tx: sys_tx } });
|
||||||
|
|
||||||
let mut monitors: Vec<Monitor> = Vec::new();
|
let mut monitors: Vec<Monitor> = Vec::new();
|
||||||
let mut watch_open = true;
|
let mut sys_open = true;
|
||||||
|
|
||||||
loop {
|
loop {
|
||||||
if monitors.is_empty() && !watch_open && infos.is_empty() {
|
if monitors.is_empty() && !sys_open && infos.is_empty() {
|
||||||
// Nothing to select over: park on the inbox alone, exactly the
|
// Nothing to select over: park on the inbox alone, exactly the
|
||||||
// pre-v0.8 loop.
|
// pre-v0.8 loop.
|
||||||
match rx.recv() {
|
match rx.recv() {
|
||||||
@@ -510,19 +564,19 @@ fn server_loop<G: GenServer>(
|
|||||||
Err(_) => break,
|
Err(_) => break,
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
// Arm priority: downs, then the control arm, then infos (each in
|
// Arm priority: downs, then the system arm, then infos (each in
|
||||||
// declaration order), then the inbox. The arm slice is rebuilt
|
// declaration order), then the inbox. The arm slice is rebuilt
|
||||||
// per iteration because every set but the inbox shrinks or grows.
|
// per iteration because every set but the inbox shrinks or grows.
|
||||||
let nd = monitors.len();
|
let nd = monitors.len();
|
||||||
let nw = watch_open as usize;
|
let nw = sys_open as usize;
|
||||||
let i = {
|
let i = {
|
||||||
let mut arms: Vec<&dyn Selectable> =
|
let mut arms: Vec<&dyn Selectable> =
|
||||||
Vec::with_capacity(nd + nw + infos.len() + 1);
|
Vec::with_capacity(nd + nw + infos.len() + 1);
|
||||||
for m in &monitors {
|
for m in &monitors {
|
||||||
arms.push(&m.rx);
|
arms.push(&m.rx);
|
||||||
}
|
}
|
||||||
if watch_open {
|
if sys_open {
|
||||||
arms.push(&watch_rx);
|
arms.push(&sys_rx);
|
||||||
}
|
}
|
||||||
for r in &infos {
|
for r in &infos {
|
||||||
arms.push(r);
|
arms.push(r);
|
||||||
@@ -538,13 +592,14 @@ fn server_loop<G: GenServer>(
|
|||||||
guard.0.handle_down(down);
|
guard.0.handle_down(down);
|
||||||
}
|
}
|
||||||
} else if i < nd + nw {
|
} else if i < nd + nw {
|
||||||
match watch_rx.try_recv() {
|
match sys_rx.try_recv() {
|
||||||
Ok(Some(m)) => monitors.push(m),
|
Ok(Some(Sys::Watch(m))) => monitors.push(m),
|
||||||
|
Ok(Some(Sys::Timer(_id, msg))) => guard.0.handle_timer(msg),
|
||||||
// Single-receiver: nothing can drain the arm between
|
// Single-receiver: nothing can drain the arm between
|
||||||
// select's ready and our try_recv.
|
// select's ready and our try_recv.
|
||||||
Ok(None) => debug_assert!(false, "ready control arm was empty"),
|
Ok(None) => debug_assert!(false, "ready system arm was empty"),
|
||||||
// Every Watcher gone: stop selecting on the arm.
|
// Every Watcher/TimerHandle gone: stop selecting on the arm.
|
||||||
Err(_) => watch_open = false,
|
Err(_) => sys_open = false,
|
||||||
}
|
}
|
||||||
} else if i < nd + nw + infos.len() {
|
} else if i < nd + nw + infos.len() {
|
||||||
let j = i - nd - nw;
|
let j = i - nd - nw;
|
||||||
|
|||||||
+9
-3
@@ -27,6 +27,7 @@ impl GenServer for Counter {
|
|||||||
type Reply = i64;
|
type Reply = i64;
|
||||||
type Cast = Op;
|
type Cast = Op;
|
||||||
type Info = ();
|
type Info = ();
|
||||||
|
type Timer = ();
|
||||||
|
|
||||||
fn handle_call(&mut self, req: Req) -> i64 {
|
fn handle_call(&mut self, req: Req) -> i64 {
|
||||||
match req {
|
match req {
|
||||||
@@ -70,8 +71,9 @@ impl GenServer for Lifecycle {
|
|||||||
type Reply = ();
|
type Reply = ();
|
||||||
type Cast = ();
|
type Cast = ();
|
||||||
type Info = ();
|
type Info = ();
|
||||||
|
type Timer = ();
|
||||||
|
|
||||||
fn init(&mut self, _ctx: &smarm::gen_server::ServerCtx) {
|
fn init(&mut self, _ctx: &smarm::gen_server::ServerCtx<Self>) {
|
||||||
self.log.lock().unwrap().push("init");
|
self.log.lock().unwrap().push("init");
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -154,6 +156,7 @@ impl GenServer for Slow {
|
|||||||
type Reply = u64;
|
type Reply = u64;
|
||||||
type Cast = ();
|
type Cast = ();
|
||||||
type Info = ();
|
type Info = ();
|
||||||
|
type Timer = ();
|
||||||
|
|
||||||
fn handle_call(&mut self, delay_ms: u64) -> u64 {
|
fn handle_call(&mut self, delay_ms: u64) -> u64 {
|
||||||
if delay_ms > 0 {
|
if delay_ms > 0 {
|
||||||
@@ -210,6 +213,7 @@ fn call_timeout_to_dead_server_is_server_down_not_timeout() {
|
|||||||
impl GenServer for Bomb {
|
impl GenServer for Bomb {
|
||||||
type Call = ();
|
type Call = ();
|
||||||
type Info = ();
|
type Info = ();
|
||||||
|
type Timer = ();
|
||||||
type Reply = ();
|
type Reply = ();
|
||||||
type Cast = ();
|
type Cast = ();
|
||||||
fn handle_call(&mut self, _: ()) {
|
fn handle_call(&mut self, _: ()) {
|
||||||
@@ -248,6 +252,7 @@ impl GenServer for Logger {
|
|||||||
type Reply = Vec<&'static str>;
|
type Reply = Vec<&'static str>;
|
||||||
type Cast = ();
|
type Cast = ();
|
||||||
type Info = &'static str;
|
type Info = &'static str;
|
||||||
|
type Timer = ();
|
||||||
|
|
||||||
fn handle_call(&mut self, _: ()) -> Vec<&'static str> {
|
fn handle_call(&mut self, _: ()) -> Vec<&'static str> {
|
||||||
self.log.clone()
|
self.log.clone()
|
||||||
@@ -350,7 +355,7 @@ use smarm::{monitor, spawn, DownReason, Pid};
|
|||||||
/// The motivating pattern: a server that spawns workers from a handler,
|
/// The motivating pattern: a server that spawns workers from a handler,
|
||||||
/// watches them, and logs their deaths.
|
/// watches them, and logs their deaths.
|
||||||
struct Pool {
|
struct Pool {
|
||||||
watcher: Option<Watcher>,
|
watcher: Option<Watcher<Self>>,
|
||||||
log: Vec<DownReason>,
|
log: Vec<DownReason>,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -364,8 +369,9 @@ impl GenServer for Pool {
|
|||||||
type Reply = Vec<DownReason>;
|
type Reply = Vec<DownReason>;
|
||||||
type Cast = PoolCast;
|
type Cast = PoolCast;
|
||||||
type Info = ();
|
type Info = ();
|
||||||
|
type Timer = ();
|
||||||
|
|
||||||
fn init(&mut self, ctx: &smarm::gen_server::ServerCtx) {
|
fn init(&mut self, ctx: &smarm::gen_server::ServerCtx<Self>) {
|
||||||
self.watcher = Some(ctx.watcher());
|
self.watcher = Some(ctx.watcher());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user