(WIP) agent-fuzz-harness #31
3 changed files with 161 additions and 12 deletions
|
|
@ -2,7 +2,7 @@
|
|||
|
||||
## Current Stage: Phase 1 — Research + First Improvement Cycle
|
||||
|
||||
### Status: Cycle 3 COMPLETE
|
||||
### Status: Cycle 4 COMPLETE
|
||||
|
||||
## Plan Overview
|
||||
1. **Phase 0**: Codebase audit — understand current swactor architecture, existing tests, benchmarks ✅
|
||||
|
|
@ -57,23 +57,26 @@
|
|||
- **Tests**: 1 new test (`mt_parked_worker_wakes_on_send`)
|
||||
- **Result**: 52 tests pass (51 + 1 new), all workspace compiles
|
||||
|
||||
### Cycle 4: Shutdown Fix + Bug-Inspired Tests
|
||||
- **Shutdown improvement**: `shutdown()` now unparks all workers for immediate exit
|
||||
- Previously, parked workers wouldn't notice shutdown until park_timeout expired
|
||||
- **Bug-inspired tests** (5 new, from competitor bug reports):
|
||||
- `stats_snapshot_is_read_only` — from ractor #310 (destructive get_children)
|
||||
- `stats_under_load_do_not_interfere_with_processing` — stats don't affect msg processing
|
||||
- `shutdown_wakes_parked_workers_immediately` — validates fast shutdown with parking
|
||||
- `mt_send_after_run_delivers_to_running_actors` — from kameo #185 (startup delivery)
|
||||
- `budget_respected_even_with_self_sends` — from actix #515 (mailbox bypass)
|
||||
- **Result**: 57 tests pass, all workspace compiles
|
||||
|
||||
### Research Notes
|
||||
- Full analysis in `CLAUDE/notes/research_synthesis.md`
|
||||
- Baseline benchmarks in `CLAUDE/notes/baseline_benchmarks.md`
|
||||
- Constraints in `CLAUDE/notes/constraints.md`
|
||||
|
||||
## Next Steps
|
||||
- [ ] **Cycle 2: Stress testing + property-based tests**
|
||||
- Concurrent spawn+send stress tests
|
||||
- Multi-threaded fairness validation
|
||||
- Property: message ordering preserved under budget
|
||||
- Property: all messages eventually delivered with budget > 0
|
||||
- [x] **Cycle 2: Stress testing + property-based tests** ✅
|
||||
- [x] **Cycle 3: Adaptive backoff with thread parking** ✅
|
||||
- [ ] **Cycle 4: Enhanced benchmarks**
|
||||
- Message size sensitivity (8B, 64B, 256B, 1KB)
|
||||
- Latency percentiles (p50, p99, p999)
|
||||
- Many-to-one fanin contention
|
||||
- Cross-worker vs same-worker delivery comparison
|
||||
- [x] **Cycle 4: Enhanced benchmarks + bug-inspired tests** ✅
|
||||
- [ ] **Cycle 5: Work stealing exploration**
|
||||
- Evaluate feasibility of actor migration between workers
|
||||
- BEAM two-tier approach: reactive steal + periodic migration
|
||||
|
|
|
|||
|
|
@ -304,12 +304,18 @@ impl Runtime {
|
|||
RuntimeStats { num_workers, uptime_ms, actors, workers, actor_details: Vec::new(), tick_timings }
|
||||
}
|
||||
|
||||
/// Signal all workers to stop
|
||||
/// Signal all workers to stop and wake any that are parked.
|
||||
pub fn shutdown(&self) {
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::info!("runtime.shutdown");
|
||||
|
||||
self.is_running.store(false, Ordering::Release);
|
||||
// Wake all parked workers so they see the shutdown flag immediately
|
||||
for thread in &self.worker_threads {
|
||||
if let Some(t) = thread.get() {
|
||||
t.unpark();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Set a stats hook to receive per-actor snapshots from workers.
|
||||
|
|
|
|||
|
|
@ -1591,3 +1591,143 @@ fn mt_parked_worker_wakes_on_send() {
|
|||
latency
|
||||
);
|
||||
}
|
||||
|
||||
// ═══════════════════════════════════════════════════════════════════════════
|
||||
// Competitor Bug-Inspired Tests
|
||||
// (from analyzing ractor, actix, kameo bug histories)
|
||||
// ═══════════════════════════════════════════════════════════════════════════
|
||||
|
||||
#[test]
|
||||
fn stats_snapshot_is_read_only() {
|
||||
// Inspired by ractor #310: get_children() was destructive (cleared on read).
|
||||
// Verify that calling stats() multiple times returns consistent data.
|
||||
let rt = Runtime::new(RuntimeConfig::default());
|
||||
let _addr = rt.spawn(PingPongActor).unwrap();
|
||||
rt.tick();
|
||||
|
||||
let s1 = rt.stats();
|
||||
let s2 = rt.stats();
|
||||
let s3 = rt.stats();
|
||||
|
||||
// All three snapshots should report the same actor count
|
||||
assert_eq!(s1.actors.len(), s2.actors.len(), "stats() should not mutate state");
|
||||
assert_eq!(s2.actors.len(), s3.actors.len(), "repeated stats() calls must be idempotent");
|
||||
assert!(s1.actors.len() >= 1, "should report at least 1 actor");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stats_under_load_do_not_interfere_with_processing() {
|
||||
// Verify that taking stats snapshots doesn't slow down or break message processing.
|
||||
let rt = Runtime::new(RuntimeConfig::default());
|
||||
let counter = Arc::new(AtomicUsize::new(0));
|
||||
let dummy = rt.new_inbox::<Pong>().unwrap();
|
||||
let addr = rt.spawn(CountingPingActor { counter: counter.clone() }).unwrap();
|
||||
|
||||
for _ in 0..100 {
|
||||
rt.send_to(addr, Ping { reply_to: *dummy.addr() }).unwrap();
|
||||
}
|
||||
|
||||
// Interleave stats calls with ticks
|
||||
for _ in 0..20 {
|
||||
rt.tick();
|
||||
let _s = rt.stats(); // should not affect processing
|
||||
}
|
||||
|
||||
let processed = counter.load(Ordering::SeqCst);
|
||||
assert_eq!(processed, 100, "stats() calls must not interfere with message processing");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn shutdown_wakes_parked_workers_immediately() {
|
||||
// Verify that shutdown unparks all workers so they exit promptly.
|
||||
let rt = Runtime::new(RuntimeConfig {
|
||||
num_threads: 4,
|
||||
..Default::default()
|
||||
});
|
||||
let handle = rt.run().unwrap();
|
||||
|
||||
// Let workers park
|
||||
std::thread::sleep(std::time::Duration::from_millis(50));
|
||||
|
||||
// Shutdown should wake all parked workers
|
||||
let before = std::time::Instant::now();
|
||||
handle.shutdown();
|
||||
handle.join();
|
||||
let shutdown_time = before.elapsed();
|
||||
|
||||
// Workers should exit quickly (well under 1 second)
|
||||
assert!(
|
||||
shutdown_time.as_millis() < 500,
|
||||
"shutdown should complete quickly with parked workers, took {:?}",
|
||||
shutdown_time
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mt_send_after_run_delivers_to_running_actors() {
|
||||
// Inspired by kameo #185: messages not delivered during startup.
|
||||
// Verify that send_to works correctly after run() is called.
|
||||
let rt = Runtime::new(RuntimeConfig {
|
||||
num_threads: 2,
|
||||
..Default::default()
|
||||
});
|
||||
let addr = rt.spawn(PingPongActor).unwrap();
|
||||
let inbox = rt.new_inbox::<Pong>().unwrap();
|
||||
|
||||
// Start the runtime FIRST, then send
|
||||
let handle = rt.run().unwrap();
|
||||
|
||||
// Give workers a moment to start
|
||||
std::thread::sleep(std::time::Duration::from_millis(10));
|
||||
|
||||
// Send after run()
|
||||
handle.runtime.send_to(addr, Ping { reply_to: *inbox.addr() }).unwrap();
|
||||
|
||||
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
|
||||
let mut received = false;
|
||||
while !received {
|
||||
if inbox.try_recv().is_some() {
|
||||
received = true;
|
||||
} else if std::time::Instant::now() > deadline {
|
||||
handle.shutdown();
|
||||
handle.join();
|
||||
panic!("Message sent after run() was not delivered");
|
||||
} else {
|
||||
std::thread::sleep(std::time::Duration::from_millis(1));
|
||||
}
|
||||
}
|
||||
|
||||
handle.shutdown();
|
||||
handle.join();
|
||||
assert!(received, "messages sent after run() must be delivered");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn budget_respected_even_with_self_sends() {
|
||||
// Inspired by actix #515: send bypassing mailbox size.
|
||||
// Verify that self-sends (pending_local) don't bypass the message budget.
|
||||
// The SelfSendActor sends to itself; each self-send goes through pending_local
|
||||
// and appears in the mailbox on the next tick. The budget should still apply.
|
||||
let rt = Runtime::new(RuntimeConfig {
|
||||
actor_message_budget: 4,
|
||||
..Default::default()
|
||||
});
|
||||
let addr = rt.spawn(SelfSendActor).unwrap();
|
||||
let inbox = rt.new_inbox::<Done>().unwrap();
|
||||
|
||||
// remaining=20 means 20 self-sends before replying Done(0)
|
||||
rt.send_to(addr, Countdown { remaining: 20, reply_to: *inbox.addr() }).unwrap();
|
||||
|
||||
// With budget=4, each tick processes at most 4 messages per actor.
|
||||
// The self-send chain should take several ticks to complete.
|
||||
for _ in 0..30 {
|
||||
rt.tick();
|
||||
}
|
||||
|
||||
let reply = inbox.try_recv();
|
||||
assert_eq!(
|
||||
reply,
|
||||
Some(Done(0)),
|
||||
"self-send chain should complete despite message budget"
|
||||
);
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue