feat: process primitives #46
22 changed files with 4594 additions and 81 deletions
|
|
@ -4,7 +4,7 @@ use std::cell::RefCell;
|
|||
use pyo3::prelude::*;
|
||||
use pyo3::types::PyModule;
|
||||
|
||||
use ::swactor::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Ctx};
|
||||
use ::swactor::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Ctx, Environment, SpawnRequest};
|
||||
use ::swactor::config::{BackoffPolicy, RuntimeConfig};
|
||||
use ::swactor::runtime::{Inbox, Runtime, RuntimeHandle};
|
||||
|
||||
|
|
@ -182,7 +182,12 @@ impl ActorInterface for PyActor {
|
|||
Effect::Spawn { addr, handler } => {
|
||||
let actor = PyActor::new(handler);
|
||||
let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor));
|
||||
let _ = ctx.raw_inner().spawn_any(addr, boxed);
|
||||
ctx.raw_inner().spawn_any(SpawnRequest {
|
||||
addr,
|
||||
actor: boxed,
|
||||
parent: Some(ctx.self_addr()),
|
||||
env: Environment::new(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
56
crates/std/src/children_registry.rs
Normal file
56
crates/std/src/children_registry.rs
Normal file
|
|
@ -0,0 +1,56 @@
|
|||
use std::sync::RwLock;
|
||||
|
||||
use swactor::actor::ActorAddress;
|
||||
use swactor::{AddrMap, AddrSet};
|
||||
|
||||
/// Tracks parent → children relationships for orphan cleanup.
|
||||
///
|
||||
/// When a parent dies, unsupervised children are stopped automatically.
|
||||
/// Entries are added in `on_spawn` and cleaned up on actor death.
|
||||
pub struct ChildrenRegistry {
|
||||
/// parent_addr → set of child addresses
|
||||
children: RwLock<AddrMap<AddrSet>>,
|
||||
}
|
||||
|
||||
impl ChildrenRegistry {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
children: RwLock::new(AddrMap::default()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Register a parent → child relationship.
|
||||
pub fn register(&self, parent: ActorAddress, child: ActorAddress) {
|
||||
self.children
|
||||
.write()
|
||||
.unwrap()
|
||||
.entry(parent)
|
||||
.or_default()
|
||||
.insert(child);
|
||||
}
|
||||
|
||||
/// Remove and return all children of a parent (for orphan handling).
|
||||
pub fn take_children(&self, parent: &ActorAddress) -> Vec<ActorAddress> {
|
||||
self.children
|
||||
.write()
|
||||
.unwrap()
|
||||
.remove(parent)
|
||||
.map(|set| set.into_iter().collect())
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
/// Clean up entries for dead actors (as both parent and child).
|
||||
pub fn cleanup(&self, dead: &[ActorAddress]) {
|
||||
let mut map = self.children.write().unwrap();
|
||||
for addr in dead {
|
||||
// Remove as parent
|
||||
map.remove(addr);
|
||||
// Remove as child from any parent's set
|
||||
for set in map.values_mut() {
|
||||
set.remove(addr);
|
||||
}
|
||||
}
|
||||
// Remove empty parent entries
|
||||
map.retain(|_, set| !set.is_empty());
|
||||
}
|
||||
}
|
||||
|
|
@ -1,10 +1,11 @@
|
|||
use swactor::actor::{ActorAddress, ActorInterface, Ctx, Message, MonitorRef};
|
||||
use swactor::actor::{ActorAddress, ActorInterface, Ctx, Environment, LogicalName, Message, MonitorRef, SystemInfo};
|
||||
use swactor::Error;
|
||||
|
||||
use crate::resource_handle::ResourceHandle;
|
||||
use crate::StdExtension;
|
||||
use crate::timer_wheel::{CloneMsg, TimerRequest};
|
||||
|
||||
fn get_ext<'a>(ctx: &'a Ctx) -> &'a StdExtension {
|
||||
pub(crate) fn get_ext<'a>(ctx: &'a Ctx) -> &'a StdExtension {
|
||||
ctx.extension()
|
||||
.expect("StdExtension not installed — use Runtime::with_extension()")
|
||||
.as_any()
|
||||
|
|
@ -18,15 +19,18 @@ fn get_ext<'a>(ctx: &'a Ctx) -> &'a StdExtension {
|
|||
pub trait CtxMonitoring {
|
||||
/// Subscribe to death notifications from `target`. Returns a [`MonitorRef`]
|
||||
/// that can be used to cancel the subscription.
|
||||
fn monitor(&self, target: ActorAddress) -> MonitorRef;
|
||||
fn monitor(&self, target: ActorAddress) -> Result<MonitorRef, Error>;
|
||||
|
||||
/// Cancel a monitor subscription.
|
||||
fn demonitor(&self, mref: MonitorRef);
|
||||
}
|
||||
|
||||
impl CtxMonitoring for Ctx<'_> {
|
||||
fn monitor(&self, target: ActorAddress) -> MonitorRef {
|
||||
get_ext(self).monitor_registry.register(self.self_addr(), target)
|
||||
fn monitor(&self, target: ActorAddress) -> Result<MonitorRef, Error> {
|
||||
if let Some(caps) = self.env::<swactor::CapabilitySet>() {
|
||||
caps.check_monitor(target)?;
|
||||
}
|
||||
Ok(get_ext(self).monitor_registry.register(self.self_addr(), target))
|
||||
}
|
||||
|
||||
fn demonitor(&self, mref: MonitorRef) {
|
||||
|
|
@ -60,7 +64,7 @@ impl CtxNaming for Ctx<'_> {
|
|||
|
||||
fn spawn_named<A: ActorInterface>(&self, name: impl Into<String>, actor: A) -> Result<ActorAddress, Error> {
|
||||
let name = name.into();
|
||||
let addr = self.spawn(actor)?;
|
||||
let addr = self.spawn_builder(actor).env(LogicalName(name.clone())).finish()?;
|
||||
if let Err(e) = get_ext(self).name_registry.register(name, addr) {
|
||||
let _ = self.stop_actor(addr);
|
||||
return Err(e);
|
||||
|
|
@ -177,3 +181,217 @@ impl CtxGroups for Ctx<'_> {
|
|||
get_ext(self).group_registry.members(group)
|
||||
}
|
||||
}
|
||||
|
||||
/// System introspection extension for [`Ctx`].
|
||||
///
|
||||
/// Provides convenience accessors for system-level information. Does NOT
|
||||
/// require [`StdExtension`] — the data comes from the core runtime.
|
||||
pub trait CtxSystem {
|
||||
/// Returns the full [`SystemInfo`] snapshot.
|
||||
fn system_info(&self) -> SystemInfo;
|
||||
|
||||
/// Index of the worker thread this actor is running on.
|
||||
fn worker_id(&self) -> usize;
|
||||
|
||||
/// Total number of worker threads in the runtime.
|
||||
fn num_workers(&self) -> usize;
|
||||
|
||||
/// Total number of live actors across all workers.
|
||||
fn total_actors(&self) -> usize;
|
||||
|
||||
/// Milliseconds since the runtime was created.
|
||||
fn uptime_ms(&self) -> u64;
|
||||
}
|
||||
|
||||
impl CtxSystem for Ctx<'_> {
|
||||
fn system_info(&self) -> SystemInfo {
|
||||
Ctx::system_info(self)
|
||||
}
|
||||
|
||||
fn worker_id(&self) -> usize {
|
||||
Ctx::system_info(self).worker_id
|
||||
}
|
||||
|
||||
fn num_workers(&self) -> usize {
|
||||
Ctx::system_info(self).num_workers
|
||||
}
|
||||
|
||||
fn total_actors(&self) -> usize {
|
||||
Ctx::system_info(self).total_actors
|
||||
}
|
||||
|
||||
fn uptime_ms(&self) -> u64 {
|
||||
Ctx::system_info(self).uptime_ms
|
||||
}
|
||||
}
|
||||
|
||||
/// Lineage extension for [`Ctx`].
|
||||
///
|
||||
/// Exposes the actor's parent and supervisor. `parent()` does NOT require
|
||||
/// [`StdExtension`] — the data is stored in core per-actor state.
|
||||
/// `supervisor()` returns `None` gracefully when StdExtension is absent.
|
||||
pub trait CtxLineage {
|
||||
/// Returns the address of the actor that spawned this one, or `None`
|
||||
/// if this actor was spawned externally via `Runtime::spawn`.
|
||||
fn parent(&self) -> Option<ActorAddress>;
|
||||
|
||||
/// Returns the address of this actor's supervisor, or `None` if
|
||||
/// unsupervised or StdExtension is not installed.
|
||||
fn supervisor(&self) -> Option<ActorAddress>;
|
||||
}
|
||||
|
||||
impl CtxLineage for Ctx<'_> {
|
||||
fn parent(&self) -> Option<ActorAddress> {
|
||||
Ctx::parent(self)
|
||||
}
|
||||
|
||||
fn supervisor(&self) -> Option<ActorAddress> {
|
||||
let ext = self.extension()?
|
||||
.as_any()
|
||||
.downcast_ref::<StdExtension>()?;
|
||||
ext.supervisor_registry.lookup(&self.self_addr())
|
||||
}
|
||||
}
|
||||
|
||||
/// Per-actor self-introspection extension for [`Ctx`].
|
||||
///
|
||||
/// Exposes the actor's own operational metrics. Does NOT require
|
||||
/// [`StdExtension`] — the data is snapshotted from core before each tick.
|
||||
pub trait CtxSelfStats {
|
||||
/// Total messages this actor has successfully processed (before the current tick).
|
||||
fn messages_processed(&self) -> u64;
|
||||
|
||||
/// Number of messages in this actor's mailbox at the start of the current tick.
|
||||
fn mailbox_depth(&self) -> usize;
|
||||
|
||||
/// Per-message-type counts for this actor, sorted descending by count.
|
||||
fn message_type_counts(&self) -> &[(&'static str, u64)];
|
||||
}
|
||||
|
||||
impl CtxSelfStats for Ctx<'_> {
|
||||
fn messages_processed(&self) -> u64 {
|
||||
Ctx::messages_processed(self)
|
||||
}
|
||||
|
||||
fn mailbox_depth(&self) -> usize {
|
||||
Ctx::mailbox_depth(self)
|
||||
}
|
||||
|
||||
fn message_type_counts(&self) -> &[(&'static str, u64)] {
|
||||
Ctx::message_type_counts(self)
|
||||
}
|
||||
}
|
||||
|
||||
/// Service resource extension for [`Ctx`].
|
||||
///
|
||||
/// Provides typed service discovery via the environment. Does NOT require
|
||||
/// [`StdExtension`] — reads from the core environment (same as [`CtxEnvironment`]).
|
||||
pub trait CtxResources {
|
||||
/// Look up a service address by marker type `S`.
|
||||
///
|
||||
/// Returns `None` if no `ServiceBinding<S>` is present in the environment.
|
||||
fn resource<S: 'static + Send + Sync>(&self) -> Option<ActorAddress>;
|
||||
}
|
||||
|
||||
impl CtxResources for Ctx<'_> {
|
||||
fn resource<S: 'static + Send + Sync>(&self) -> Option<ActorAddress> {
|
||||
if let Some(caps) = self.env::<swactor::CapabilitySet>() {
|
||||
if caps.check_service::<S>().is_err() {
|
||||
return None;
|
||||
}
|
||||
}
|
||||
self.env::<swactor::ServiceBinding<S>>().map(|b| b.addr)
|
||||
}
|
||||
}
|
||||
|
||||
/// Environment extension for [`Ctx`].
|
||||
///
|
||||
/// Provides access to the actor's inherited typed key-value environment.
|
||||
/// Does NOT require [`StdExtension`] — the data is stored in core per-actor state.
|
||||
pub trait CtxEnvironment {
|
||||
/// Read a typed value from this actor's environment.
|
||||
fn env<T: std::any::Any + Send + Sync>(&self) -> Option<&T>;
|
||||
|
||||
/// Access this actor's full environment.
|
||||
fn environment(&self) -> &Environment;
|
||||
}
|
||||
|
||||
impl CtxEnvironment for Ctx<'_> {
|
||||
fn env<T: std::any::Any + Send + Sync>(&self) -> Option<&T> {
|
||||
Ctx::env(self)
|
||||
}
|
||||
|
||||
fn environment(&self) -> &Environment {
|
||||
Ctx::environment(self)
|
||||
}
|
||||
}
|
||||
|
||||
/// Resource handle extension for [`Ctx`].
|
||||
///
|
||||
/// Provides `handle::<H>()` to construct typed proxy structs wrapping service
|
||||
/// addresses for ergonomic domain-specific APIs. See [`ResourceHandle`] for
|
||||
/// how to define a handle type.
|
||||
pub trait CtxHandles {
|
||||
/// Construct a typed resource handle from the service registry.
|
||||
///
|
||||
/// Returns `None` if no `ServiceBinding<H::Service>` is present in the
|
||||
/// actor's environment (consistent with `ctx.resource()`, `ctx.where_is()`, etc).
|
||||
fn handle<H: ResourceHandle>(&self) -> Option<H>;
|
||||
}
|
||||
|
||||
impl CtxHandles for Ctx<'_> {
|
||||
fn handle<H: ResourceHandle>(&self) -> Option<H> {
|
||||
let binding = self.env::<swactor::ServiceBinding<H::Service>>()?;
|
||||
Some(H::from_parts(binding.addr, self.self_addr()))
|
||||
}
|
||||
}
|
||||
|
||||
/// Lifecycle extension for [`Ctx`].
|
||||
///
|
||||
/// Provides suspend/resume capabilities with authorization:
|
||||
/// only the actor itself or its supervisor can resume it.
|
||||
pub trait CtxLifecycle {
|
||||
/// Suspend this actor. Messages continue to queue but are not processed
|
||||
/// until resumed by self or supervisor.
|
||||
fn suspend_self(&self);
|
||||
|
||||
/// Resume a suspended actor. Only the actor itself or its supervisor
|
||||
/// may call this. Returns `Err` if the caller is not authorized.
|
||||
fn resume(&self, target: ActorAddress) -> Result<(), Error>;
|
||||
}
|
||||
|
||||
impl CtxLifecycle for Ctx<'_> {
|
||||
fn suspend_self(&self) {
|
||||
Ctx::suspend_self(self);
|
||||
}
|
||||
|
||||
fn resume(&self, target: ActorAddress) -> Result<(), Error> {
|
||||
// Self-resume is always allowed
|
||||
if target == self.self_addr() {
|
||||
self.raw_inner().request_resume(target);
|
||||
return Ok(());
|
||||
}
|
||||
// Supervisor can resume its child
|
||||
let ext = get_ext(self);
|
||||
if ext.supervisor_registry.lookup(&target) == Some(self.self_addr()) {
|
||||
self.raw_inner().request_resume(target);
|
||||
return Ok(());
|
||||
}
|
||||
Err(Error::from("resume denied: caller is not self or supervisor"))
|
||||
}
|
||||
}
|
||||
|
||||
/// Capability introspection extension for [`Ctx`].
|
||||
pub trait CtxCapabilities {
|
||||
fn capabilities(&self) -> Option<&swactor::CapabilitySet>;
|
||||
fn is_restricted(&self) -> bool;
|
||||
}
|
||||
|
||||
impl CtxCapabilities for Ctx<'_> {
|
||||
fn capabilities(&self) -> Option<&swactor::CapabilitySet> {
|
||||
Ctx::env(self)
|
||||
}
|
||||
fn is_restricted(&self) -> bool {
|
||||
self.env::<swactor::CapabilitySet>().is_some()
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,11 +1,14 @@
|
|||
use std::any::Any;
|
||||
|
||||
use swactor::actor::{ActorAddress, Down, ExitReason, StopReason};
|
||||
use swactor::actor::{ActorAddress, Down, Environment, EnvironmentBuilder, ExitReason, ExitValue, SpawnTimestamp, StopReason, StopSignal};
|
||||
use swactor::extension::{RuntimeExtension, WorkerExtension};
|
||||
|
||||
use crate::children_registry::ChildrenRegistry;
|
||||
use crate::group_registry::GroupRegistry;
|
||||
use crate::monitor_registry::MonitorRegistry;
|
||||
use crate::name_registry::NameRegistry;
|
||||
use crate::service_registry::ServiceRegistry;
|
||||
use crate::supervisor_registry::SupervisorRegistry;
|
||||
use crate::timer_wheel::TimerWheel;
|
||||
use crate::watch_registry::WatchRegistry;
|
||||
|
||||
|
|
@ -17,6 +20,9 @@ pub struct StdExtension {
|
|||
pub(crate) monitor_registry: MonitorRegistry,
|
||||
pub(crate) watch_registry: WatchRegistry,
|
||||
pub(crate) group_registry: GroupRegistry,
|
||||
pub(crate) supervisor_registry: SupervisorRegistry,
|
||||
pub(crate) service_registry: ServiceRegistry,
|
||||
pub(crate) children_registry: ChildrenRegistry,
|
||||
}
|
||||
|
||||
impl StdExtension {
|
||||
|
|
@ -26,6 +32,9 @@ impl StdExtension {
|
|||
monitor_registry: MonitorRegistry::new(),
|
||||
watch_registry: WatchRegistry::new(),
|
||||
group_registry: GroupRegistry::new(),
|
||||
supervisor_registry: SupervisorRegistry::new(),
|
||||
service_registry: ServiceRegistry::new(),
|
||||
children_registry: ChildrenRegistry::new(),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -33,6 +42,14 @@ impl StdExtension {
|
|||
pub fn resolve_name(&self, addr: &ActorAddress) -> Option<String> {
|
||||
self.name_registry.lookup_by_addr(addr)
|
||||
}
|
||||
|
||||
/// Register a supervisor → child relationship.
|
||||
///
|
||||
/// This is used by the built-in [`Supervisor`](crate::Supervisor) and can
|
||||
/// also be called by custom supervisor implementations.
|
||||
pub fn register_supervisor(&self, supervisor: ActorAddress, child: ActorAddress) {
|
||||
self.supervisor_registry.register(supervisor, child);
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for StdExtension {
|
||||
|
|
@ -46,40 +63,54 @@ fn stop_to_exit(reason: StopReason) -> ExitReason {
|
|||
match reason {
|
||||
StopReason::Normal => ExitReason::Stopped,
|
||||
StopReason::Panicked => ExitReason::Panicked,
|
||||
StopReason::Completed => ExitReason::Completed,
|
||||
}
|
||||
}
|
||||
|
||||
impl RuntimeExtension for StdExtension {
|
||||
fn on_actor_death(
|
||||
&self,
|
||||
dead: &[(ActorAddress, StopReason)],
|
||||
dead: &[(ActorAddress, StopReason, Option<ExitValue>)],
|
||||
) -> Vec<(ActorAddress, Box<dyn Any + Send>)> {
|
||||
let mut notifications = Vec::new();
|
||||
|
||||
for &(addr, reason) in dead {
|
||||
for (addr, reason, exit_value) in dead {
|
||||
let addr = *addr;
|
||||
let reason = *reason;
|
||||
|
||||
// Monitor notifications (Down)
|
||||
let watchers = self.monitor_registry.take_monitors(&addr);
|
||||
for (_mref, watcher) in watchers {
|
||||
let down = Down { addr, reason };
|
||||
let down = Down { addr, reason, exit_value: exit_value.clone() };
|
||||
notifications.push((watcher, Box::new(down) as Box<dyn Any + Send>));
|
||||
}
|
||||
|
||||
// Watch notifications (ActorExited)
|
||||
let watch_notifications = self.watch_registry.notify_death(addr, stop_to_exit(reason));
|
||||
let watch_notifications = self.watch_registry.notify_death(addr, stop_to_exit(reason), exit_value.clone());
|
||||
for (watcher, exited) in watch_notifications {
|
||||
notifications.push((watcher, Box::new(exited) as Box<dyn Any + Send>));
|
||||
}
|
||||
|
||||
// Orphan handling: kill unsupervised children
|
||||
let children = self.children_registry.take_children(&addr);
|
||||
for child in children {
|
||||
if self.supervisor_registry.lookup(&child).is_none() {
|
||||
notifications.push((child, Box::new(StopSignal) as Box<dyn Any + Send>));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
notifications
|
||||
}
|
||||
|
||||
fn cleanup_dead(&self, dead: &[ActorAddress]) {
|
||||
self.children_registry.cleanup(dead);
|
||||
for addr in dead {
|
||||
self.name_registry.unregister_by_addr(addr);
|
||||
self.group_registry.cleanup(addr);
|
||||
self.monitor_registry.remove_watcher(addr);
|
||||
self.watch_registry.cleanup_watcher(addr);
|
||||
self.supervisor_registry.cleanup(addr);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -87,6 +118,17 @@ impl RuntimeExtension for StdExtension {
|
|||
self
|
||||
}
|
||||
|
||||
fn on_spawn(&self, child: ActorAddress, parent: Option<ActorAddress>, env: Environment, uptime_ms: u64) -> Environment {
|
||||
// Register parent → child relationship for orphan cleanup
|
||||
if let Some(parent_addr) = parent {
|
||||
self.children_registry.register(parent_addr, child);
|
||||
}
|
||||
let env = self.service_registry.inject_into(env);
|
||||
EnvironmentBuilder::from_env(&env)
|
||||
.set(SpawnTimestamp(uptime_ms))
|
||||
.build()
|
||||
}
|
||||
|
||||
fn create_worker_extension(&self) -> Option<Box<dyn WorkerExtension>> {
|
||||
Some(Box::new(TimerWheel::new()))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,6 +4,10 @@ pub mod name_registry;
|
|||
pub mod monitor_registry;
|
||||
pub mod watch_registry;
|
||||
pub mod group_registry;
|
||||
pub mod supervisor_registry;
|
||||
pub mod service_registry;
|
||||
pub mod resource_handle;
|
||||
pub mod children_registry;
|
||||
pub(crate) mod timer_wheel;
|
||||
mod extension;
|
||||
mod ctx_ext;
|
||||
|
|
@ -12,5 +16,6 @@ mod runtime_ext;
|
|||
pub use supervisor::{ChildSpec, RestartPolicy, Supervisor, SupervisorStrategy};
|
||||
pub use router::{Router, RoutingStrategy};
|
||||
pub use extension::StdExtension;
|
||||
pub use ctx_ext::{CtxMonitoring, CtxNaming, CtxGroups, CtxWatching, CtxTimers};
|
||||
pub use runtime_ext::{RuntimeNaming, RuntimeGroups, RuntimeWatching};
|
||||
pub use ctx_ext::{CtxMonitoring, CtxNaming, CtxGroups, CtxWatching, CtxTimers, CtxSystem, CtxSelfStats, CtxLineage, CtxEnvironment, CtxResources, CtxHandles, CtxLifecycle, CtxCapabilities};
|
||||
pub use resource_handle::ResourceHandle;
|
||||
pub use runtime_ext::{RuntimeNaming, RuntimeGroups, RuntimeWatching, RuntimeResources};
|
||||
|
|
|
|||
44
crates/std/src/resource_handle.rs
Normal file
44
crates/std/src/resource_handle.rs
Normal file
|
|
@ -0,0 +1,44 @@
|
|||
use swactor::actor::ActorAddress;
|
||||
|
||||
/// Typed proxy wrapping a service address for ergonomic domain-specific APIs.
|
||||
///
|
||||
/// Implement this trait on a struct that wraps a service address and provides
|
||||
/// domain-specific methods. Methods take `&self` + `&Ctx` (not stored `&Ctx` —
|
||||
/// avoids lifetime issues with `&mut self` in handlers).
|
||||
///
|
||||
/// # Example
|
||||
///
|
||||
/// ```ignore
|
||||
/// struct CounterHandle {
|
||||
/// service: ActorAddress,
|
||||
/// self_addr: ActorAddress,
|
||||
/// }
|
||||
///
|
||||
/// impl ResourceHandle for CounterHandle {
|
||||
/// type Service = CounterService;
|
||||
/// fn from_parts(service_addr: ActorAddress, self_addr: ActorAddress) -> Self {
|
||||
/// Self { service: service_addr, self_addr }
|
||||
/// }
|
||||
/// fn service_addr(&self) -> ActorAddress { self.service }
|
||||
/// fn self_addr(&self) -> ActorAddress { self.self_addr }
|
||||
/// }
|
||||
///
|
||||
/// impl CounterHandle {
|
||||
/// pub fn increment(&self, ctx: &Ctx) -> Result<(), Error> {
|
||||
/// ctx.send(self.service_addr(), Increment { reply_to: self.self_addr() })
|
||||
/// }
|
||||
/// }
|
||||
/// ```
|
||||
pub trait ResourceHandle: Sized {
|
||||
/// Marker type identifying the service (same `S` used with `ServiceRegistry`).
|
||||
type Service: 'static + Send + Sync;
|
||||
|
||||
/// Construct a handle from a service address and the calling actor's address.
|
||||
fn from_parts(service_addr: ActorAddress, self_addr: ActorAddress) -> Self;
|
||||
|
||||
/// The address of the underlying service actor.
|
||||
fn service_addr(&self) -> ActorAddress;
|
||||
|
||||
/// The address of the actor holding this handle (for reply_to patterns).
|
||||
fn self_addr(&self) -> ActorAddress;
|
||||
}
|
||||
|
|
@ -69,7 +69,7 @@ impl<M: Message> Router<M> {
|
|||
|
||||
fn start_worker(&mut self, ctx: &Ctx, idx: usize) -> Result<(), Error> {
|
||||
let addr = (self.factory)(ctx)?;
|
||||
let mref = ctx.monitor(addr);
|
||||
let mref = ctx.monitor(addr)?;
|
||||
self.workers[idx] = Some(ActiveChild {
|
||||
addr,
|
||||
_monitor_ref: mref,
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
use swactor::actor::{ActorAddress, ActorInterface, Message};
|
||||
use swactor::actor::{ActorAddress, ActorInterface, EnvironmentBuilder, LogicalName, Message};
|
||||
use swactor::runtime::Runtime;
|
||||
use swactor::Error;
|
||||
|
||||
|
|
@ -40,7 +40,10 @@ impl RuntimeNaming for Runtime {
|
|||
|
||||
fn spawn_named<A: ActorInterface>(&self, name: impl Into<String>, actor: A) -> Result<ActorAddress, Error> {
|
||||
let name = name.into();
|
||||
let addr = self.spawn(actor)?;
|
||||
let env = EnvironmentBuilder::new()
|
||||
.set(LogicalName(name.clone()))
|
||||
.build();
|
||||
let addr = self.spawn_with_env(actor, env)?;
|
||||
if let Err(e) = get_ext(self).name_registry.register(name, addr) {
|
||||
let _ = self.stop_actor(addr);
|
||||
return Err(e);
|
||||
|
|
@ -132,3 +135,21 @@ impl RuntimeGroups for Runtime {
|
|||
get_ext(self).group_registry.group_names()
|
||||
}
|
||||
}
|
||||
|
||||
/// Service registry extension for [`Runtime`].
|
||||
///
|
||||
/// Allows registering typed service bindings that are automatically injected
|
||||
/// into every actor's environment at spawn time.
|
||||
pub trait RuntimeResources {
|
||||
/// Register a service address under marker type `S`.
|
||||
///
|
||||
/// All actors spawned after this call will have `ServiceBinding<S>` in
|
||||
/// their environment (unless overridden via `spawn_builder`).
|
||||
fn register_service<S: 'static + Send + Sync>(&self, addr: ActorAddress);
|
||||
}
|
||||
|
||||
impl RuntimeResources for Runtime {
|
||||
fn register_service<S: 'static + Send + Sync>(&self, addr: ActorAddress) {
|
||||
get_ext(self).service_registry.register::<S>(addr);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
52
crates/std/src/service_registry.rs
Normal file
52
crates/std/src/service_registry.rs
Normal file
|
|
@ -0,0 +1,52 @@
|
|||
use std::any::{Any, TypeId};
|
||||
use std::collections::HashMap;
|
||||
use std::sync::{Arc, RwLock};
|
||||
|
||||
use swactor::actor::{Environment, EnvironmentBuilder};
|
||||
|
||||
/// Stores typed service bindings for injection into actor environments.
|
||||
///
|
||||
/// Bindings are registered at the runtime level (e.g., during startup) and
|
||||
/// automatically injected into every actor's environment via the `on_spawn`
|
||||
/// hook. Existing environment keys are **not** overwritten — this preserves
|
||||
/// per-subtree overrides set via `spawn_builder`.
|
||||
pub struct ServiceRegistry {
|
||||
bindings: RwLock<HashMap<TypeId, Arc<dyn Any + Send + Sync>>>,
|
||||
}
|
||||
|
||||
impl ServiceRegistry {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
bindings: RwLock::new(HashMap::new()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Register a service binding by marker type `S`.
|
||||
///
|
||||
/// Overwrites any previous binding for the same marker type.
|
||||
pub fn register<S: 'static + Send + Sync>(&self, addr: swactor::actor::ActorAddress) {
|
||||
let binding = swactor::actor::ServiceBinding::<S>::new(addr);
|
||||
let type_id = TypeId::of::<swactor::actor::ServiceBinding<S>>();
|
||||
self.bindings
|
||||
.write()
|
||||
.unwrap()
|
||||
.insert(type_id, Arc::new(binding));
|
||||
}
|
||||
|
||||
/// Merge all registered bindings into an environment, skipping keys
|
||||
/// that are already present (preserves spawn_builder overrides).
|
||||
pub fn inject_into(&self, env: Environment) -> Environment {
|
||||
let bindings = self.bindings.read().unwrap();
|
||||
if bindings.is_empty() {
|
||||
return env;
|
||||
}
|
||||
|
||||
let mut builder = EnvironmentBuilder::from_env(&env);
|
||||
for (&type_id, value) in bindings.iter() {
|
||||
if !env.contains_type_id(type_id) {
|
||||
builder.set_raw(type_id, Arc::clone(value));
|
||||
}
|
||||
}
|
||||
builder.build()
|
||||
}
|
||||
}
|
||||
|
|
@ -3,6 +3,7 @@ use std::sync::Arc;
|
|||
use swactor::actor::{ActorAddress, ActorInterface, Ctx, Down, MonitorRef, StopReason};
|
||||
use swactor::Error;
|
||||
|
||||
use crate::ctx_ext::get_ext;
|
||||
use crate::CtxMonitoring;
|
||||
|
||||
/// How a child should be restarted when it dies.
|
||||
|
|
@ -140,7 +141,8 @@ impl Supervisor {
|
|||
|
||||
fn start_child(&mut self, ctx: &Ctx, idx: usize) -> Result<(), Error> {
|
||||
let addr = (self.specs[idx].start)(ctx)?;
|
||||
let mref = ctx.monitor(addr);
|
||||
let mref = ctx.monitor(addr)?;
|
||||
get_ext(ctx).supervisor_registry.register(ctx.self_addr(), addr);
|
||||
self.children[idx] = Some(ActiveChild {
|
||||
addr,
|
||||
_monitor_ref: mref,
|
||||
|
|
|
|||
40
crates/std/src/supervisor_registry.rs
Normal file
40
crates/std/src/supervisor_registry.rs
Normal file
|
|
@ -0,0 +1,40 @@
|
|||
use std::sync::RwLock;
|
||||
|
||||
use swactor::actor::ActorAddress;
|
||||
use swactor::AddrMap;
|
||||
|
||||
/// Maps supervised children to their supervisor.
|
||||
///
|
||||
/// Follows the same pattern as `MonitorRegistry`, `GroupRegistry`, etc.
|
||||
/// Entries are added in `Supervisor::start_child` and cleaned up on actor death.
|
||||
pub struct SupervisorRegistry {
|
||||
/// child_addr → supervisor_addr
|
||||
children: RwLock<AddrMap<ActorAddress>>,
|
||||
}
|
||||
|
||||
impl SupervisorRegistry {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
children: RwLock::new(AddrMap::default()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Register a supervisor → child relationship.
|
||||
pub fn register(&self, supervisor: ActorAddress, child: ActorAddress) {
|
||||
self.children.write().unwrap().insert(child, supervisor);
|
||||
}
|
||||
|
||||
/// Look up the supervisor of a child actor.
|
||||
pub fn lookup(&self, child: &ActorAddress) -> Option<ActorAddress> {
|
||||
self.children.read().unwrap().get(child).copied()
|
||||
}
|
||||
|
||||
/// Remove entries where `dead_addr` is either a child or a supervisor.
|
||||
pub fn cleanup(&self, dead_addr: &ActorAddress) {
|
||||
let mut map = self.children.write().unwrap();
|
||||
// Remove the dead actor as a child
|
||||
map.remove(dead_addr);
|
||||
// Remove all children supervised by the dead actor
|
||||
map.retain(|_, supervisor| supervisor != dead_addr);
|
||||
}
|
||||
}
|
||||
|
|
@ -1,7 +1,7 @@
|
|||
use std::collections::{HashMap, HashSet};
|
||||
use std::sync::Mutex;
|
||||
|
||||
use swactor::actor::{ActorAddress, ActorExited, ExitReason};
|
||||
use swactor::actor::{ActorAddress, ActorExited, ExitReason, ExitValue};
|
||||
|
||||
/// Tracks watch relationships between actors.
|
||||
///
|
||||
|
|
@ -55,11 +55,13 @@ impl WatchRegistry {
|
|||
&self,
|
||||
target: ActorAddress,
|
||||
reason: ExitReason,
|
||||
exit_value: Option<ExitValue>,
|
||||
) -> Vec<(ActorAddress, ActorExited)> {
|
||||
let mut state = self.inner.lock().unwrap();
|
||||
let notification = ActorExited {
|
||||
addr: target,
|
||||
reason,
|
||||
exit_value,
|
||||
};
|
||||
let mut result = Vec::new();
|
||||
|
||||
|
|
|
|||
542
docs/development_history/PROCESS_PRIMITIVES.md
Normal file
542
docs/development_history/PROCESS_PRIMITIVES.md
Normal file
|
|
@ -0,0 +1,542 @@
|
|||
# Process Abstraction for Swactor — Development History
|
||||
|
||||
> Design and implementation record for the "process" abstraction layer built
|
||||
> on top of swactor's actor primitives. This work ran across items 1–9 and
|
||||
> added 9 extension traits, 3 registries, and ~70 scenario tests.
|
||||
|
||||
## Context
|
||||
|
||||
Swactor is a distributed actor runtime with local primitives (spawn, send, stop, monitor,
|
||||
supervise) and distributed primitives (SWIM membership, Kademlia directory, cluster-wide naming,
|
||||
content-addressed datastore). The goal was to design a "process" abstraction that sits on top of
|
||||
these primitives, making the experience of running code on a swactor network feel closer to what
|
||||
an OS process feels like -- with access to an API for requesting resources and querying system
|
||||
state.
|
||||
|
||||
---
|
||||
## Part 1: OS Process Mapping
|
||||
|
||||
### Already strong (direct OS equivalents exist)
|
||||
|
||||
OS Concept: PID
|
||||
Swactor Equivalent: ActorAddress (32-byte random)
|
||||
Where: src/actor.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: fork+exec
|
||||
Swactor Equivalent: ctx.spawn(), Runtime::spawn()
|
||||
Where: src/actor.rs, src/runtime.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: exit(0)
|
||||
Swactor Equivalent: ctx.stop_self()
|
||||
Where: src/actor.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: kill(pid, SIGTERM)
|
||||
Swactor Equivalent: ctx.stop_actor(addr)
|
||||
Where: src/actor.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: SIGCHLD / waitpid
|
||||
Swactor Equivalent: ctx.monitor() -> Down, ctx.watch() -> ActorExited
|
||||
Where: crates/std/src/ctx_ext.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: IPC (message queues)
|
||||
Swactor Equivalent: Typed message passing (local + cross-worker + cross-runtime)
|
||||
Where: src/actor.rs, src/transport.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: Service names
|
||||
Swactor Equivalent: NameRegistry (local), ClusterRegistry (cluster CRDT)
|
||||
Where: crates/std/src/name_registry.rs, crates/distribution/src/registry.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: Process groups
|
||||
Swactor Equivalent: GroupRegistry (join/leave/publish/members)
|
||||
Where: crates/std/src/ctx_ext.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: init/systemd
|
||||
Swactor Equivalent: Supervisor with restart strategies
|
||||
Where: crates/std/src/supervisor.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: Scheduler
|
||||
Swactor Equivalent: Worker pool with load-aware placement + per-actor message budgets
|
||||
Where: src/worker.rs, src/delivery.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: Machine identity
|
||||
Swactor Equivalent: NodeId (ed25519 public key)
|
||||
Where: crates/distribution/src/types.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: Cluster membership
|
||||
Swactor Equivalent: SWIM protocol
|
||||
Where: crates/distribution/src/swim/
|
||||
────────────────────────────────────────
|
||||
OS Concept: /proc, top, ps
|
||||
Swactor Equivalent: RuntimeStats, StatsHook, Dashboard, Investigate protocol
|
||||
Where: src/stats.rs, crates/dashboard/
|
||||
|
||||
### Implemented during this work
|
||||
|
||||
OS Concept: System introspection from inside
|
||||
Swactor Equivalent: CtxSystem (worker_id, num_workers, total_actors, uptime_ms) + SystemInfo
|
||||
Where: src/actor.rs, crates/std/src/ctx_ext.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: Per-actor introspection
|
||||
Swactor Equivalent: CtxSelfStats (messages_processed, mailbox_depth, message_type_counts)
|
||||
Where: src/actor.rs, src/worker.rs, crates/std/src/ctx_ext.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: Process lineage (getppid)
|
||||
Swactor Equivalent: CtxLineage (ctx.parent(), ctx.supervisor())
|
||||
Where: src/actor.rs, src/worker.rs, crates/std/src/ctx_ext.rs, crates/std/src/supervisor_registry.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: Process environment (environ/getenv)
|
||||
Swactor Equivalent: CtxEnvironment (ctx.env::<T>(), ctx.environment(), SpawnBuilder for overrides)
|
||||
Where: src/actor.rs, src/worker.rs, crates/std/src/ctx_ext.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: Well-known environment keys (spawn metadata)
|
||||
Swactor Equivalent: SpawnTimestamp(u64) injected by StdExtension on_spawn hook;
|
||||
LogicalName(String) injected by spawn_named (ctx and runtime level)
|
||||
Where: src/actor.rs, src/extension.rs, src/worker.rs, crates/std/src/extension.rs,
|
||||
crates/std/src/ctx_ext.rs, crates/std/src/runtime_ext.rs, src/runtime.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: Service discovery
|
||||
Swactor Equivalent: ServiceRegistry + CtxResources (ctx.resource::<S>() -> Option<ActorAddress>)
|
||||
Where: src/actor.rs, crates/std/src/service_registry.rs, crates/std/src/ctx_ext.rs,
|
||||
crates/std/src/runtime_ext.rs, crates/std/src/extension.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: Resource request API (typed handles)
|
||||
Swactor Equivalent: ResourceHandle trait + CtxHandles (ctx.handle::<H>() -> Option<H>)
|
||||
Where: crates/std/src/resource_handle.rs, crates/std/src/ctx_ext.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: Exit codes / rich exit values
|
||||
Swactor Equivalent: ExitValue(Arc<dyn Any + Send + Sync>), ctx.stop_with(value),
|
||||
StopReason::Completed, ExitReason::Completed. Exit values propagated via Down/ActorExited.
|
||||
Where: src/actor.rs, src/worker.rs, crates/std/src/extension.rs, crates/std/src/watch_registry.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: Parent-child hierarchy + orphan handling
|
||||
Swactor Equivalent: ChildrenRegistry tracks parent->children. On parent death, unsupervised
|
||||
children are killed (StopSignal). Supervised children are left to their supervisor. Cascades
|
||||
naturally across generations via tick-based cleanup.
|
||||
Where: crates/std/src/children_registry.rs, crates/std/src/extension.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: Suspend/resume (SIGSTOP/SIGCONT)
|
||||
Swactor Equivalent: ctx.suspend_self(), ctx.resume(target) with auth (self or supervisor only).
|
||||
Suspended actors queue messages but don't process them. ResumeSignal via transfer queue for
|
||||
cross-worker resume.
|
||||
Where: src/actor.rs, src/worker.rs, src/runtime.rs, crates/std/src/ctx_ext.rs
|
||||
────────────────────────────────────────
|
||||
OS Concept: Capability model / sandboxing
|
||||
Swactor Equivalent: CapabilitySet stored in actor's Environment. Enforced at Ctx level (send,
|
||||
spawn, stop_actor, monitor, resource). Opt-in: actors without a CapabilitySet are unrestricted.
|
||||
Where: src/actor.rs, crates/std/src/ctx_ext.rs
|
||||
|
||||
### Still partially there
|
||||
|
||||
OS Concept: Resource limits
|
||||
What Exists: Mailbox capacity + message budget
|
||||
What's Missing: No per-actor memory/CPU/fd limits
|
||||
────────────────────────────────────────
|
||||
OS Concept: Auth/permissions
|
||||
What Exists: Datastore ACL + node-level peer auth + actor-level CapabilitySet
|
||||
What's Missing: Cluster-level capability propagation (local-only today)
|
||||
|
||||
---
|
||||
## Part 2: Design Primitives
|
||||
|
||||
The design followed the existing extension pattern: new capabilities were added as extension traits
|
||||
on Ctx<'_>, backed by registries in the extension system. This preserved backwards compatibility
|
||||
and kept the core minimal.
|
||||
|
||||
### 2.1 System Queries (CtxSystem)
|
||||
|
||||
What it enables: An actor can ask about the system it's running in.
|
||||
|
||||
Implemented queries (available via ctx.system_info() or the CtxSystem extension trait):
|
||||
- ctx.worker_id() -> usize -- which worker thread am I on?
|
||||
- ctx.num_workers() -> usize -- how many worker threads exist?
|
||||
- ctx.total_actors() -> usize -- live actors across all workers
|
||||
- ctx.uptime_ms() -> u64 -- milliseconds since runtime creation
|
||||
|
||||
Implementation: SystemInfo struct in src/actor.rs. ContextInner::system_info() implemented on
|
||||
both Runtime (for spawn-time context) and WorkerContext (for handler context). Data flows through
|
||||
TickContext (worker_stats + created_at fields in src/delivery.rs). The CtxSystem extension trait
|
||||
in crates/std/src/ctx_ext.rs provides ergonomic per-field accessors.
|
||||
|
||||
Future cluster-level queries (not yet implemented):
|
||||
- What is my node's identity (NodeId)?
|
||||
- How many cluster nodes are alive?
|
||||
- Who are the cluster members?
|
||||
|
||||
These require the distribution crate's DistributedNode state to be exposed through the extension
|
||||
system. The CtxSystem trait can be extended with these when the distribution integration is ready.
|
||||
|
||||
### 2.2 Process Environment (CtxEnvironment)
|
||||
|
||||
What it enables: Typed configuration that flows from parent to child at spawn time.
|
||||
|
||||
Properties:
|
||||
- Inherited: When actor A spawns actor B via ctx.spawn(), B gets A's environment (Arc clone)
|
||||
- Overridable: ctx.spawn_builder(actor).env(Key(val)).finish() lazily clones the parent's map
|
||||
on first override (copy-on-write), leaving the common case (no overrides) allocation-free
|
||||
- Immutable after spawn: Set at creation, read-only thereafter. Mutable config goes through
|
||||
messages.
|
||||
- Typed values: TypeId-keyed (like http::Extensions), not string-to-string
|
||||
- Runtime-spawned actors start with an empty environment
|
||||
|
||||
Implementation: Environment is Arc<HashMap<TypeId, Arc<dyn Any + Send + Sync>>> -- clone is an
|
||||
Arc bump (zero allocation). EnvironmentBuilder provides from_env() for copy-on-write overrides
|
||||
(cloning individual entries is cheap since values are also Arc-wrapped). The spawn channel was
|
||||
replaced with a SpawnRequest struct (addr, actor, parent, env) to avoid further tuple growth.
|
||||
ActorSlot stores env, and Ctx receives it at both construction sites (tick_all and cleanup_dead).
|
||||
SpawnBuilder provides the ergonomic override API. The CtxEnvironment extension trait in
|
||||
crates/std/src/ctx_ext.rs provides the import path, following the same pattern as CtxLineage
|
||||
(no StdExtension dependency required). Python crate spawns with Environment::new(). 6 scenario
|
||||
tests in tests/std_extension.rs cover: inheritance, empty for runtime-spawned, grandchild chain,
|
||||
override-one-inherit-others, readable in on_stop, and sibling independence.
|
||||
|
||||
Well-known keys:
|
||||
- SpawnTimestamp(u64): Injected by StdExtension's on_spawn hook. Milliseconds since runtime
|
||||
creation, same time base as SystemInfo::uptime_ms. Opt-in at runtime level (present when
|
||||
StdExtension is installed). Read via ctx.env::<SpawnTimestamp>().
|
||||
- LogicalName(String): Injected by spawn_named() at both ctx and Runtime levels. Inherited by
|
||||
children via normal environment inheritance. Read via ctx.env::<LogicalName>().
|
||||
- ServiceBinding<S>(ActorAddress): Injected by ServiceRegistry's inject_into() hook during
|
||||
on_spawn. Registered at runtime level via rt.register_service::<S>(addr). Read via
|
||||
ctx.resource::<S>() (CtxResources trait). Overridable per-subtree via spawn_builder.
|
||||
- CapabilitySet: Granted at spawn time (via environment or spawn_builder). Inherited by children.
|
||||
Enforced at Ctx level. See section 2.7.
|
||||
|
||||
Analogy: Unix environ -- inherited by default, augmented at fork/exec time, readable via getenv().
|
||||
|
||||
### 2.3 Service Discovery (CtxResources)
|
||||
|
||||
What it enables: Actors can discover system services by type, not by knowing raw addresses.
|
||||
|
||||
How it differs from NameRegistry: NameRegistry maps strings to addresses. CtxResources maps
|
||||
service marker types to addresses. Looking up "datastore" by name gives you a raw ActorAddress and
|
||||
you must know what messages it accepts. ctx.resource::<Datastore>() gives you the address of the
|
||||
service registered under that marker type.
|
||||
|
||||
Implementation: Three layers compose the feature:
|
||||
|
||||
1. Core type: ServiceBinding<S>(ActorAddress) in src/actor.rs -- a generic environment key
|
||||
parameterized by a zero-sized marker type. Any struct satisfying 'static + Send + Sync works
|
||||
as a marker (no special Service trait required, consistent with Environment's existing API).
|
||||
|
||||
2. Registry + injection: ServiceRegistry in crates/std/src/service_registry.rs stores registered
|
||||
bindings as RwLock<HashMap<TypeId, Arc<dyn Any + Send + Sync>>> (same thread-safety pattern
|
||||
as SupervisorRegistry). StdExtension's on_spawn hook calls inject_into() before adding
|
||||
SpawnTimestamp -- this merges all registered bindings into the actor's environment, skipping
|
||||
keys already present (preserves per-subtree overrides set via spawn_builder). Helper methods
|
||||
on Environment (contains_type_id) and EnvironmentBuilder (set_raw) support type-erased
|
||||
injection without knowing concrete types at compile time.
|
||||
|
||||
3. Read API: CtxResources trait in crates/std/src/ctx_ext.rs provides ctx.resource::<S>() ->
|
||||
Option<ActorAddress>, a thin wrapper around ctx.env::<ServiceBinding<S>>().map(|b| b.addr).
|
||||
Does NOT require StdExtension -- reads from core environment (same pattern as CtxEnvironment).
|
||||
When a CapabilitySet is present, resource() checks check_service::<S>() and returns None if
|
||||
denied. RuntimeResources trait in crates/std/src/runtime_ext.rs provides
|
||||
rt.register_service::<S>(addr) for startup-time registration.
|
||||
|
||||
Key design decisions:
|
||||
- No Service marker trait: S: 'static + Send + Sync is sufficient. Any zero-size struct works.
|
||||
- "Skip if present" injection: The registry doesn't overwrite env keys set by spawn_builder,
|
||||
enabling per-subtree service overrides (e.g., test doubles, staging vs production services).
|
||||
- No cleanup on service actor death: A dead service's binding stays in the registry (stale
|
||||
address). Sends to it will fail. Service lifecycle management is a higher-level concern.
|
||||
|
||||
6 scenario tests in tests/std_extension.rs cover: discovery by marker type, child inherits
|
||||
binding from parent, multiple services each accessible by marker, unregistered returns None,
|
||||
overridable via spawn_builder, accessible in on_start and on_stop lifecycle hooks.
|
||||
|
||||
Well-known services that could be registered (when swactor-node is updated):
|
||||
- Storage -- content-addressed datastore (currently wired manually in swactor-node)
|
||||
- Directory -- actor location resolution (currently locked inside DistributedNode)
|
||||
- Cluster -- membership/topology info (currently snapshot-only for dashboard)
|
||||
- Metrics -- runtime stats (currently StatsHook push-only)
|
||||
|
||||
### 2.4 Resource Handles (CtxHandles)
|
||||
|
||||
What it enables: Domain-specific typed proxies that wrap service addresses and provide ergonomic
|
||||
APIs.
|
||||
|
||||
The pattern: A handle wraps (service_address, self_address) and provides methods that construct
|
||||
and send the right messages, embedding self_addr as reply_to. Responses arrive as normal messages
|
||||
in the actor's handle().
|
||||
|
||||
Implementation: The ResourceHandle trait in crates/std/src/resource_handle.rs defines the contract:
|
||||
- type Service: 'static + Send + Sync -- the marker type used for service discovery
|
||||
- from_parts(service_addr, self_addr) -> Self -- construct from addresses
|
||||
- service_addr() -> ActorAddress -- the underlying service address
|
||||
- self_addr() -> ActorAddress -- the actor's own address (for reply_to)
|
||||
|
||||
The CtxHandles extension trait in crates/std/src/ctx_ext.rs provides ctx.handle::<H>() -> Option<H>,
|
||||
which looks up ServiceBinding<H::Service> from the actor's environment and constructs the handle.
|
||||
Returns None if the service is not registered (consistent with ctx.resource(), ctx.where_is()).
|
||||
|
||||
Handle methods take &self + &Ctx (not stored &Ctx -- avoids lifetime issues with &mut self in
|
||||
handlers). Example:
|
||||
impl MyHandle {
|
||||
pub fn do_work(&self, ctx: &Ctx, data: Vec<u8>) -> Result<(), Error> {
|
||||
ctx.send(self.service_addr(), MyMsg::DoWork { data, reply_to: self.self_addr() })
|
||||
}
|
||||
}
|
||||
|
||||
Key design tension: Handles can't block (no await in swactor). The response arrives asynchronously
|
||||
as a message. This is inherent to the actor model and not something to "fix" -- the handle just
|
||||
makes the send side ergonomic.
|
||||
|
||||
5 scenario tests: handle wraps service and sends ergonomically, returns None when service not
|
||||
registered, inherits service binding from parent, constructible in on_start, two actors with same
|
||||
handle type each get responses at their own address.
|
||||
|
||||
### 2.5 Process Lineage (CtxLineage)
|
||||
|
||||
What it enables: Actors know their ancestry.
|
||||
|
||||
Implemented queries:
|
||||
- ctx.parent() -> Option<ActorAddress> (who spawned me?)
|
||||
Returns Some(spawner_addr) for actor-spawned children, None for Runtime::spawn().
|
||||
Available in handle(), on_start(), and on_stop().
|
||||
- ctx.supervisor() -> Option<ActorAddress> (who supervises me, if anyone?)
|
||||
Returns Some(supervisor_addr) for supervised children, None for unsupervised actors.
|
||||
Gracefully returns None when StdExtension is absent (no panic).
|
||||
|
||||
Implementation (parent): The spawn channel uses a SpawnRequest struct (addr, actor, parent, env) --
|
||||
the original 3-tuple was replaced when CtxEnvironment was added. When Ctx::spawn is called, the
|
||||
spawning actor's self_addr is passed as Some(parent). Runtime::spawn passes None. The parent is
|
||||
stored in ActorSlot::parent_addr and threaded into Ctx::self_parent_addr at both construction sites
|
||||
(tick_all and cleanup_dead). 4 scenario tests cover: child knows parent, runtime-spawned has no
|
||||
parent, grandchild sees immediate parent (not grandparent), and parent is visible in on_stop.
|
||||
|
||||
Implementation (supervisor): SupervisorRegistry in crates/std/src/supervisor_registry.rs stores a
|
||||
child_addr -> supervisor_addr map (RwLock<AddrMap<ActorAddress>>). Supervisor::start_child calls
|
||||
register(self_addr, child_addr) after spawning and monitoring. cleanup() removes entries where the
|
||||
dead address is either child or supervisor (O(n) scan for supervisor death, acceptable since
|
||||
supervisor death is rare and the map is small). CtxLineage::supervisor() downcasts the extension
|
||||
gracefully (returns None if StdExtension is absent). 5 scenario tests cover: supervised child knows
|
||||
supervisor, unsupervised actor returns None, supervisor survives child restart, grandchild not
|
||||
supervised but parent is, OneForAll restart re-registers all children.
|
||||
|
||||
The CtxLineage extension trait in crates/std/src/ctx_ext.rs provides the ergonomic import path.
|
||||
|
||||
Orphan handling was implemented as part of item 8 (Lifecycle Enrichment) -- see section 2.8.
|
||||
|
||||
### 2.6 Self-Introspection (CtxSelfStats)
|
||||
|
||||
What it enables: Actors can see their own operational metrics.
|
||||
|
||||
Implemented queries (available directly on Ctx or via the CtxSelfStats extension trait):
|
||||
- ctx.messages_processed() -> u64 -- total successfully processed before current tick
|
||||
- ctx.mailbox_depth() -> usize -- messages queued at start of current tick (pre-dequeue)
|
||||
- ctx.message_type_counts() -> &[(&str, u64)] -- per-type counts, sorted descending
|
||||
|
||||
Implementation: Stats are snapshotted from ActorSlot fields into Ctx before each tick_all
|
||||
iteration (src/worker.rs). The snapshot captures the state before any messages are dequeued
|
||||
in the current tick, giving actors a consistent view. The same snapshot is provided during
|
||||
on_stop callbacks in cleanup_dead. The CtxSelfStats extension trait in crates/std/src/ctx_ext.rs
|
||||
provides the ergonomic import path.
|
||||
|
||||
The Vec allocation for type counts is bounded (max 32 entries from ActorSlot's msg_type_counts
|
||||
cap) and negligible relative to handle_any cost.
|
||||
|
||||
### 2.7 Capability Model (CapabilitySet + CtxCapabilities)
|
||||
|
||||
What it enables: Controlled access to system resources and other actors. Primarily important for
|
||||
sandboxing untrusted code (wasm actors in crates/bin-runner/).
|
||||
|
||||
Approach: A single CapabilitySet stored in the actor's Environment. When present, enforcement is
|
||||
active -- the actor can only perform operations granted by the set. When absent, the actor is
|
||||
unrestricted (backward compatible). Capabilities inherit from parent to child via normal
|
||||
environment inheritance.
|
||||
|
||||
Capability grants (all in CapabilitySet):
|
||||
- with_send(addr) -- send any message type to a specific address
|
||||
- with_send_typed::<M>(addr) -- send only messages of type M to a specific address
|
||||
- with_spawn() -- permission to spawn new actors
|
||||
- with_service::<S>() -- permission to access system service S via ctx.resource::<S>()
|
||||
- with_monitor(addr) -- permission to monitor a specific actor
|
||||
|
||||
Enforcement points (all in src/actor.rs Ctx methods or crates/std/src/ctx_ext.rs):
|
||||
- ctx.send::<M>(addr, msg) -- checks check_send::<M>(addr); self-send always allowed
|
||||
- ctx.spawn() / SpawnBuilder::finish() -- checks check_spawn()
|
||||
- ctx.stop_actor(addr) -- checks check_send_addr(addr) (stop is a send of StopSignal)
|
||||
- ctx.monitor(addr) -- checks check_monitor(addr); returns Result<MonitorRef, Error>
|
||||
- ctx.resource::<S>() -- checks check_service::<S>(); returns None if denied
|
||||
|
||||
Key design decisions:
|
||||
- Opt-in: No CapabilitySet in environment means unrestricted. Zero behavioral change for existing
|
||||
actors. The only cost is an Option check (env.get::<CapabilitySet>()) at each enforcement point.
|
||||
- Enforcement at Ctx level only: The core ContextInner::send_any is not gated. This means
|
||||
extension code (supervisors, timers, etc.) that calls send_any directly bypasses capability
|
||||
checks, which is intentional -- system infrastructure is trusted.
|
||||
- Dual send granularity: with_send(addr) grants all message types to an address.
|
||||
with_send_typed::<M>(addr) grants only type M. The check tries address-only first, then typed.
|
||||
This allows coarse grants for trusted peers and fine-grained grants for untrusted actors.
|
||||
- Self-send always allowed: A restricted actor can always send to its own address. This prevents
|
||||
capabilities from breaking actors that use self-messaging patterns (timers, state machines).
|
||||
- monitor() returns Result: Changed from -> MonitorRef to -> Result<MonitorRef, Error>. This was
|
||||
a breaking change to all callers (supervisor.rs, router.rs, test files), fixed mechanically by
|
||||
adding ? or .unwrap().
|
||||
|
||||
Builder API: Fluent (CapabilitySet::new().with_send(addr).with_spawn()) and mutable
|
||||
(caps.grant_send(addr)) variants. Mutable methods return &mut Self for chaining.
|
||||
|
||||
Introspection: CtxCapabilities extension trait in crates/std/src/ctx_ext.rs provides:
|
||||
- ctx.capabilities() -> Option<&CapabilitySet> -- access the raw set
|
||||
- ctx.is_restricted() -> bool -- quick check
|
||||
|
||||
Implementation locations:
|
||||
- src/actor.rs: CapabilitySet struct, builder methods, check methods, Ctx::capabilities() helper,
|
||||
enforcement in send/spawn/stop_actor/SpawnBuilder::finish
|
||||
- src/lib.rs: CapabilitySet re-export
|
||||
- crates/std/src/ctx_ext.rs: CtxCapabilities trait, monitor() enforcement, resource() enforcement
|
||||
- crates/std/src/lib.rs: CtxCapabilities re-export
|
||||
|
||||
11 scenario tests in tests/std_extension.rs cover: unrestricted actor sends freely (backward
|
||||
compat), restricted actor denied send, restricted actor allowed send, typed send grant (Ping
|
||||
allowed / Pong denied), spawn denied, spawn allowed, capability inheritance (child inherits
|
||||
parent's CapabilitySet), monitor denied, service access denied, self-send always allowed, stop
|
||||
requires send permission.
|
||||
|
||||
### 2.8 Lifecycle Enrichment
|
||||
|
||||
Rich exit values: ExitValue(Arc<dyn Any + Send + Sync>) is an opaque typed wrapper. Actors stop
|
||||
with ctx.stop_with(value) which stores the value and triggers StopReason::Completed. The value
|
||||
is propagated through Down (monitors) and ActorExited (watchers) via the exit_value: Option<ExitValue>
|
||||
field. Manual PartialEq/Eq on ExitValue (always false -- opaque blob), so Down/ActorExited compare
|
||||
by addr+reason only.
|
||||
|
||||
Implementation: StopWithSignal(ExitValue) is a sentinel message intercepted in tick_all (like
|
||||
StopSignal). ActorSlot gains exit_value: Option<ExitValue>. cleanup_dead returns
|
||||
Vec<(ActorAddress, StopReason, Option<ExitValue>)> with StopReason::Completed when exit_value is
|
||||
present. The on_actor_death extension hook receives and propagates exit values to monitors/watchers.
|
||||
|
||||
7 scenario tests: stop_with value received in Down, received in ActorExited, normal stop has None,
|
||||
panic has None, multiple monitors receive cloned value, stop_with from on_start, supervisor receives
|
||||
rich exit in handle_down (graceful handoff pattern).
|
||||
|
||||
Orphan handling: ChildrenRegistry tracks parent -> set of children. Populated in on_spawn when a
|
||||
parent is present. On parent death (on_actor_death), unsupervised children receive StopSignal.
|
||||
Supervised children are left to their supervisor. Cascades naturally: parent dies -> children killed
|
||||
next tick -> grandchildren killed the tick after that. StopSignal made pub (was pub(crate)) to
|
||||
enable this -- it's not Message (not Clone) so can't be sent via ctx.send().
|
||||
|
||||
4 scenario tests: unsupervised children killed on parent death, supervised children not killed,
|
||||
cascading cleanup across generations, runtime-spawned actors unaffected.
|
||||
|
||||
Suspend/resume: ActorSlot gains a suspended: bool flag. Suspended actors queue messages but don't
|
||||
process them (tick_all skips them). ctx.suspend_self() sets the flag via a suspend_requests buffer.
|
||||
ResumeSignal is intercepted in deliver() to clear the flag. StopSignal/StopWithSignal are also
|
||||
intercepted for suspended actors (so stop_actor works on them). Cross-worker resume sends
|
||||
ResumeSignal via the transfer queue.
|
||||
|
||||
Authorization: CtxLifecycle extension trait provides ctx.suspend_self() (always allowed) and
|
||||
ctx.resume(target) which checks: target == self (self-resume) OR caller is the target's supervisor
|
||||
via SupervisorRegistry. Returns Err if unauthorized.
|
||||
|
||||
5 scenario tests: suspended actor queues then resume processes, supervisor can resume, non-supervisor
|
||||
cannot resume, suspended actor can be stopped, cross-worker resume via runtime.
|
||||
|
||||
Graceful handoff: Built on rich exit values. An outgoing actor stops with its state via
|
||||
ctx.stop_with(state); the supervisor receives it in handle_down's Down message and can pass it
|
||||
to the replacement's constructor. Enables zero-downtime upgrades. No additional mechanism needed --
|
||||
the pattern composes from existing primitives.
|
||||
|
||||
---
|
||||
## Part 3: How These Compose
|
||||
|
||||
The primitives form a layered system:
|
||||
|
||||
Layer 3: Integration (swactor-node wires services at startup)
|
||||
Layer 2: Process (CapabilitySet, ProcessBuilder)
|
||||
Layer 1: Std (CtxSystem, CtxEnvironment, CtxLineage, CtxSelfStats, Well-known env keys,
|
||||
SupervisorRegistry, CtxResources, CtxHandles, CtxLifecycle, ChildrenRegistry,
|
||||
CtxCapabilities)
|
||||
Layer 0: Core (SystemInfo, Ctx self-stats, parent tracking, Environment + SpawnRequest,
|
||||
on_spawn hook, spawn_with_env, ServiceBinding, suspend flag, rich exit, orphan
|
||||
handling, CapabilitySet)
|
||||
|
||||
A "process" in swactor is an actor that has:
|
||||
1. An identity (ActorAddress) and a name (NameRegistry)
|
||||
2. A parent and supervisor it can query (CtxLineage)
|
||||
3. An environment inherited from its spawner, with well-known keys (CtxEnvironment)
|
||||
4. Access to system services through discovery (CtxResources)
|
||||
5. The ability to query the system it lives in (CtxSystem)
|
||||
6. Awareness of its own operational state (CtxSelfStats)
|
||||
7. Typed resource handles for ergonomic service interaction (CtxHandles)
|
||||
8. Rich lifecycle support including typed exit values, orphan handling, and suspend/resume
|
||||
9. Controlled permissions for what it can access (CapabilitySet)
|
||||
|
||||
What stayed the same: The core actor model (message passing, mailboxes, workers, tick-based
|
||||
execution) was unchanged. ActorInterface, Ctx, Runtime remained the foundation. The process
|
||||
abstraction was additive -- existing actors continued to work exactly as before.
|
||||
|
||||
---
|
||||
## Part 4: Implementation Sequence
|
||||
|
||||
Each item was implemented and merged in dependency order. Earlier items established the
|
||||
infrastructure (Environment, extension hooks) that later items built on.
|
||||
|
||||
1. **CtxSystem + CtxSelfStats** -- Exposed existing internal data to actors. SystemInfo struct,
|
||||
ContextInner::system_info(), Ctx self-stats snapshot fields. Extension traits CtxSystem and
|
||||
CtxSelfStats in swactor-std. Covered by 3 scenario tests.
|
||||
|
||||
2. **CtxLineage (parent tracking)** -- Option<ActorAddress> threaded through the spawn path.
|
||||
ContextInner::spawn_any gained a parent parameter. ActorSlot stores parent_addr. Ctx exposes
|
||||
parent(). CtxLineage extension trait in swactor-std. 4 scenario tests.
|
||||
Python crate updated to pass parent on spawn.
|
||||
|
||||
3. **CtxEnvironment (process environment)** -- Typed key-value map inherited from parent to
|
||||
child at spawn time. Environment is Arc<HashMap<TypeId, Arc<dyn Any + Send + Sync>>> -- clone
|
||||
is an Arc bump. EnvironmentBuilder supports copy-on-write overrides via from_env(). The spawn
|
||||
channel 3-tuple was replaced with a SpawnRequest struct (addr, actor, parent, env) to stop
|
||||
tuple growth. ActorSlot stores env. Ctx gains env::<T>(), environment(), and spawn_builder().
|
||||
SpawnBuilder lazily clones the parent's map on first .env() call. CtxEnvironment extension trait
|
||||
in swactor-std (no StdExtension dependency). Python crate spawns with Environment::new().
|
||||
6 scenario tests: inheritance, empty for runtime-spawned, grandchild chain,
|
||||
override-one-inherit-others, readable in on_stop, sibling independence.
|
||||
|
||||
4. **Well-known environment keys** -- SpawnTimestamp(u64) and LogicalName(String) types in
|
||||
src/actor.rs, exported from src/lib.rs. SpawnTimestamp is opt-in at runtime level: injected by
|
||||
StdExtension's on_spawn hook (new RuntimeExtension::on_spawn hook with default no-op in
|
||||
src/extension.rs). Worker::drain_spawns now takes &TickContext and calls on_spawn for each
|
||||
spawn request, passing uptime_ms to avoid exposing the pub(crate) Instant type. LogicalName is
|
||||
injected by spawn_named at both ctx level (via spawn_builder + env override) and runtime level
|
||||
(via new Runtime::spawn_with_env method). LogicalName inherits to children automatically via
|
||||
normal environment inheritance. 7 scenario tests.
|
||||
|
||||
5. **Supervisor lineage (ctx.supervisor())** -- SupervisorRegistry in
|
||||
crates/std/src/supervisor_registry.rs stores child_addr -> supervisor_addr as
|
||||
RwLock<AddrMap<ActorAddress>>. Supervisor::start_child calls register() after spawning and
|
||||
monitoring. cleanup() removes entries for dead actors (both as child and as supervisor).
|
||||
CtxLineage::supervisor() gracefully returns None when StdExtension is absent (downcasts via
|
||||
as_any, no panic). Distinct from parent() because not every parent is a supervisor. get_ext
|
||||
made pub(crate) so supervisor.rs can access it. 5 scenario tests.
|
||||
|
||||
6. **Service Registry + CtxResources** -- Actors discover system services by type
|
||||
(ctx.resource::<Datastore>()) rather than by raw address. ServiceBinding<S>(ActorAddress)
|
||||
is a generic environment key parameterized by a marker type. ServiceRegistry in StdExtension
|
||||
stores bindings and injects them into every actor's environment via on_spawn (skipping keys
|
||||
already present to preserve spawn_builder overrides). CtxResources trait provides
|
||||
ctx.resource::<S>() sugar. RuntimeResources trait provides rt.register_service::<S>(addr).
|
||||
6 scenario tests.
|
||||
|
||||
7. **Resource Handles (CtxHandles)** -- ResourceHandle trait + CtxHandles extension trait.
|
||||
ctx.handle::<H>() -> Option<H> constructs typed proxies from ServiceBinding<H::Service> in the
|
||||
actor's environment. Handle methods take &self + &Ctx for ergonomic domain-specific APIs.
|
||||
5 scenario tests.
|
||||
|
||||
8. **Lifecycle enrichment** -- Three sub-features:
|
||||
a) Rich exit values: ExitValue(Arc<dyn Any + Send + Sync>), ctx.stop_with(value),
|
||||
StopReason::Completed, ExitReason::Completed. Propagated through Down/ActorExited.
|
||||
7 scenario tests.
|
||||
b) Orphan handling: ChildrenRegistry tracks parent->children. Unsupervised children killed
|
||||
on parent death. Supervised children left to their supervisor. Natural cascade.
|
||||
4 scenario tests.
|
||||
c) Suspend/resume: ActorSlot::suspended flag, ctx.suspend_self(), ctx.resume(target) with
|
||||
auth (self or supervisor only). ResumeSignal for cross-worker resume.
|
||||
5 scenario tests.
|
||||
|
||||
9. **Capability model (CapabilitySet)** -- Per-actor permission set stored in the Environment.
|
||||
Grants: with_send(addr), with_send_typed::<M>(addr), with_spawn(), with_service::<S>(),
|
||||
with_monitor(addr). Enforced at Ctx level in send, spawn, stop_actor, monitor, and resource.
|
||||
Opt-in: actors without a CapabilitySet are unrestricted (zero behavioral change). Self-send
|
||||
always allowed. monitor() changed from -> MonitorRef to -> Result<MonitorRef, Error> (breaking
|
||||
change, fixed mechanically in supervisor.rs, router.rs, and all test files). CtxCapabilities
|
||||
extension trait for introspection. 11 scenario tests.
|
||||
505
src/actor.rs
505
src/actor.rs
|
|
@ -1,7 +1,60 @@
|
|||
use std::any::Any;
|
||||
use std::any::TypeId;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::sync::Arc;
|
||||
|
||||
use crate::Error;
|
||||
|
||||
// ─── ExitValue ──────────────────────────────────────────────────────────────
|
||||
|
||||
/// Opaque typed value attached to a completed actor's exit.
|
||||
///
|
||||
/// Created via [`Ctx::stop_with`]. Delivered to monitors/watchers in
|
||||
/// [`Down::exit_value`] and [`ActorExited::exit_value`].
|
||||
///
|
||||
/// Clone is an `Arc` bump (zero allocation).
|
||||
#[derive(Clone)]
|
||||
pub struct ExitValue(Arc<dyn Any + Send + Sync>);
|
||||
|
||||
impl ExitValue {
|
||||
/// Wrap a typed value as an opaque exit value.
|
||||
pub fn new<T: Any + Send + Sync>(value: T) -> Self {
|
||||
Self(Arc::new(value))
|
||||
}
|
||||
|
||||
/// Attempt to downcast to a concrete type by reference.
|
||||
pub fn downcast_ref<T: Any>(&self) -> Option<&T> {
|
||||
self.0.downcast_ref::<T>()
|
||||
}
|
||||
}
|
||||
|
||||
impl PartialEq for ExitValue {
|
||||
fn eq(&self, _other: &Self) -> bool {
|
||||
false // opaque blob — always not equal
|
||||
}
|
||||
}
|
||||
|
||||
impl Eq for ExitValue {}
|
||||
|
||||
impl std::fmt::Debug for ExitValue {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.write_str("ExitValue(..)")
|
||||
}
|
||||
}
|
||||
|
||||
/// System-level information visible to actors.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct SystemInfo {
|
||||
/// Index of the worker thread this actor is running on.
|
||||
pub worker_id: usize,
|
||||
/// Total number of worker threads in the runtime.
|
||||
pub num_workers: usize,
|
||||
/// Total number of live actors across all workers.
|
||||
pub total_actors: usize,
|
||||
/// Milliseconds since the runtime was created.
|
||||
pub uptime_ms: u64,
|
||||
}
|
||||
|
||||
/// Why an actor exited.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
|
||||
|
|
@ -12,6 +65,8 @@ pub enum ExitReason {
|
|||
Panicked,
|
||||
/// The node hosting the actor left the cluster (SWIM Dead).
|
||||
NodeDown,
|
||||
/// Actor stopped with a typed exit value (via [`Ctx::stop_with`]).
|
||||
Completed,
|
||||
}
|
||||
|
||||
/// Delivered to watchers when a watched actor exits.
|
||||
|
|
@ -25,8 +80,19 @@ pub struct ActorExited {
|
|||
pub addr: ActorAddress,
|
||||
/// Why it exited.
|
||||
pub reason: ExitReason,
|
||||
/// Typed exit value if the actor called [`Ctx::stop_with`].
|
||||
#[cfg_attr(feature = "serde", serde(skip))]
|
||||
pub exit_value: Option<ExitValue>,
|
||||
}
|
||||
|
||||
impl PartialEq for ActorExited {
|
||||
fn eq(&self, other: &Self) -> bool {
|
||||
self.addr == other.addr && self.reason == other.reason
|
||||
}
|
||||
}
|
||||
|
||||
impl Eq for ActorExited {}
|
||||
|
||||
/// The primary trait defining data that can be passed to and from actor processes
|
||||
pub trait Message: 'static + Sized + Clone + Send + Sync {}
|
||||
impl<T: 'static + Sized + Clone + Send + Sync> Message for T {}
|
||||
|
|
@ -100,6 +166,250 @@ impl ActorAddress {
|
|||
}
|
||||
}
|
||||
|
||||
// ─── Environment ─────────────────────────────────────────────────────────────
|
||||
|
||||
/// A typed key-value map that flows from parent to child at spawn time.
|
||||
///
|
||||
/// Analogous to Unix `environ` — provides inherited configuration without
|
||||
/// threading values through every constructor. Clone is an Arc bump (zero allocation).
|
||||
///
|
||||
/// Values are stored as `Arc<dyn Any>` so that [`EnvironmentBuilder::from_env`]
|
||||
/// can clone individual entries cheaply (Arc bump) for copy-on-write overrides.
|
||||
#[derive(Clone, Default)]
|
||||
pub struct Environment {
|
||||
inner: Arc<HashMap<TypeId, Arc<dyn Any + Send + Sync>>>,
|
||||
}
|
||||
|
||||
impl Environment {
|
||||
/// Create an empty environment.
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
inner: Arc::new(HashMap::new()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Read a typed value from the environment.
|
||||
pub fn get<T: Any + Send + Sync>(&self) -> Option<&T> {
|
||||
self.inner
|
||||
.get(&TypeId::of::<T>())
|
||||
.and_then(|v| v.downcast_ref::<T>())
|
||||
}
|
||||
|
||||
/// Check if the environment contains a value of type `T`.
|
||||
pub fn contains<T: Any + Send + Sync>(&self) -> bool {
|
||||
self.inner.contains_key(&TypeId::of::<T>())
|
||||
}
|
||||
|
||||
/// Returns `true` if the environment has no values.
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.inner.is_empty()
|
||||
}
|
||||
|
||||
/// Number of typed values in the environment.
|
||||
pub fn len(&self) -> usize {
|
||||
self.inner.len()
|
||||
}
|
||||
|
||||
/// Check if the environment contains a value with the given `TypeId`.
|
||||
///
|
||||
/// Type-erased version of [`contains`](Self::contains) — used by
|
||||
/// `ServiceRegistry::inject_into` to skip keys already present.
|
||||
pub fn contains_type_id(&self, type_id: TypeId) -> bool {
|
||||
self.inner.contains_key(&type_id)
|
||||
}
|
||||
}
|
||||
|
||||
/// Builder for constructing an [`Environment`].
|
||||
///
|
||||
/// Allows inserting/replacing typed values before freezing into an immutable `Environment`.
|
||||
pub struct EnvironmentBuilder {
|
||||
map: HashMap<TypeId, Arc<dyn Any + Send + Sync>>,
|
||||
}
|
||||
|
||||
impl EnvironmentBuilder {
|
||||
/// Create an empty builder.
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
map: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Create a builder pre-populated with values from an existing environment.
|
||||
///
|
||||
/// This enables copy-on-write overrides: clone the parent's map, modify, then freeze.
|
||||
/// Cloning entries is cheap — each value is `Arc`-wrapped.
|
||||
pub fn from_env(env: &Environment) -> Self {
|
||||
let map = env.inner.as_ref().clone();
|
||||
Self { map }
|
||||
}
|
||||
|
||||
/// Insert or replace a typed value.
|
||||
pub fn set<T: Any + Send + Sync>(mut self, value: T) -> Self {
|
||||
self.map.insert(TypeId::of::<T>(), Arc::new(value));
|
||||
self
|
||||
}
|
||||
|
||||
/// Insert or replace a typed value (mutable reference version).
|
||||
pub fn set_mut<T: Any + Send + Sync>(&mut self, value: T) -> &mut Self {
|
||||
self.map.insert(TypeId::of::<T>(), Arc::new(value));
|
||||
self
|
||||
}
|
||||
|
||||
/// Insert a type-erased value by `TypeId`.
|
||||
///
|
||||
/// Used by `ServiceRegistry::inject_into` to merge pre-built bindings
|
||||
/// without knowing concrete types at compile time.
|
||||
pub fn set_raw(&mut self, type_id: TypeId, value: Arc<dyn Any + Send + Sync>) -> &mut Self {
|
||||
self.map.insert(type_id, value);
|
||||
self
|
||||
}
|
||||
|
||||
/// Freeze the builder into an immutable `Environment`.
|
||||
pub fn build(self) -> Environment {
|
||||
Environment {
|
||||
inner: Arc::new(self.map),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for EnvironmentBuilder {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Well-Known Environment Keys ─────────────────────────────────────────────
|
||||
|
||||
/// Milliseconds since runtime creation when this actor was spawned.
|
||||
/// Injected by StdExtension (opt-in at runtime level). Read via `ctx.env::<SpawnTimestamp>()`.
|
||||
/// Uses the same time base as `SystemInfo::uptime_ms`.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
|
||||
pub struct SpawnTimestamp(pub u64);
|
||||
|
||||
/// Logical name assigned via `spawn_named()`. Read via `ctx.env::<LogicalName>()`.
|
||||
/// None for unnamed actors.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
|
||||
pub struct LogicalName(pub String);
|
||||
|
||||
impl LogicalName {
|
||||
pub fn as_str(&self) -> &str {
|
||||
&self.0
|
||||
}
|
||||
}
|
||||
|
||||
/// A typed service binding stored in the environment.
|
||||
///
|
||||
/// `S` is a zero-sized marker type that identifies the service (e.g., `struct Datastore;`).
|
||||
/// Stored via `ServiceRegistry` and read via `ctx.resource::<S>()`.
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct ServiceBinding<S: 'static + Send + Sync> {
|
||||
pub addr: ActorAddress,
|
||||
_marker: std::marker::PhantomData<S>,
|
||||
}
|
||||
|
||||
impl<S: 'static + Send + Sync> ServiceBinding<S> {
|
||||
pub fn new(addr: ActorAddress) -> Self {
|
||||
Self {
|
||||
addr,
|
||||
_marker: std::marker::PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Capabilities ────────────────────────────────────────────────────────────
|
||||
|
||||
/// Per-actor capability set controlling what operations the actor can perform.
|
||||
///
|
||||
/// When present in an actor's [`Environment`], enforcement is active — the actor
|
||||
/// can only perform operations granted by the set. When absent, the actor is
|
||||
/// unrestricted (backward compatible). Inherits from parent to child via normal
|
||||
/// environment inheritance.
|
||||
///
|
||||
/// Built via fluent API: `CapabilitySet::new().with_send(addr).with_spawn()`.
|
||||
#[derive(Clone, Default)]
|
||||
pub struct CapabilitySet {
|
||||
send_any: HashSet<ActorAddress, crate::delivery::AddrBuildHasher>,
|
||||
send_typed: HashSet<(TypeId, ActorAddress)>,
|
||||
can_spawn: bool,
|
||||
service_types: HashSet<TypeId>,
|
||||
monitor_targets: HashSet<ActorAddress, crate::delivery::AddrBuildHasher>,
|
||||
}
|
||||
|
||||
impl CapabilitySet {
|
||||
pub fn new() -> Self { Self::default() }
|
||||
|
||||
// ── Builder methods (fluent) ──
|
||||
pub fn with_send(mut self, addr: ActorAddress) -> Self {
|
||||
self.send_any.insert(addr); self
|
||||
}
|
||||
pub fn with_send_typed<M: Message>(mut self, addr: ActorAddress) -> Self {
|
||||
self.send_typed.insert((TypeId::of::<M>(), addr)); self
|
||||
}
|
||||
pub fn with_spawn(mut self) -> Self {
|
||||
self.can_spawn = true; self
|
||||
}
|
||||
pub fn with_service<S: 'static + Send + Sync>(mut self) -> Self {
|
||||
self.service_types.insert(TypeId::of::<S>()); self
|
||||
}
|
||||
pub fn with_monitor(mut self, addr: ActorAddress) -> Self {
|
||||
self.monitor_targets.insert(addr); self
|
||||
}
|
||||
|
||||
// ── Mutable builder methods ──
|
||||
pub fn grant_send(&mut self, addr: ActorAddress) -> &mut Self {
|
||||
self.send_any.insert(addr); self
|
||||
}
|
||||
pub fn grant_send_typed<M: Message>(&mut self, addr: ActorAddress) -> &mut Self {
|
||||
self.send_typed.insert((TypeId::of::<M>(), addr)); self
|
||||
}
|
||||
|
||||
// ── Check methods ──
|
||||
pub fn check_send<M: Message>(&self, addr: ActorAddress) -> Result<(), crate::Error> {
|
||||
if self.send_any.contains(&addr) { return Ok(()); }
|
||||
if self.send_typed.contains(&(TypeId::of::<M>(), addr)) { return Ok(()); }
|
||||
Err(crate::Error::from("capability denied: send"))
|
||||
}
|
||||
pub fn check_send_addr(&self, addr: ActorAddress) -> Result<(), crate::Error> {
|
||||
if self.send_any.contains(&addr) { return Ok(()); }
|
||||
Err(crate::Error::from("capability denied: send"))
|
||||
}
|
||||
pub fn check_spawn(&self) -> Result<(), crate::Error> {
|
||||
if self.can_spawn { Ok(()) } else { Err(crate::Error::from("capability denied: spawn")) }
|
||||
}
|
||||
pub fn check_service<S: 'static + Send + Sync>(&self) -> Result<(), crate::Error> {
|
||||
if self.service_types.contains(&TypeId::of::<S>()) { Ok(()) }
|
||||
else { Err(crate::Error::from("capability denied: service")) }
|
||||
}
|
||||
pub fn check_monitor(&self, addr: ActorAddress) -> Result<(), crate::Error> {
|
||||
if self.monitor_targets.contains(&addr) { Ok(()) }
|
||||
else { Err(crate::Error::from("capability denied: monitor")) }
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for CapabilitySet {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("CapabilitySet")
|
||||
.field("send_any", &self.send_any.len())
|
||||
.field("send_typed", &self.send_typed.len())
|
||||
.field("can_spawn", &self.can_spawn)
|
||||
.field("services", &self.service_types.len())
|
||||
.field("monitors", &self.monitor_targets.len())
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
// ─── SpawnRequest ────────────────────────────────────────────────────────────
|
||||
|
||||
/// Bundled arguments for spawning an actor.
|
||||
///
|
||||
/// Replaces the spawn channel 3-tuple to stop tuple growth as new fields are added.
|
||||
pub struct SpawnRequest {
|
||||
pub addr: ActorAddress,
|
||||
pub actor: Box<dyn AnyActor>,
|
||||
pub parent: Option<ActorAddress>,
|
||||
pub env: Environment,
|
||||
}
|
||||
|
||||
/// The actor process as represented in the Runtime — thin wrapper around user state.
|
||||
pub struct Actor<A: ActorInterface> {
|
||||
inner: A,
|
||||
|
|
@ -181,23 +491,45 @@ pub enum StopReason {
|
|||
Normal,
|
||||
/// Actor panicked and could not be restarted.
|
||||
Panicked,
|
||||
/// Actor stopped with a typed exit value (via [`Ctx::stop_with`]).
|
||||
Completed,
|
||||
}
|
||||
|
||||
/// Death notification delivered as a normal message when a monitored actor dies.
|
||||
///
|
||||
/// Subscribe via [`Ctx::monitor`]. The `Down` message arrives in the watcher's
|
||||
/// regular `handle()` method — no special callback needed.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Down {
|
||||
/// Address of the dead actor.
|
||||
pub addr: ActorAddress,
|
||||
/// Why it died.
|
||||
pub reason: StopReason,
|
||||
/// Typed exit value if the actor called [`Ctx::stop_with`].
|
||||
pub exit_value: Option<ExitValue>,
|
||||
}
|
||||
|
||||
impl PartialEq for Down {
|
||||
fn eq(&self, other: &Self) -> bool {
|
||||
self.addr == other.addr && self.reason == other.reason
|
||||
}
|
||||
}
|
||||
|
||||
impl Eq for Down {}
|
||||
|
||||
/// Internal sentinel message for graceful actor stop.
|
||||
/// Not a `Message` — intercepted in `tick_all` before reaching `handle_any`.
|
||||
pub(crate) struct StopSignal;
|
||||
/// Not a `Message` (not Clone) — intercepted in `tick_all` / `deliver` before
|
||||
/// reaching `handle_any`. Public so extension crates can construct it for
|
||||
/// orphan cleanup, but users cannot send it via `ctx.send()`.
|
||||
pub struct StopSignal;
|
||||
|
||||
/// Like [`StopSignal`] but carries a typed exit value.
|
||||
/// Intercepted in `tick_all` / `deliver`.
|
||||
pub struct StopWithSignal(pub ExitValue);
|
||||
|
||||
/// Internal sentinel message for resuming a suspended actor.
|
||||
/// Intercepted in [`ActorPool::deliver`] — not delivered to user code.
|
||||
pub(crate) struct ResumeSignal;
|
||||
|
||||
/// Object-safe inner trait for sending type-erased messages.
|
||||
///
|
||||
|
|
@ -207,13 +539,21 @@ pub(crate) struct StopSignal;
|
|||
#[allow(private_interfaces)]
|
||||
pub trait ContextInner {
|
||||
fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> Result<(), Error>;
|
||||
fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>);
|
||||
fn spawn_any(&self, request: SpawnRequest);
|
||||
/// Request graceful stop for an actor. Takes effect after the current message.
|
||||
fn request_stop(&self, addr: ActorAddress);
|
||||
/// Request graceful stop with a typed exit value. Takes effect after the current message.
|
||||
fn request_stop_with(&self, addr: ActorAddress, value: ExitValue);
|
||||
/// Request suspension for an actor. Takes effect after the current message.
|
||||
fn request_suspend(&self, addr: ActorAddress);
|
||||
/// Request resumption for a suspended actor. Sends a [`ResumeSignal`].
|
||||
fn request_resume(&self, addr: ActorAddress);
|
||||
/// Post a request to the per-worker extension (e.g., timer scheduling).
|
||||
fn post_worker_request(&self, request: Box<dyn Any + Send>);
|
||||
/// Access the runtime extension (if installed).
|
||||
fn extension(&self) -> Option<&dyn crate::extension::RuntimeExtension>;
|
||||
/// Return system-level information (worker count, actor count, uptime).
|
||||
fn system_info(&self) -> SystemInfo;
|
||||
}
|
||||
|
||||
/// Actor syscall interface — passed to `ActorInterface::handle()`.
|
||||
|
|
@ -224,11 +564,32 @@ pub trait ContextInner {
|
|||
pub struct Ctx<'a> {
|
||||
inner: &'a dyn ContextInner,
|
||||
self_addr: ActorAddress,
|
||||
self_parent_addr: Option<ActorAddress>,
|
||||
self_env: Environment,
|
||||
self_messages_processed: u64,
|
||||
self_mailbox_depth: usize,
|
||||
self_msg_type_counts: Vec<(&'static str, u64)>,
|
||||
}
|
||||
|
||||
impl<'a> Ctx<'a> {
|
||||
pub(crate) fn new(inner: &'a dyn ContextInner, self_addr: ActorAddress) -> Self {
|
||||
Self { inner, self_addr }
|
||||
pub(crate) fn new(
|
||||
inner: &'a dyn ContextInner,
|
||||
self_addr: ActorAddress,
|
||||
self_parent_addr: Option<ActorAddress>,
|
||||
self_env: Environment,
|
||||
self_messages_processed: u64,
|
||||
self_mailbox_depth: usize,
|
||||
self_msg_type_counts: Vec<(&'static str, u64)>,
|
||||
) -> Self {
|
||||
Self {
|
||||
inner,
|
||||
self_addr,
|
||||
self_parent_addr,
|
||||
self_env,
|
||||
self_messages_processed,
|
||||
self_mailbox_depth,
|
||||
self_msg_type_counts,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn raw_inner(&self) -> &dyn ContextInner {
|
||||
|
|
@ -240,24 +601,92 @@ impl<'a> Ctx<'a> {
|
|||
self.self_addr
|
||||
}
|
||||
|
||||
/// Returns the address of the actor that spawned this one, or `None`
|
||||
/// if this actor was spawned externally via `Runtime::spawn`.
|
||||
pub fn parent(&self) -> Option<ActorAddress> {
|
||||
self.self_parent_addr
|
||||
}
|
||||
|
||||
/// Access the runtime extension (if installed).
|
||||
pub fn extension(&self) -> Option<&dyn crate::extension::RuntimeExtension> {
|
||||
self.inner.extension()
|
||||
}
|
||||
|
||||
/// Return system-level information (worker count, actor count, uptime).
|
||||
pub fn system_info(&self) -> SystemInfo {
|
||||
self.inner.system_info()
|
||||
}
|
||||
|
||||
/// Total messages this actor has successfully processed (before the current tick).
|
||||
pub fn messages_processed(&self) -> u64 {
|
||||
self.self_messages_processed
|
||||
}
|
||||
|
||||
/// Number of messages in this actor's mailbox at the start of the current tick.
|
||||
pub fn mailbox_depth(&self) -> usize {
|
||||
self.self_mailbox_depth
|
||||
}
|
||||
|
||||
/// Per-message-type counts for this actor, sorted descending by count.
|
||||
pub fn message_type_counts(&self) -> &[(&'static str, u64)] {
|
||||
&self.self_msg_type_counts
|
||||
}
|
||||
|
||||
/// Check if this actor has a capability set (i.e., is restricted).
|
||||
fn capabilities(&self) -> Option<&CapabilitySet> {
|
||||
self.self_env.get::<CapabilitySet>()
|
||||
}
|
||||
|
||||
/// Send a typed message to an actor address.
|
||||
pub fn send<M: Message>(&self, addr: ActorAddress, msg: M) -> Result<(), Error> {
|
||||
if let Some(caps) = self.capabilities() {
|
||||
if addr != self.self_addr {
|
||||
caps.check_send::<M>(addr)?;
|
||||
}
|
||||
}
|
||||
self.inner.send_any(addr, Box::new(msg))
|
||||
}
|
||||
|
||||
/// Read a typed value from this actor's environment.
|
||||
pub fn env<T: Any + Send + Sync>(&self) -> Option<&T> {
|
||||
self.self_env.get::<T>()
|
||||
}
|
||||
|
||||
/// Access this actor's full environment.
|
||||
pub fn environment(&self) -> &Environment {
|
||||
&self.self_env
|
||||
}
|
||||
|
||||
/// Spawn a new actor, returning its address.
|
||||
///
|
||||
/// The child inherits this actor's environment (Arc clone — zero allocation).
|
||||
pub fn spawn<A: ActorInterface>(&self, actor: A) -> Result<ActorAddress, Error> {
|
||||
if let Some(caps) = self.capabilities() {
|
||||
caps.check_spawn()?;
|
||||
}
|
||||
let addr = ActorAddress::new_random();
|
||||
let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor));
|
||||
self.inner.spawn_any(addr, boxed);
|
||||
self.inner.spawn_any(SpawnRequest {
|
||||
addr,
|
||||
actor: boxed,
|
||||
parent: Some(self.self_addr),
|
||||
env: self.self_env.clone(),
|
||||
});
|
||||
Ok(addr)
|
||||
}
|
||||
|
||||
/// Create a [`SpawnBuilder`] to spawn an actor with environment overrides.
|
||||
///
|
||||
/// Common case (`ctx.spawn(actor)`) is unchanged — this is for when you
|
||||
/// need to add or replace environment values for the child.
|
||||
pub fn spawn_builder<A: ActorInterface>(&self, actor: A) -> SpawnBuilder<'_, A> {
|
||||
SpawnBuilder {
|
||||
ctx: self,
|
||||
actor,
|
||||
env_builder: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Request graceful stop for this actor after the current message completes.
|
||||
///
|
||||
/// The actor's `on_stop()` hook is called and the actor is removed from the
|
||||
|
|
@ -272,7 +701,69 @@ impl<'a> Ctx<'a> {
|
|||
/// the stop signal, then its `on_stop()` hook is called and it is removed.
|
||||
/// Uses PoisonPill semantics — queued after existing messages.
|
||||
pub fn stop_actor(&self, addr: ActorAddress) -> Result<(), Error> {
|
||||
if let Some(caps) = self.capabilities() {
|
||||
caps.check_send_addr(addr)?;
|
||||
}
|
||||
self.inner.send_any(addr, Box::new(StopSignal))
|
||||
}
|
||||
|
||||
/// Stop this actor with a typed exit value.
|
||||
///
|
||||
/// Like [`stop_self`](Self::stop_self), but the value is delivered to
|
||||
/// monitors (in [`Down::exit_value`]) and watchers (in [`ActorExited::exit_value`]).
|
||||
/// The stop reason is [`StopReason::Completed`].
|
||||
pub fn stop_with<T: Any + Send + Sync + 'static>(&self, value: T) {
|
||||
self.inner.request_stop_with(self.self_addr, ExitValue::new(value));
|
||||
}
|
||||
|
||||
/// Suspend this actor. Messages continue to queue but are not processed
|
||||
/// until a supervisor (or self) calls resume.
|
||||
pub fn suspend_self(&self) {
|
||||
self.inner.request_suspend(self.self_addr);
|
||||
}
|
||||
}
|
||||
|
||||
// ─── SpawnBuilder ────────────────────────────────────────────────────────────
|
||||
|
||||
/// Builder for spawning an actor with environment overrides.
|
||||
///
|
||||
/// Created via [`Ctx::spawn_builder`]. Lazily clones the parent environment
|
||||
/// on the first `.env()` call to avoid allocation when no overrides are needed.
|
||||
pub struct SpawnBuilder<'a, A: ActorInterface> {
|
||||
ctx: &'a Ctx<'a>,
|
||||
actor: A,
|
||||
env_builder: Option<EnvironmentBuilder>,
|
||||
}
|
||||
|
||||
impl<'a, A: ActorInterface> SpawnBuilder<'a, A> {
|
||||
/// Add or replace a typed environment value for the child.
|
||||
///
|
||||
/// On the first call, lazily clones the parent's environment map.
|
||||
pub fn env<T: Any + Send + Sync>(mut self, value: T) -> Self {
|
||||
let builder = self.env_builder.get_or_insert_with(|| {
|
||||
EnvironmentBuilder::from_env(self.ctx.environment())
|
||||
});
|
||||
builder.set_mut(value);
|
||||
self
|
||||
}
|
||||
|
||||
/// Spawn the actor, returning its address.
|
||||
pub fn finish(self) -> Result<ActorAddress, Error> {
|
||||
if let Some(caps) = self.ctx.capabilities() {
|
||||
caps.check_spawn()?;
|
||||
}
|
||||
let addr = ActorAddress::new_random();
|
||||
let boxed: Box<dyn AnyActor> = Box::new(Actor::new(self.actor));
|
||||
let env = match self.env_builder {
|
||||
Some(builder) => builder.build(),
|
||||
None => self.ctx.self_env.clone(),
|
||||
};
|
||||
self.ctx.inner.spawn_any(SpawnRequest {
|
||||
addr,
|
||||
actor: boxed,
|
||||
parent: Some(self.ctx.self_addr),
|
||||
env,
|
||||
});
|
||||
Ok(addr)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
|
|||
use std::sync::{Arc, OnceLock, RwLock};
|
||||
use std::thread::Thread;
|
||||
|
||||
use crate::actor::{ActorAddress, AnyActor, Message};
|
||||
use crate::actor::{ActorAddress, Message, SpawnRequest};
|
||||
use crate::channel::Sender;
|
||||
use crate::config::RuntimeConfig;
|
||||
use crate::stats::WorkerStats;
|
||||
|
|
@ -235,7 +235,7 @@ impl InboxRegistry {
|
|||
pub(crate) struct TickContext<'a> {
|
||||
pub(crate) address_map: &'a AddressMap,
|
||||
pub(crate) transfer_txs: &'a [Sender<Envelope>],
|
||||
pub(crate) spawn_txs: &'a [Sender<(ActorAddress, Box<dyn AnyActor>)>],
|
||||
pub(crate) spawn_txs: &'a [Sender<SpawnRequest>],
|
||||
pub(crate) placement: &'a Placement,
|
||||
pub(crate) inbox_registry: &'a InboxRegistry,
|
||||
pub(crate) config: &'a RuntimeConfig,
|
||||
|
|
@ -243,6 +243,10 @@ pub(crate) struct TickContext<'a> {
|
|||
pub(crate) stats_hook: Option<&'a dyn crate::stats::StatsHook>,
|
||||
/// Thread handles for waking parked workers on cross-worker sends.
|
||||
pub(crate) worker_threads: &'a [OnceLock<Thread>],
|
||||
/// Per-worker stats for summing total_actors across workers.
|
||||
pub(crate) worker_stats: &'a [Arc<WorkerStats>],
|
||||
/// Runtime creation time for computing uptime_ms.
|
||||
pub(crate) created_at: crate::Instant,
|
||||
#[cfg(feature = "transport")]
|
||||
pub(crate) codec_registry: Option<&'a crate::transport::CodecRegistry>,
|
||||
#[cfg(feature = "transport")]
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
use std::any::Any;
|
||||
|
||||
use crate::actor::{ActorAddress, StopReason};
|
||||
use crate::actor::{ActorAddress, Environment, ExitValue, StopReason};
|
||||
|
||||
/// Extension hook for runtime lifecycle events.
|
||||
///
|
||||
|
|
@ -17,12 +17,23 @@ pub trait RuntimeExtension: Send + Sync {
|
|||
/// The core delivers these through normal routing (pending_local or transfer queue).
|
||||
fn on_actor_death(
|
||||
&self,
|
||||
dead: &[(ActorAddress, StopReason)],
|
||||
dead: &[(ActorAddress, StopReason, Option<ExitValue>)],
|
||||
) -> Vec<(ActorAddress, Box<dyn Any + Send>)>;
|
||||
|
||||
/// Clean up extension state for dead actors (names, groups, monitors).
|
||||
fn cleanup_dead(&self, dead: &[ActorAddress]);
|
||||
|
||||
/// Called for each newly spawned actor, before it enters the pool.
|
||||
/// Extensions can enrich the actor's environment (e.g., inject SpawnTimestamp).
|
||||
/// `child` is the address of the newly spawned actor.
|
||||
/// `parent` is the address of the spawning actor, or `None` for runtime-spawned actors.
|
||||
/// `uptime_ms` is milliseconds since runtime creation.
|
||||
/// Default: no-op (returns env unchanged).
|
||||
fn on_spawn(&self, child: ActorAddress, parent: Option<ActorAddress>, env: Environment, uptime_ms: u64) -> Environment {
|
||||
let _ = (child, parent, uptime_ms);
|
||||
env
|
||||
}
|
||||
|
||||
/// Downcast support for Ctx extension traits.
|
||||
fn as_any(&self) -> &dyn Any;
|
||||
|
||||
|
|
|
|||
|
|
@ -2,6 +2,9 @@ pub mod actor;
|
|||
pub mod extension;
|
||||
pub mod worker;
|
||||
|
||||
// Re-export well-known environment key types for convenient access.
|
||||
pub use actor::{SpawnTimestamp, LogicalName, ServiceBinding, ExitValue, CapabilitySet};
|
||||
|
||||
pub(crate) mod channel;
|
||||
pub(crate) mod error;
|
||||
pub use error::Error;
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ use std::thread::{self, JoinHandle};
|
|||
use std::thread::Thread;
|
||||
use crate::Instant;
|
||||
|
||||
use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message, StopSignal};
|
||||
use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Environment, ExitValue, Message, ResumeSignal, SpawnRequest, StopSignal, StopWithSignal, SystemInfo};
|
||||
use crate::channel::{Receiver, Sender};
|
||||
// Re-export config types so existing code using `runtime::RuntimeConfig` still works
|
||||
pub use crate::config::{BackoffPolicy, MailboxOverflow, RuntimeConfig};
|
||||
|
|
@ -102,7 +102,7 @@ pub struct Runtime {
|
|||
inbox_registry: Arc<InboxRegistry>,
|
||||
extension: Option<Arc<dyn RuntimeExtension>>,
|
||||
transfer_txs: Vec<Sender<Envelope>>,
|
||||
spawn_txs: Vec<Sender<(ActorAddress, Box<dyn AnyActor>)>>,
|
||||
spawn_txs: Vec<Sender<SpawnRequest>>,
|
||||
placement: Placement,
|
||||
is_running: AtomicBool,
|
||||
worker_stats: Vec<Arc<WorkerStats>>,
|
||||
|
|
@ -170,7 +170,7 @@ impl Runtime {
|
|||
transfer_txs.push(transfer_tx);
|
||||
|
||||
let spawn_rx =
|
||||
Receiver::<(ActorAddress, Box<dyn AnyActor>)>::new(config.max_actors);
|
||||
Receiver::<SpawnRequest>::new(config.max_actors);
|
||||
let spawn_tx = spawn_rx.new_sender();
|
||||
spawn_txs.push(spawn_tx);
|
||||
|
||||
|
|
@ -228,7 +228,26 @@ impl Runtime {
|
|||
self.address_map.insert(addr, worker_id);
|
||||
let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor));
|
||||
self.spawn_txs[worker_id.as_usize()]
|
||||
.send((addr, boxed));
|
||||
.send(SpawnRequest { addr, actor: boxed, parent: None, env: Environment::new() });
|
||||
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::info!(
|
||||
actor_addr = %addr,
|
||||
worker_id = worker_id.as_usize(),
|
||||
"actor.spawned"
|
||||
);
|
||||
|
||||
Ok(addr)
|
||||
}
|
||||
|
||||
/// Spawn an actor with a pre-built environment, returns its address.
|
||||
pub fn spawn_with_env<A: ActorInterface>(&self, actor: A, env: Environment) -> Result<ActorAddress, Error> {
|
||||
let addr = ActorAddress::new_random();
|
||||
let worker_id = self.placement.next_worker();
|
||||
self.address_map.insert(addr, worker_id);
|
||||
let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor));
|
||||
self.spawn_txs[worker_id.as_usize()]
|
||||
.send(SpawnRequest { addr, actor: boxed, parent: None, env });
|
||||
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::info!(
|
||||
|
|
@ -309,6 +328,8 @@ impl Runtime {
|
|||
extension: self.extension.as_deref(),
|
||||
stats_hook: self.stats_hook.as_deref(),
|
||||
worker_threads: &self.worker_threads,
|
||||
worker_stats: &self.worker_stats,
|
||||
created_at: self.created_at,
|
||||
#[cfg(feature = "transport")]
|
||||
codec_registry: self.codec_registry.as_deref(),
|
||||
#[cfg(feature = "transport")]
|
||||
|
|
@ -487,11 +508,11 @@ impl ContextInner for Runtime {
|
|||
}
|
||||
}
|
||||
|
||||
fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>) {
|
||||
fn spawn_any(&self, request: SpawnRequest) {
|
||||
let worker_id = self.placement.next_worker();
|
||||
self.address_map.insert(addr, worker_id);
|
||||
self.address_map.insert(request.addr, worker_id);
|
||||
self.spawn_txs[worker_id.as_usize()]
|
||||
.send((addr, actor));
|
||||
.send(request);
|
||||
notify_worker(&self.worker_threads, worker_id.as_usize());
|
||||
}
|
||||
|
||||
|
|
@ -504,6 +525,27 @@ impl ContextInner for Runtime {
|
|||
}
|
||||
}
|
||||
|
||||
fn request_stop_with(&self, addr: ActorAddress, value: ExitValue) {
|
||||
if let Some(wid) = self.address_map.lookup(&addr) {
|
||||
self.transfer_txs[wid.as_usize()]
|
||||
.send(Envelope::new(addr, Box::new(StopWithSignal(value))));
|
||||
notify_worker(&self.worker_threads, wid.as_usize());
|
||||
}
|
||||
}
|
||||
|
||||
fn request_suspend(&self, addr: ActorAddress) {
|
||||
// Outside worker context — not supported (suspend is per-actor, from handler)
|
||||
eprintln!("swactor: request_suspend called outside worker context for {addr} — ignored");
|
||||
}
|
||||
|
||||
fn request_resume(&self, addr: ActorAddress) {
|
||||
if let Some(wid) = self.address_map.lookup(&addr) {
|
||||
self.transfer_txs[wid.as_usize()]
|
||||
.send(Envelope::new(addr, Box::new(ResumeSignal)));
|
||||
notify_worker(&self.worker_threads, wid.as_usize());
|
||||
}
|
||||
}
|
||||
|
||||
fn post_worker_request(&self, _request: Box<dyn Any + Send>) {
|
||||
// Worker requests (e.g., timers) are per-worker; posting from outside
|
||||
// a worker context (e.g., rt.spawn() callback) is not supported.
|
||||
|
|
@ -513,4 +555,17 @@ impl ContextInner for Runtime {
|
|||
fn extension(&self) -> Option<&dyn RuntimeExtension> {
|
||||
self.extension.as_deref()
|
||||
}
|
||||
|
||||
fn system_info(&self) -> SystemInfo {
|
||||
let num_workers = self.config.num_threads.max(1);
|
||||
let total_actors: usize = self.worker_stats.iter()
|
||||
.map(|ws| ws.num_actors.load(Ordering::Relaxed))
|
||||
.sum();
|
||||
SystemInfo {
|
||||
worker_id: 0,
|
||||
num_workers,
|
||||
total_actors,
|
||||
uptime_ms: self.created_at.elapsed().as_millis() as u64,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
227
src/worker.rs
227
src/worker.rs
|
|
@ -6,7 +6,7 @@ use std::sync::Arc;
|
|||
use std::thread;
|
||||
use crate::Instant;
|
||||
|
||||
use crate::actor::{ActorAddress, AnyActor, ContextInner, Ctx, StopReason, StopSignal};
|
||||
use crate::actor::{ActorAddress, AnyActor, ContextInner, Ctx, Environment, ExitValue, ResumeSignal, SpawnRequest, StopReason, StopSignal, StopWithSignal, SystemInfo};
|
||||
use crate::channel::Receiver;
|
||||
use crate::config::MailboxOverflow;
|
||||
use crate::delivery::{AddrBuildHasher, AddrMap, Envelope, TickContext, WorkerId};
|
||||
|
|
@ -45,7 +45,7 @@ pub(crate) struct Worker {
|
|||
pub(crate) id: WorkerId,
|
||||
pub(crate) pool: ActorPool,
|
||||
transfer_rx: Receiver<Envelope>,
|
||||
spawn_rx: Receiver<(ActorAddress, Box<dyn AnyActor>)>,
|
||||
spawn_rx: Receiver<SpawnRequest>,
|
||||
stats: Arc<WorkerStats>,
|
||||
/// Reusable scratch buffer for building per-actor snapshots.
|
||||
snapshot_buf: Vec<ActorSnapshot>,
|
||||
|
|
@ -57,7 +57,7 @@ impl Worker {
|
|||
pub(crate) fn new(
|
||||
id: WorkerId,
|
||||
transfer_rx: Receiver<Envelope>,
|
||||
spawn_rx: Receiver<(ActorAddress, Box<dyn AnyActor>)>,
|
||||
spawn_rx: Receiver<SpawnRequest>,
|
||||
stats: Arc<WorkerStats>,
|
||||
default_mailbox_capacity: usize,
|
||||
default_overflow_policy: MailboxOverflow,
|
||||
|
|
@ -76,12 +76,15 @@ impl Worker {
|
|||
/// Run one iteration of the worker loop. Returns `true` if any work was done.
|
||||
/// Drain the spawn queue, inserting new actors into the pool.
|
||||
/// Used in phases 1 and 4 of tick_once.
|
||||
fn drain_spawns(&mut self) -> bool {
|
||||
fn drain_spawns(&mut self, tc: &TickContext) -> bool {
|
||||
let mut did_work = false;
|
||||
#[cfg(feature = "tracing")]
|
||||
let mut spawn_count: usize = 0;
|
||||
while let Some((addr, actor)) = self.spawn_rx.try_recv() {
|
||||
self.pool.insert(addr, actor);
|
||||
while let Some(mut req) = self.spawn_rx.try_recv() {
|
||||
if let Some(ext) = tc.extension {
|
||||
req.env = ext.on_spawn(req.addr, req.parent, req.env, tc.created_at.elapsed().as_millis() as u64);
|
||||
}
|
||||
self.pool.insert(req);
|
||||
#[cfg(feature = "tracing")]
|
||||
{ spawn_count += 1; }
|
||||
did_work = true;
|
||||
|
|
@ -98,6 +101,8 @@ impl Worker {
|
|||
let cleanup_pending: RefCell<Vec<(ActorAddress, Box<dyn Any + Send>)>> =
|
||||
RefCell::new(Vec::new());
|
||||
let cleanup_stops: RefCell<Vec<ActorAddress>> = RefCell::new(Vec::new());
|
||||
let cleanup_stop_withs: RefCell<Vec<(ActorAddress, ExitValue)>> = RefCell::new(Vec::new());
|
||||
let cleanup_suspends: RefCell<Vec<ActorAddress>> = RefCell::new(Vec::new());
|
||||
let cleanup_requests: RefCell<Vec<Box<dyn Any + Send>>> = RefCell::new(Vec::new());
|
||||
let dead = {
|
||||
let cleanup_ctx = WorkerContext {
|
||||
|
|
@ -105,6 +110,8 @@ impl Worker {
|
|||
tc,
|
||||
pending_local: &cleanup_pending,
|
||||
stop_requests: &cleanup_stops,
|
||||
stop_with_values: &cleanup_stop_withs,
|
||||
suspend_requests: &cleanup_suspends,
|
||||
worker_requests: &cleanup_requests,
|
||||
stats: &self.stats,
|
||||
};
|
||||
|
|
@ -113,13 +120,13 @@ impl Worker {
|
|||
|
||||
let had_dead = !dead.is_empty();
|
||||
if had_dead {
|
||||
for &(addr, _) in &dead {
|
||||
tc.address_map.remove(&addr);
|
||||
for (addr, _, _) in &dead {
|
||||
tc.address_map.remove(addr);
|
||||
}
|
||||
|
||||
if let Some(ext) = tc.extension {
|
||||
let notifications = ext.on_actor_death(&dead);
|
||||
let dead_addrs: Vec<_> = dead.iter().map(|(a, _)| *a).collect();
|
||||
let dead_addrs: Vec<_> = dead.iter().map(|(a, _, _)| *a).collect();
|
||||
ext.cleanup_dead(&dead_addrs);
|
||||
for (dest, msg) in notifications {
|
||||
route_to_pool_or_remote(&mut self.pool, tc, dest, msg);
|
||||
|
|
@ -136,7 +143,7 @@ impl Worker {
|
|||
|
||||
// GC per-worker extension state for dead actors
|
||||
if let Some(ext) = &mut self.worker_ext {
|
||||
let dead_addrs: Vec<ActorAddress> = dead.iter().map(|(a, _)| *a).collect();
|
||||
let dead_addrs: Vec<ActorAddress> = dead.iter().map(|(a, _, _)| *a).collect();
|
||||
ext.gc_dead(&dead_addrs);
|
||||
}
|
||||
|
||||
|
|
@ -151,7 +158,7 @@ impl Worker {
|
|||
let t0 = Instant::now();
|
||||
|
||||
// 1. Drain spawn queue → add actors to pool
|
||||
did_work |= self.drain_spawns();
|
||||
did_work |= self.drain_spawns(tc);
|
||||
let t1 = Instant::now();
|
||||
|
||||
// 2. Drain transfer queue → deliver envelopes to actors
|
||||
|
|
@ -176,6 +183,8 @@ impl Worker {
|
|||
let pending_local: RefCell<Vec<(ActorAddress, Box<dyn Any + Send>)>> =
|
||||
RefCell::new(Vec::new());
|
||||
let stop_requests: RefCell<Vec<ActorAddress>> = RefCell::new(Vec::new());
|
||||
let stop_with_values: RefCell<Vec<(ActorAddress, ExitValue)>> = RefCell::new(Vec::new());
|
||||
let suspend_requests: RefCell<Vec<ActorAddress>> = RefCell::new(Vec::new());
|
||||
let worker_requests: RefCell<Vec<Box<dyn Any + Send>>> = RefCell::new(Vec::new());
|
||||
|
||||
let processed;
|
||||
|
|
@ -185,10 +194,12 @@ impl Worker {
|
|||
tc,
|
||||
pending_local: &pending_local,
|
||||
stop_requests: &stop_requests,
|
||||
stop_with_values: &stop_with_values,
|
||||
suspend_requests: &suspend_requests,
|
||||
worker_requests: &worker_requests,
|
||||
stats: &self.stats,
|
||||
};
|
||||
processed = self.pool.tick_all(&worker_ctx, &self.stats, tc.config.actor_message_budget, &stop_requests);
|
||||
processed = self.pool.tick_all(&worker_ctx, &self.stats, tc.config.actor_message_budget, &stop_requests, &stop_with_values, &suspend_requests);
|
||||
if processed > 0 {
|
||||
did_work = true;
|
||||
}
|
||||
|
|
@ -206,7 +217,7 @@ impl Worker {
|
|||
|
||||
// 4. Drain spawn queue again — actors spawned during step 3
|
||||
// must be in the pool before pending_local delivery.
|
||||
did_work |= self.drain_spawns();
|
||||
did_work |= self.drain_spawns(tc);
|
||||
let t4 = Instant::now();
|
||||
|
||||
// 5. Drain pending_local buffer → deliver to local actors
|
||||
|
|
@ -316,6 +327,8 @@ struct WorkerContext<'a> {
|
|||
tc: &'a TickContext<'a>,
|
||||
pending_local: &'a RefCell<Vec<(ActorAddress, Box<dyn Any + Send>)>>,
|
||||
stop_requests: &'a RefCell<Vec<ActorAddress>>,
|
||||
stop_with_values: &'a RefCell<Vec<(ActorAddress, ExitValue)>>,
|
||||
suspend_requests: &'a RefCell<Vec<ActorAddress>>,
|
||||
worker_requests: &'a RefCell<Vec<Box<dyn Any + Send>>>,
|
||||
stats: &'a WorkerStats,
|
||||
}
|
||||
|
|
@ -341,11 +354,11 @@ impl ContextInner for WorkerContext<'_> {
|
|||
}
|
||||
}
|
||||
|
||||
fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>) {
|
||||
fn spawn_any(&self, request: SpawnRequest) {
|
||||
let worker_id = self.tc.placement.next_worker();
|
||||
self.tc.address_map.insert(addr, worker_id);
|
||||
self.tc.address_map.insert(request.addr, worker_id);
|
||||
self.tc.spawn_txs[worker_id.as_usize()]
|
||||
.send((addr, actor));
|
||||
.send(request);
|
||||
crate::runtime::notify_worker(self.tc.worker_threads, worker_id.as_usize());
|
||||
}
|
||||
|
||||
|
|
@ -353,6 +366,20 @@ impl ContextInner for WorkerContext<'_> {
|
|||
self.stop_requests.borrow_mut().push(addr);
|
||||
}
|
||||
|
||||
fn request_stop_with(&self, addr: ActorAddress, value: ExitValue) {
|
||||
self.stop_with_values.borrow_mut().push((addr, value));
|
||||
}
|
||||
|
||||
fn request_suspend(&self, addr: ActorAddress) {
|
||||
self.suspend_requests.borrow_mut().push(addr);
|
||||
}
|
||||
|
||||
fn request_resume(&self, addr: ActorAddress) {
|
||||
// Same-worker: buffer as pending_local ResumeSignal
|
||||
// Cross-worker: would go through transfer queue (handled by Runtime impl)
|
||||
self.pending_local.borrow_mut().push((addr, Box::new(ResumeSignal)));
|
||||
}
|
||||
|
||||
fn post_worker_request(&self, request: Box<dyn Any + Send>) {
|
||||
self.worker_requests.borrow_mut().push(request);
|
||||
}
|
||||
|
|
@ -360,6 +387,19 @@ impl ContextInner for WorkerContext<'_> {
|
|||
fn extension(&self) -> Option<&dyn crate::extension::RuntimeExtension> {
|
||||
self.tc.extension
|
||||
}
|
||||
|
||||
fn system_info(&self) -> SystemInfo {
|
||||
let num_workers = self.tc.config.num_threads.max(1);
|
||||
let total_actors: usize = self.tc.worker_stats.iter()
|
||||
.map(|ws| ws.num_actors.load(Ordering::Relaxed))
|
||||
.sum();
|
||||
SystemInfo {
|
||||
worker_id: self.worker_id.0,
|
||||
num_workers,
|
||||
total_actors,
|
||||
uptime_ms: self.tc.created_at.elapsed().as_millis() as u64,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct ActorSlot {
|
||||
|
|
@ -370,6 +410,8 @@ struct ActorSlot {
|
|||
stopping: bool,
|
||||
/// Whether on_start has been called for this actor.
|
||||
started: bool,
|
||||
/// Actor is suspended — messages queue but are not processed.
|
||||
suspended: bool,
|
||||
last_msg_type: Option<&'static str>,
|
||||
messages_processed: u64,
|
||||
/// Per-message-type counters (bounded to 32 entries).
|
||||
|
|
@ -377,6 +419,12 @@ struct ActorSlot {
|
|||
/// Per-actor mailbox capacity. 0 = unbounded.
|
||||
mailbox_capacity: usize,
|
||||
overflow_policy: MailboxOverflow,
|
||||
/// Address of the actor that spawned this one, or `None` for externally-spawned actors.
|
||||
parent_addr: Option<ActorAddress>,
|
||||
/// Inherited environment from parent (or empty for runtime-spawned actors).
|
||||
env: Environment,
|
||||
/// Typed exit value set by `ctx.stop_with()`.
|
||||
exit_value: Option<ExitValue>,
|
||||
}
|
||||
|
||||
/// Per-worker actor storage. Owns per-actor mailboxes.
|
||||
|
|
@ -398,20 +446,24 @@ impl ActorPool {
|
|||
}
|
||||
}
|
||||
|
||||
pub fn insert(&mut self, addr: ActorAddress, actor: Box<dyn AnyActor>) {
|
||||
pub fn insert(&mut self, req: SpawnRequest) {
|
||||
let cap = self.default_mailbox_capacity;
|
||||
let prealloc = if cap > 0 { cap.min(64) } else { 16 };
|
||||
self.actors.insert(addr, ActorSlot {
|
||||
self.actors.insert(req.addr, ActorSlot {
|
||||
mailbox: VecDeque::with_capacity(prealloc),
|
||||
actor,
|
||||
actor: req.actor,
|
||||
poisoned: false,
|
||||
stopping: false,
|
||||
started: false,
|
||||
suspended: false,
|
||||
last_msg_type: None,
|
||||
messages_processed: 0,
|
||||
msg_type_counts: HashMap::new(),
|
||||
mailbox_capacity: self.default_mailbox_capacity,
|
||||
overflow_policy: self.default_overflow_policy,
|
||||
parent_addr: req.parent,
|
||||
env: req.env,
|
||||
exit_value: None,
|
||||
});
|
||||
}
|
||||
|
||||
|
|
@ -419,6 +471,27 @@ impl ActorPool {
|
|||
/// Returns `true` if the actor exists (message handled or dropped; type check deferred to tick).
|
||||
pub fn deliver(&mut self, addr: &ActorAddress, msg: Box<dyn Any + Send>) -> bool {
|
||||
if let Some(slot) = self.actors.get_mut(addr) {
|
||||
// Intercept control signals for suspended actors: they skip tick_all
|
||||
// so we must handle resume/stop at delivery time.
|
||||
if slot.suspended {
|
||||
if msg.is::<ResumeSignal>() {
|
||||
slot.suspended = false;
|
||||
return true;
|
||||
}
|
||||
if msg.is::<StopSignal>() {
|
||||
slot.stopping = true;
|
||||
slot.mailbox.clear();
|
||||
return true;
|
||||
}
|
||||
if msg.is::<StopWithSignal>() {
|
||||
if let Ok(sig) = msg.downcast::<StopWithSignal>() {
|
||||
slot.exit_value = Some(sig.0);
|
||||
}
|
||||
slot.stopping = true;
|
||||
slot.mailbox.clear();
|
||||
return true;
|
||||
}
|
||||
}
|
||||
if slot.mailbox_capacity > 0 && slot.mailbox.len() >= slot.mailbox_capacity {
|
||||
match slot.overflow_policy {
|
||||
MailboxOverflow::DropNewest => {
|
||||
|
|
@ -453,6 +526,8 @@ impl ActorPool {
|
|||
stats: &WorkerStats,
|
||||
budget: usize,
|
||||
stop_requests: &RefCell<Vec<ActorAddress>>,
|
||||
stop_with_values: &RefCell<Vec<(ActorAddress, ExitValue)>>,
|
||||
suspend_requests: &RefCell<Vec<ActorAddress>>,
|
||||
) -> usize {
|
||||
let mut count = 0;
|
||||
for (&addr, slot) in self.actors.iter_mut() {
|
||||
|
|
@ -462,10 +537,22 @@ impl ActorPool {
|
|||
continue;
|
||||
}
|
||||
|
||||
// Skip suspended actors — messages keep queueing
|
||||
if slot.suspended {
|
||||
continue;
|
||||
}
|
||||
|
||||
#[cfg(feature = "tracing")]
|
||||
let _actor_span = tracing::trace_span!("actor.tick", actor_addr = %addr).entered();
|
||||
|
||||
let ctx = Ctx::new(inner, addr);
|
||||
// Snapshot self-stats before creating Ctx
|
||||
let snap_processed = slot.messages_processed;
|
||||
let snap_depth = slot.mailbox.len();
|
||||
let mut snap_type_counts: Vec<(&'static str, u64)> =
|
||||
slot.msg_type_counts.iter().map(|(&k, &v)| (k, v)).collect();
|
||||
snap_type_counts.sort_by(|a, b| b.1.cmp(&a.1));
|
||||
|
||||
let ctx = Ctx::new(inner, addr, slot.parent_addr, slot.env.clone(), snap_processed, snap_depth, snap_type_counts);
|
||||
|
||||
// Call on_start once, before first message
|
||||
if !slot.started {
|
||||
|
|
@ -482,7 +569,7 @@ impl ActorPool {
|
|||
slot.mailbox.clear();
|
||||
continue;
|
||||
}
|
||||
// Check if on_start requested stop
|
||||
// Check if on_start requested stop or stop_with
|
||||
{
|
||||
let stops = stop_requests.borrow();
|
||||
if !stops.is_empty() && stops.contains(&addr) {
|
||||
|
|
@ -490,6 +577,33 @@ impl ActorPool {
|
|||
slot.stopping = true;
|
||||
stats.stops.fetch_add(1, Ordering::Relaxed);
|
||||
slot.mailbox.clear();
|
||||
// Check for stop_with value
|
||||
let mut sws = stop_with_values.borrow_mut();
|
||||
if let Some(pos) = sws.iter().position(|(a, _)| *a == addr) {
|
||||
let (_, val) = sws.swap_remove(pos);
|
||||
slot.exit_value = Some(val);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
}
|
||||
// Check if on_start requested stop_with (without plain stop)
|
||||
{
|
||||
let mut sws = stop_with_values.borrow_mut();
|
||||
if let Some(pos) = sws.iter().position(|(a, _)| *a == addr) {
|
||||
let (_, val) = sws.swap_remove(pos);
|
||||
slot.exit_value = Some(val);
|
||||
slot.stopping = true;
|
||||
stats.stops.fetch_add(1, Ordering::Relaxed);
|
||||
slot.mailbox.clear();
|
||||
continue;
|
||||
}
|
||||
}
|
||||
// Check if on_start requested suspend
|
||||
{
|
||||
let suspends = suspend_requests.borrow();
|
||||
if !suspends.is_empty() && suspends.contains(&addr) {
|
||||
drop(suspends);
|
||||
slot.suspended = true;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
|
@ -507,6 +621,17 @@ impl ActorPool {
|
|||
break;
|
||||
}
|
||||
|
||||
// Intercept StopWithSignal (from external runtime)
|
||||
if msg.is::<StopWithSignal>() {
|
||||
if let Ok(sig) = msg.downcast::<StopWithSignal>() {
|
||||
slot.exit_value = Some(sig.0);
|
||||
}
|
||||
slot.stopping = true;
|
||||
stats.stops.fetch_add(1, Ordering::Relaxed);
|
||||
slot.mailbox.clear();
|
||||
break;
|
||||
}
|
||||
|
||||
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
|
||||
slot.actor.handle_any(&ctx, msg)
|
||||
}));
|
||||
|
|
@ -536,16 +661,37 @@ impl ActorPool {
|
|||
count += 1;
|
||||
actor_count += 1;
|
||||
|
||||
// Check if handler requested self-stop (via ctx.stop_self())
|
||||
// Check if handler requested self-stop or stop_with
|
||||
{
|
||||
let stops = stop_requests.borrow();
|
||||
if !stops.is_empty() && stops.contains(&addr) {
|
||||
drop(stops);
|
||||
let has_stop = !stops.is_empty() && stops.contains(&addr);
|
||||
drop(stops);
|
||||
|
||||
let mut sws = stop_with_values.borrow_mut();
|
||||
let sw_pos = sws.iter().position(|(a, _)| *a == addr);
|
||||
|
||||
if has_stop || sw_pos.is_some() {
|
||||
if let Some(pos) = sw_pos {
|
||||
let (_, val) = sws.swap_remove(pos);
|
||||
slot.exit_value = Some(val);
|
||||
}
|
||||
drop(sws);
|
||||
slot.stopping = true;
|
||||
stats.stops.fetch_add(1, Ordering::Relaxed);
|
||||
slot.mailbox.clear();
|
||||
break;
|
||||
}
|
||||
drop(sws);
|
||||
}
|
||||
|
||||
// Check if handler requested suspend
|
||||
{
|
||||
let suspends = suspend_requests.borrow();
|
||||
if !suspends.is_empty() && suspends.contains(&addr) {
|
||||
drop(suspends);
|
||||
slot.suspended = true;
|
||||
break; // stop processing this actor's messages this tick
|
||||
}
|
||||
}
|
||||
|
||||
if budget > 0 && actor_count >= budget {
|
||||
|
|
@ -568,30 +714,45 @@ impl ActorPool {
|
|||
self.actors.values().map(|slot| slot.mailbox.len()).sum()
|
||||
}
|
||||
|
||||
/// Remove poisoned and stopping actors, returning their addresses and stop reasons.
|
||||
/// Remove poisoned and stopping actors, returning their addresses, stop reasons,
|
||||
/// and optional exit values.
|
||||
/// Called after tick_all so the caller can clean up the address map.
|
||||
///
|
||||
/// For stopping actors: calls `on_stop()` before removal (wrapped in catch_unwind).
|
||||
/// For poisoned actors: `on_stop()` is NOT called (state may be corrupt).
|
||||
pub fn cleanup_dead(&mut self, inner: &dyn ContextInner) -> Vec<(ActorAddress, StopReason)> {
|
||||
let dead: Vec<(ActorAddress, StopReason)> = self
|
||||
pub fn cleanup_dead(&mut self, inner: &dyn ContextInner) -> Vec<(ActorAddress, StopReason, Option<ExitValue>)> {
|
||||
let dead_addrs: Vec<ActorAddress> = self
|
||||
.actors
|
||||
.iter()
|
||||
.filter(|(_, slot)| slot.poisoned || slot.stopping)
|
||||
.map(|(&addr, slot)| {
|
||||
let reason = if slot.poisoned { StopReason::Panicked } else { StopReason::Normal };
|
||||
(addr, reason)
|
||||
})
|
||||
.map(|(&addr, _)| addr)
|
||||
.collect();
|
||||
for &(addr, _) in &dead {
|
||||
let mut dead = Vec::with_capacity(dead_addrs.len());
|
||||
for addr in dead_addrs {
|
||||
if let Some(mut slot) = self.actors.remove(&addr) {
|
||||
let reason = if slot.poisoned {
|
||||
StopReason::Panicked
|
||||
} else if slot.exit_value.is_some() {
|
||||
StopReason::Completed
|
||||
} else {
|
||||
StopReason::Normal
|
||||
};
|
||||
// Call on_stop for gracefully stopping actors only
|
||||
if slot.stopping && !slot.poisoned {
|
||||
let ctx = Ctx::new(inner, addr);
|
||||
let mut type_counts: Vec<(&'static str, u64)> =
|
||||
slot.msg_type_counts.iter().map(|(&k, &v)| (k, v)).collect();
|
||||
type_counts.sort_by(|a, b| b.1.cmp(&a.1));
|
||||
let ctx = Ctx::new(
|
||||
inner, addr, slot.parent_addr, slot.env.clone(),
|
||||
slot.messages_processed,
|
||||
slot.mailbox.len(),
|
||||
type_counts,
|
||||
);
|
||||
let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
|
||||
slot.actor.on_stop(&ctx);
|
||||
}));
|
||||
}
|
||||
dead.push((addr, reason, slot.exit_value.take()));
|
||||
// slot is dropped here — actor resources freed
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -216,7 +216,7 @@ impl ActorInterface for MonitorWatcherActor {
|
|||
type Incoming = Down;
|
||||
type Response = ();
|
||||
fn on_start(&mut self, ctx: &Ctx) {
|
||||
self.mref = Some(ctx.monitor(self.watch_target));
|
||||
self.mref = Some(ctx.monitor(self.watch_target).unwrap());
|
||||
}
|
||||
fn handle(&mut self, ctx: &Ctx, msg: Down) {
|
||||
ctx.send(self.reply_to, msg).unwrap();
|
||||
|
|
@ -233,7 +233,7 @@ impl ActorInterface for DemonitorActor {
|
|||
type Incoming = Ping;
|
||||
type Response = ();
|
||||
fn on_start(&mut self, ctx: &Ctx) {
|
||||
self.mref = Some(ctx.monitor(self.watch_target));
|
||||
self.mref = Some(ctx.monitor(self.watch_target).unwrap());
|
||||
}
|
||||
fn handle(&mut self, ctx: &Ctx, _msg: Ping) {
|
||||
if let Some(mref) = self.mref.take() {
|
||||
|
|
@ -705,8 +705,8 @@ fn monitor_death_notification_contract() {
|
|||
type Incoming = Down;
|
||||
type Response = ();
|
||||
fn on_start(&mut self, ctx: &Ctx) {
|
||||
ctx.monitor(self.target);
|
||||
ctx.monitor(self.target);
|
||||
ctx.monitor(self.target).unwrap();
|
||||
ctx.monitor(self.target).unwrap();
|
||||
}
|
||||
fn handle(&mut self, ctx: &Ctx, msg: Down) {
|
||||
ctx.send(self.reply_to, msg).unwrap();
|
||||
|
|
@ -734,7 +734,7 @@ fn monitor_death_notification_contract() {
|
|||
type Incoming = Ping;
|
||||
type Response = ();
|
||||
fn on_start(&mut self, ctx: &Ctx) {
|
||||
ctx.monitor(self.target);
|
||||
ctx.monitor(self.target).unwrap();
|
||||
}
|
||||
fn handle(&mut self, ctx: &Ctx, _msg: Ping) {
|
||||
let _ = ctx.send(self.inbox, Count(self.downs.len()));
|
||||
|
|
@ -768,7 +768,7 @@ fn monitor_death_notification_contract() {
|
|||
type Incoming = Down;
|
||||
type Response = ();
|
||||
fn on_start(&mut self, ctx: &Ctx) {
|
||||
ctx.monitor(self.target);
|
||||
ctx.monitor(self.target).unwrap();
|
||||
}
|
||||
fn handle(&mut self, ctx: &Ctx, msg: Down) {
|
||||
let _ = ctx.send(self.inbox, msg);
|
||||
|
|
|
|||
|
|
@ -6,13 +6,16 @@ use std::sync::atomic::{AtomicUsize, Ordering};
|
|||
use std::sync::Arc;
|
||||
|
||||
pub use swactor::actor::{
|
||||
ActorAddress, ActorExited, ActorInterface, Down, ExitReason, MonitorRef, StopReason,
|
||||
ActorAddress, ActorExited, ActorInterface, CapabilitySet, Down, Environment, EnvironmentBuilder,
|
||||
ExitReason, ExitValue, LogicalName, MonitorRef, ServiceBinding, SpawnBuilder, SpawnTimestamp,
|
||||
StopReason,
|
||||
};
|
||||
pub use swactor::runtime::{Ctx, Inbox, MailboxOverflow, Runtime, RuntimeConfig};
|
||||
pub use swactor_std::{
|
||||
ChildSpec, CtxGroups, CtxMonitoring, CtxNaming, CtxTimers, CtxWatching, RestartPolicy, Router,
|
||||
RoutingStrategy, RuntimeGroups, RuntimeNaming, RuntimeWatching, StdExtension, Supervisor,
|
||||
SupervisorStrategy,
|
||||
ChildSpec, CtxCapabilities, CtxEnvironment, CtxGroups, CtxHandles, CtxLifecycle, CtxLineage,
|
||||
CtxMonitoring, CtxNaming, CtxResources, CtxSelfStats, CtxSystem, CtxTimers, CtxWatching,
|
||||
ResourceHandle, RestartPolicy, Router, RoutingStrategy, RuntimeGroups, RuntimeNaming,
|
||||
RuntimeResources, RuntimeWatching, StdExtension, Supervisor, SupervisorStrategy,
|
||||
};
|
||||
|
||||
// ── Messages ────────────────────────────────────────────────────────────────
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
Loading…
Reference in a new issue