feat: hybrid channels

Add a mutex-locked Dequeue to prevent panics and failures on overflow.

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"

DESIGN.md

@@ -74,4 +74,4 @@ impl<T> HybridChannel<T> {
         }
     }
 }
-```
+```

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<Duration>,
+    /// Optional: elements processed (for throughput calculation)
+    pub elements: Option<u64>,
+}
+
+/// 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<f64>, // elements per second
+}
+
+impl BenchResult {
+    /// Calculate statistics from the raw timing data
+    pub fn stats(&self) -> Stats {
+        let mut sorted: Vec<Duration> = 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::<f64>()
+            / 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<u64>,
+}
+
+impl Bench {
+    pub fn new(name: impl Into<String>) -> 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<S, T, F>(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<BenchResult>,
+}
+
+impl BenchSuite {
+    pub fn new(name: impl Into<String>) -> 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<T>(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) }
+}

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<String> = 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.");
+}

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<std::sync::atomic::AtomicUsize>,
+}
+
+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<u8>);
+
+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::<Increment>(*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::<Increment>(*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::<Increment>(*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::<Payload>(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
+}

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<ActorAddress>,
+}
+
+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::<Ping>(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::<Ping>(*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::<Ping>(*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<ActorAddress> = 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::<Ping>(*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
+}

examples/hello.rs

@@ -1,4 +1,7 @@
-use swactor::{ActorAddress, ActorInterface, Message, Runtime, RuntimeFlavor};
+use swactor::{
+    actor::{ActorAddress, ActorInterface},
+    runtime::{Runtime, RuntimeConfig},
+};
 
 #[derive(Debug, Default)]
 struct Greeter {
@@ -13,7 +16,9 @@ struct GreetMessage {
     /// who do we send out greeting back to?
     return_addr: ActorAddress,
 }
-impl Message for GreetMessage {}
+
+#[derive(Debug, Default, Clone)]
+struct GreetResponse(String);
 
 impl ActorInterface for Greeter {
     type Incoming = GreetMessage;
@@ -29,17 +34,18 @@ impl ActorInterface for Greeter {
     }
 }
 
-#[derive(Debug, Default, Clone)]
-struct GreetResponse(String);
-impl Message for GreetResponse {}
-
 fn main() {
-    let mut rt = Runtime::new(100, Some(RuntimeFlavor::SingleThreaded));
+    let rt = Runtime::new(RuntimeConfig::default());
+
+    // spawn a `Greeter` in the runtime, returning an address to contact it with
     let addr = rt
         .spawn(Greeter::default())
         .expect("failed to spawn greeter");
-    let inbox = rt.new_inbox::<GreetResponse>();
 
+    // create an `Inbox` that allows us to receive messages from the runtime
+    let inbox = rt.new_inbox::<GreetResponse>().unwrap();
+
+    // send a message to the `Greeter` we spawned
     rt.send_to(
         addr,
         GreetMessage {
@@ -48,10 +54,11 @@ fn main() {
         },
     )
     .unwrap();
+
+    // default runtime is single threaded, and requires the parent process to drive
     for _ in 0..3 {
         rt.tick();
     }
-
     let resp = inbox.try_recv().expect("greeter should have said hello");
 
     println!("{}", resp.0);

examples/ring.rs

@@ -0,0 +1,71 @@
+use swactor::{
+    actor::{ActorAddress, ActorInterface},
+    runtime::{Inbox, Runtime, RuntimeConfig},
+};
+
+#[derive(Debug, Default, Clone)]
+pub struct RingMessage {
+    count: usize,
+}
+
+impl RingMessage {
+    pub fn next(self) -> Self {
+        Self {
+            count: self.count + 1,
+        }
+    }
+}
+
+#[derive(Debug, Default)]
+struct RingActor {
+    next: ActorAddress,
+}
+
+impl RingActor {
+    pub fn new(next: ActorAddress) -> Self {
+        Self { next }
+    }
+}
+
+impl ActorInterface for RingActor {
+    type Incoming = RingMessage;
+    type Response = ();
+    fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming) {
+        if let Err(_) = ctx.send_to(self.next, msg.next()) {
+            // do nothing
+        }
+    }
+}
+
+fn main() {
+    let config = RuntimeConfig::default();
+    let rt = Runtime::new(config);
+    let inbox: Inbox<RingMessage> = rt.new_inbox().unwrap();
+
+    let mut next = rt
+        .spawn(RingActor::new(*inbox.addr()))
+        .expect("failed to spawn");
+    let num_passes = 500;
+    for _ in 0..num_passes {
+        let new = rt.spawn(RingActor::new(next)).expect("failed to spawn");
+        next = new;
+    }
+    rt.send_to(next, RingMessage { count: 0 })
+        .expect("failed to start message ring");
+
+    let msg: RingMessage;
+    loop {
+        match inbox.try_recv() {
+            Some(m) => {
+                msg = m;
+                break;
+            }
+            None => {
+                rt.tick();
+            }
+        }
+    }
+    assert_eq!(msg.count, num_passes + 1); // count should equal the number of passes plus the return to main process inbox
+
+    println!("{msg:?}");
+}

src/actor.rs

@@ -0,0 +1,106 @@
+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<T: 'static + Sized + Clone + Send + Sync> 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;
+///     
+///     fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming) {
+///         let response = GreetResponse(format!("Hello, {}!", msg.who).to_string());
+///         if let Ok(_) = ctx.send_to(msg.return_addr, response) {
+///             self.num_greeted += 1;
+///         }
+///     }
+/// }
+/// ```
+pub trait ActorInterface: 'static + Send {
+    type Incoming: Message;
+    type Response: Message;
+    fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming);
+}
+
+/// A unique address for this actor. 32 bytes is overkill for a small application,
+/// but most systems are powerful, and this allows us to create a global map of
+/// actor processes in the future, without worrying about collision.
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct ActorAddress(pub [u8; 32]);
+impl ActorAddress {
+    pub fn new_random() -> Self {
+        let mut bytes = [0u8; 32];
+        get_random(&mut bytes);
+        Self(bytes)
+    }
+}
+
+/// The actor process as represented in the Runtime, with the actor state stored with it's inbox.
+pub(crate) struct Actor<A>
+where
+    A: ActorInterface,
+{
+    inbox: Receiver<A::Incoming>,
+    inner: A,
+}
+
+impl<A: ActorInterface> Actor<A> {
+    pub(crate) fn new(inbox: Receiver<A::Incoming>, inner: A) -> Self {
+        Self { inbox, inner }
+    }
+}
+
+/// Trait for type-erased actors
+pub(crate) trait AnyActor: Send {
+    fn tick(&mut self, ctx: &Runtime);
+}
+
+impl<A> AnyActor for Actor<A>
+where
+    A: ActorInterface,
+{
+    fn tick(&mut self, ctx: &Runtime) {
+        // TODO: WATERLEVEL is hard coded, and so is this message handling scheme. We should
+        // make it so both are more flexible, with sane defaults.
+        let total_messages = self.inbox.len();
+        let messages_to_process = if total_messages < WATERLEVEL {
+            total_messages
+        } else {
+            total_messages >> 1
+        };
+
+        for _ in 0..messages_to_process {
+            match self.inbox.try_recv() {
+                Some(msg) => self.inner.handle(ctx, msg),
+                None => unreachable!(
+                    "We checked number of unprocessed messages in the queue ahead of processing"
+                ),
+            }
+        }
+    }
+}

src/channel.rs

@@ -0,0 +1,76 @@
+use std::{collections::VecDeque, sync::{Arc, Mutex}};
+
+use crossbeam_queue::ArrayQueue;
+
+pub struct HybridChannel<T> {
+    ring: ArrayQueue<T>,
+    overflow: Mutex<VecDeque<T>>,
+}
+
+impl<T> HybridChannel<T> {
+    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<T> {
+        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<T> {
+    queue: Arc<HybridChannel<T>>,
+}
+
+impl<T> Receiver<T> {
+    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<T> {
+        return self.queue.pop();
+    }
+
+    pub fn new_sender(&self) -> Sender<T> {
+        Sender {
+            queue: self.queue.clone(),
+        }
+    }
+}
+
+pub(crate) struct Sender<T> {
+    queue: Arc<HybridChannel<T>>,
+}
+
+impl<T> Sender<T> {
+    pub fn try_send(&self, value: T) -> Result<(), T> {
+        return self.queue.push(value);
+    }
+}
+

src/error.rs

@@ -1,6 +1,19 @@
+/// Simple, ergonomic, local `Error` type.
+/// # Usage
+/// ```
+/// use swactor::Error;
+///
+/// fn foo_if_even(num: u64) -> Result<String, Error> {
+///     if num % 2 == 0 {
+///         return Ok("foo".into());
+///     }
+///     else {
+///         return Err(Error::from("baz"));
+///     }
+/// }
+/// ```
 #[derive(Debug)]
 pub struct Error(Box<dyn std::error::Error + Send + Sync + 'static>);
-pub type Result<T> = std::result::Result<T, Error>;
 pub(crate) fn convert_err<E: std::fmt::Debug>(e: E) -> Error {
     Error(format!("{e:?}").into())
 }

src/lib.rs

@@ -1,251 +1,38 @@
-mod ring_buffer;
+pub mod actor;
 
-use std::collections::HashMap;
+mod channel;
+pub(crate) mod error;
+pub use error::Error;
 
-use crossbeam_queue::ArrayQueue;
+mod router;
-use ring_buffer::{Receiver, Sender};
+pub mod runtime;
-
-pub mod error;
-use error::Error;
 
 #[cfg(feature = "getrandom")]
-pub fn get_random(buf: &mut [u8]) {
+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::<usize>()];
+    }
+}
+
+/// FIXME: remove hard coded defaults
 /// The strategy for message processing is such:
+///
+/// ```ignore
 /// if total_messages < WATERLEVEL:
 ///     process all
 /// else
-///     process total_messages // 2
+///     process total_messages >> 1
+/// ```
 const WATERLEVEL: usize = 10;
-
-const DEFAULT_INBOX_CAPACITY: usize = 100;
-
-pub trait Message: 'static + Sized + Clone + Send {}
-pub type Envelope = Box<dyn std::any::Any + Send>;
-
-pub trait ActorInterface: 'static + Send {
-    type Incoming: Message;
-    type Response: Message;
-    fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming);
-}
-
-pub type ActorAddress = u64;
-
-pub struct Actor<A>
-where
-    A: ActorInterface,
-{
-    _addr: ActorAddress,
-    inbox: Receiver<A::Incoming>,
-    inner: A,
-}
-
-/// Trait for type-erased actors
-trait AnyActor: Send {
-    fn tick(&mut self, ctx: &Runtime);
-}
-
-impl<A> AnyActor for Actor<A>
-where
-    A: ActorInterface,
-{
-    fn tick(&mut self, ctx: &Runtime) {
-        let total_messages = self.inbox.len();
-        let messages_to_process = if total_messages < WATERLEVEL {
-            total_messages
-        } else {
-            total_messages >> 1
-        };
-
-        for _ in 0..messages_to_process {
-            match self.inbox.try_recv() {
-                Some(msg) => self.inner.handle(ctx, msg),
-                None => unreachable!(
-                    "We checked number of unprocessed messages in the queue ahead of processing"
-                ),
-            }
-        }
-    }
-}
-
-pub struct Inbox<M: Message> {
-    addr: ActorAddress,
-    inner: Receiver<M>,
-}
-
-impl<M: Message> Inbox<M> {
-    pub fn addr(&self) -> &ActorAddress {
-        &self.addr
-    }
-
-    pub fn try_recv(&self) -> Option<M> {
-        self.inner.try_recv()
-    }
-}
-
-#[derive(Debug, Default)]
-pub enum RuntimeFlavor {
-    #[default]
-    SingleThreaded,
-    Multithreaded(usize),
-}
-
-pub struct Runtime {
-    flavor: RuntimeFlavor,
-    router: Router,
-    router_inbox: Sender<RouterMessage>,
-    actor_queue: ArrayQueue<Box<dyn AnyActor>>,
-}
-
-impl Runtime {
-    pub fn new(capacity: usize, flavor: Option<RuntimeFlavor>) -> Self {
-        let router = Router::new(DEFAULT_INBOX_CAPACITY);
-        let router_inbox = router.new_sender();
-        Self {
-            flavor: flavor.unwrap_or_default(),
-            router,
-            router_inbox,
-            actor_queue: ArrayQueue::new(capacity),
-        }
-    }
-
-    pub fn spawn<A: ActorInterface>(&self, actor: A) -> Result<ActorAddress, Error> {
-        let addr = {
-            let mut bytes = u64::to_le_bytes(0);
-            get_random(&mut bytes);
-            u64::from_le_bytes(bytes)
-        };
-        let inbox = Receiver::<A::Incoming>::new(DEFAULT_INBOX_CAPACITY);
-        let sender = inbox.new_sender();
-
-        // Register the sender with the router
-        let _ = self
-            .router_inbox
-            .try_send(RouterMessage::AddAddr(addr, Box::new(sender)));
-
-        self.actor_queue
-            .push(Box::new(Actor {
-                _addr: addr,
-                inbox,
-                inner: actor,
-            }))
-            .map_err(|_| Error::from("Runtime error: Failed to spawn actor."))?;
-
-        Ok(addr)
-    }
-
-    pub fn send_to<M: Message>(&self, addr: ActorAddress, msg: M) -> Result<(), ()> {
-        let envelope: Envelope = Box::new(msg);
-        self.router_inbox
-            .try_send(RouterMessage::SendToAddr {
-                addr,
-                msg: envelope,
-            })
-            .map_err(|_| ())
-    }
-
-    pub fn tick(&mut self) {
-        // Pop actor, tick it, push it back
-        if let Some(mut actor) = self.actor_queue.pop() {
-            actor.tick(self);
-            let _ = self.actor_queue.push(actor);
-        }
-
-        match self.flavor {
-            RuntimeFlavor::Multithreaded(_) => (), // router has its own thread
-            RuntimeFlavor::SingleThreaded => self.router.tick(),
-        }
-    }
-
-    pub fn new_inbox<M: Message>(&self) -> Inbox<M> {
-        let addr = {
-            let mut bytes = u64::to_le_bytes(0);
-            get_random(&mut bytes);
-            u64::from_le_bytes(bytes)
-        };
-        let receiver = Receiver::<M>::new(DEFAULT_INBOX_CAPACITY);
-        let sender = receiver.new_sender();
-        // Register the sender with the router
-        let _ = self
-            .router_inbox
-            .try_send(RouterMessage::AddAddr(addr, Box::new(sender)));
-        Inbox {
-            addr,
-            inner: receiver,
-        }
-    }
-}
-
-pub trait SenderT: Send {
-    fn try_send(&self, envelope: Envelope);
-}
-
-impl<M: Message> SenderT for Sender<M> {
-    fn try_send(&self, envelope: Envelope) {
-        if let Ok(msg) = envelope.downcast::<M>() {
-            let _ = Sender::try_send(self, *msg);
-        }
-    }
-}
-
-/// Internal messages for the Router's own inbox
-pub enum RouterMessage {
-    /// register addrs <addr> with sender <sender>
-    AddAddr(ActorAddress, Box<dyn SenderT>),
-    /// remove an actor from the address book
-    RemoveAddr(ActorAddress),
-    /// send <msg> to <addr>
-    SendToAddr { addr: ActorAddress, msg: Envelope },
-}
-
-struct Router {
-    directory: HashMap<ActorAddress, Box<dyn SenderT>>,
-    inbox: Receiver<RouterMessage>,
-}
-
-impl Router {
-    pub fn new(cap: usize) -> Self {
-        Self {
-            directory: HashMap::new(),
-            inbox: Receiver::new(cap),
-        }
-    }
-
-    pub fn tick(&mut self) {
-        let total_messages = self.inbox.len();
-        let messages_to_process = if total_messages < WATERLEVEL {
-            total_messages
-        } else {
-            total_messages >> 1
-        };
-
-        for _ in 0..messages_to_process {
-            match self.inbox.try_recv() {
-                Some(msg) => self.handle(msg),
-                None => unreachable!("We ran checks on total messages before processing."),
-            }
-        }
-    }
-
-    pub fn new_sender(&self) -> Sender<RouterMessage> {
-        self.inbox.new_sender()
-    }
-
-    fn handle(&mut self, msg: RouterMessage) {
-        match msg {
-            RouterMessage::AddAddr(addr, sender) => {
-                self.directory.insert(addr, sender);
-            }
-            RouterMessage::RemoveAddr(addr) => {
-                self.directory.remove(&addr);
-            }
-            RouterMessage::SendToAddr { addr, msg } => {
-                if let Some(sender) = self.directory.get(&addr) {
-                    sender.try_send(msg);
-                }
-            }
-        }
-    }
-}

src/ring_buffer.rs

@@ -1,56 +0,0 @@
-pub use crossbeam_queue::ArrayQueue;
-use std::sync::Arc;
-
-/// The receiving end of a `crossbeam_queue::ArrayQueue`, a lock-free mpsc 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<T> {
-    queue: Arc<ArrayQueue<T>>,
-}
-
-impl<T> Receiver<T> {
-    /// 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<T> {
-        self.queue.pop()
-    }
-
-    /// Construct a new `Sender` assosciated with this queue.
-    pub fn new_sender(&self) -> Sender<T> {
-        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::<T>::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<T> {
-    queue: Arc<ArrayQueue<T>>,
-}
-
-impl<T> Sender<T> {
-    /// 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)
-    }
-}

src/router.rs

@@ -0,0 +1,75 @@
+use std::{collections::HashMap, sync::Arc};
+
+use crate::{
+    actor::{ActorAddress, ActorInterface, Message},
+    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<dyn std::any::Any + Send + Sync>;
+
+pub(crate) trait SenderT: Send + Sync {
+    fn try_send(&self, envelope: Envelope);
+}
+
+impl<M: Message> SenderT for Sender<M> {
+    fn try_send(&self, envelope: Envelope) {
+        if let Some(msg) = envelope.downcast_ref::<M>() {
+            let _ = Sender::try_send(self, msg.clone());
+        }
+    }
+}
+
+/// Internal messages for the Router's own inbox
+#[derive(Clone)]
+pub(crate) enum RouterMessage {
+    /// register addrs <addr> with sender <sender>
+    AddAddr(ActorAddress, Arc<dyn SenderT>),
+
+    /// FIXME: this will be active when we allow actors to shut themselves
+    /// down. For now, disable the warning.
+    #[allow(dead_code)]
+    /// remove an actor from the address book
+    RemoveAddr(ActorAddress),
+
+    /// send <msg> to <addr>
+    SendToAddr { addr: ActorAddress, msg: Envelope },
+}
+
+/// The `Router` is responsible for taking in and delivering all messages in the runtime.
+pub(crate) struct Router {
+    directory: HashMap<ActorAddress, Arc<dyn SenderT>>,
+}
+
+impl Router {
+    pub fn new() -> Self {
+        Self {
+            directory: HashMap::new(),
+        }
+    }
+}
+
+impl ActorInterface for Router {
+    type Incoming = RouterMessage;
+    type Response = ();
+
+    fn handle(&mut self, _ctx: &Runtime, msg: Self::Incoming) {
+        match msg {
+            RouterMessage::AddAddr(addr, sender) => {
+                self.directory.insert(addr, sender);
+            }
+            RouterMessage::RemoveAddr(addr) => {
+                self.directory.remove(&addr);
+            }
+            RouterMessage::SendToAddr { addr, msg } => {
+                if let Some(sender) = self.directory.get(&addr) {
+                    sender.try_send(msg);
+                }
+            }
+        }
+    }
+}

src/runtime.rs

@@ -0,0 +1,260 @@
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::thread::{self, JoinHandle};
+
+use crate::channel::HybridChannel;
+use crate::{
+    actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message},
+    channel::{Receiver, Sender},
+    router::{Router, RouterMessage},
+    Error,
+};
+
+/// Generic message inbox for receiving messages outside of the runtime.
+pub struct Inbox<M: Message> {
+    addr: ActorAddress,
+    inner: Receiver<M>,
+}
+
+impl<M: Message> Inbox<M> {
+    pub fn addr(&self) -> &ActorAddress {
+        &self.addr
+    }
+
+    pub fn try_recv(&self) -> Option<M> {
+        self.inner.try_recv()
+    }
+}
+
+/// The tunable settings for the runtime.
+pub struct RuntimeConfig {
+    pub max_actors: usize,
+    pub router_max_messages: usize,
+    pub actor_max_messages: usize,
+    pub num_threads: usize,
+}
+
+/// 8kB for the `Box<..>` before counting the rest of the memory
+const DEFAULT_MAX_ACTORS: usize = 1_000;
+
+/// 160kB for the `Arc<..>` before counting the rest of the memory
+const DEFAULT_ROUTER_MAX_MESSAGES: usize = 10_000;
+
+/// 16kB PER ACTOR to alloc space for storing the `Arc<..>` pointers
+/// With default setting of [DEFAULT_MAX_ACTORS] this is:
+/// 1_000 * 16kB = 16MB
+const DEFAULT_ACTOR_MAX_MESSAGES: usize = 1_000;
+
+impl Default for RuntimeConfig {
+    fn default() -> Self {
+        Self {
+            max_actors: DEFAULT_MAX_ACTORS,
+            router_max_messages: DEFAULT_ROUTER_MAX_MESSAGES,
+            actor_max_messages: DEFAULT_ACTOR_MAX_MESSAGES,
+            num_threads: 1,
+        }
+    }
+}
+
+/// The `Runtime` struct is the primary gateway for interacting with the framework.
+pub struct Runtime {
+    config: RuntimeConfig,
+    actor_queue: HybridChannel<Box<dyn AnyActor>>,
+    router_interface: Sender<RouterMessage>,
+    router: Option<Actor<Router>>, // `None` if single-threaded
+
+    // for multithreaded contexts
+    is_running: AtomicBool,
+}
+
+/// Handle for dealing with a runtime that has started via the `Runtime::run()` method.
+pub struct RuntimeHandle {
+    pub runtime: Arc<Runtime>,
+    threads: Vec<JoinHandle<()>>,
+}
+
+impl RuntimeHandle {
+    pub fn join(self) {
+        for handle in self.threads {
+            let _ = handle.join();
+        }
+    }
+
+    /// Simple helper, calls the inner `Runtime::shutdown()` method
+    pub fn shutdown(&self) {
+        self.runtime.shutdown();
+    }
+}
+
+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 = 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();
+        let router_inbox: Receiver<RouterMessage> =
+            Receiver::<<Router as ActorInterface>::Incoming>::new(config.router_max_messages);
+        let router_sender = router_inbox.new_sender();
+        let router = Actor::new(router_inbox, router_inner);
+
+        // Single-threaded: router goes in queue. Multi-threaded: stays in Option
+        let router_option = if config.num_threads < 2 {
+            actor_queue
+                .push(Box::new(router) as Box<dyn AnyActor>)
+                .map_err(|_| "failed to add router to actor queue")
+                .expect("failed to spawn router at runtime initialization.");
+            None
+        } else {
+            Some(router)
+        };
+
+        Self {
+            config,
+            actor_queue,
+            router_interface: router_sender,
+            is_running: AtomicBool::new(false),
+            router: router_option,
+        }
+    }
+
+    /// Spawn an actor, returns its address
+    pub fn spawn<A: ActorInterface>(&self, actor: A) -> Result<ActorAddress, Error> {
+        // assign a stochastic
+        let addr = ActorAddress::new_random();
+        let inbox = Receiver::<A::Incoming>::new(self.config.actor_max_messages);
+        let sender = inbox.new_sender();
+
+        // Register the sender with the router
+        self.router_interface
+            .try_send(RouterMessage::AddAddr(addr, Arc::new(sender)))
+            .map_err(|_| {
+                Error::from("Runtime error: failed to add actor to router. Router inbox full")
+            })?;
+
+        self.actor_queue
+            .push(Box::new(Actor::new(inbox, actor)))
+            .map_err(|_| Error::from("Runtime error: Failed to spawn actor. Queue full."))?;
+
+        Ok(addr)
+    }
+
+    /// Send a message to an actor address
+    pub fn send_to<M: Message>(&self, addr: ActorAddress, msg: M) -> Result<(), Error> {
+        self.router_interface
+            .try_send(RouterMessage::SendToAddr {
+                addr,
+                msg: Arc::new(msg),
+            })
+            .map_err(|_| Error::from("Failed to send message to router."))
+    }
+
+    /// Create an external inbox for receiving messages in the outer process containing the runtime
+    pub fn new_inbox<M: Message>(&self) -> Result<Inbox<M>, Error> {
+        let addr = ActorAddress::new_random();
+
+        let receiver = Receiver::<M>::new(self.config.actor_max_messages);
+        let sender = receiver.new_sender();
+
+        // Register the sender with the router
+        self.router_interface
+            .try_send(RouterMessage::AddAddr(addr, Arc::new(sender)))
+            .map_err(|_| {
+                Error::from(
+                    "Runtime error: failed to add a new inbox channel. Router inbox is full.",
+                )
+            })?;
+
+        Ok(Inbox {
+            addr,
+            inner: receiver,
+        })
+    }
+
+    /// Spawn worker threads and start processing, returning a set of handles and
+    /// a Runtime object to interface with.
+    ///
+    /// ### WARN:
+    /// ##### This function panics if the configuration is set as single threaded
+    /// `config.num_threads == 1`
+    pub fn run(mut self) -> Result<RuntimeHandle, Error> {
+        if self.config.num_threads < 2 {
+            return Err(Error::from(
+                "Runtime error: cannot call `Runtime::run()` from a single-threaded context.",
+            ));
+        }
+
+        self.is_running.store(true, Ordering::Release);
+
+        // Take router out before wrapping in Arc - it will be owned by router thread
+        let mut router = self
+            .router
+            .take()
+            .expect("Router must be present for multi-threaded runtime");
+
+        let rt = Arc::new(self);
+        let mut handles: Vec<JoinHandle<()>> = vec![];
+
+        // Router thread owns the router directly - no synchronization needed
+        let router_handle = {
+            let ctx = rt.clone();
+            thread::spawn(move || {
+                while ctx.is_running.load(Ordering::Acquire) {
+                    router.tick(&ctx);
+                    thread::yield_now();
+                }
+            })
+        };
+        handles.push(router_handle);
+
+        // Spawn worker threads
+        let num_workers = rt.config.num_threads - 1;
+        for _ in 0..num_workers {
+            let ctx = rt.clone();
+            let handle = thread::spawn(move || {
+                while ctx.is_running.load(Ordering::Acquire) {
+                    if let Some(mut actor) = ctx.actor_queue.pop() {
+                        actor.tick(&ctx);
+                        // 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();
+                    }
+                }
+            });
+            handles.push(handle);
+        }
+
+        Ok(RuntimeHandle {
+            runtime: rt,
+            threads: handles,
+        })
+    }
+
+    /// Pop the actor off the top of the queue and process it's messages, returning it to the back of
+    /// the queue upon completion.
+    pub fn tick(&self) {
+        if let Some(mut actor) = self.actor_queue.pop() {
+            actor.tick(&self);
+            let _ = self.actor_queue.push(actor);
+        }
+    }
+
+    /// Signal all workers to stop
+    pub fn shutdown(&self) {
+        self.is_running.store(false, Ordering::Release);
+    }
+}

tests/runtime_tests.rs

@@ -0,0 +1,130 @@
+use swactor::{actor::{ActorAddress, ActorInterface}, runtime::{Inbox, Runtime, RuntimeConfig}};
+
+#[derive(Clone)]
+struct PingMessage {
+    reply_to: ActorAddress,
+}
+
+#[derive(Clone)]
+struct PongMessage;
+
+struct PongActor;
+
+impl ActorInterface for PongActor {
+    type Incoming = PingMessage;
+    type Response = PongMessage;
+
+    fn handle(&mut self, ctx: &Runtime, msg: PingMessage) {
+        let _ = ctx.send_to(msg.reply_to, PongMessage);
+    }
+}
+
+/// An actor that forwards messages to another address
+struct ForwarderActor {
+    target: ActorAddress,
+}
+
+#[derive(Clone)]
+struct ForwardMessage(usize);
+
+impl ActorInterface for ForwarderActor {
+    type Incoming = ForwardMessage;
+    type Response = ();
+
+    fn handle(&mut self, ctx: &Runtime, msg: ForwardMessage) {
+        let _ = ctx.send_to(self.target, msg);
+    }
+}
+
+#[test]
+fn test_single_threaded_ping_pong() {
+    let rt = Runtime::new(RuntimeConfig::default());
+    let inbox: Inbox<PongMessage> = rt.new_inbox().unwrap();
+
+    let pong_addr = rt.spawn(PongActor).expect("spawn pong");
+
+    // Send ping
+    rt.send_to(
+        pong_addr,
+        PingMessage {
+            reply_to: *inbox.addr(),
+        },
+    )
+    .unwrap();
+
+    // Tick until we get a response
+    for _ in 0..10 {
+        rt.tick();
+        if inbox.try_recv().is_some() {
+            return; // Success!
+        }
+    }
+
+    panic!("Did not receive pong response");
+}
+
+#[test]
+fn test_single_threaded_message_chain() {
+    let rt = Runtime::new(RuntimeConfig::default());
+    let inbox: Inbox<ForwardMessage> = rt.new_inbox().unwrap();
+
+    // Create a chain: A -> B -> C -> inbox
+    let c_addr = rt
+        .spawn(ForwarderActor {
+            target: *inbox.addr(),
+        })
+        .unwrap();
+    let b_addr = rt.spawn(ForwarderActor { target: c_addr }).unwrap();
+    let a_addr = rt.spawn(ForwarderActor { target: b_addr }).unwrap();
+
+    // Send message to start of chain
+    rt.send_to(a_addr, ForwardMessage(42)).unwrap();
+
+    // Tick until message arrives
+    for _ in 0..20 {
+        rt.tick();
+        if let Some(ForwardMessage(val)) = inbox.try_recv() {
+            assert_eq!(val, 42);
+            return;
+        }
+    }
+
+    panic!("Message did not traverse the chain");
+}
+
+#[test]
+fn test_multithreaded_message_passing() {
+    let config = RuntimeConfig {
+        num_threads: 4,
+        ..Default::default()
+    };
+    let rt = Runtime::new(config);
+    let inbox: Inbox<ForwardMessage> = rt.new_inbox().unwrap();
+
+    // Create a longer chain to exercise multi-threading
+    let mut target = *inbox.addr();
+    for _ in 0..20 {
+        target = rt.spawn(ForwarderActor { target }).unwrap();
+    }
+    let start_addr = target;
+
+    // Send message
+    rt.send_to(start_addr, ForwardMessage(999)).unwrap();
+
+    // Spawn thread to check for result and shutdown
+    let ctx = rt.run().unwrap();
+    let inbox_check = std::thread::spawn(move || {
+        for _ in 0..100 {
+            std::thread::sleep(std::time::Duration::from_millis(10));
+            if let Some(ForwardMessage(val)) = inbox.try_recv() {
+                ctx.shutdown();
+                return Some(val);
+            }
+        }
+        ctx.shutdown();
+        None
+    });
+
+    let result = inbox_check.join().unwrap();
+    assert_eq!(result, Some(999));
+}

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::<Msg>(*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::<Msg>(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::<Msg>(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");
+    }
+}

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<String>,
+}
+
+impl StressResult {
+    pub fn new(name: impl Into<String>) -> 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<String>) {
+        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<Duration>,
+    iterations: Option<u64>,
+}
+
+impl Stress {
+    pub fn new(name: impl Into<String>) -> 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<F>(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;
+    }
+}

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::<Msg>(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::<Msg>(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::<Msg>(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");
+}

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;