diff --git a/src/channel.rs b/src/channel.rs index 29ef06a..8c48a8a 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -137,6 +137,14 @@ impl Drop for Receiver { } impl Sender { + /// Number of messages currently queued behind this channel. Introspection + /// only (RFC 016 mailbox depth); takes the channel lock, so callers reach + /// it under the registry Leaf (Leaf → Channel) via the erased probe in + /// `registry.rs`, never on a hot path. + pub(crate) fn queued_len(&self) -> usize { + self.inner.lock().queue.len() + } + pub fn send(&self, value: T) -> Result<(), SendError> { let unpark = { let mut g = self.inner.lock(); diff --git a/src/introspect.rs b/src/introspect.rs new file mode 100644 index 0000000..5d37f0b --- /dev/null +++ b/src/introspect.rs @@ -0,0 +1,180 @@ +//! RFC 016 — runtime introspection (Chunk 1: the read primitive). +//! +//! A synchronous, internal read of the slab that returns *owned* data. This is +//! the mechanism the whole RFC hangs off: tests, the future observer +//! gen_server (Chunk 4), and a later control plane (RFC 003) are all consumers +//! of [`snapshot`] / [`actor_info`], never of the runtime internals directly. +//! +//! ## Consistency (DECISION D2 — per-slot tearing, `ps` semantics) +//! +//! [`snapshot`] is point-in-time and mildly racy *across* actors: each slot's +//! scheduling state is a lock-free word load, so an actor reported `Running` +//! may already be `Parked`, and an actor can die mid-scan. This is the cheap, +//! useful model (a coherent stop-the-world cut is expensive and rarely wanted). +//! [`actor_info`] is coherent for the single actor it names. +//! +//! ## Locking +//! +//! The lock order is **Leaf → Channel, at most one of each** (`raw_mutex.rs`); +//! cold locks, the registry, and the free list are all Leaves, so we may never +//! hold two at once. The read is therefore phased: first a single registry-leaf +//! pass for names and mailbox depth (the per-channel length read is a Channel +//! lock taken under that Leaf — legal), released before the slab scan takes any +//! per-slot cold Leaf. + +use crate::pid::Pid; +use crate::registry::MailboxInfo; +use crate::runtime::{Slot, ROOT_PID}; +use crate::scheduler::with_runtime; +use crate::slot_state::{ + word_gen, word_state, ST_DONE, ST_PARKED, ST_QUEUED, ST_RUNNING, ST_RUNNING_NOTIFIED, +}; + +/// Snapshot wire-format version (DECISION D1). [`RuntimeSnapshot`] is treated as +/// a stable type from day one: it becomes the observer protocol (Chunk 4) and +/// crosses a version boundary the moment a remote observer attaches (RFC 011), +/// so the version travels with the data from the start. +pub const SNAPSHOT_FORMAT_VERSION: u16 = 1; + +/// Fine-grained scheduling state, mapped from the packed slot word with no new +/// storage. `RunningNotified` collapses into `Notified` — a wake landed while +/// the actor was on-CPU and it will re-queue when it yields. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ActorState { + Queued, + Running, + Notified, + Parked, + Done, +} + +/// Classify a packed state word. `None` for a Vacant slot (skipped by the scan) +/// — the only state that is not an actor. +fn classify(w: u64) -> Option { + Some(match word_state(w) { + ST_QUEUED => ActorState::Queued, + ST_RUNNING => ActorState::Running, + ST_RUNNING_NOTIFIED => ActorState::Notified, + ST_PARKED => ActorState::Parked, + ST_DONE => ActorState::Done, + _ => return None, // ST_VACANT + }) +} + +/// Owned, point-in-time view of one actor — no borrows of runtime internals, so +/// it is safe to hand to any consumer. +#[derive(Debug, Clone)] +pub struct ActorInfo { + pub pid: Pid, + /// Registered names, inverted from the registry (usually 0 or 1). + pub names: Vec<&'static str>, + pub state: ActorState, + /// Spawn-time parent edge (DECISION D9): `spawn_under` sets it to the + /// supervisor, plain `spawn` to the spawning actor — so it is parentage, + /// not necessarily a supervision relationship. `ROOT_PID` for the run's + /// root actor and for `Done` tombstones (whose `Actor` is already gone). + pub supervisor: Pid, + pub trap_exit: bool, + pub monitors: u32, + pub links: u32, + pub joiners: u32, + /// Queued messages summed over the actor's *published* channels (register / + /// install / spawn_addr / gen_server). 0 for an actor that holds only a + /// private `channel()` receiver — those are invisible to the registry. + pub mailbox_depth: u32, +} + +/// A whole-runtime snapshot. See the module docs for the D2 tearing model. +#[derive(Debug, Clone)] +pub struct RuntimeSnapshot { + pub format_version: u16, + pub actors: Vec, +} + +/// Snapshot every live (and `Done`-but-not-yet-reclaimed) actor on the slab. +/// O(n) over the slot table, running with preemption disabled (like every +/// runtime primitive) but holding no lock across the scan. Panics outside +/// `Runtime::run()`; callable from actor code and the run thread. +pub fn snapshot() -> RuntimeSnapshot { + with_runtime(|inner| { + // Phase A: one registry-leaf pass for names + mailbox depth, released + // before any cold leaf (no two Leaves at once). + let mail = inner.registry.lock().introspect_map(); + + // Phase B: lock-free slab scan; per-slot cold leaf only to copy cold + // fields. Tearing across slots is intentional (D2). + let mut actors = Vec::new(); + for (idx, slot) in inner.slots.iter().enumerate() { + let idx = idx as u32; + if let Some(info) = read_slot(slot, idx, mail.get(&idx)) { + actors.push(info); + } + } + RuntimeSnapshot { format_version: SNAPSHOT_FORMAT_VERSION, actors } + }) +} + +/// Coherent view of a single actor, or `None` if the pid is stale, out of +/// range, or names a Vacant slot. +pub fn actor_info(pid: Pid) -> Option { + with_runtime(|inner| { + let slot = inner.slot_at(pid)?; + let mail = inner.registry.lock().introspect_one(pid.index()); + let info = read_slot(slot, pid.index(), mail.as_ref())?; + // read_slot keys on the slab's *current* generation; reject if that + // isn't the incarnation the caller asked about. + (info.pid.generation() == pid.generation()).then_some(info) + }) +} + +/// Build one `ActorInfo` for slot `idx`, or `None` if Vacant or +/// racing-reclaimed. State is classified from a lock-free word load (the torn +/// read); the cold lock then pins the generation (reclaim bumps it under that +/// same lock) so the cold fields are coherent for this incarnation. `mail` is +/// this slot's registry entry, if any. +fn read_slot(slot: &Slot, idx: u32, mail: Option<&MailboxInfo>) -> Option { + let w = slot.state_word(); + let state = classify(w)?; + let gen = word_gen(w); + let pid = Pid::new(idx, gen); + + let cold = slot.cold.lock(); + // If the generation moved between the lock-free load and acquiring the cold + // lock, the slot was reclaimed (and maybe reused) — drop it rather than mix + // one incarnation's state with another's cold data. (ps semantics: a racing + // actor may simply be missed mid-scan.) + if word_gen(slot.state_word()) != gen { + return None; + } + let (supervisor, trap_exit) = match cold.actor.as_ref() { + // Live incarnation: parent + trap live on the Actor. + Some(actor) => (actor.supervisor, actor.trap.is_some()), + // Done tombstone: the Actor was taken at finalize and the collections + // cleared, so report it root-less with empty counts. + None => (ROOT_PID, false), + }; + let monitors = cold.monitors.len() as u32; + let links = cold.links.len() as u32; + let joiners = cold.waiters.len() as u32; + drop(cold); + + // Names + depth belong to this incarnation only if the registry mailbox's + // pid matches the slab generation; a stale registry entry (dead prior + // occupant, not yet pruned) contributes nothing. + let (names, mailbox_depth) = match mail { + Some(mi) if mi.pid.generation() == gen => (mi.names.clone(), mi.depth), + _ => (Vec::new(), 0), + }; + + Some(ActorInfo { + pid, + names, + state, + supervisor, + trap_exit, + monitors, + links, + joiners, + mailbox_depth, + }) +} diff --git a/src/lib.rs b/src/lib.rs index 1793ec8..45bd3a0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -27,6 +27,7 @@ pub mod registry; pub mod pg; pub mod link; pub mod gen_server; +pub mod introspect; pub mod runtime; pub(crate) mod raw_mutex; pub(crate) mod slot_state; @@ -54,6 +55,9 @@ pub use gen_server::{ call, cast, shutdown, whereis_server, CallError, CallTimeoutError, CastError, GenServer, NamedServerBuilder, ServerBuilder, ServerCtx, ServerName, ServerRef, TimerHandle, Watcher, }; +pub use introspect::{ + actor_info, snapshot, ActorInfo, ActorState, RuntimeSnapshot, SNAPSHOT_FORMAT_VERSION, +}; pub use link::{link, trap_exit, unlink, ExitSignal}; pub use monitor::{demonitor, monitor, Down, DownReason, Monitor, MonitorId}; pub use mutex::{LockTimeout, Mutex, MutexGuard}; diff --git a/src/registry.rs b/src/registry.rs index 0025eb6..6648cc9 100644 --- a/src/registry.rs +++ b/src/registry.rs @@ -148,11 +148,30 @@ impl std::fmt::Display for SendError { impl std::error::Error for SendError {} +/// A registry-stored channel, type-erased over its message type. The stored +/// object must serve two readers: `clone_sender` (downcast back to the concrete +/// `Sender`) and the RFC 016 snapshot (queued length without knowing `M`). +/// A bare `Box` gives the first but not the second, so we erase behind +/// this small trait instead. +trait ErasedSender: Send { + fn as_any(&self) -> &dyn Any; + fn queued_len(&self) -> usize; +} + +impl ErasedSender for Sender { + fn as_any(&self) -> &dyn Any { + self + } + fn queued_len(&self) -> usize { + Sender::queued_len(self) + } +} + /// One typed channel of an actor, type-erased. Concretely a `Sender` filed /// under `TypeId::of::()`; `msg_type` is `type_name::()`, kept for /// observers (RFC 014 §4.5) and as the debug cross-check on the downcast. struct Channel { - sender: Box, + sender: Box, msg_type: &'static str, } @@ -176,6 +195,7 @@ impl Mailbox { let ch = self.channels.get(&TypeId::of::())?; let tx = ch .sender + .as_any() .downcast_ref::>() .expect("channel keyed by TypeId but downcast to its own type failed — smarm bug"); debug_assert_eq!(ch.msg_type, type_name::(), "msg_type / TypeId disagree"); @@ -183,6 +203,18 @@ impl Mailbox { } } +/// Per-actor registry view handed to RFC 016 introspection: registered names +/// and summed mailbox depth, tagged with the mailbox's `pid` so a stale +/// incarnation can be filtered against the slab. Covers only *published* +/// channels (`register` / `install` / `spawn_addr` / gen_server start); an +/// actor that holds only a private `channel()` receiver is invisible here and +/// reports depth 0. +pub(crate) struct MailboxInfo { + pub(crate) pid: Pid, + pub(crate) names: Vec<&'static str>, + pub(crate) depth: u32, +} + /// The directory. Invariant (held under the registry lock): every value in /// `by_name` is the index of a [`Mailbox`] present in `by_index`, and that /// mailbox's `pid.index()` equals the key. Stale entries (dead actors) violate @@ -204,6 +236,49 @@ impl Registry { self.by_index.remove(&index); self.by_name.retain(|_, idx| *idx != index); } + + /// RFC 016 snapshot input: per-slot-index registry view — the actor's + /// registered names (inverted from `by_name`) and its mailbox depth (queued + /// messages summed across every published typed channel). Built in one pass + /// under the registry Leaf; the per-channel `queued_len` takes a Channel + /// lock, legal under the Leaf (Leaf → Channel). Carries each mailbox's full + /// `pid` so the caller can discard a stale incarnation's entry against the + /// slab's live generation. Names that dangle (point at no mailbox) are + /// dropped — they violate no invariant and get pruned on next contact. + pub(crate) fn introspect_map(&self) -> HashMap { + let mut names: HashMap> = HashMap::new(); + for (&name, &idx) in &self.by_name { + names.entry(idx).or_default().push(name); + } + let mut out: HashMap = HashMap::with_capacity(self.by_index.len()); + for (&idx, mb) in &self.by_index { + let depth: usize = mb.channels.values().map(|c| c.sender.queued_len()).sum(); + out.insert( + idx, + MailboxInfo { + pid: mb.pid, + names: names.remove(&idx).unwrap_or_default(), + depth: depth.min(u32::MAX as usize) as u32, + }, + ); + } + out + } + + /// Single-actor form of [`introspect_map`](Self::introspect_map): the + /// registry view for one slot index, or `None` if no mailbox is published + /// there. Used by `actor_info` so its cost stays proportional to the one + /// actor rather than locking every channel in the runtime. + pub(crate) fn introspect_one(&self, idx: u32) -> Option { + let mb = self.by_index.get(&idx)?; + let depth: usize = mb.channels.values().map(|c| c.sender.queued_len()).sum(); + let names = self + .by_name + .iter() + .filter_map(|(&n, &i)| (i == idx).then_some(n)) + .collect(); + Some(MailboxInfo { pid: mb.pid, names, depth: depth.min(u32::MAX as usize) as u32 }) + } } /// Is `pid` a live actor right now? Atomic slot-word read; no lock. diff --git a/src/runtime.rs b/src/runtime.rs index 6a95e3d..8f8d8a8 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -457,6 +457,14 @@ impl Slot { self.word.generation() } + /// Raw packed state word, for introspection's lock-free classify + /// (`introspect.rs`). The coarse `status_for` only distinguishes + /// Live/Done/Stale; the snapshot needs the fine scheduling state. + #[inline] + pub(crate) fn state_word(&self) -> u64 { + self.word.load() + } + /// A pid's-eye snapshot of the slot. Cold paths re-read this under the /// cold lock (generation can't change while it is held). #[inline] diff --git a/tests/introspect.rs b/tests/introspect.rs new file mode 100644 index 0000000..29cf082 --- /dev/null +++ b/tests/introspect.rs @@ -0,0 +1,184 @@ +//! RFC 016 Chunk 1 — the read primitive. These exercise exactly what the RFC +//! promised: tests that assert an actor's state, parentage, names, mailbox +//! depth, and lifecycle counts directly off `snapshot()` / `actor_info()` +//! instead of sleeping-and-hoping. + +use smarm::{ + actor_info, channel, monitor, register, run, self_pid, send, snapshot, spawn, ActorState, Name, + Pid, SNAPSHOT_FORMAT_VERSION, +}; + +const SVC: Name = Name::new("svc"); + +/// Bounded poll on the introspection result itself (not a wall-clock sleep): +/// yield until `pred` holds for the given pid, panicking if it never does. +fn spin_until(pid: Pid, mut pred: impl FnMut(&smarm::ActorInfo) -> bool) -> smarm::ActorInfo { + for _ in 0..100_000 { + if let Some(info) = actor_info(pid) { + if pred(&info) { + return info; + } + } + smarm::yield_now(); + } + panic!("actor {pid:?} never reached the expected state"); +} + +#[test] +fn snapshot_lists_actors_with_parent_edge() { + run(|| { + let (ready_tx, ready_rx) = channel::<()>(); + let (gate_tx, gate_rx) = channel::<()>(); + let (_cmd_tx, cmd_rx) = channel::(); + + let h = spawn(move || { + register(Name::::new("w"), _cmd_tx).unwrap(); + ready_tx.send(()).unwrap(); + gate_rx.recv().unwrap(); // park here until released + drop(cmd_rx); + }); + ready_rx.recv().unwrap(); + + let me = self_pid(); + let snap = snapshot(); + assert_eq!(snap.format_version, SNAPSHOT_FORMAT_VERSION); + + let worker = snap + .actors + .iter() + .find(|a| a.pid == h.pid()) + .expect("worker present in snapshot"); + assert_eq!(worker.names, vec!["w"]); + // `spawn` records the spawning actor as the parent (D9). + assert_eq!(worker.supervisor, me); + assert!(!worker.trap_exit); + assert_eq!((worker.monitors, worker.links, worker.joiners), (0, 0, 0)); + + // The root itself is on-CPU (it's running this code) and rooted under + // the forest sentinel. + let root = snap.actors.iter().find(|a| a.pid == me).expect("root present"); + assert_eq!(root.state, ActorState::Running); + assert_eq!(root.supervisor, smarm::Pid::new(u32::MAX, u32::MAX)); + + gate_tx.send(()).unwrap(); + h.join().unwrap(); + }); +} + +#[test] +fn parked_state_is_observable() { + run(|| { + let (ready_tx, ready_rx) = channel::<()>(); + let (gate_tx, gate_rx) = channel::<()>(); + let h = spawn(move || { + ready_tx.send(()).unwrap(); + gate_rx.recv().unwrap(); + }); + ready_rx.recv().unwrap(); + // The worker has nothing to do but block on the empty gate channel, so + // it must reach Parked. + let info = spin_until(h.pid(), |a| a.state == ActorState::Parked); + assert_eq!(info.state, ActorState::Parked); + gate_tx.send(()).unwrap(); + h.join().unwrap(); + }); +} + +#[test] +fn mailbox_depth_counts_queued_messages() { + run(|| { + let (ready_tx, ready_rx) = channel::<()>(); + let (gate_tx, gate_rx) = channel::<()>(); + let (cmd_tx, _cmd_rx) = channel::(); + + let h = spawn(move || { + // Publish the command inbox, then block on an unrelated gate so the + // queued commands are never drained while we observe them. + register(SVC, cmd_tx).unwrap(); + ready_tx.send(()).unwrap(); + gate_rx.recv().unwrap(); + drop(_cmd_rx); + }); + ready_rx.recv().unwrap(); + + for i in 0..3 { + send(SVC, i).unwrap(); + } + + // Depth is a property of the queue, set synchronously by `send`, so it + // reads 3 regardless of the worker's scheduling state. + let via_snapshot = snapshot() + .actors + .into_iter() + .find(|a| a.pid == h.pid()) + .expect("worker present"); + assert_eq!(via_snapshot.mailbox_depth, 3); + assert_eq!(actor_info(h.pid()).unwrap().mailbox_depth, 3); + + gate_tx.send(()).unwrap(); + h.join().unwrap(); + }); +} + +#[test] +fn monitor_count_is_visible() { + run(|| { + let (ready_tx, ready_rx) = channel::<()>(); + let (gate_tx, gate_rx) = channel::<()>(); + let h = spawn(move || { + ready_tx.send(()).unwrap(); + gate_rx.recv().unwrap(); + }); + ready_rx.recv().unwrap(); + + let _m = monitor(h.pid()); + let info = actor_info(h.pid()).expect("worker present"); + assert_eq!(info.monitors, 1); + + gate_tx.send(()).unwrap(); + h.join().unwrap(); + }); +} + +#[test] +fn stale_and_forged_pids_return_none() { + run(|| { + let (ready_tx, ready_rx) = channel::<()>(); + let (gate_tx, gate_rx) = channel::<()>(); + let h = spawn(move || { + ready_tx.send(()).unwrap(); + gate_rx.recv().unwrap(); + }); + ready_rx.recv().unwrap(); + let live = h.pid(); + + // Same slot index, wrong generation → stale, no such incarnation. + let stale = Pid::new(live.index(), live.generation().wrapping_add(7)); + assert!(actor_info(stale).is_none()); + + // Out-of-range index → not in the slab at all. + let forged = Pid::new(u32::MAX - 1, 0); + assert!(actor_info(forged).is_none()); + + gate_tx.send(()).unwrap(); + h.join().unwrap(); + }); +} + +#[test] +fn done_actor_is_a_tombstone() { + run(|| { + // Hold the join handle so the slot is NOT reclaimed when the actor + // exits: outstanding_handles stays > 0, leaving a Done tombstone to + // observe. + let h = spawn(|| {}); + let pid = h.pid(); + let info = spin_until(pid, |a| a.state == ActorState::Done); + assert_eq!(info.state, ActorState::Done); + // The Actor record is gone at finalize, so a tombstone reports root-less + // with empty lifecycle counts. + assert_eq!(info.supervisor, Pid::new(u32::MAX, u32::MAX)); + assert_eq!((info.monitors, info.links, info.joiners), (0, 0, 0)); + h.join().unwrap(); + }); +}