(WIP) agent-fuzz-harness #31
4 changed files with 127 additions and 17 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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<ActorAddress> {
|
||||
let dead: Vec<ActorAddress> = 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<ActorSnapshot>) {
|
||||
out.clear();
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
Loading…
Reference in a new issue