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"]

benches/harness.rs

@@ -197,37 +197,6 @@ impl Bench {
             elements: self.elements_per_iter,
         }
     }
-
-    /// Run the benchmark (no setup between iterations)
-    pub fn run<F>(self, mut f: F) -> BenchResult
-    where
-        F: FnMut(),
-    {
-        // Warmup
-        for _ in 0..self.warmup_iters {
-            f();
-        }
-
-        // Benchmark
-        let mut times = Vec::with_capacity(self.bench_iters);
-        let total_start = Instant::now();
-
-        for _ in 0..self.bench_iters {
-            let start = Instant::now();
-            f();
-            times.push(start.elapsed());
-        }
-
-        let total_time = total_start.elapsed();
-
-        BenchResult {
-            name: self.name,
-            iterations: self.bench_iters,
-            total_time,
-            times,
-            elements: self.elements_per_iter,
-        }
-    }
 }
 
 /// A collection of benchmarks to run together

benches/scaling.rs

@@ -7,7 +7,6 @@
 
 use crate::harness::{black_box, Bench, BenchSuite};
 use std::thread;
-use std::time::Duration;
 use swactor::{
     actor::ActorInterface,
     runtime::{Runtime, RuntimeConfig},

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<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;
@@ -71,10 +71,7 @@ where
 
 impl<A: ActorInterface> Actor<A> {
     pub(crate) fn new(inbox: Receiver<A::Incoming>, inner: A) -> Self {
-        Self {
-            inbox,
-            inner,
-        }
+        Self { inbox, inner }
     }
 }
 

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

@@ -2,7 +2,7 @@
 /// # Usage
 /// ```
 /// use swactor::Error;
-/// 
+///
 /// fn foo_if_even(num: u64) -> Result<String, Error> {
 ///     if num % 2 == 0 {
 ///         return Ok("foo".into());

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;
 
@@ -15,12 +15,12 @@ pub(crate) fn get_random(buf: &mut [u8]) {
 #[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>()];
     }

src/ring_buffer.rs

@@ -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<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

@@ -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<dyn std::any::Any + Send + Sync>;
 
@@ -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);
                 }
-            },
+            }
         }
     }
 }

src/runtime.rs

@@ -2,11 +2,10 @@ 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::{
     actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message},
-    ring_buffer::{Receiver, Sender},
+    channel::{Receiver, Sender},
     router::{Router, RouterMessage},
     Error,
 };
@@ -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<Box<dyn AnyActor>>,
+    actor_queue: HybridChannel<Box<dyn AnyActor>>,
     router_interface: Sender<RouterMessage>,
     router: Option<Actor<Router>>, // `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();

tests/stress/concurrency.rs

@@ -127,7 +127,15 @@ fn send_to_newborn() {
     println!("  Total spawned:    {}", total_spawned);
     println!("  Sends succeeded:  {}", total_send_ok);
     println!("  Sends failed:     {}", total_send_fail);
-    println!("  Note: Failures expected - message may arrive before registration");
+
+    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");
 }
 
@@ -258,6 +266,7 @@ fn inbox_contention() {
     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]

tests/stress/saturation.rs

@@ -6,9 +6,7 @@ use super::{BlackHole, Counter, Msg, Stress, StressResult};
 use std::time::Duration;
 use swactor::runtime::{Runtime, RuntimeConfig};
 
