diff --git a/CLAUDE/notes/progress.md b/CLAUDE/notes/progress.md index 819db08..412e9bd 100644 --- a/CLAUDE/notes/progress.md +++ b/CLAUDE/notes/progress.md @@ -2,7 +2,7 @@ ## Current Stage: Phase 1 — Research + First Improvement Cycle -### Status: Cycle 7 COMPLETE +### Status: Cycle 8 COMPLETE ## Plan Overview 1. **Phase 0**: Codebase audit — understand current swactor architecture, existing tests, benchmarks ✅ @@ -89,6 +89,23 @@ - **Benchmarks**: 1 new group — `placement/spawn_under_load` (2t, 4t) - **Result**: 60 tests pass, all workspace compiles +### Cycle 8: Dead Actor Cleanup (Memory Leak Fix) +- **Research**: Audited remaining improvement gaps, found ActorPool and AddressMap both leak + permanently poisoned actors. Known class of bug in Akka (#22990), CAF (#420). +- **Implementation**: Automatic cleanup of poisoned actors after tick_all + - `AddressMap::remove()` added to delivery.rs + - `ActorPool::cleanup_dead()` collects and removes poisoned actors, returns their addresses + - Phase 7 in tick_once: cleanup_dead → remove from address_map → update num_actors stat + - Re-publish num_actors after cleanup so stats immediately reflect removal +- **Behavior change**: Sends to poisoned actors now return Err (address not found) instead of + silently discarding. This is better — callers learn the actor is gone. +- **Tests**: 2 new tests + 2 existing tests updated + - `dead_actor_cleaned_up_from_stats_and_address_map` — good actor persists, bad actor removed + - `bulk_dead_actor_cleanup` — 20 panicked actors all cleaned up + - Updated `send_to_poisoned_actor_is_a_silent_black_hole` — now asserts send returns Err + - Updated `poisoned_actor_messages_not_counted_as_processed` — sends fail to cleaned-up actor +- **Result**: 70 tests pass, all workspace compiles + ### Cycle 7: Actor Recovery (Factory Restart) - **Research**: Deep analysis of supervision/recovery across Erlang (supervision trees, restart intensity), Akka (Resume/Restart/Stop/Escalate), Kameo (on_panic hook), Actix (Supervised trait), Ractor (SupervisionEvent) @@ -140,8 +157,9 @@ - [x] **Cycle 5: Work stealing research + load-aware placement** ✅ - [x] **Cycle 6: Mailbox backpressure** ✅ - [x] **Cycle 7: Actor recovery (factory restart)** ✅ -- [ ] **Cycle 8: Next improvement** - - Candidates: arena-allocated ActorPool, per-actor mailbox config, tracing integration +- [x] **Cycle 8: Dead actor cleanup** ✅ +- [ ] **Cycle 9: Next improvement** + - Candidates: lifecycle hooks, graceful stop, arena-allocated ActorPool - LIFO slot rejected (0-5% benefit for typical workloads, not worth complexity) ## Open Questions diff --git a/src/delivery.rs b/src/delivery.rs index 931d6bc..0d08ef8 100644 --- a/src/delivery.rs +++ b/src/delivery.rs @@ -44,6 +44,11 @@ impl AddressMap { self.inner.read().unwrap().get(addr).copied() } + /// Remove an actor address from the map (e.g., after permanent poisoning). + pub fn remove(&self, addr: &ActorAddress) { + self.inner.write().unwrap().remove(addr); + } + /// Returns a snapshot of all (address, worker) pairs. pub fn snapshot(&self) -> Vec<(ActorAddress, WorkerId)> { self.inner diff --git a/src/worker.rs b/src/worker.rs index 73923be..486e382 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -165,6 +165,17 @@ impl Worker { ); } + // 7. Clean up permanently poisoned actors + let dead = self.pool.cleanup_dead(); + if !dead.is_empty() { + for addr in &dead { + tc.address_map.remove(addr); + } + // Re-publish num_actors after cleanup so stats reflect removal + self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed); + did_work = true; + } + did_work } @@ -374,6 +385,21 @@ impl ActorPool { self.actors.values().map(|slot| slot.mailbox.len()).sum() } + /// Remove permanently poisoned actors and return their addresses. + /// Called after tick_all so the caller can clean up the address map. + pub fn cleanup_dead(&mut self) -> Vec { + let dead: Vec = self + .actors + .iter() + .filter(|(_, slot)| slot.poisoned) + .map(|(&addr, _)| addr) + .collect(); + for addr in &dead { + self.actors.remove(addr); + } + dead + } + /// Fill `out` with per-actor snapshots, reusing the existing allocation. pub fn mailbox_depths_into(&self, out: &mut Vec) { out.clear(); diff --git a/tests/runtime_api.rs b/tests/runtime_api.rs index 086329b..5c0e086 100644 --- a/tests/runtime_api.rs +++ b/tests/runtime_api.rs @@ -930,22 +930,19 @@ fn send_to_poisoned_actor_is_a_silent_black_hole() { rt.tick(); } - // When I send more messages to it + // When I send more messages to it (after cleanup, address is removed) let result = rt.send_to(panic_addr, PanicMsg); - // Then send_to succeeds (address is still in address_map) + // Then send_to returns an error (actor has been cleaned up and removed) assert!( - result.is_ok(), - "send_to poisoned actor should succeed from sender's POV" + result.is_err(), + "send_to cleaned-up actor should return error" ); - // And ticking doesn't produce new panics — messages are discarded in tick_all - for _ in 0..10 { - rt.tick(); - } + // And the original panic was recorded let s = rt.stats(); let panics: u64 = s.workers.iter().map(|w| w.panics).sum(); - assert_eq!(panics, 1, "poisoned actor should not produce new panics"); + assert_eq!(panics, 1, "poisoned actor should have recorded one panic"); } #[test] @@ -1112,7 +1109,7 @@ fn multiple_inbox_types_coexist() { #[test] fn poisoned_actor_messages_not_counted_as_processed() { - // Given a poisoned actor that then receives 10 more messages + // Given a poisoned actor that has been cleaned up let rt = Runtime::new(RuntimeConfig::default()); let panic_addr = rt.spawn(PanicActor).unwrap(); rt.send_to(panic_addr, PanicMsg).unwrap(); @@ -1122,9 +1119,13 @@ fn poisoned_actor_messages_not_counted_as_processed() { let s1 = rt.stats(); let processed_before: u64 = s1.workers.iter().map(|w| w.messages_processed).sum(); - // When I send 10 messages to the poisoned actor and tick + // When I try to send 10 messages to the cleaned-up actor + // (sends will fail because actor is removed from address map) + let mut send_failures = 0; for _ in 0..10 { - rt.send_to(panic_addr, PanicMsg).unwrap(); + if rt.send_to(panic_addr, PanicMsg).is_err() { + send_failures += 1; + } } for _ in 0..20 { rt.tick(); @@ -1132,10 +1133,11 @@ fn poisoned_actor_messages_not_counted_as_processed() { let s2 = rt.stats(); let processed_after: u64 = s2.workers.iter().map(|w| w.messages_processed).sum(); - // Then the 10 discarded messages should NOT increase the processed count + // Then sends fail (actor cleaned up) and processed count unchanged + assert_eq!(send_failures, 10, "all sends should fail to cleaned-up actor"); assert_eq!( processed_before, processed_after, - "messages discarded by poisoned actors should not be counted as processed" + "no additional messages should be processed after cleanup" ); } @@ -1994,6 +1996,65 @@ fn bounded_mailbox_refills_after_processing() { assert_eq!(total_drops, 0, "no drops when mailbox drains between batches"); } +// ── Dead Actor Cleanup Tests ─────────────────────────────────────────────── + +/// Given an actor that panics and is poisoned, +/// when ticks continue, +/// then the actor is removed from stats and sends to its address fail. +#[test] +fn dead_actor_cleaned_up_from_stats_and_address_map() { + let rt = Runtime::new(RuntimeConfig::default()); + + let good = rt.spawn(PingPongActor).unwrap(); + let bad = rt.spawn(PanicActor).unwrap(); + + // Trigger panic + let _ = rt.send_to(bad, PanicMsg); + for _ in 0..5 { rt.tick(); } + + let stats = rt.stats(); + // Good actor still present, bad actor cleaned up + assert_eq!(stats.workers[0].num_actors, 1, "only the healthy actor should remain"); + assert!( + stats.actors.iter().any(|(a, _)| *a == good), + "good actor should be in address map" + ); + assert!( + !stats.actors.iter().any(|(a, _)| *a == bad), + "poisoned actor should be removed from address map" + ); + + // Sends to cleaned-up actor fail + let result = rt.send_to(bad, PanicMsg); + assert!(result.is_err(), "send to cleaned-up actor should fail"); +} + +/// Given many actors that all panic, +/// when ticks proceed, +/// then all are cleaned up and stats reflect zero actors. +#[test] +fn bulk_dead_actor_cleanup() { + let rt = Runtime::new(RuntimeConfig::default()); + + let mut addrs = Vec::new(); + for _ in 0..20 { + addrs.push(rt.spawn(PanicActor).unwrap()); + } + + // Trigger all panics + for &addr in &addrs { + let _ = rt.send_to(addr, PanicMsg); + } + for _ in 0..10 { rt.tick(); } + + let stats = rt.stats(); + assert_eq!(stats.workers[0].num_actors, 0, "all poisoned actors should be cleaned up"); + assert_eq!( + stats.actors.len(), 0, + "address map should be empty after all actors poisoned" + ); +} + // ── Actor Recovery Helpers ────────────────────────────────────────────────── /// Handles Forward messages, replies Done(value * 2), panics on the panic_at-th message.