zacheryasc/swactor
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: decompose tick_once into helper methods

Browse source 
Extract drain_spawns() (phases 1+4, dedup) and cleanup_dead_actors()
    (phase 7, 48 lines of complex death notification + GC logic).

    tick_once: 196 → 133 lines (−32%), max nesting depth: 5 → 3.

    Authored by Claude, lovingly guided by Zachery Aaron Shores-Chmielewski
This commit is contained in:
Claude 2026-02-13 20:34:53 +07:00
parent 5a8aeaf428
commit 41af749bb4
1 changed files with 67 additions and 62 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

129
src/worker.rs
View file

@ -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<Vec<(ActorAddress, Box<dyn Any + Send>)>> =
RefCell::new(Vec::new());
let cleanup_stops: 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 {
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<ActorAddress> = 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<Vec<(ActorAddress, Box<dyn Any + Send>)>> =
RefCell::new(Vec::new());
let cleanup_stops: 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 {
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<ActorAddress> = dead.iter().map(|(a, _)| *a).collect();
ext.gc_dead(&dead_addrs);
}
did_work |= self.cleanup_dead_actors(tc);
did_work
}
@ -496,7 +502,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;