-/// FIXME: does this even make sense to test?
-/// Blast the router inbox until it overflows.
-/// Documents: What happens when router can't keep up?
+/// Blast the router inbox
 #[test]
 #[cfg(feature = "stress")]
 fn router_inbox_overflow() {
@@ -38,20 +36,14 @@ fn router_inbox_overflow() {
 
     result.duration = start.elapsed();
     result.note(format!("Router buffer: 100, Messages sent: 10,000"));
-    result.note(format!(
-        "Expected: ~99% failure rate (buffer fills immediately)"
-    ));
-    result.print();
 
-    // Verify we actually saw failures
-    assert!(result.failures > 0, "Expected router to reject messages");
-    println!(">>> PASS: Router correctly rejects messages when full\n");
+    // 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");
 }
 
-/// FIXME: This test makes no sense until we make sure panics happen when
-/// actor inbox buffers are full.
-/// Blast a single actor's inbox until it overflows.
-/// Documents: What happens when actor can't keep up?
+/// Blast a single actor's inbox
 #[test]
 #[cfg(feature = "stress")]
 fn actor_inbox_overflow() {
@@ -64,16 +56,19 @@ fn actor_inbox_overflow() {
         num_threads: 1,
     };
     let runtime = Runtime::new(config);
-    let sink = runtime.spawn(BlackHole).unwrap();
+    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 {
@@ -81,18 +76,19 @@ fn actor_inbox_overflow() {
         }
     }
 
-    // The router accepted messages, but many were dropped at actor inbox
-    // We can't easily count these drops from outside, but we can document the behavior
-    println!("  Router accepted {} messages", sent);
-    println!("  Actor inbox capacity: 100");
-    println!("  Note: Messages beyond inbox capacity are silently dropped");
-    println!(">>> This is a known limitation - bounded queues drop overflow\n");
+    // 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");
 }
 
-/// FIXME: Does it make sense to have this as a test? Yes the runtime
-/// fails if you try and spawn actors when the queue is full.
-/// Spawn actors until the queue rejects.
-/// Documents: What happens when actor queue fills?
+/// Blast the runtime with actor spawns
 #[test]
 #[cfg(feature = "stress")]
 fn actor_queue_overflow() {
@@ -120,17 +116,13 @@ fn actor_queue_overflow() {
 
     result.duration = start.elapsed();
     result.note(format!("Queue capacity: 100, Spawn attempts: 500"));
-    result.note(format!(
-        "Expected: ~80% failure rate (queue fills after ~100)"
-    ));
     result.print();
 
     // Note: Router also takes a slot, so we expect ~99 actors max
-    assert!(
-        result.successes <= 100,
+    assert_eq!(
+        result.failures, 0,
         "Spawned more actors than queue capacity"
     );
-    assert!(result.failures > 0, "Expected spawn failures");
     println!(">>> PASS: Actor queue correctly rejects when full\n");
 }
 
@@ -179,121 +171,3 @@ fn sustained_overload() {
     result.print();
     println!(">>> System survived sustained overload without panic\n");
 }
-
-/// FIXME: The logic for timing recovery does not make sense
-/// Burst traffic - idle to 100x normal, back to idle.
-/// Documents: Recovery behavior after traffic spikes.
-#[test]
-#[cfg(feature = "stress")]
-fn burst_traffic() {
-    println!("\n>>> STRESS: Burst Traffic");
-
-    let config = RuntimeConfig {
-        max_actors: 100,
-        router_max_messages: 10_000,
-        actor_max_messages: 1000,
-        num_threads: 1,
-    };
-    let runtime = Runtime::new(config);
-    let sink = runtime.spawn(Counter::new()).unwrap();
-
-    // Process registration
-    for _ in 0..10 {
-        runtime.tick();
-    }
-
-    let mut total_sent = 0u64;
-    let mut total_failed = 0u64;
-
-    // 5 burst cycles
-    for cycle in 0..5 {
-        // Burst: send 1000 messages as fast as possible
-        let mut burst_sent = 0;
-        let mut burst_failed = 0;
-        for _ in 0..1000 {
-            if runtime.send_to::<Msg>(sink, Msg).is_ok() {
-                burst_sent += 1;
-            } else {
-                burst_failed += 1;
-            }
-        }
-        total_sent += burst_sent;
-        total_failed += burst_failed;
-
-        // Recovery: process until queue is drained
-        let recovery_start = std::time::Instant::now();
-        for _ in 0..5000 {
-            runtime.tick();
-        }
-        let recovery_time = recovery_start.elapsed();
-
-        println!(
-            "  Cycle {}: sent={}, failed={}, recovery={:?}",
-            cycle + 1,
-            burst_sent,
-            burst_failed,
-            recovery_time
-        );
-    }
-
-    println!(
-        "\n  Total sent: {}, Total failed: {}",
-        total_sent, total_failed
-    );
-    println!(">>> Burst traffic test complete\n");
-}
-
-/// Find the message drop cliff - at what load factor do drops spike?
-#[test]
-#[cfg(feature = "stress")]
-fn message_drop_curve() {
-    println!("\n>>> STRESS: Message Drop Curve");
-    println!("  Testing drop rate at various load factors...\n");
-
-    // Test at different load factors (messages per tick)
-    for msgs_per_tick in [1, 5, 10, 20, 50, 100] {
-        let config = RuntimeConfig {
-            max_actors: 10,
-            router_max_messages: 1000,
-            actor_max_messages: 500,
-            num_threads: 1,
-        };
-        let runtime = Runtime::new(config);
-        let sink = runtime.spawn(BlackHole).unwrap();
-
-        // Warmup
-        for _ in 0..10 {
-            runtime.tick();
-        }
-
-        let mut sent = 0u64;
-        let mut failed = 0u64;
-
-        // Run for fixed iterations
-        for _ in 0..100 {
-            // Send burst
-            for _ in 0..msgs_per_tick {
-                if runtime.send_to::<Msg>(sink, Msg).is_ok() {
-                    sent += 1;
-                } else {
-                    failed += 1;
-                }
-            }
-            // Process one tick
-            runtime.tick();
-        }
-
-        let drop_rate = if sent + failed > 0 {
-            (failed as f64 / (sent + failed) as f64) * 100.0
-        } else {
-            0.0
-        };
-
-        println!(
-            "  msgs/tick={:3}  sent={:5}  failed={:5}  drop_rate={:.1}%",
-            msgs_per_tick, sent, failed, drop_rate
-        );
-    }
-
-    println!("\n>>> Message drop curve test complete\n");
-}