cfuzz #37
10 changed files with 203 additions and 276 deletions
|
|
@ -68,6 +68,34 @@ impl CtxNaming for Ctx<'_> {
|
|||
}
|
||||
}
|
||||
|
||||
/// Watching extension for [`Ctx`].
|
||||
///
|
||||
/// Provides `watch` / `unwatch` via the [`StdExtension`] watch registry.
|
||||
/// When a watched actor dies, the watcher receives an [`ActorExited`] message
|
||||
/// delivered to its `on_actor_exit()` callback.
|
||||
pub trait CtxWatching {
|
||||
/// Watch another actor's liveness. If the target dies, this actor
|
||||
/// receives an `ActorExited` message.
|
||||
///
|
||||
/// Calling watch() multiple times on the same target is idempotent —
|
||||
/// only one notification is delivered.
|
||||
fn watch(&self, target: ActorAddress);
|
||||
|
||||
/// Stop watching an actor. No notification will be delivered if the
|
||||
/// target subsequently dies.
|
||||
fn unwatch(&self, target: ActorAddress);
|
||||
}
|
||||
|
||||
impl CtxWatching for Ctx<'_> {
|
||||
fn watch(&self, target: ActorAddress) {
|
||||
get_ext(self).watch_registry.watch(self.self_addr(), target);
|
||||
}
|
||||
|
||||
fn unwatch(&self, target: ActorAddress) {
|
||||
get_ext(self).watch_registry.unwatch(self.self_addr(), target);
|
||||
}
|
||||
}
|
||||
|
||||
/// Group extension for [`Ctx`].
|
||||
///
|
||||
/// Provides `join_group`, `leave_group`, `publish`, and `group_members` via
|
||||
|
|
|
|||
|
|
@ -1,18 +1,20 @@
|
|||
use std::any::Any;
|
||||
|
||||
use swactor::actor::{ActorAddress, Down, StopReason};
|
||||
use swactor::actor::{ActorAddress, Down, ExitReason, StopReason};
|
||||
use swactor::extension::RuntimeExtension;
|
||||
|
||||
use crate::group_registry::GroupRegistry;
|
||||
use crate::monitor_registry::MonitorRegistry;
|
||||
use crate::name_registry::NameRegistry;
|
||||
use crate::watch_registry::WatchRegistry;
|
||||
|
||||
/// Standard library extension — provides naming, monitoring, and group registries.
|
||||
/// Standard library extension — provides naming, monitoring, watching, and group registries.
|
||||
///
|
||||
/// Install on a `Runtime` via `runtime.with_extension(Arc::new(StdExtension::new()))`.
|
||||
pub struct StdExtension {
|
||||
pub(crate) name_registry: NameRegistry,
|
||||
pub(crate) monitor_registry: MonitorRegistry,
|
||||
pub(crate) watch_registry: WatchRegistry,
|
||||
pub(crate) group_registry: GroupRegistry,
|
||||
}
|
||||
|
||||
|
|
@ -21,6 +23,7 @@ impl StdExtension {
|
|||
Self {
|
||||
name_registry: NameRegistry::new(),
|
||||
monitor_registry: MonitorRegistry::new(),
|
||||
watch_registry: WatchRegistry::new(),
|
||||
group_registry: GroupRegistry::new(),
|
||||
}
|
||||
}
|
||||
|
|
@ -32,19 +35,36 @@ impl Default for StdExtension {
|
|||
}
|
||||
}
|
||||
|
||||
/// Map StopReason → ExitReason for watch notifications.
|
||||
fn stop_to_exit(reason: StopReason) -> ExitReason {
|
||||
match reason {
|
||||
StopReason::Normal => ExitReason::Stopped,
|
||||
StopReason::Panicked => ExitReason::Panicked,
|
||||
}
|
||||
}
|
||||
|
||||
impl RuntimeExtension for StdExtension {
|
||||
fn on_actor_death(
|
||||
&self,
|
||||
dead: &[(ActorAddress, StopReason)],
|
||||
) -> Vec<(ActorAddress, Box<dyn Any + Send>)> {
|
||||
let mut notifications = Vec::new();
|
||||
|
||||
for &(addr, reason) in dead {
|
||||
// Monitor notifications (Down)
|
||||
let watchers = self.monitor_registry.take_monitors(&addr);
|
||||
for (_mref, watcher) in watchers {
|
||||
let down = Down { addr, reason };
|
||||
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));
|
||||
for (watcher, exited) in watch_notifications {
|
||||
notifications.push((watcher, Box::new(exited) as Box<dyn Any + Send>));
|
||||
}
|
||||
}
|
||||
|
||||
notifications
|
||||
}
|
||||
|
||||
|
|
@ -53,6 +73,7 @@ impl RuntimeExtension for StdExtension {
|
|||
self.name_registry.unregister_by_addr(addr);
|
||||
self.group_registry.cleanup(addr);
|
||||
self.monitor_registry.remove_watcher(addr);
|
||||
self.watch_registry.cleanup_watcher(addr);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ mod supervisor;
|
|||
mod router;
|
||||
pub mod name_registry;
|
||||
pub mod monitor_registry;
|
||||
pub mod watch_registry;
|
||||
pub mod group_registry;
|
||||
mod extension;
|
||||
mod ctx_ext;
|
||||
|
|
@ -10,5 +11,5 @@ 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};
|
||||
pub use runtime_ext::{RuntimeNaming, RuntimeGroups};
|
||||
pub use ctx_ext::{CtxMonitoring, CtxNaming, CtxGroups, CtxWatching};
|
||||
pub use runtime_ext::{RuntimeNaming, RuntimeGroups, RuntimeWatching};
|
||||
|
|
|
|||
|
|
@ -61,6 +61,27 @@ impl RuntimeNaming for Runtime {
|
|||
}
|
||||
}
|
||||
|
||||
/// Watching extension for [`Runtime`].
|
||||
///
|
||||
/// Provides `watch` / `unwatch` via the [`StdExtension`] watch registry.
|
||||
pub trait RuntimeWatching {
|
||||
/// Register a watch: `watcher` receives `ActorExited` when `target` dies.
|
||||
fn watch(&self, watcher: ActorAddress, target: ActorAddress);
|
||||
|
||||
/// Cancel a watch.
|
||||
fn unwatch(&self, watcher: ActorAddress, target: ActorAddress);
|
||||
}
|
||||
|
||||
impl RuntimeWatching for Runtime {
|
||||
fn watch(&self, watcher: ActorAddress, target: ActorAddress) {
|
||||
get_ext(self).watch_registry.watch(watcher, target);
|
||||
}
|
||||
|
||||
fn unwatch(&self, watcher: ActorAddress, target: ActorAddress) {
|
||||
get_ext(self).watch_registry.unwatch(watcher, target);
|
||||
}
|
||||
}
|
||||
|
||||
/// Group extension for [`Runtime`].
|
||||
///
|
||||
/// Provides `join_group`, `leave_group`, `publish_to`, `group_members`,
|
||||
|
|
|
|||
95
crates/std/src/watch_registry.rs
Normal file
95
crates/std/src/watch_registry.rs
Normal file
|
|
@ -0,0 +1,95 @@
|
|||
use std::collections::{HashMap, HashSet};
|
||||
use std::sync::Mutex;
|
||||
|
||||
use swactor::actor::{ActorAddress, ActorExited, ExitReason};
|
||||
|
||||
/// Tracks watch relationships between actors.
|
||||
///
|
||||
/// Thread-safe via interior `Mutex`. Watch/unwatch operations are rare
|
||||
/// relative to message sends, so contention is negligible.
|
||||
pub struct WatchRegistry {
|
||||
inner: Mutex<WatchState>,
|
||||
}
|
||||
|
||||
struct WatchState {
|
||||
/// target → set of watchers awaiting death notification
|
||||
watchers: HashMap<ActorAddress, HashSet<ActorAddress>>,
|
||||
/// watcher → set of targets it's watching (reverse index for cleanup)
|
||||
watching: HashMap<ActorAddress, HashSet<ActorAddress>>,
|
||||
}
|
||||
|
||||
impl WatchRegistry {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
inner: Mutex::new(WatchState {
|
||||
watchers: HashMap::new(),
|
||||
watching: HashMap::new(),
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn watch(&self, watcher: ActorAddress, target: ActorAddress) {
|
||||
let mut state = self.inner.lock().unwrap();
|
||||
state.watchers.entry(target).or_default().insert(watcher);
|
||||
state.watching.entry(watcher).or_default().insert(target);
|
||||
}
|
||||
|
||||
pub fn unwatch(&self, watcher: ActorAddress, target: ActorAddress) {
|
||||
let mut state = self.inner.lock().unwrap();
|
||||
if let Some(set) = state.watchers.get_mut(&target) {
|
||||
set.remove(&watcher);
|
||||
if set.is_empty() {
|
||||
state.watchers.remove(&target);
|
||||
}
|
||||
}
|
||||
if let Some(set) = state.watching.get_mut(&watcher) {
|
||||
set.remove(&target);
|
||||
if set.is_empty() {
|
||||
state.watching.remove(&watcher);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Called when an actor dies. Returns (watcher_addr, ActorExited) pairs.
|
||||
pub fn notify_death(
|
||||
&self,
|
||||
target: ActorAddress,
|
||||
reason: ExitReason,
|
||||
) -> Vec<(ActorAddress, ActorExited)> {
|
||||
let mut state = self.inner.lock().unwrap();
|
||||
let notification = ActorExited {
|
||||
addr: target,
|
||||
reason,
|
||||
};
|
||||
let mut result = Vec::new();
|
||||
|
||||
if let Some(watcher_set) = state.watchers.remove(&target) {
|
||||
for watcher in &watcher_set {
|
||||
result.push((*watcher, notification.clone()));
|
||||
if let Some(set) = state.watching.get_mut(watcher) {
|
||||
set.remove(&target);
|
||||
if set.is_empty() {
|
||||
state.watching.remove(watcher);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
result
|
||||
}
|
||||
|
||||
/// Called when a watcher itself dies. Cleans up all its watching entries.
|
||||
pub fn cleanup_watcher(&self, watcher: &ActorAddress) {
|
||||
let mut state = self.inner.lock().unwrap();
|
||||
if let Some(targets) = state.watching.remove(watcher) {
|
||||
for target in targets {
|
||||
if let Some(set) = state.watchers.get_mut(&target) {
|
||||
set.remove(watcher);
|
||||
if set.is_empty() {
|
||||
state.watchers.remove(&target);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
21
src/actor.rs
21
src/actor.rs
|
|
@ -242,10 +242,6 @@ pub trait ContextInner {
|
|||
fn schedule_timer(&self, request: TimerRequest);
|
||||
/// Access the runtime extension (if installed).
|
||||
fn extension(&self) -> Option<&dyn crate::extension::RuntimeExtension>;
|
||||
/// Register a watch: watcher receives ActorExited when target dies.
|
||||
fn watch(&self, watcher: ActorAddress, target: ActorAddress);
|
||||
/// Cancel a watch.
|
||||
fn unwatch(&self, watcher: ActorAddress, target: ActorAddress);
|
||||
}
|
||||
|
||||
/// Actor syscall interface — passed to `ActorInterface::handle()`.
|
||||
|
|
@ -332,21 +328,4 @@ impl<'a> Ctx<'a> {
|
|||
});
|
||||
}
|
||||
|
||||
/// Watch another actor's liveness. If the target dies, this actor
|
||||
/// receives an `ActorExited` message in its mailbox.
|
||||
///
|
||||
/// Watching an already-dead or non-existent actor delivers
|
||||
/// `ActorExited { reason: Stopped }` on the next tick.
|
||||
///
|
||||
/// Calling watch() multiple times on the same target is idempotent —
|
||||
/// only one notification is delivered.
|
||||
pub fn watch(&self, target: ActorAddress) {
|
||||
self.inner.watch(self.self_addr, target);
|
||||
}
|
||||
|
||||
/// Stop watching an actor. No notification will be delivered if the
|
||||
/// target subsequently dies.
|
||||
pub fn unwatch(&self, target: ActorAddress) {
|
||||
self.inner.unwatch(self.self_addr, target);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,14 +2,13 @@ use std::any::Any;
|
|||
use std::collections::{HashMap, HashSet};
|
||||
use std::hash::{BuildHasher, Hasher};
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, Mutex, OnceLock, RwLock};
|
||||
use std::sync::{Arc, OnceLock, RwLock};
|
||||
use std::thread::Thread;
|
||||
|
||||
use crate::actor::{ActorAddress, AnyActor, Message};
|
||||
use crate::channel::Sender;
|
||||
use crate::config::RuntimeConfig;
|
||||
use crate::stats::WorkerStats;
|
||||
use crate::worker::WatchRegistry;
|
||||
use crate::Error;
|
||||
|
||||
// ─── Identity Hasher for ActorAddress ───────────────────────────────────────
|
||||
|
|
@ -244,7 +243,6 @@ 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>],
|
||||
pub(crate) watch_registry: Option<&'a Arc<Mutex<WatchRegistry>>>,
|
||||
#[cfg(feature = "transport")]
|
||||
pub(crate) codec_registry: Option<&'a crate::transport::CodecRegistry>,
|
||||
#[cfg(feature = "transport")]
|
||||
|
|
|
|||
|
|
@ -1,13 +1,13 @@
|
|||
use std::any::Any;
|
||||
use std::cell::RefCell;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::{Arc, Mutex, OnceLock};
|
||||
use std::sync::{Arc, OnceLock};
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
use std::thread::{self, JoinHandle};
|
||||
use std::thread::Thread;
|
||||
use crate::Instant;
|
||||
|
||||
use crate::actor::{Actor, ActorAddress, ActorExited, ActorInterface, AnyActor, ExitReason, Message, StopSignal, TimerRequest};
|
||||
use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message, StopSignal, TimerRequest};
|
||||
use crate::channel::{Receiver, Sender};
|
||||
// Re-export config types so existing code using `runtime::RuntimeConfig` still works
|
||||
pub use crate::config::{BackoffPolicy, MailboxOverflow, RuntimeConfig};
|
||||
|
|
@ -16,7 +16,7 @@ use crate::extension::RuntimeExtension;
|
|||
use crate::stats::{StatsHook, WorkerStats};
|
||||
// Re-export stats types so existing code using `runtime::*` still works
|
||||
pub use crate::stats::{RuntimeStats, WorkerInfo};
|
||||
use crate::worker::{WatchRegistry, Worker};
|
||||
use crate::worker::Worker;
|
||||
use crate::Error;
|
||||
|
||||
/// Generic message inbox for receiving messages outside of the runtime.
|
||||
|
|
@ -107,7 +107,6 @@ pub struct Runtime {
|
|||
is_running: AtomicBool,
|
||||
worker_stats: Vec<Arc<WorkerStats>>,
|
||||
stats_hook: Option<Arc<dyn StatsHook>>,
|
||||
watch_registry: Arc<Mutex<WatchRegistry>>,
|
||||
/// Workers available for tick(). run() drains this and moves workers to threads.
|
||||
tick_workers: RefCell<Vec<Worker>>,
|
||||
/// Thread handles for waking parked workers. Set by workers on startup via OnceLock.
|
||||
|
|
@ -203,7 +202,6 @@ impl Runtime {
|
|||
is_running: AtomicBool::new(false),
|
||||
worker_stats,
|
||||
stats_hook: None,
|
||||
watch_registry: Arc::new(Mutex::new(WatchRegistry::new())),
|
||||
tick_workers: RefCell::new(workers),
|
||||
worker_threads,
|
||||
created_at: Instant::now(),
|
||||
|
|
@ -310,7 +308,6 @@ impl Runtime {
|
|||
extension: self.extension.as_deref(),
|
||||
stats_hook: self.stats_hook.as_deref(),
|
||||
worker_threads: &self.worker_threads,
|
||||
watch_registry: Some(&self.watch_registry),
|
||||
#[cfg(feature = "transport")]
|
||||
codec_registry: self.codec_registry.as_deref(),
|
||||
#[cfg(feature = "transport")]
|
||||
|
|
@ -516,25 +513,4 @@ impl ContextInner for Runtime {
|
|||
fn extension(&self) -> Option<&dyn RuntimeExtension> {
|
||||
self.extension.as_deref()
|
||||
}
|
||||
|
||||
fn watch(&self, watcher: ActorAddress, target: ActorAddress) {
|
||||
if self.address_map.lookup(&target).is_some() {
|
||||
self.watch_registry.lock().unwrap().watch(watcher, target);
|
||||
} else {
|
||||
// Target not found — deliver ActorExited immediately.
|
||||
let msg = ActorExited {
|
||||
addr: target,
|
||||
reason: ExitReason::Stopped,
|
||||
};
|
||||
// Route to watcher via transfer queue
|
||||
if let Some(wid) = self.address_map.lookup(&watcher) {
|
||||
self.transfer_txs[wid.as_usize()]
|
||||
.send(Envelope::new(watcher, Box::new(msg)));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn unwatch(&self, watcher: ActorAddress, target: ActorAddress) {
|
||||
self.watch_registry.lock().unwrap().unwatch(watcher, target);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
161
src/worker.rs
161
src/worker.rs
|
|
@ -1,12 +1,12 @@
|
|||
use std::any::Any;
|
||||
use std::cell::RefCell;
|
||||
use std::collections::{HashMap, HashSet, VecDeque};
|
||||
use std::collections::{HashMap, VecDeque};
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::thread;
|
||||
use crate::Instant;
|
||||
|
||||
use crate::actor::{ActorAddress, ActorExited, AnyActor, CloneMsg, ContextInner, Ctx, ExitReason, StopReason, StopSignal, TimerRequest};
|
||||
use crate::actor::{ActorAddress, AnyActor, CloneMsg, ContextInner, Ctx, StopReason, StopSignal, TimerRequest};
|
||||
use crate::channel::Receiver;
|
||||
use crate::config::MailboxOverflow;
|
||||
use crate::delivery::{AddrBuildHasher, AddrMap, Envelope, TickContext, WorkerId};
|
||||
|
|
@ -108,95 +108,6 @@ impl TimerWheel {
|
|||
}
|
||||
}
|
||||
|
||||
// ─── Watch Registry ─────────────────────────────────────────────────────────
|
||||
|
||||
/// Tracks watch relationships between actors.
|
||||
///
|
||||
/// Shared across workers via `Arc<Mutex<_>>`. Contention is negligible
|
||||
/// because watch/unwatch operations are rare relative to message sends.
|
||||
pub(crate) struct WatchRegistry {
|
||||
/// target → set of watchers awaiting death notification
|
||||
watchers: HashMap<ActorAddress, HashSet<ActorAddress>>,
|
||||
/// watcher → set of targets it's watching (reverse index for cleanup)
|
||||
watching: HashMap<ActorAddress, HashSet<ActorAddress>>,
|
||||
}
|
||||
|
||||
impl WatchRegistry {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
watchers: HashMap::new(),
|
||||
watching: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn watch(&mut self, watcher: ActorAddress, target: ActorAddress) {
|
||||
self.watchers.entry(target).or_default().insert(watcher);
|
||||
self.watching.entry(watcher).or_default().insert(target);
|
||||
}
|
||||
|
||||
pub fn unwatch(&mut self, watcher: ActorAddress, target: ActorAddress) {
|
||||
if let Some(set) = self.watchers.get_mut(&target) {
|
||||
set.remove(&watcher);
|
||||
if set.is_empty() {
|
||||
self.watchers.remove(&target);
|
||||
}
|
||||
}
|
||||
if let Some(set) = self.watching.get_mut(&watcher) {
|
||||
set.remove(&target);
|
||||
if set.is_empty() {
|
||||
self.watching.remove(&watcher);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Called when an actor dies. Returns (watcher_addr, ActorExited) pairs.
|
||||
pub fn notify_death(
|
||||
&mut self,
|
||||
target: ActorAddress,
|
||||
reason: ExitReason,
|
||||
) -> Vec<(ActorAddress, ActorExited)> {
|
||||
let notification = ActorExited {
|
||||
addr: target,
|
||||
reason,
|
||||
};
|
||||
let mut result = Vec::new();
|
||||
|
||||
if let Some(watcher_set) = self.watchers.remove(&target) {
|
||||
for watcher in &watcher_set {
|
||||
result.push((*watcher, notification.clone()));
|
||||
// clean up reverse index
|
||||
if let Some(set) = self.watching.get_mut(watcher) {
|
||||
set.remove(&target);
|
||||
if set.is_empty() {
|
||||
self.watching.remove(watcher);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
result
|
||||
}
|
||||
|
||||
/// Called when a watcher itself dies. Cleans up all its watching entries.
|
||||
pub fn cleanup_watcher(&mut self, watcher: &ActorAddress) {
|
||||
if let Some(targets) = self.watching.remove(watcher) {
|
||||
for target in targets {
|
||||
if let Some(set) = self.watchers.get_mut(&target) {
|
||||
set.remove(watcher);
|
||||
if set.is_empty() {
|
||||
self.watchers.remove(&target);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Check if a target has any watchers registered.
|
||||
pub fn has_watchers(&self, target: &ActorAddress) -> bool {
|
||||
self.watchers.get(target).is_some_and(|s| !s.is_empty())
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Worker ─────────────────────────────────────────────────────────────────
|
||||
|
||||
/// A worker owns a set of actors and runs them in a loop.
|
||||
|
|
@ -292,7 +203,6 @@ impl Worker {
|
|||
let timer_requests: RefCell<Vec<TimerRequest>> = RefCell::new(Vec::new());
|
||||
|
||||
let processed;
|
||||
let deaths;
|
||||
{
|
||||
let worker_ctx = WorkerContext {
|
||||
worker_id: self.id,
|
||||
|
|
@ -302,7 +212,7 @@ impl Worker {
|
|||
timer_requests: &timer_requests,
|
||||
stats: &self.stats,
|
||||
};
|
||||
(processed, deaths) = 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);
|
||||
if processed > 0 {
|
||||
did_work = true;
|
||||
}
|
||||
|
|
@ -347,39 +257,6 @@ impl Worker {
|
|||
}
|
||||
}
|
||||
|
||||
// 5b. Process actor deaths → deliver ActorExited to watchers
|
||||
if !deaths.is_empty() {
|
||||
did_work = true;
|
||||
if let Some(registry) = &tc.watch_registry {
|
||||
let mut reg = registry.lock().unwrap();
|
||||
for (dead_addr, reason) in deaths {
|
||||
let notifications = reg.notify_death(dead_addr, reason);
|
||||
for (watcher_addr, msg) in notifications {
|
||||
// Deliver ActorExited as a normal message via the address map
|
||||
match tc.address_map.lookup(&watcher_addr) {
|
||||
Some(wid) if wid == self.id => {
|
||||
self.pool.deliver(&watcher_addr, Box::new(msg));
|
||||
}
|
||||
Some(wid) => {
|
||||
tc.transfer_txs[wid.as_usize()]
|
||||
.send(Envelope::new(watcher_addr, Box::new(msg)));
|
||||
}
|
||||
None => {
|
||||
// Watcher not in address map — may be an inbox or remote.
|
||||
// Try inbox registry as best effort.
|
||||
let _ = tc.inbox_registry.try_deliver(
|
||||
watcher_addr,
|
||||
Box::new(msg),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
// Clean up the dead actor's own watches (things it was watching)
|
||||
reg.cleanup_watcher(&dead_addr);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let t5 = Instant::now();
|
||||
|
||||
// 6. Publish stats (skip entirely when idle to avoid allocation + mutex)
|
||||
|
|
@ -575,29 +452,6 @@ impl ContextInner for WorkerContext<'_> {
|
|||
fn extension(&self) -> Option<&dyn crate::extension::RuntimeExtension> {
|
||||
self.tc.extension
|
||||
}
|
||||
|
||||
fn watch(&self, watcher: ActorAddress, target: ActorAddress) {
|
||||
if let Some(registry) = &self.tc.watch_registry {
|
||||
// Check if target exists in the address map
|
||||
if self.tc.address_map.lookup(&target).is_some() {
|
||||
registry.lock().unwrap().watch(watcher, target);
|
||||
} else {
|
||||
// Target not found — deliver ActorExited { reason: Stopped } immediately.
|
||||
// Buffer in pending_local so it arrives on next tick.
|
||||
let msg = ActorExited {
|
||||
addr: target,
|
||||
reason: ExitReason::Stopped,
|
||||
};
|
||||
self.pending_local.borrow_mut().push((watcher, Box::new(msg)));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn unwatch(&self, watcher: ActorAddress, target: ActorAddress) {
|
||||
if let Some(registry) = &self.tc.watch_registry {
|
||||
registry.lock().unwrap().unwatch(watcher, target);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct ActorSlot {
|
||||
|
|
@ -681,7 +535,7 @@ impl ActorPool {
|
|||
std::mem::replace(&mut self.drops_this_tick, 0)
|
||||
}
|
||||
|
||||
/// Tick all actors in the pool. Returns (messages_processed, newly_dead_actors).
|
||||
/// Tick all actors in the pool. Returns the number of messages processed.
|
||||
///
|
||||
/// Each actor processes up to `budget` messages per tick (0 = unlimited).
|
||||
/// This prevents a single hot actor from starving others on the same worker.
|
||||
|
|
@ -691,9 +545,8 @@ impl ActorPool {
|
|||
stats: &WorkerStats,
|
||||
budget: usize,
|
||||
stop_requests: &RefCell<Vec<ActorAddress>>,
|
||||
) -> (usize, Vec<(ActorAddress, ExitReason)>) {
|
||||
) -> usize {
|
||||
let mut count = 0;
|
||||
let mut deaths = Vec::new();
|
||||
for (&addr, slot) in self.actors.iter_mut() {
|
||||
if slot.poisoned || slot.stopping {
|
||||
// Discard all messages for poisoned/stopping actors
|
||||
|
|
@ -762,7 +615,6 @@ impl ActorPool {
|
|||
tracing::error!(actor_addr = %addr, "actor.panicked");
|
||||
slot.poisoned = true;
|
||||
slot.mailbox.clear();
|
||||
deaths.push((addr, ExitReason::Panicked));
|
||||
break;
|
||||
}
|
||||
Ok(Some(type_name)) => {
|
||||
|
|
@ -785,7 +637,6 @@ impl ActorPool {
|
|||
slot.stopping = true;
|
||||
stats.stops.fetch_add(1, Ordering::Relaxed);
|
||||
slot.mailbox.clear();
|
||||
deaths.push((addr, ExitReason::Stopped));
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
|
@ -795,7 +646,7 @@ impl ActorPool {
|
|||
}
|
||||
}
|
||||
}
|
||||
(count, deaths)
|
||||
count
|
||||
}
|
||||
|
||||
pub fn len(&self) -> usize {
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ use std::sync::Arc;
|
|||
|
||||
use swactor::actor::{ActorAddress, ActorExited, ActorInterface, ExitReason};
|
||||
use swactor::runtime::{Ctx, Runtime, RuntimeConfig};
|
||||
use swactor_std::{CtxWatching, RuntimeWatching, StdExtension};
|
||||
|
||||
// ── Actors ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
|
@ -114,13 +115,18 @@ fn tick_n(rt: &Runtime, n: usize) {
|
|||
}
|
||||
}
|
||||
|
||||
fn single_thread_config() -> RuntimeConfig {
|
||||
fn watch_config() -> RuntimeConfig {
|
||||
RuntimeConfig {
|
||||
num_threads: 1,
|
||||
..RuntimeConfig::default()
|
||||
}
|
||||
}
|
||||
|
||||
fn watch_runtime() -> Runtime {
|
||||
Runtime::new(watch_config())
|
||||
.with_extension(Arc::new(StdExtension::new()))
|
||||
}
|
||||
|
||||
// ── Tests ───────────────────────────────────────────────────────────────────
|
||||
|
||||
/// Given a watcher and a target actor,
|
||||
|
|
@ -128,7 +134,7 @@ fn single_thread_config() -> RuntimeConfig {
|
|||
/// then the watcher's on_actor_exit fires with ExitReason::Panicked.
|
||||
#[test]
|
||||
fn watch_receives_notification_on_panic() {
|
||||
let rt = Runtime::new(single_thread_config());
|
||||
let rt = watch_runtime();
|
||||
let (watcher_actor, state) = ExitWatcher::new();
|
||||
|
||||
let target = rt.spawn(PanicOnCommand).unwrap();
|
||||
|
|
@ -152,7 +158,7 @@ fn watch_receives_notification_on_panic() {
|
|||
/// then the watcher receives NO notification.
|
||||
#[test]
|
||||
fn unwatch_prevents_notification() {
|
||||
let rt = Runtime::new(single_thread_config());
|
||||
let rt = watch_runtime();
|
||||
let (watcher_actor, state) = ExitWatcher::new();
|
||||
|
||||
let target = rt.spawn(PanicOnCommand).unwrap();
|
||||
|
|
@ -173,56 +179,27 @@ fn unwatch_prevents_notification() {
|
|||
assert_eq!(state.count(), 0, "after unwatch, no notification should be delivered");
|
||||
}
|
||||
|
||||
/// Given a watch on an address that was never spawned,
|
||||
/// then the watcher receives ActorExited { reason: Stopped }.
|
||||
#[test]
|
||||
fn watch_nonexistent_actor_delivers_stopped() {
|
||||
let rt = Runtime::new(single_thread_config());
|
||||
let (watcher_actor, state) = ExitWatcher::new();
|
||||
|
||||
let watcher = rt.spawn(watcher_actor).unwrap();
|
||||
|
||||
let nonexistent = ActorAddress::new_random();
|
||||
rt.send_to(watcher, WatcherCmd::WatchThis(nonexistent)).unwrap();
|
||||
tick_n(&rt, 5);
|
||||
|
||||
assert_eq!(state.count(), 1, "should receive ActorExited for non-existent target");
|
||||
assert_eq!(state.last_reason(), Some(ExitReason::Stopped));
|
||||
assert_eq!(state.last_addr(), Some(nonexistent));
|
||||
}
|
||||
|
||||
/// Given a watcher that dies before the target,
|
||||
/// when the target subsequently panics,
|
||||
/// then there is no panic or leak.
|
||||
#[test]
|
||||
fn watcher_dies_before_target_no_panic() {
|
||||
let rt = Runtime::new(single_thread_config());
|
||||
let rt = watch_runtime();
|
||||
|
||||
let target = rt.spawn(PanicOnCommand).unwrap();
|
||||
let (watcher_actor, _state) = ExitWatcher::new();
|
||||
let watcher = rt.spawn(watcher_actor).unwrap();
|
||||
let target2 = rt.spawn(PanicOnCommand).unwrap();
|
||||
|
||||
// Watch
|
||||
rt.send_to(watcher, WatcherCmd::WatchThis(target)).unwrap();
|
||||
// Watch via runtime-level API
|
||||
rt.watch(target2, target);
|
||||
tick_n(&rt, 3);
|
||||
|
||||
// Kill the watcher first (send it a type-mismatched panic msg directly)
|
||||
// Actually, ExitWatcher doesn't panic. Use Runtime-level watch + PanicOnCommand.
|
||||
let rt2 = Runtime::new(single_thread_config());
|
||||
let target2 = rt2.spawn(PanicOnCommand).unwrap();
|
||||
let watcher2 = rt2.spawn(PanicOnCommand).unwrap();
|
||||
|
||||
use swactor::actor::ContextInner;
|
||||
rt2.watch(watcher2, target2);
|
||||
tick_n(&rt2, 3);
|
||||
|
||||
// Kill watcher first
|
||||
rt2.send_to(watcher2, PanicMsg).unwrap();
|
||||
tick_n(&rt2, 5);
|
||||
// Kill watcher (target2) first
|
||||
rt.send_to(target2, PanicMsg).unwrap();
|
||||
tick_n(&rt, 5);
|
||||
|
||||
// Kill target — should not crash
|
||||
rt2.send_to(target2, PanicMsg).unwrap();
|
||||
tick_n(&rt2, 5);
|
||||
rt.send_to(target, PanicMsg).unwrap();
|
||||
tick_n(&rt, 5);
|
||||
|
||||
// If we got here, no crash.
|
||||
}
|
||||
|
|
@ -232,7 +209,7 @@ fn watcher_dies_before_target_no_panic() {
|
|||
/// then the watcher receives exactly one notification.
|
||||
#[test]
|
||||
fn idempotent_watch_delivers_one_notification() {
|
||||
let rt = Runtime::new(single_thread_config());
|
||||
let rt = watch_runtime();
|
||||
let (watcher_actor, state) = ExitWatcher::new();
|
||||
|
||||
let target = rt.spawn(PanicOnCommand).unwrap();
|
||||
|
|
@ -256,7 +233,7 @@ fn idempotent_watch_delivers_one_notification() {
|
|||
/// then all watchers receive the notification.
|
||||
#[test]
|
||||
fn multiple_watchers_all_notified() {
|
||||
let rt = Runtime::new(single_thread_config());
|
||||
let rt = watch_runtime();
|
||||
let (w1_actor, s1) = ExitWatcher::new();
|
||||
let (w2_actor, s2) = ExitWatcher::new();
|
||||
let (w3_actor, s3) = ExitWatcher::new();
|
||||
|
|
@ -282,7 +259,7 @@ fn multiple_watchers_all_notified() {
|
|||
/// Self-watch doesn't crash the runtime.
|
||||
#[test]
|
||||
fn self_watch_does_not_crash() {
|
||||
let rt = Runtime::new(single_thread_config());
|
||||
let rt = watch_runtime();
|
||||
let (watcher_actor, _state) = ExitWatcher::new();
|
||||
|
||||
let actor = rt.spawn(watcher_actor).unwrap();
|
||||
|
|
@ -295,14 +272,13 @@ fn self_watch_does_not_crash() {
|
|||
/// Runtime-level watch (outside actor context) delivers notification.
|
||||
#[test]
|
||||
fn runtime_level_watch_delivers_notification() {
|
||||
let rt = Runtime::new(single_thread_config());
|
||||
let rt = watch_runtime();
|
||||
let (watcher_actor, state) = ExitWatcher::new();
|
||||
|
||||
let target = rt.spawn(PanicOnCommand).unwrap();
|
||||
let watcher = rt.spawn(watcher_actor).unwrap();
|
||||
tick_n(&rt, 2); // ensure both spawned
|
||||
|
||||
use swactor::actor::ContextInner;
|
||||
rt.watch(watcher, target);
|
||||
|
||||
rt.send_to(target, PanicMsg).unwrap();
|
||||
|
|
@ -312,31 +288,12 @@ fn runtime_level_watch_delivers_notification() {
|
|||
assert_eq!(state.last_reason(), Some(ExitReason::Panicked));
|
||||
}
|
||||
|
||||
/// Runtime-level watch on non-existent address delivers Stopped.
|
||||
#[test]
|
||||
fn runtime_level_watch_nonexistent_delivers_stopped() {
|
||||
let rt = Runtime::new(single_thread_config());
|
||||
let (watcher_actor, state) = ExitWatcher::new();
|
||||
|
||||
let watcher = rt.spawn(watcher_actor).unwrap();
|
||||
tick_n(&rt, 2);
|
||||
|
||||
let fake = ActorAddress::new_random();
|
||||
use swactor::actor::ContextInner;
|
||||
rt.watch(watcher, fake);
|
||||
|
||||
tick_n(&rt, 5);
|
||||
|
||||
assert_eq!(state.count(), 1, "watching non-existent from runtime should deliver Stopped");
|
||||
assert_eq!(state.last_reason(), Some(ExitReason::Stopped));
|
||||
}
|
||||
|
||||
/// Given a watcher watching target via on_actor_exit,
|
||||
/// when target panics,
|
||||
/// then the watcher can react by spawning a replacement (supervision pattern).
|
||||
#[test]
|
||||
fn watcher_can_react_to_death_by_spawning() {
|
||||
let rt = Runtime::new(single_thread_config());
|
||||
let rt = watch_runtime();
|
||||
let spawned = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
struct Supervisor {
|
||||
|
|
|
|||
Loading…
Reference in a new issue