diff --git a/Cargo.toml b/Cargo.toml index 1c39c18..b18f9eb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,7 +10,12 @@ crate-type = ["cdylib", "rlib"] default = ["getrandom"] getrandom = ["dep:getrandom"] no_random = [] # compile without access to a source of randomness +stress = [] # Enable stress tests [dependencies] getrandom = { version = "0.2", optional = true } crossbeam-queue = "0.3.12" + +[[bin]] +name = "bench" +path = "benches/main.rs" diff --git a/DESIGN.md b/DESIGN.md index 7870a2e..c43a407 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -75,197 +75,3 @@ impl HybridChannel { } } ``` - -### Kimi Suggestions - -IMPROVEMENTS FITTING DESIGN GOALS -Here are improvements aligned with the stated goals: "maximum usability and speed while keeping line count low" and "no footguns." -Priority 1: Critical Bug Fixes & MVP Completion -1. Fix Runtime Constructor (~5 lines) - - Implement Runtime::new() - - Implement Builder::build() - - Fix examples to compile -2. Handle Full Inboxes Gracefully (~15 lines) - - Return Result<(), Error> from send_to on full inbox - - Provide backpressure signal instead of silent drop - - Add try_send() vs send() distinction -3. Implement Multithreaded Runtime (~30-40 lines) - - Complete threading infrastructure (already partially designed) - - Router runs in separate thread - - Actor processing pool with work-stealing (simple round-robin first) -Priority 2: Usability Enhancements (Low Line Count) -4. Ergonomic Macros (~20-30 lines procedural macro crate) - #[derive(Actor)] - struct MyActor { ... } - - Auto-impl ActorInterface for simple cases - - Reduces boilerplate significantly -5. Request/Response Helper (~15 lines) - let resp: Response = rt.request(addr, msg).await?; - - Common pattern many users need - - Maintains simplicity -6. Inbox Capacity Configuration (~5 lines) - - Per-actor capacity instead of global constant - - Allow spawn_with_capacity() -Priority 3: Performance Optimizations -7. Sharded Router (~30-40 lines) - - Multiple HashMaps based on address hash - - Reduces contention on messaging hot path - - Maintains O(1) lookups -8. Actor Work Stealing (~40-50 lines) - - Multiple actor queues instead of single global queue - - Threads steal work when idle - - Improves cache locality -9. Hybrid Channel (from DESIGN.md) (~25 lines) - - Implements the overflow mechanism described - - Ring buffer + Mutex for emergencies - - Prevent message loss under burst loads -10. Actor State Colocation (~15 lines) - - Group related actors by affinity - - Optional "actor system" or "node" concept - - Better cache locality -Priority 4: Observability (Minimal Overhead) -11. Lightweight Metrics (~15-20 lines) - - Message counts per actor (atomic counters) - - Overflow/drop tracking - - Optional, compile-time feature flag -12. Message Tracing (~10-15 lines opt-in) - - Optional trace ID in envelope - - Zero-cost when disabled (feature flag) -Priority 5: Reliability Patterns -13. Bounded Channels with Overflow (~20 lines) - - Implement HybridChannel from design doc - - Graceful degradation under load -14. Watchdog Timer Pattern (~15 lines example) - - Show pattern: actor checking heartbeats - - Keep library simple, document patterns ---- -SPECIFIC CODE IMPROVEMENTS -Fix Silent Failures (Priority: CRITICAL) -Current (src/runtime.rs:85-93): -pub fn send_to(&self, addr: ActorAddress, msg: M) -> Result<(), ()> { - let envelope: Envelope = Box::new(msg); - self.router_inbox - .try_send(RouterMessage::SendToAddr { addr, msg: envelope }) - .map_err(|_| ()) -} -Improved: -pub fn try_send(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { - let envelope: Envelope = Box::new(msg); - self.router_inbox - .try_send(RouterMessage::SendToAddr { addr, msg: envelope }) - .map_err(|_| Error::from("Router inbox full")) -} -// Add send that blocks/resizes -pub fn send(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { ... } -Implement HybridChannel (Priority: HIGH) -From design doc, add to ring_buffer.rs: -pub struct HybridChannel { - ring: ArrayQueue, - overflow: Mutex>, - overflow_count: AtomicUsize, -} -impl HybridChannel { - fn push(&self, value: T) -> Result<(), T> { - if self.ring.push(value).is_err() { - self.overflow.lock().push_back(value); - self.overflow_count.fetch_add(1, Relaxed); - // Optionally resize ring if overflow_count > threshold - } - Ok(()) - } -} -Fix Runtime Construction (Priority: CRITICAL) -Current: Runtime::new() doesn't exist but examples use it. -Add to runtime.rs: -impl Runtime { - pub fn new(capacity: usize, flavor: Option) -> Self { - let router = Router::new(capacity); - let router_inbox = router.new_sender(); - - Self { - flavor: flavor.unwrap_or_default(), - router, - router_inbox, - actor_queue: ArrayQueue::new(capacity), - thread_pool: Vec::new(), - } - } -} -Add Sharded Router (Priority: MEDIUM) -Current: Single HashMap for all addresses -Improved: N HashMaps based on address modulo -pub(crate) struct Router { - shards: Vec>>, - shard_mask: usize, // shards.len() - 1 (power of 2) - inbox: Receiver, -} -impl Router { - fn get_shard(&self, addr: ActorAddress) -> &HashMap<...> { - &self.shards[(addr as usize) & self.shard_mask] - } -} ---- -RECOMMENDED ROADMAP -Phase 1: Bug Fixes & MVP (1-2 days) -1. Implement Runtime::new() -2. Implement Builder::build() -3. Fix compilation errors -4. Add error handling for full inboxes -5. Document API -Phase 2: Single-Threaded Polish (1 week) -1. Ergonomic macros -2. Request/response helpers -3. Inbox capacity configuration -4. Example improvements -5. Basic tests -Phase 3: Multi-Threaded (2 weeks) -1. Implement threaded runtime -2. Worker thread pool -3. Router in separate thread -4. Work-stealing queues -5. Performance benchmarks -Phase 4: Production Hardening (2 weeks) -1. Sharded router -2. Hybrid channels -3. Metrics (opt-in) -4. Message tracing (opt-in) -5. Stress testing -Phase 5: Documentation & Patterns (1 week) -1. Actor patterns guide -2. Performance tuning guide -3. WASM integration examples -4. Best practices documentation ---- -ALTERNATIVE ARCHITECTURES TO CONSIDER -For Even Smaller Line Count -If the goal is absolutely minimal code, consider: -- Single-threaded only: Remove multi-threading complexity entirely -- No router: Direct mpsc channels between actors (more Erlang-like) -- Simpler scheduling: Round-robin over actors array -Tradeoff: Less flexible, but potentially <200 lines total. -For Better Performance -If performance outweighs minimalism: -- Lock-free HashMap: Use dashmap or equivalent for router -- SegQueue: Better for work-stealing than ArrayQueue -- Pre-allocated: Fixed-size actor pool with object pool pattern -- Lock-free message passing: Use crossbeam or tokio channels throughout -Tradeoff: More dependencies, larger binary size. -For Better Ergonomics -If usability is primary goal: -- Async/Await native: Integrate with tokio or async-std -- Actor supervision: Basic supervisor trees (small implementation) -- Message DSL: Macro for pattern-matching message handlers -Tradeoff: Increases complexity substantially, may conflict with "small" goal. ---- -CONCLUSION -swactor has a solid, minimalist architecture that delivers on its core promise: a small, WASM-compatible actor library. The design is clean, modular, and avoids unnecessary complexity. -Current Grade: C+ (Incomplete MVP) -- Architecture: B+ -- Ease of Use: D (examples don't compile, silent failures) -- Performance: B (good primitives but scalability concerns) -Potential Grade with improvements: A- -- Fixing critical bugs would make it immediately usable -- Sharded router + work-stealing would address scalability -- Ergonomic macros would dramatically improve UX -- Hybrid channels would solve burst-load scenarios -Recommendation: Focus on completing Phase 1 (bug fixes) and Phase 2 (usability). The architecture is sound—it's just incomplete. Avoid premature optimization; measure performance first, then implement sharding/work-stealing if benchmarks show contention. \ No newline at end of file diff --git a/benches/harness.rs b/benches/harness.rs new file mode 100644 index 0000000..b399a5e --- /dev/null +++ b/benches/harness.rs @@ -0,0 +1,301 @@ +//! Manual benchmark harness - zero dependencies, full control. +//! +//! Provides statistical analysis of benchmark runs including: +//! - Mean, median, min, max +//! - Standard deviation +//! - Percentiles (P50, P90, P99, P99.9) +//! - Throughput calculations +//! - Outlier detection and removal + +use std::time::{Duration, Instant}; + +/// Results from a single benchmark run +#[derive(Debug, Clone)] +pub struct BenchResult { + pub name: String, + pub iterations: usize, + pub total_time: Duration, + pub times: Vec, + /// Optional: elements processed (for throughput calculation) + pub elements: Option, +} + +/// Statistical summary of benchmark results +#[derive(Debug)] +pub struct Stats { + pub mean: Duration, + pub median: Duration, + pub min: Duration, + pub max: Duration, + pub std_dev: Duration, + pub p50: Duration, + pub p90: Duration, + pub p99: Duration, + pub p999: Duration, + pub throughput: Option, // elements per second +} + +impl BenchResult { + /// Calculate statistics from the raw timing data + pub fn stats(&self) -> Stats { + let mut sorted: Vec = self.times.clone(); + sorted.sort(); + + let n = sorted.len(); + assert!(n > 0, "Cannot compute stats on empty results"); + + let sum: Duration = sorted.iter().sum(); + let mean = sum / n as u32; + + let median = if n % 2 == 0 { + (sorted[n / 2 - 1] + sorted[n / 2]) / 2 + } else { + sorted[n / 2] + }; + + // Standard deviation + let mean_nanos = mean.as_nanos() as f64; + let variance: f64 = sorted + .iter() + .map(|t| { + let diff = t.as_nanos() as f64 - mean_nanos; + diff * diff + }) + .sum::() + / n as f64; + let std_dev = Duration::from_nanos(variance.sqrt() as u64); + + // Percentiles + let percentile = |p: f64| -> Duration { + let idx = ((p / 100.0) * (n - 1) as f64).round() as usize; + sorted[idx.min(n - 1)] + }; + + let throughput = self.elements.map(|e| { + let secs = self.total_time.as_secs_f64(); + if secs > 0.0 { + (e * self.iterations as u64) as f64 / secs + } else { + 0.0 + } + }); + + Stats { + mean, + median, + min: sorted[0], + max: sorted[n - 1], + std_dev, + p50: percentile(50.0), + p90: percentile(90.0), + p99: percentile(99.0), + p999: percentile(99.9), + throughput, + } + } + + /// Pretty print the results + pub fn print(&self) { + let stats = self.stats(); + + println!("\n{}", "=".repeat(60)); + println!(" {}", self.name); + println!("{}", "=".repeat(60)); + println!(" Iterations: {}", self.iterations); + println!(" Total time: {:?}", self.total_time); + println!(); + println!(" Mean: {:?}", stats.mean); + println!(" Median: {:?}", stats.median); + println!(" Std Dev: {:?}", stats.std_dev); + println!(" Min: {:?}", stats.min); + println!(" Max: {:?}", stats.max); + println!(); + println!(" P50: {:?}", stats.p50); + println!(" P90: {:?}", stats.p90); + println!(" P99: {:?}", stats.p99); + println!(" P99.9: {:?}", stats.p999); + + if let Some(throughput) = stats.throughput { + println!(); + println!(" Throughput: {:.2} ops/sec", throughput); + if throughput > 1_000_000.0 { + println!(" {:.2} M ops/sec", throughput / 1_000_000.0); + } else if throughput > 1_000.0 { + println!(" {:.2} K ops/sec", throughput / 1_000.0); + } + } + println!("{}", "=".repeat(60)); + } +} + +/// A benchmark builder for configuring and running benchmarks +pub struct Bench { + name: String, + warmup_iters: usize, + bench_iters: usize, + elements_per_iter: Option, +} + +impl Bench { + pub fn new(name: impl Into) -> Self { + Self { + name: name.into(), + warmup_iters: 3, + bench_iters: 100, + elements_per_iter: None, + } + } + + /// Set number of warmup iterations (default: 3) + pub fn warmup(mut self, n: usize) -> Self { + self.warmup_iters = n; + self + } + + /// Set number of benchmark iterations (default: 100) + pub fn iters(mut self, n: usize) -> Self { + self.bench_iters = n; + self + } + + /// Set elements per iteration for throughput calculation + pub fn elements(mut self, n: u64) -> Self { + self.elements_per_iter = Some(n); + self + } + + /// Run the benchmark with setup before each iteration + pub fn run_with_setup(self, mut setup: S, mut f: F) -> BenchResult + where + S: FnMut() -> T, + F: FnMut(T), + { + // Warmup + for _ in 0..self.warmup_iters { + let state = setup(); + f(state); + } + + // Benchmark + let mut times = Vec::with_capacity(self.bench_iters); + let total_start = Instant::now(); + + for _ in 0..self.bench_iters { + let state = setup(); + let start = Instant::now(); + f(state); + times.push(start.elapsed()); + } + + let total_time = total_start.elapsed(); + + BenchResult { + name: self.name, + iterations: self.bench_iters, + total_time, + times, + elements: self.elements_per_iter, + } + } + + /// Run the benchmark (no setup between iterations) + pub fn run(self, mut f: F) -> BenchResult + where + F: FnMut(), + { + // Warmup + for _ in 0..self.warmup_iters { + f(); + } + + // Benchmark + let mut times = Vec::with_capacity(self.bench_iters); + let total_start = Instant::now(); + + for _ in 0..self.bench_iters { + let start = Instant::now(); + f(); + times.push(start.elapsed()); + } + + let total_time = total_start.elapsed(); + + BenchResult { + name: self.name, + iterations: self.bench_iters, + total_time, + times, + elements: self.elements_per_iter, + } + } +} + +/// A collection of benchmarks to run together +pub struct BenchSuite { + name: String, + results: Vec, +} + +impl BenchSuite { + pub fn new(name: impl Into) -> Self { + Self { + name: name.into(), + results: Vec::new(), + } + } + + pub fn add(&mut self, result: BenchResult) { + self.results.push(result); + } + + pub fn print_summary(&self) { + println!("\n{}", "#".repeat(70)); + println!("# BENCHMARK SUITE: {}", self.name); + println!("{}", "#".repeat(70)); + + for result in &self.results { + result.print(); + } + + // Summary table + println!("\n{}", "-".repeat(70)); + println!(" SUMMARY"); + println!("{}", "-".repeat(70)); + println!( + " {:30} {:>12} {:>12} {:>12}", + "Benchmark", "Mean", "P99", "Throughput" + ); + println!("{}", "-".repeat(70)); + + for result in &self.results { + let stats = result.stats(); + let throughput_str = stats + .throughput + .map(|t| { + if t > 1_000_000.0 { + format!("{:.2}M/s", t / 1_000_000.0) + } else if t > 1_000.0 { + format!("{:.2}K/s", t / 1_000.0) + } else { + format!("{:.2}/s", t) + } + }) + .unwrap_or_else(|| "-".to_string()); + + println!( + " {:30} {:>12.2?} {:>12.2?} {:>12}", + result.name, stats.mean, stats.p99, throughput_str + ); + } + println!("{}", "-".repeat(70)); + } +} + +/// Prevent the compiler from optimizing away a value +#[inline(never)] +pub fn black_box(x: T) -> T { + // Use inline assembly to prevent optimization + // This is a simplified version - in practice, reads from the value + let ptr = &x as *const T; + unsafe { std::ptr::read_volatile(ptr) } +} diff --git a/benches/main.rs b/benches/main.rs new file mode 100644 index 0000000..41f32fd --- /dev/null +++ b/benches/main.rs @@ -0,0 +1,46 @@ +//! Swactor Benchmark Suite +//! +//! A manual benchmark harness for measuring runtime performance. +//! Zero external dependencies - just std::time. +//! +//! Run with: cargo run --bin bench --release +//! +//! Options: +//! --throughput Run throughput benchmarks only +//! --scaling Run scaling benchmarks only +//! --all Run all benchmarks (default) + +mod harness; +mod throughput; +mod scaling; + +use std::env; + +fn main() { + let args: Vec = env::args().collect(); + + println!("============================================================"); + println!(" SWACTOR BENCHMARK SUITE"); + println!("============================================================"); + println!(); + + // Parse arguments + let run_throughput = args.contains(&"--throughput".to_string()) + || args.contains(&"--all".to_string()) + || args.len() == 1; + let run_scaling = args.contains(&"--scaling".to_string()) + || args.contains(&"--all".to_string()) + || args.len() == 1; + + if run_throughput { + let suite = throughput::run_all(); + suite.print_summary(); + } + + if run_scaling { + let suite = scaling::run_all(); + suite.print_summary(); + } + + println!("\nBenchmarks complete."); +} diff --git a/benches/scaling.rs b/benches/scaling.rs new file mode 100644 index 0000000..8443891 --- /dev/null +++ b/benches/scaling.rs @@ -0,0 +1,281 @@ +//! Scaling benchmarks for the swactor runtime. +//! +//! These benchmarks measure how performance scales with: +//! - Number of actors +//! - Number of worker threads +//! - Message payload size + +use crate::harness::{black_box, Bench, BenchSuite}; +use std::thread; +use std::time::Duration; +use swactor::{ + actor::ActorInterface, + runtime::{Runtime, RuntimeConfig}, +}; + +// ============================================================================ +// Test Actors +// ============================================================================ + +/// A counter actor that just increments on each message +struct CounterActor { + count: usize, +} + +impl CounterActor { + fn new() -> Self { + Self { count: 0 } + } +} + +#[derive(Clone)] +struct Increment; + +impl ActorInterface for CounterActor { + type Incoming = Increment; + type Response = (); + + fn handle(&mut self, _ctx: &Runtime, _msg: Increment) { + self.count += 1; + } +} + +struct SharedCounter { + count: std::sync::Arc, +} + +impl ActorInterface for SharedCounter { + type Incoming = Increment; + type Response = (); + + fn handle(&mut self, _ctx: &Runtime, _msg: Increment) { + self.count + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } +} + +/// An actor that handles variable-sized payloads +struct PayloadActor { + bytes_received: usize, +} + +impl PayloadActor { + fn new() -> Self { + Self { bytes_received: 0 } + } +} + +#[derive(Clone)] +struct Payload(Vec); + +impl ActorInterface for PayloadActor { + type Incoming = Payload; + type Response = (); + + fn handle(&mut self, _ctx: &Runtime, msg: Payload) { + self.bytes_received += msg.0.len(); + black_box(&msg.0); + } +} + +// ============================================================================ +// Benchmarks +// ============================================================================ + +/// Benchmark: How throughput scales with actor count +pub fn bench_actor_count_scaling(suite: &mut BenchSuite) { + let messages_per_actor = 100u64; + + for actor_count in [10u64, 100, 500, 1000] { + let name = format!("scaling_{}_actors", actor_count); + let total_messages = actor_count * messages_per_actor; + + let result = Bench::new(&name) + .warmup(2) + .iters(10) + .elements(total_messages) + .run_with_setup( + || { + let config = RuntimeConfig { + max_actors: (actor_count as usize) + 100, + router_max_messages: (total_messages as usize) * 3, + actor_max_messages: (messages_per_actor as usize) * 2, + num_threads: 1, + }; + let runtime = Runtime::new(config); + + // Spawn actors + let mut actors = Vec::with_capacity(actor_count as usize); + for _ in 0..actor_count { + let addr = runtime.spawn(CounterActor::new()).unwrap(); + actors.push(addr); + } + + // Process registrations + for _ in 0..(actor_count * 2) { + runtime.tick(); + } + + (runtime, actors, messages_per_actor) + }, + |(runtime, actors, msgs_per)| { + // Distribute messages across all actors + for _ in 0..msgs_per { + for actor in &actors { + let _ = runtime.send_to::(*actor, Increment); + } + } + + // Process all + let total = actors.len() as u64 * msgs_per; + for _ in 0..(total * 3) { + runtime.tick(); + } + black_box(()); + }, + ); + + suite.add(result); + } +} + +/// Benchmark: How throughput scales with thread count (multithreaded runtime) +pub fn bench_thread_count_scaling(suite: &mut BenchSuite) { + let actor_count = 100u64; + let messages_per_actor = 500u64; + let total_messages = actor_count * messages_per_actor; + + for thread_count in [2usize, 4, 8] { + let name = format!("scaling_{}_threads", thread_count); + + let result = Bench::new(&name) + .warmup(1) + .iters(5) + .elements(total_messages) + .run_with_setup( + || { + let counter = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let config = RuntimeConfig { + max_actors: (actor_count as usize) + 100, + router_max_messages: (total_messages as usize) * 3, + actor_max_messages: (messages_per_actor as usize) * 2, + num_threads: thread_count, + }; + let runtime = Runtime::new(config); + + let mut actors = Vec::with_capacity(actor_count as usize); + for _ in 0..actor_count { + let addr = runtime + .spawn(SharedCounter { + count: counter.clone(), + }) + .unwrap(); + actors.push(addr); + } + + let handle = runtime.run().unwrap(); + + for actor in &actors { + loop { + if handle + .runtime + .send_to::(*actor, Increment) + .is_ok() + { + break; + } + thread::yield_now(); + } + } + + while counter.load(std::sync::atomic::Ordering::Relaxed) < actors.len() { + thread::yield_now(); + } + counter.store(0, std::sync::atomic::Ordering::Relaxed); + + (handle, actors, counter) + }, + |(handle, actors, counter)| { + for _ in 0..messages_per_actor { + for actor in &actors { + let _ = handle.runtime.send_to::(*actor, Increment); + } + } + + while counter.load(std::sync::atomic::Ordering::Relaxed) + < total_messages as usize + { + thread::yield_now(); + } + + handle.shutdown(); + handle.join(); + black_box(()); + }, + ); + + suite.add(result); + } +} + +/// Benchmark: How throughput scales with message payload size +pub fn bench_payload_size_scaling(suite: &mut BenchSuite) { + let message_count = 1_000u64; + + for payload_size in [64usize, 1024, 16384, 65536] { + let name = format!("payload_{}B", payload_size); + let payload = vec![0u8; payload_size]; + + let result = Bench::new(&name) + .warmup(2) + .iters(20) + .elements(message_count) + .run_with_setup( + || { + let config = RuntimeConfig { + max_actors: 10, + router_max_messages: (message_count as usize) * 2, + actor_max_messages: (message_count as usize) * 2, + num_threads: 1, + }; + let runtime = Runtime::new(config); + let sink = runtime.spawn(PayloadActor::new()).unwrap(); + + // Process registration + for _ in 0..10 { + runtime.tick(); + } + + (runtime, sink, payload.clone()) + }, + |(runtime, sink, payload)| { + for _ in 0..message_count { + let _ = runtime.send_to::(sink, Payload(payload.clone())); + } + + for _ in 0..(message_count * 3) { + runtime.tick(); + } + black_box(()); + }, + ); + + suite.add(result); + } +} + +/// Run all scaling benchmarks +pub fn run_all() -> BenchSuite { + let mut suite = BenchSuite::new("Scaling Benchmarks"); + + println!("\nRunning actor count scaling benchmarks..."); + bench_actor_count_scaling(&mut suite); + + println!("Running thread count scaling benchmarks..."); + bench_thread_count_scaling(&mut suite); + + println!("Running payload size scaling benchmarks..."); + bench_payload_size_scaling(&mut suite); + + suite +} diff --git a/benches/throughput.rs b/benches/throughput.rs new file mode 100644 index 0000000..58c67b3 --- /dev/null +++ b/benches/throughput.rs @@ -0,0 +1,343 @@ +//! Core throughput benchmarks for the swactor runtime. +//! +//! These benchmarks measure: +//! - Message passing throughput +//! - Actor spawn rate +//! - Fan-out and fan-in patterns +//! - Ping-pong latency + +use crate::harness::{black_box, Bench, BenchSuite}; +use swactor::{ + actor::{ActorAddress, ActorInterface}, + runtime::{Runtime, RuntimeConfig}, +}; + +// ============================================================================ +// Test Actors +// ============================================================================ + +/// A sink actor that counts messages received +struct SinkActor { + count: usize, +} + +impl SinkActor { + fn new() -> Self { + Self { count: 0 } + } +} + +#[derive(Clone)] +struct Ping; + +impl ActorInterface for SinkActor { + type Incoming = Ping; + type Response = (); + + fn handle(&mut self, _ctx: &Runtime, _msg: Ping) { + self.count += 1; + } +} + +/// A forwarding actor that passes messages along a chain +struct ForwardActor { + next: Option, +} + +impl ForwardActor { + fn new() -> Self { + Self { next: None } + } + + fn with_next(next: ActorAddress) -> Self { + Self { next: Some(next) } + } +} + +impl ActorInterface for ForwardActor { + type Incoming = Ping; + type Response = Ping; + + fn handle(&mut self, ctx: &Runtime, msg: Ping) { + if let Some(next) = self.next { + let _ = ctx.send_to(next, msg); + } + } +} + +// ============================================================================ +// Benchmarks +// ============================================================================ + +/// Benchmark: Messages sent through the router to a single sink actor +pub fn bench_message_throughput(suite: &mut BenchSuite) { + for msg_count in [1_000u64, 10_000, 100_000] { + let name = format!("message_throughput_{}", msg_count); + + let result = Bench::new(&name) + .warmup(3) + .iters(20) + .elements(msg_count) + .run_with_setup( + || { + // Setup: create runtime and sink actor + let config = RuntimeConfig { + max_actors: 100, + router_max_messages: (msg_count as usize) * 2, + actor_max_messages: (msg_count as usize) * 2, + num_threads: 1, + }; + let runtime = Runtime::new(config); + let sink = runtime.spawn(SinkActor::new()).unwrap(); + (runtime, sink, msg_count) + }, + |(runtime, sink, count)| { + // Send all messages + for _ in 0..count { + let _ = runtime.send_to::(sink, Ping); + } + // Process until done + // Tick enough times to process all messages + // (router tick + actor tick) * messages / WATERLEVEL + for _ in 0..(count * 3) { + runtime.tick(); + } + black_box(()); + }, + ); + + suite.add(result); + } +} + +/// Benchmark: Actor spawn rate +pub fn bench_spawn_rate(suite: &mut BenchSuite) { + for actor_count in [100u64, 500, 900] { + let name = format!("spawn_rate_{}_actors", actor_count); + + let result = Bench::new(&name) + .warmup(3) + .iters(50) + .elements(actor_count) + .run_with_setup( + || { + let config = RuntimeConfig { + max_actors: 1000, + router_max_messages: 10_000, + actor_max_messages: 100, + num_threads: 1, + }; + Runtime::new(config) + }, + |runtime| { + for _ in 0..actor_count { + let _ = runtime.spawn(SinkActor::new()); + } + // Process router messages to register all actors + for _ in 0..(actor_count * 2) { + runtime.tick(); + } + black_box(()); + }, + ); + + suite.add(result); + } +} + +/// Benchmark: Fan-out (1 sender to N receivers) +pub fn bench_fanout(suite: &mut BenchSuite) { + for fan_count in [10u64, 100, 500] { + let name = format!("fanout_1_to_{}", fan_count); + let messages_per_receiver = 100u64; + + let result = Bench::new(&name) + .warmup(2) + .iters(20) + .elements(fan_count * messages_per_receiver) + .run_with_setup( + || { + let config = RuntimeConfig { + max_actors: (fan_count as usize) + 10, + router_max_messages: (fan_count as usize) + * (messages_per_receiver as usize) + * 2, + actor_max_messages: (messages_per_receiver as usize) * 2, + num_threads: 1, + }; + let runtime = Runtime::new(config); + + // Spawn N sink actors + let mut sinks = Vec::with_capacity(fan_count as usize); + for _ in 0..fan_count { + let addr = runtime.spawn(SinkActor::new()).unwrap(); + sinks.push(addr); + } + + // Process router registrations + for _ in 0..(fan_count * 2) { + runtime.tick(); + } + + (runtime, sinks, messages_per_receiver) + }, + |(runtime, sinks, msgs_per)| { + // Send messages to all sinks + for _ in 0..msgs_per { + for sink in &sinks { + let _ = runtime.send_to::(*sink, Ping); + } + } + + // Process all messages + let total_msgs = sinks.len() as u64 * msgs_per; + for _ in 0..(total_msgs * 3) { + runtime.tick(); + } + black_box(()); + }, + ); + + suite.add(result); + } +} + +/// Benchmark: Fan-in (N senders to 1 receiver) +pub fn bench_fanin(suite: &mut BenchSuite) { + for sender_count in [10u64, 100, 500] { + let name = format!("fanin_{}_to_1", sender_count); + let messages_per_sender = 100u64; + + let result = Bench::new(&name) + .warmup(2) + .iters(20) + .elements(sender_count * messages_per_sender) + .run_with_setup( + || { + let total_messages = (sender_count * messages_per_sender) as usize; + let config = RuntimeConfig { + max_actors: (sender_count as usize) + 10, + router_max_messages: total_messages * 3, + actor_max_messages: total_messages * 2, + num_threads: 1, + }; + let runtime = Runtime::new(config); + + // Spawn the sink + let sink = runtime.spawn(SinkActor::new()).unwrap(); + + // Spawn N forwarders pointing at sink + let mut senders = Vec::with_capacity(sender_count as usize); + for _ in 0..sender_count { + let addr = runtime.spawn(ForwardActor::with_next(sink)).unwrap(); + senders.push(addr); + } + + // Process router registrations + for _ in 0..((sender_count + 1) * 2) { + runtime.tick(); + } + + (runtime, senders, sink, messages_per_sender) + }, + |(runtime, senders, _sink, msgs_per)| { + // Each sender forwards msgs_per messages to the sink + for _ in 0..msgs_per { + for sender in &senders { + let _ = runtime.send_to::(*sender, Ping); + } + } + + // Process all messages (forwarder receives + forwards, sink receives) + let total_msgs = senders.len() as u64 * msgs_per; + for _ in 0..(total_msgs * 6) { + runtime.tick(); + } + black_box(()); + }, + ); + + suite.add(result); + } +} + +/// Benchmark: Ring topology (message passed around N actors in a circle) +pub fn bench_ring(suite: &mut BenchSuite) { + for ring_size in [10u64, 100, 500] { + let name = format!("ring_{}_actors", ring_size); + let laps = 10u64; // How many times around the ring + + let result = Bench::new(&name) + .warmup(2) + .iters(20) + .elements(ring_size * laps) + .run_with_setup( + || { + let config = RuntimeConfig { + max_actors: (ring_size as usize) + 10, + router_max_messages: 10_000, + actor_max_messages: 1_000, + num_threads: 1, + }; + let runtime = Runtime::new(config); + + // First, spawn all actors without links + let mut actors: Vec = Vec::with_capacity(ring_size as usize); + for _ in 0..ring_size { + let addr = runtime.spawn(ForwardActor::new()).unwrap(); + actors.push(addr); + } + + // We can't update their `next` field after spawn in this design, + // so instead we'll use an inbox to receive the final message + // For now, we'll just measure message passing through a chain + + // Process registrations + for _ in 0..(ring_size * 2) { + runtime.tick(); + } + + (runtime, actors, laps) + }, + |(runtime, actors, laps)| { + // Send to first actor (even though they don't forward, we're + // measuring the router + inbox overhead) + for _ in 0..laps { + for actor in &actors { + let _ = runtime.send_to::(*actor, Ping); + } + } + + let total = actors.len() as u64 * laps; + for _ in 0..(total * 3) { + runtime.tick(); + } + black_box(()); + }, + ); + + suite.add(result); + } +} + +/// Run all throughput benchmarks +pub fn run_all() -> BenchSuite { + let mut suite = BenchSuite::new("Throughput Benchmarks"); + + println!("\nRunning message throughput benchmarks..."); + bench_message_throughput(&mut suite); + + println!("Running spawn rate benchmarks..."); + bench_spawn_rate(&mut suite); + + println!("Running fan-out benchmarks..."); + bench_fanout(&mut suite); + + println!("Running fan-in benchmarks..."); + bench_fanin(&mut suite); + + println!("Running ring topology benchmarks..."); + bench_ring(&mut suite); + + suite +} diff --git a/src/runtime.rs b/src/runtime.rs index 6a7a76a..ebf7b6b 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,14 +1,14 @@ -use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; use std::thread::{self, JoinHandle}; use crossbeam_queue::ArrayQueue; use crate::{ - Error, actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}, ring_buffer::{Receiver, Sender}, router::{Router, RouterMessage}, + Error, }; /// Generic message inbox for receiving messages outside of the runtime. @@ -176,7 +176,7 @@ impl Runtime { /// Spawn worker threads and start processing, returning a set of handles and /// a Runtime object to interface with. /// - /// ### WARN: + /// ### WARN: /// ##### This function panics if the configuration is set as single threaded /// `config.num_threads == 1` pub fn run(mut self) -> Result { @@ -217,10 +217,19 @@ impl Runtime { while ctx.is_running.load(Ordering::Acquire) { if let Some(mut actor) = ctx.actor_queue.pop() { actor.tick(&ctx); - if let Err(_) = ctx.actor_queue.push(actor) { - panic!( - "Runtime panic: attempted to return an actor to the queue, but queue was full." - ) + // FIXME: Justify this loop. It is here to prevent panics when the + // actor queue is full, but results in a spinlock. + loop { + match ctx.actor_queue.push(actor) { + Ok(()) => break, + Err(a) => { + actor = a; + if !ctx.is_running.load(Ordering::Acquire) { + break; + } + thread::yield_now(); + } + } } } else { thread::yield_now(); diff --git a/tests/stress/concurrency.rs b/tests/stress/concurrency.rs new file mode 100644 index 0000000..654300e --- /dev/null +++ b/tests/stress/concurrency.rs @@ -0,0 +1,323 @@ +//! Concurrency stress tests - hunt for race conditions. +//! +//! These tests target the shutdown races and concurrent access patterns +//! that are most likely to expose bugs. + +use super::{BlackHole, Msg}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::thread; +use std::time::Duration; +use swactor::runtime::{Runtime, RuntimeConfig}; + +/// Shutdown while messages are in flight. +/// Target: AtomicBool ordering bugs, use-after-shutdown. +#[test] +#[cfg(feature = "stress")] +fn shutdown_under_load() { + println!("\n>>> STRESS: Shutdown Under Load"); + + let mut panics = 0; + let mut successes = 0; + + // Run many iterations to catch rare races + for iteration in 0..100 { + let result = std::panic::catch_unwind(|| { + let config = RuntimeConfig { + max_actors: 100, + router_max_messages: 10_000, + actor_max_messages: 1000, + num_threads: 4, + }; + let runtime = Runtime::new(config); + + // Spawn actors + let mut actors = Vec::new(); + for _ in 0..50 { + if let Ok(addr) = runtime.spawn(BlackHole) { + actors.push(addr); + } + } + + let handle = runtime.run().unwrap(); + let rt = handle.runtime.clone(); + + // Sender thread - blast messages + let actors_clone = actors.clone(); + let rt_send = rt.clone(); + let sender = thread::spawn(move || { + for _ in 0..1000 { + for actor in &actors_clone { + let _ = rt_send.send_to::(*actor, Msg); + } + } + }); + + // Random delay before shutdown + let delay = Duration::from_micros((iteration * 17) % 500); + thread::sleep(delay); + + // Shutdown while sender is still going + handle.shutdown(); + + // Wait for sender (it should not panic) + let _ = sender.join(); + + // Join should complete (not hang) + handle.join(); + }); + + match result { + Ok(_) => successes += 1, + Err(_) => panics += 1, + } + } + + println!(" Iterations: 100"); + println!(" Successes: {}", successes); + println!(" Panics: {}", panics); + + if panics > 0 { + println!(">>> FAIL: {} panics detected during shutdown\n", panics); + } else { + println!(">>> PASS: No panics during shutdown under load\n"); + } + + assert_eq!(panics, 0, "Shutdown under load caused panics"); +} + +/// Send to actor immediately after spawn. +/// Target: Race between spawn registration and first message. +#[test] +#[cfg(feature = "stress")] +fn send_to_newborn() { + println!("\n>>> STRESS: Send to Newborn Actor"); + + let mut total_spawned = 0; + let mut total_send_ok = 0; + let mut total_send_fail = 0; + + for _ in 0..100 { + let config = RuntimeConfig { + max_actors: 1000, + router_max_messages: 10_000, + actor_max_messages: 100, + num_threads: 4, + }; + let runtime = Runtime::new(config); + let handle = runtime.run().unwrap(); + + // Immediately spawn and send + for _ in 0..50 { + if let Ok(addr) = handle.runtime.spawn(BlackHole) { + total_spawned += 1; + // Send immediately - actor may not be registered yet + if handle.runtime.send_to::(addr, Msg).is_ok() { + total_send_ok += 1; + } else { + total_send_fail += 1; + } + } + } + + handle.shutdown(); + handle.join(); + } + + println!(" Total spawned: {}", total_spawned); + println!(" Sends succeeded: {}", total_send_ok); + println!(" Sends failed: {}", total_send_fail); + println!(" Note: Failures expected - message may arrive before registration"); + println!(">>> Test complete\n"); +} + +/// FIXME: This test means nothing until we allow killing off actor processes +/// Rapid spawn/despawn cycles. +/// Target: Queue management under churn. +#[test] +#[cfg(feature = "stress")] +fn rapid_spawn_churn() { + println!("\n>>> STRESS: Rapid Spawn Churn"); + + let config = RuntimeConfig { + max_actors: 100, + router_max_messages: 10_000, + actor_max_messages: 100, + num_threads: 4, + }; + let runtime = Runtime::new(config); + let handle = runtime.run().unwrap(); + + let spawn_count = Arc::new(AtomicUsize::new(0)); + let fail_count = Arc::new(AtomicUsize::new(0)); + + // Multiple threads spawning actors + let mut threads = Vec::new(); + for _ in 0..4 { + let rt = handle.runtime.clone(); + let spawns = spawn_count.clone(); + let fails = fail_count.clone(); + + threads.push(thread::spawn(move || { + for _ in 0..500 { + match rt.spawn(BlackHole) { + Ok(_) => { + spawns.fetch_add(1, Ordering::Relaxed); + } + Err(_) => { + fails.fetch_add(1, Ordering::Relaxed); + } + } + // Small yield to increase interleaving + thread::yield_now(); + } + })); + } + + // Let it churn + thread::sleep(Duration::from_millis(100)); + + handle.shutdown(); + + for t in threads { + let _ = t.join(); + } + handle.join(); + + let total_spawns = spawn_count.load(Ordering::Relaxed); + let total_fails = fail_count.load(Ordering::Relaxed); + + println!(" Spawn attempts: {}", total_spawns + total_fails); + println!(" Successes: {}", total_spawns); + println!(" Failures: {} (expected - queue fills)", total_fails); + println!(">>> Test complete - no panics\n"); +} + +/// Multiple threads sending to same actor. +/// Target: Inbox contention, message ordering. +#[test] +#[cfg(feature = "stress")] +fn inbox_contention() { + println!("\n>>> STRESS: Inbox Contention"); + + let config = RuntimeConfig { + max_actors: 10, + router_max_messages: 100_000, + actor_max_messages: 10_000, + num_threads: 4, + }; + let runtime = Runtime::new(config); + let target = runtime.spawn(BlackHole).unwrap(); + let handle = runtime.run().unwrap(); + + // Wait for registration + thread::sleep(Duration::from_millis(10)); + + let send_count = Arc::new(AtomicUsize::new(0)); + let fail_count = Arc::new(AtomicUsize::new(0)); + + // 8 threads all sending to same actor + let mut threads = Vec::new(); + for _ in 0..8 { + let rt = handle.runtime.clone(); + let sends = send_count.clone(); + let fails = fail_count.clone(); + + threads.push(thread::spawn(move || { + for _ in 0..10_000 { + if rt.send_to::(target, Msg).is_ok() { + sends.fetch_add(1, Ordering::Relaxed); + } else { + fails.fetch_add(1, Ordering::Relaxed); + } + } + })); + } + + for t in threads { + let _ = t.join(); + } + + // Let messages process + thread::sleep(Duration::from_millis(50)); + + handle.shutdown(); + handle.join(); + + let total_sends = send_count.load(Ordering::Relaxed); + let total_fails = fail_count.load(Ordering::Relaxed); + + println!(" Threads: 8"); + println!(" Msgs per thread: 10,000"); + println!(" Total sent: {}", total_sends); + println!(" Total failed: {}", total_fails); + println!( + " Success rate: {:.1}%", + (total_sends as f64 / (total_sends + total_fails) as f64) * 100.0 + ); + println!(">>> Test complete - no panics\n"); +} + +/// Shutdown timing fuzz - randomize when shutdown is called. +/// Target: Edge cases in shutdown state machine. +#[test] +#[cfg(feature = "stress")] +fn shutdown_timing_fuzz() { + println!("\n>>> STRESS: Shutdown Timing Fuzz"); + + let mut results = Vec::new(); + + for delay_us in [0, 1, 10, 100, 1000, 5000] { + let mut ok = 0; + let mut fail = 0; + + for _ in 0..20 { + let result = std::panic::catch_unwind(|| { + let config = RuntimeConfig { + max_actors: 50, + router_max_messages: 1000, + actor_max_messages: 100, + num_threads: 4, + }; + let runtime = Runtime::new(config); + + for _ in 0..20 { + let _ = runtime.spawn(BlackHole); + } + + let handle = runtime.run().unwrap(); + + // Specific delay + if delay_us > 0 { + thread::sleep(Duration::from_micros(delay_us)); + } + + handle.shutdown(); + handle.join(); + }); + + match result { + Ok(_) => ok += 1, + Err(_) => fail += 1, + } + } + + results.push((delay_us, ok, fail)); + } + + println!(" delay_us ok fail"); + println!(" -------- -- ----"); + for (delay, ok, fail) in &results { + println!(" {:>8} {:>2} {:>4}", delay, ok, fail); + } + + let total_fails: i32 = results.iter().map(|(_, _, f)| *f).sum(); + if total_fails > 0 { + println!( + "\n>>> FAIL: {} panics across timing variations", + total_fails + ); + } else { + println!("\n>>> PASS: All timing variations succeeded"); + } +} diff --git a/tests/stress/mod.rs b/tests/stress/mod.rs new file mode 100644 index 0000000..be625c4 --- /dev/null +++ b/tests/stress/mod.rs @@ -0,0 +1,199 @@ +//! Stress test utilities and result reporting. +//! +//! Provides a simple framework for stress tests with JSON + pretty output. + +#![allow(dead_code)] // Utilities may not all be used in every test + +pub mod concurrency; +pub mod saturation; + +use std::time::{Duration, Instant}; + +/// Results from a stress test +#[derive(Debug)] +pub struct StressResult { + pub name: String, + pub duration: Duration, + pub operations: u64, + pub successes: u64, + pub failures: u64, + pub notes: Vec, +} + +impl StressResult { + pub fn new(name: impl Into) -> Self { + Self { + name: name.into(), + duration: Duration::ZERO, + operations: 0, + successes: 0, + failures: 0, + notes: Vec::new(), + } + } + + pub fn failure_rate(&self) -> f64 { + if self.operations == 0 { + 0.0 + } else { + (self.failures as f64 / self.operations as f64) * 100.0 + } + } + + pub fn throughput(&self) -> f64 { + let secs = self.duration.as_secs_f64(); + if secs > 0.0 { + self.operations as f64 / secs + } else { + 0.0 + } + } + + pub fn note(&mut self, msg: impl Into) { + self.notes.push(msg.into()); + } + + pub fn print(&self) { + println!("\n{}", "=".repeat(60)); + println!(" STRESS: {}", self.name); + println!("{}", "=".repeat(60)); + println!(" Duration: {:?}", self.duration); + println!(" Operations: {}", self.operations); + println!(" Successes: {}", self.successes); + println!(" Failures: {}", self.failures); + println!(" Failure Rate: {:.2}%", self.failure_rate()); + println!(" Throughput: {:.2} ops/sec", self.throughput()); + + if !self.notes.is_empty() { + println!(); + println!(" Notes:"); + for note in &self.notes { + println!(" - {}", note); + } + } + println!("{}", "=".repeat(60)); + } + + pub fn to_json(&self) -> String { + format!( + r#"{{"name":"{}","duration_ms":{},"operations":{},"successes":{},"failures":{},"failure_rate_pct":{:.2},"throughput":{:.2},"notes":{:?}}}"#, + self.name, + self.duration.as_millis(), + self.operations, + self.successes, + self.failures, + self.failure_rate(), + self.throughput(), + self.notes + ) + } +} + +/// A simple stress test runner +pub struct Stress { + name: String, + duration: Option, + iterations: Option, +} + +impl Stress { + pub fn new(name: impl Into) -> Self { + Self { + name: name.into(), + duration: None, + iterations: None, + } + } + + /// Run for a fixed duration + pub fn for_duration(mut self, d: Duration) -> Self { + self.duration = Some(d); + self + } + + /// Run for a fixed number of iterations + pub fn for_iterations(mut self, n: u64) -> Self { + self.iterations = Some(n); + self + } + + /// Run the stress test, counting successes and failures + pub fn run(self, mut f: F) -> StressResult + where + F: FnMut() -> bool, // returns true on success, false on failure + { + let mut result = StressResult::new(&self.name); + let start = Instant::now(); + + match (self.duration, self.iterations) { + (Some(duration), _) => { + while start.elapsed() < duration { + if f() { + result.successes += 1; + } else { + result.failures += 1; + } + result.operations += 1; + } + } + (None, Some(iterations)) => { + for _ in 0..iterations { + if f() { + result.successes += 1; + } else { + result.failures += 1; + } + result.operations += 1; + } + } + (None, None) => { + // Default: 1000 iterations + for _ in 0..1000 { + if f() { + result.successes += 1; + } else { + result.failures += 1; + } + result.operations += 1; + } + } + } + + result.duration = start.elapsed(); + result + } +} + +// Test actors used across stress tests +use swactor::{actor::ActorInterface, runtime::Runtime}; + +/// An actor that just absorbs messages +pub struct BlackHole; + +#[derive(Clone)] +pub struct Msg; + +impl ActorInterface for BlackHole { + type Incoming = Msg; + type Response = (); + fn handle(&mut self, _ctx: &Runtime, _msg: Msg) {} +} + +/// An actor that counts messages received +pub struct Counter { + pub count: usize, +} + +impl Counter { + pub fn new() -> Self { + Self { count: 0 } + } +} + +impl ActorInterface for Counter { + type Incoming = Msg; + type Response = (); + fn handle(&mut self, _ctx: &Runtime, _msg: Msg) { + self.count += 1; + } +} diff --git a/tests/stress/saturation.rs b/tests/stress/saturation.rs new file mode 100644 index 0000000..6301ae3 --- /dev/null +++ b/tests/stress/saturation.rs @@ -0,0 +1,299 @@ +//! Saturation stress tests - find where the runtime breaks. +//! +//! These tests intentionally push past limits to document failure modes. + +use super::{BlackHole, Counter, Msg, Stress, StressResult}; +use std::time::Duration; +use swactor::runtime::{Runtime, RuntimeConfig}; + +/// FIXME: does this even make sense to test? +/// Blast the router inbox until it overflows. +/// Documents: What happens when router can't keep up? +#[test] +#[cfg(feature = "stress")] +fn router_inbox_overflow() { + println!("\n>>> STRESS: Router Inbox Overflow"); + + let config = RuntimeConfig { + max_actors: 10, + router_max_messages: 100, // Tiny buffer + actor_max_messages: 1000, + num_threads: 1, + }; + let runtime = Runtime::new(config); + let sink = runtime.spawn(BlackHole).unwrap(); + + // Blast messages without processing + let mut result = StressResult::new("router_inbox_overflow"); + let start = std::time::Instant::now(); + + for _ in 0..10_000 { + result.operations += 1; + if runtime.send_to::(sink, Msg).is_ok() { + result.successes += 1; + } else { + result.failures += 1; + } + } + + result.duration = start.elapsed(); + result.note(format!("Router buffer: 100, Messages sent: 10,000")); + result.note(format!( + "Expected: ~99% failure rate (buffer fills immediately)" + )); + result.print(); + + // Verify we actually saw failures + assert!(result.failures > 0, "Expected router to reject messages"); + println!(">>> PASS: Router correctly rejects messages when full\n"); +} + +/// FIXME: This test makes no sense until we make sure panics happen when +/// actor inbox buffers are full. +/// Blast a single actor's inbox until it overflows. +/// Documents: What happens when actor can't keep up? +#[test] +#[cfg(feature = "stress")] +fn actor_inbox_overflow() { + println!("\n>>> STRESS: Actor Inbox Overflow"); + + let config = RuntimeConfig { + max_actors: 10, + router_max_messages: 100_000, // Large router buffer + actor_max_messages: 100, // Tiny actor inbox + num_threads: 1, + }; + let runtime = Runtime::new(config); + let sink = runtime.spawn(BlackHole).unwrap(); + + // Process router registration + runtime.tick(); + + // Now blast messages - router will accept them but actor inbox will fill + let mut sent = 0u64; + for _ in 0..10_000 { + if runtime.send_to::(sink, Msg).is_ok() { + sent += 1; + } + // Tick occasionally to let router deliver + if sent % 100 == 0 { + runtime.tick(); + } + } + + // The router accepted messages, but many were dropped at actor inbox + // We can't easily count these drops from outside, but we can document the behavior + println!(" Router accepted {} messages", sent); + println!(" Actor inbox capacity: 100"); + println!(" Note: Messages beyond inbox capacity are silently dropped"); + println!(">>> This is a known limitation - bounded queues drop overflow\n"); +} + +/// FIXME: Does it make sense to have this as a test? Yes the runtime +/// fails if you try and spawn actors when the queue is full. +/// Spawn actors until the queue rejects. +/// Documents: What happens when actor queue fills? +#[test] +#[cfg(feature = "stress")] +fn actor_queue_overflow() { + println!("\n>>> STRESS: Actor Queue Overflow"); + + let config = RuntimeConfig { + max_actors: 100, // Small actor queue + router_max_messages: 10_000, + actor_max_messages: 100, + num_threads: 1, + }; + let runtime = Runtime::new(config); + + let mut result = StressResult::new("actor_queue_overflow"); + let start = std::time::Instant::now(); + + // Try to spawn 500 actors into 100-slot queue + for _ in 0..500 { + result.operations += 1; + match runtime.spawn(BlackHole) { + Ok(_) => result.successes += 1, + Err(_) => result.failures += 1, + } + } + + result.duration = start.elapsed(); + result.note(format!("Queue capacity: 100, Spawn attempts: 500")); + result.note(format!( + "Expected: ~80% failure rate (queue fills after ~100)" + )); + result.print(); + + // Note: Router also takes a slot, so we expect ~99 actors max + assert!( + result.successes <= 100, + "Spawned more actors than queue capacity" + ); + assert!(result.failures > 0, "Expected spawn failures"); + println!(">>> PASS: Actor queue correctly rejects when full\n"); +} + +/// FIXME: IS this actually testing what it should be? +/// Sustained overload - run at 2x capacity for extended period. +/// Documents: Does the system degrade gracefully or crash? +#[test] +#[cfg(feature = "stress")] +fn sustained_overload() { + println!("\n>>> STRESS: Sustained Overload"); + + let config = RuntimeConfig { + max_actors: 100, + router_max_messages: 1000, + actor_max_messages: 100, + num_threads: 1, + }; + let runtime = Runtime::new(config); + + // Spawn some actors + let mut actors = Vec::new(); + for _ in 0..50 { + if let Ok(addr) = runtime.spawn(Counter::new()) { + actors.push(addr); + } + } + + // Process registrations + for _ in 0..200 { + runtime.tick(); + } + + let result = Stress::new("sustained_overload") + .for_duration(Duration::from_secs(2)) + .run(|| { + // Send to random actor + let idx = (std::time::Instant::now().elapsed().as_nanos() as usize) % actors.len(); + let success = runtime.send_to::(actors[idx], Msg).is_ok(); + + // Process some (but not all) - simulating overload + runtime.tick(); + + success + }); + + result.print(); + println!(">>> System survived sustained overload without panic\n"); +} + +/// FIXME: The logic for timing recovery does not make sense +/// Burst traffic - idle to 100x normal, back to idle. +/// Documents: Recovery behavior after traffic spikes. +#[test] +#[cfg(feature = "stress")] +fn burst_traffic() { + println!("\n>>> STRESS: Burst Traffic"); + + let config = RuntimeConfig { + max_actors: 100, + router_max_messages: 10_000, + actor_max_messages: 1000, + num_threads: 1, + }; + let runtime = Runtime::new(config); + let sink = runtime.spawn(Counter::new()).unwrap(); + + // Process registration + for _ in 0..10 { + runtime.tick(); + } + + let mut total_sent = 0u64; + let mut total_failed = 0u64; + + // 5 burst cycles + for cycle in 0..5 { + // Burst: send 1000 messages as fast as possible + let mut burst_sent = 0; + let mut burst_failed = 0; + for _ in 0..1000 { + if runtime.send_to::(sink, Msg).is_ok() { + burst_sent += 1; + } else { + burst_failed += 1; + } + } + total_sent += burst_sent; + total_failed += burst_failed; + + // Recovery: process until queue is drained + let recovery_start = std::time::Instant::now(); + for _ in 0..5000 { + runtime.tick(); + } + let recovery_time = recovery_start.elapsed(); + + println!( + " Cycle {}: sent={}, failed={}, recovery={:?}", + cycle + 1, + burst_sent, + burst_failed, + recovery_time + ); + } + + println!( + "\n Total sent: {}, Total failed: {}", + total_sent, total_failed + ); + println!(">>> Burst traffic test complete\n"); +} + +/// Find the message drop cliff - at what load factor do drops spike? +#[test] +#[cfg(feature = "stress")] +fn message_drop_curve() { + println!("\n>>> STRESS: Message Drop Curve"); + println!(" Testing drop rate at various load factors...\n"); + + // Test at different load factors (messages per tick) + for msgs_per_tick in [1, 5, 10, 20, 50, 100] { + let config = RuntimeConfig { + max_actors: 10, + router_max_messages: 1000, + actor_max_messages: 500, + num_threads: 1, + }; + let runtime = Runtime::new(config); + let sink = runtime.spawn(BlackHole).unwrap(); + + // Warmup + for _ in 0..10 { + runtime.tick(); + } + + let mut sent = 0u64; + let mut failed = 0u64; + + // Run for fixed iterations + for _ in 0..100 { + // Send burst + for _ in 0..msgs_per_tick { + if runtime.send_to::(sink, Msg).is_ok() { + sent += 1; + } else { + failed += 1; + } + } + // Process one tick + runtime.tick(); + } + + let drop_rate = if sent + failed > 0 { + (failed as f64 / (sent + failed) as f64) * 100.0 + } else { + 0.0 + }; + + println!( + " msgs/tick={:3} sent={:5} failed={:5} drop_rate={:.1}%", + msgs_per_tick, sent, failed, drop_rate + ); + } + + println!("\n>>> Message drop curve test complete\n"); +} diff --git a/tests/stress_tests.rs b/tests/stress_tests.rs new file mode 100644 index 0000000..216aea2 --- /dev/null +++ b/tests/stress_tests.rs @@ -0,0 +1,11 @@ +//! Stress test suite for swactor runtime. +//! +//! Run with: cargo test --features stress stress_ -- --nocapture +//! +//! These tests are hidden behind the `stress` feature flag because they: +//! - Take longer to run +//! - Intentionally push the system to failure +//! - May produce different results on different machines + +#[cfg(feature = "stress")] +mod stress;