diff --git a/src/worker.rs b/src/worker.rs index a64d62e..fa5313b 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -74,14 +74,10 @@ impl Worker { } /// Run one iteration of the worker loop. Returns `true` if any work was done. - pub(crate) fn tick_once(&mut self, tc: &TickContext) -> bool { - #[cfg(feature = "tracing")] - let _span = tracing::trace_span!("worker.tick", worker_id = self.id.0).entered(); - + /// 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 { let mut did_work = false; - let t0 = Instant::now(); - - // 1. Drain spawn queue → add actors to pool #[cfg(feature = "tracing")] let mut spawn_count: usize = 0; while let Some((addr, actor)) = self.spawn_rx.try_recv() { @@ -94,6 +90,68 @@ impl Worker { if spawn_count > 0 { tracing::debug!(worker_id = self.id.0, count = spawn_count, "worker.spawns_drained"); } + did_work + } + + /// Phase 7: clean up dead actors, deliver death notifications, GC extension state. + fn cleanup_dead_actors(&mut self, tc: &TickContext) -> bool { + let cleanup_pending: RefCell)>> = + RefCell::new(Vec::new()); + let cleanup_stops: RefCell> = RefCell::new(Vec::new()); + let cleanup_requests: RefCell>> = RefCell::new(Vec::new()); + let dead = { + let cleanup_ctx = WorkerContext { + worker_id: self.id, + tc, + pending_local: &cleanup_pending, + stop_requests: &cleanup_stops, + worker_requests: &cleanup_requests, + stats: &self.stats, + }; + self.pool.cleanup_dead(&cleanup_ctx) + }; + + let had_dead = !dead.is_empty(); + if had_dead { + 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(); + ext.cleanup_dead(&dead_addrs); + for (dest, msg) in notifications { + route_to_pool_or_remote(&mut self.pool, tc, dest, msg); + } + } + + self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed); + } + + // Deliver any messages sent during on_stop callbacks + for (addr, msg) in cleanup_pending.into_inner() { + self.pool.deliver(&addr, msg); + } + + // GC per-worker extension state for dead actors + if let Some(ext) = &mut self.worker_ext { + let dead_addrs: Vec = dead.iter().map(|(a, _)| *a).collect(); + ext.gc_dead(&dead_addrs); + } + + had_dead + } + + pub(crate) fn tick_once(&mut self, tc: &TickContext) -> bool { + #[cfg(feature = "tracing")] + let _span = tracing::trace_span!("worker.tick", worker_id = self.id.0).entered(); + + let mut did_work = false; + let t0 = Instant::now(); + + // 1. Drain spawn queue → add actors to pool + did_work |= self.drain_spawns(); let t1 = Instant::now(); // 2. Drain transfer queue → deliver envelopes to actors @@ -148,10 +206,7 @@ impl Worker { // 4. Drain spawn queue again — actors spawned during step 3 // must be in the pool before pending_local delivery. - while let Some((addr, actor)) = self.spawn_rx.try_recv() { - self.pool.insert(addr, actor); - did_work = true; - } + did_work |= self.drain_spawns(); let t4 = Instant::now(); // 5. Drain pending_local buffer → deliver to local actors @@ -217,56 +272,7 @@ impl Worker { } // 7. Clean up poisoned and stopping actors - // on_stop() may send messages, so provide a fresh pending_local buffer. - let cleanup_pending: RefCell)>> = - RefCell::new(Vec::new()); - let cleanup_stops: RefCell> = RefCell::new(Vec::new()); - let cleanup_requests: RefCell>> = RefCell::new(Vec::new()); - let dead = { - let cleanup_ctx = WorkerContext { - worker_id: self.id, - tc, - pending_local: &cleanup_pending, - stop_requests: &cleanup_stops, - worker_requests: &cleanup_requests, - stats: &self.stats, - }; - let dead = self.pool.cleanup_dead(&cleanup_ctx); - if !dead.is_empty() { - for &(addr, _) in &dead { - tc.address_map.remove(&addr); - } - - if let Some(ext) = tc.extension { - // Get death notifications (monitors) before cleaning up state - let notifications = ext.on_actor_death(&dead); - - // Clean up extension state (names, groups, dead watcher monitors) - let dead_addrs: Vec<_> = dead.iter().map(|(a, _)| *a).collect(); - ext.cleanup_dead(&dead_addrs); - - // Deliver death notifications through normal routing - for (dest, msg) in notifications { - route_to_pool_or_remote(&mut self.pool, tc, dest, msg); - } - } - - // Re-publish num_actors after cleanup so stats reflect removal - self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed); - did_work = true; - } - dead - }; - // Deliver any messages sent during on_stop callbacks - for (addr, msg) in cleanup_pending.into_inner() { - self.pool.deliver(&addr, msg); - } - - // GC per-worker extension state for dead actors (e.g., orphaned interval timers) - if let Some(ext) = &mut self.worker_ext { - let dead_addrs: Vec = dead.iter().map(|(a, _)| *a).collect(); - ext.gc_dead(&dead_addrs); - } + did_work |= self.cleanup_dead_actors(tc); did_work } @@ -490,7 +496,6 @@ impl ActorPool { slot.stopping = true; stats.stops.fetch_add(1, Ordering::Relaxed); slot.mailbox.clear(); - deaths.push((addr, ExitReason::Stopped)); #[cfg(feature = "tracing")] tracing::info!(actor_addr = %addr, "actor.stop_requested"); break;