diff --git a/Cargo.toml b/Cargo.toml index 8b21fb6..c8963bf 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,7 @@ name = "swactor" version = "0.1.0" edition = "2024" +autobenches = false [lib] crate-type = ["cdylib", "rlib"] @@ -9,7 +10,13 @@ crate-type = ["cdylib", "rlib"] [features] 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..8eddc47 --- /dev/null +++ b/benches/harness.rs @@ -0,0 +1,270 @@ +//! 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, + } + } +} + +/// 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..e54bf64 --- /dev/null +++ b/benches/scaling.rs @@ -0,0 +1,280 @@ +//! 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 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/actor.rs b/src/actor.rs index 2b80d40..3450bee 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -1,34 +1,34 @@ -use crate::{runtime::Runtime, WATERLEVEL, get_random, ring_buffer::Receiver}; +use crate::{WATERLEVEL, channel::Receiver, get_random, runtime::Runtime}; /// The primary trait defining data that can be passed to and from actor processes pub trait Message: 'static + Sized + Clone + Send + Sync {} impl Message for T {} /// The trait that needs to be implemented in order to run a process as an `Actor` -/// +/// /// The `Incoming` type represents `Messages` that can be delivered to the `Actor`. -/// +/// /// The `Response` type represents possible `Messages` the actor may attempt to reply with. -/// +/// /// The `fn handle(..)` is where you implement the logic for handling `Incoming` messages -/// +/// /// # Example /// ``` /// use swactor::{actor::{ActorAddress, ActorInterface}, runtime::Runtime}; -/// +/// /// struct Greeter { /// num_greeted: usize, /// } -/// +/// /// #[derive(Clone)] // required to auto implement `Message` /// struct GreetMessage { /// who: String, /// return_addr: ActorAddress, /// } -/// +/// /// #[derive(Clone)] /// struct GreetResponse(String); -/// +/// /// impl ActorInterface for Greeter { /// type Incoming = GreetMessage; /// type Response = GreetResponse; @@ -71,10 +71,7 @@ where impl Actor { pub(crate) fn new(inbox: Receiver, inner: A) -> Self { - Self { - inbox, - inner, - } + Self { inbox, inner } } } diff --git a/src/channel.rs b/src/channel.rs new file mode 100644 index 0000000..a210a93 --- /dev/null +++ b/src/channel.rs @@ -0,0 +1,76 @@ +use std::{collections::VecDeque, sync::{Arc, Mutex}}; + +use crossbeam_queue::ArrayQueue; + +pub struct HybridChannel { + ring: ArrayQueue, + overflow: Mutex>, +} + +impl HybridChannel { + pub fn new(capacity: usize) -> Self { + Self { + ring: ArrayQueue::new(capacity), + overflow: Mutex::new(VecDeque::new()), + } + } + + pub fn push(&self, value: T) -> Result<(), T> { + match self.ring.push(value) { + Ok(()) => Ok(()), + Err(v) => { + self.overflow.lock().unwrap().push_back(v); + Ok(()) + } + } + } + + pub fn pop(&self) -> Option { + if let Some(value) = self.ring.pop() { + return Some(value); + } + + self.overflow.lock().unwrap().pop_front() + } + + pub fn len(&self) -> usize { + self.ring.len() + self.overflow.lock().unwrap().len() + } +} + +pub(crate) struct Receiver { + queue: Arc>, +} + +impl Receiver { + pub fn new(capacity: usize) -> Self { + let queue = Arc::new(HybridChannel::new(capacity)); + + Self { queue } + } + + pub fn len(&self) -> usize { + self.queue.len() + } + + pub fn try_recv(&self) -> Option { + return self.queue.pop(); + } + + pub fn new_sender(&self) -> Sender { + Sender { + queue: self.queue.clone(), + } + } +} + +pub(crate) struct Sender { + queue: Arc>, +} + +impl Sender { + pub fn try_send(&self, value: T) -> Result<(), T> { + return self.queue.push(value); + } +} + diff --git a/src/error.rs b/src/error.rs index 12183e2..9aa626a 100644 --- a/src/error.rs +++ b/src/error.rs @@ -2,7 +2,7 @@ /// # Usage /// ``` /// use swactor::Error; -/// +/// /// fn foo_if_even(num: u64) -> Result { /// if num % 2 == 0 { /// return Ok("foo".into()); diff --git a/src/lib.rs b/src/lib.rs index 5ea2d6b..1d67009 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,9 +1,9 @@ pub mod actor; +mod channel; pub(crate) mod error; pub use error::Error; -mod ring_buffer; mod router; pub mod runtime; @@ -12,6 +12,20 @@ pub(crate) fn get_random(buf: &mut [u8]) { getrandom::getrandom(buf).unwrap() } +#[cfg(feature = "no_random")] +pub(crate) fn get_random(buf: &mut [u8]) { + use core::sync::atomic::{AtomicUsize, Ordering}; + + static COUNTER: AtomicUsize = AtomicUsize::new(0); + + let value = COUNTER.fetch_add(1, Ordering::Relaxed); + let bytes = value.to_ne_bytes(); + + for (i, byte) in buf.iter_mut().enumerate() { + *byte = bytes[i % core::mem::size_of::()]; + } +} + /// FIXME: remove hard coded defaults /// The strategy for message processing is such: /// diff --git a/src/ring_buffer.rs b/src/ring_buffer.rs deleted file mode 100644 index b016b3e..0000000 --- a/src/ring_buffer.rs +++ /dev/null @@ -1,57 +0,0 @@ -//! Shallow wrapper around the `crossbeam_queue::ArrayQueue` implementation of a mpmc ring buffer. -use std::sync::Arc; -pub use crossbeam_queue::ArrayQueue; - -/// The receiving end of a `crossbeam_queue::ArrayQueue`, a lock-free mpmc queue. -/// The queue is constructed by the `Receiver::new()` method. -/// Responsible for creating the `Sender` ends of itself. -/// -/// Notably: The `Receiver` provides no guarentees that a sending end of the channel exists. -pub(crate) struct Receiver { - queue: Arc>, -} - -impl Receiver { - /// Constructs a new `ArrayQueue` with given capacity. - /// - /// # Panics - /// Will panic if capacity is passed as 0 - pub fn new(capacity: usize) -> Self { - Self { - queue: Arc::new(ArrayQueue::new(capacity)), - } - } - - /// Returns the number of elements in the inner queue - pub fn len(&self) -> usize { - self.queue.len() - } - - /// Attempt to retrieve a value from the queue. Returns `None` if empty - pub fn try_recv(&self) -> Option { - self.queue.pop() - } - - /// Construct a new `Sender` assosciated with this queue. - pub fn new_sender(&self) -> Sender { - Sender { - queue: self.queue.clone(), - } - } -} - -/// The sending end of a `crossbeam_queue::ArrayQueue`, a lock free mpsc queue. -/// The queue is initialized via calling the corresponding `Receiver::::new()` method, -/// and the sending end of the queue is constructed via calling `receiver.new_sender()`. -/// -/// Notably: The `Sender` provides no guarentees that a receiving end of the channel exists. -pub(crate) struct Sender { - queue: Arc>, -} - -impl Sender { - /// Attempt to push a value to the queue. Returns Err(value) if the queue is full. - pub fn try_send(&self, value: T) -> Result<(), T> { - self.queue.push(value) - } -} diff --git a/src/router.rs b/src/router.rs index 82fb10a..55296f0 100644 --- a/src/router.rs +++ b/src/router.rs @@ -2,12 +2,13 @@ use std::{collections::HashMap, sync::Arc}; use crate::{ actor::{ActorAddress, ActorInterface, Message}, - ring_buffer::Sender, runtime::Runtime, + channel::Sender, + runtime::Runtime, }; /// FIXME: Go over with a fine-toothed comb and reassure yourself this typing /// makes sense, that we are not doing loads of indirection on a hot path. -/// +/// /// A type erased `Message` to be routed between actor processes. pub(crate) type Envelope = Arc; @@ -60,13 +61,15 @@ impl ActorInterface for Router { match msg { RouterMessage::AddAddr(addr, sender) => { self.directory.insert(addr, sender); - }, - RouterMessage::RemoveAddr(addr) => { self.directory.remove(&addr); }, + } + RouterMessage::RemoveAddr(addr) => { + self.directory.remove(&addr); + } RouterMessage::SendToAddr { addr, msg } => { if let Some(sender) = self.directory.get(&addr) { sender.try_send(msg); } - }, + } } } } diff --git a/src/runtime.rs b/src/runtime.rs index 6a7a76a..7a9f768 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,14 +1,13 @@ -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::channel::HybridChannel; use crate::{ - Error, actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}, - ring_buffer::{Receiver, Sender}, + channel::{Receiver, Sender}, router::{Router, RouterMessage}, + Error, }; /// Generic message inbox for receiving messages outside of the runtime. @@ -60,7 +59,7 @@ impl Default for RuntimeConfig { /// The `Runtime` struct is the primary gateway for interacting with the framework. pub struct Runtime { config: RuntimeConfig, - actor_queue: ArrayQueue>, + actor_queue: HybridChannel>, router_interface: Sender, router: Option>, // `None` if single-threaded @@ -91,7 +90,7 @@ impl Runtime { /// Builds a new `Runtime` struct, but does not yet run anything. If multithreaded, call /// `run()`, if single threaded, needs to be driven by calls to the `tick()` method. pub fn new(config: RuntimeConfig) -> Self { - let actor_queue = ArrayQueue::new(config.max_actors); + let actor_queue = HybridChannel::new(config.max_actors); // router is a unique actor in that the runtime needs access to it's `Sender` handle let router_inner = Router::new(); @@ -176,7 +175,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 +216,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..2511745 --- /dev/null +++ b/tests/stress/concurrency.rs @@ -0,0 +1,332 @@ +//! 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); + + if total_send_fail > 0 { + println!(">>> FAIL: {} messages failed to send\n", total_send_fail); + } else { + println!(">>> PASS: All messages succeeded\n"); + } + + assert_eq!(total_send_fail, 0, "Race condition caused failed message delivery"); + + 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"); +} + +/// FIXME: Not sure this test is meaningful. +/// 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..a8fc97e --- /dev/null +++ b/tests/stress/saturation.rs @@ -0,0 +1,173 @@ +//! 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}; + +/// Blast the router inbox +#[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")); + + // With hybrid, no failures expected + assert_eq!(result.failures, 0, "Hybrid channel should not reject"); + result.print(); + println!(">>> PASS: Hybrid channel prevented router overflow\n"); +} + +/// Blast a single actor's inbox +#[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(Counter::new()).unwrap(); + + // Process router registration + runtime.tick(); + + // Now blast messages - router will accept them but actor inbox will fill + let mut sent = 0u64; + let mut router_failed = 0u64; + for _ in 0..10_000 { + if runtime.send_to::(sink, Msg).is_ok() { + sent += 1; + } else { + router_failed += 1; + } + // Tick occasionally to let router deliver + if sent % 100 == 0 { + runtime.tick(); + } + } + + // Process all remaining messages + for _ in 0..5000 { + runtime.tick(); + } + + println!(" Router accepted: {}", sent); + println!(" Router rejected: {}", router_failed); + + assert_eq!(router_failed, 0, "Router rejected message under load"); + println!(">>> PASS: No message loss with hybrid channel\n"); +} + +/// Blast the runtime with actor spawns +#[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.print(); + + // Note: Router also takes a slot, so we expect ~99 actors max + assert_eq!( + result.failures, 0, + "Spawned more actors than queue capacity" + ); + 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"); +} 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;