diff --git a/crates/std/src/ctx_ext.rs b/crates/std/src/ctx_ext.rs index 4db2b15..cf856bd 100644 --- a/crates/std/src/ctx_ext.rs +++ b/crates/std/src/ctx_ext.rs @@ -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 diff --git a/crates/std/src/extension.rs b/crates/std/src/extension.rs index fa7d20c..a7a86f2 100644 --- a/crates/std/src/extension.rs +++ b/crates/std/src/extension.rs @@ -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)> { 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)); } + + // 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)); + } } + 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); } } diff --git a/crates/std/src/lib.rs b/crates/std/src/lib.rs index 3481957..4153dfb 100644 --- a/crates/std/src/lib.rs +++ b/crates/std/src/lib.rs @@ -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}; diff --git a/crates/std/src/runtime_ext.rs b/crates/std/src/runtime_ext.rs index f75b0d7..3cd72c6 100644 --- a/crates/std/src/runtime_ext.rs +++ b/crates/std/src/runtime_ext.rs @@ -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`, diff --git a/crates/std/src/watch_registry.rs b/crates/std/src/watch_registry.rs new file mode 100644 index 0000000..4ebaf51 --- /dev/null +++ b/crates/std/src/watch_registry.rs @@ -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, +} + +struct WatchState { + /// target → set of watchers awaiting death notification + watchers: HashMap>, + /// watcher → set of targets it's watching (reverse index for cleanup) + watching: HashMap>, +} + +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); + } + } + } + } + } +} diff --git a/src/actor.rs b/src/actor.rs index 64b620c..55d6a71 100644 --- a/src/actor.rs +++ b/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); - } } diff --git a/src/delivery.rs b/src/delivery.rs index b21bcc2..b1f3aeb 100644 --- a/src/delivery.rs +++ b/src/delivery.rs @@ -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], - pub(crate) watch_registry: Option<&'a Arc>>, #[cfg(feature = "transport")] pub(crate) codec_registry: Option<&'a crate::transport::CodecRegistry>, #[cfg(feature = "transport")] diff --git a/src/runtime.rs b/src/runtime.rs index a412f91..7af01b0 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -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>, stats_hook: Option>, - watch_registry: Arc>, /// Workers available for tick(). run() drains this and moves workers to threads. tick_workers: RefCell>, /// 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); - } } diff --git a/src/worker.rs b/src/worker.rs index 884d247..9f6fc09 100644 --- a/src/worker.rs +++ b/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>`. 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>, - /// watcher → set of targets it's watching (reverse index for cleanup) - watching: HashMap>, -} - -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> = 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 { @@ -678,7 +532,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. @@ -688,9 +542,8 @@ impl ActorPool { stats: &WorkerStats, budget: usize, stop_requests: &RefCell>, - ) -> (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 @@ -756,7 +609,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)) => { @@ -775,7 +627,6 @@ impl ActorPool { slot.stopping = true; stats.stops.fetch_add(1, Ordering::Relaxed); slot.mailbox.clear(); - deaths.push((addr, ExitReason::Stopped)); break; } } @@ -785,7 +636,7 @@ impl ActorPool { } } } - (count, deaths) + count } pub fn len(&self) -> usize { diff --git a/tests/watch_api.rs b/tests/watch_api.rs index 0ee45ce..27b75c9 100644 --- a/tests/watch_api.rs +++ b/tests/watch_api.rs @@ -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 {