From 3c5e3d2ece93367fd26dfe787db677ccd8e9d8ec Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Mon, 26 Jan 2026 14:00:25 +0700 Subject: [PATCH] feat: hybrid channels Add a mutex-locked Dequeue to prevent panics and failures on overflow. --- Cargo.toml | 1 + benches/harness.rs | 31 ------- benches/scaling.rs | 1 - src/actor.rs | 23 +++-- src/channel.rs | 76 ++++++++++++++++ src/error.rs | 2 +- src/lib.rs | 8 +- src/ring_buffer.rs | 57 ------------ src/router.rs | 13 +-- src/runtime.rs | 9 +- tests/stress/concurrency.rs | 11 ++- tests/stress/saturation.rs | 172 +++++------------------------------- 12 files changed, 137 insertions(+), 267 deletions(-) create mode 100644 src/channel.rs delete mode 100644 src/ring_buffer.rs diff --git a/Cargo.toml b/Cargo.toml index b18f9eb..c8963bf 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,7 @@ name = "swactor" version = "0.1.0" edition = "2024" +autobenches = false [lib] crate-type = ["cdylib", "rlib"] diff --git a/benches/harness.rs b/benches/harness.rs index b399a5e..8eddc47 100644 --- a/benches/harness.rs +++ b/benches/harness.rs @@ -197,37 +197,6 @@ impl Bench { elements: self.elements_per_iter, } } - - /// Run the benchmark (no setup between iterations) - pub fn run(self, mut f: F) -> BenchResult - where - F: FnMut(), - { - // Warmup - for _ in 0..self.warmup_iters { - f(); - } - - // Benchmark - let mut times = Vec::with_capacity(self.bench_iters); - let total_start = Instant::now(); - - for _ in 0..self.bench_iters { - let start = Instant::now(); - f(); - times.push(start.elapsed()); - } - - let total_time = total_start.elapsed(); - - BenchResult { - name: self.name, - iterations: self.bench_iters, - total_time, - times, - elements: self.elements_per_iter, - } - } } /// A collection of benchmarks to run together diff --git a/benches/scaling.rs b/benches/scaling.rs index 8443891..e54bf64 100644 --- a/benches/scaling.rs +++ b/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}, diff --git a/src/actor.rs b/src/actor.rs index 2b80d40..3450bee 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -1,34 +1,34 @@ -use crate::{runtime::Runtime, WATERLEVEL, get_random, ring_buffer::Receiver}; +use crate::{WATERLEVEL, channel::Receiver, get_random, runtime::Runtime}; /// The primary trait defining data that can be passed to and from actor processes pub trait Message: 'static + Sized + Clone + Send + Sync {} impl Message for T {} /// The trait that needs to be implemented in order to run a process as an `Actor` -/// +/// /// The `Incoming` type represents `Messages` that can be delivered to the `Actor`. -/// +/// /// The `Response` type represents possible `Messages` the actor may attempt to reply with. -/// +/// /// The `fn handle(..)` is where you implement the logic for handling `Incoming` messages -/// +/// /// # Example /// ``` /// use swactor::{actor::{ActorAddress, ActorInterface}, runtime::Runtime}; -/// +/// /// struct Greeter { /// num_greeted: usize, /// } -/// +/// /// #[derive(Clone)] // required to auto implement `Message` /// struct GreetMessage { /// who: String, /// return_addr: ActorAddress, /// } -/// +/// /// #[derive(Clone)] /// struct GreetResponse(String); -/// +/// /// impl ActorInterface for Greeter { /// type Incoming = GreetMessage; /// type Response = GreetResponse; @@ -71,10 +71,7 @@ where impl Actor { pub(crate) fn new(inbox: Receiver, inner: A) -> Self { - Self { - inbox, - inner, - } + Self { inbox, inner } } } diff --git a/src/channel.rs b/src/channel.rs new file mode 100644 index 0000000..a210a93 --- /dev/null +++ b/src/channel.rs @@ -0,0 +1,76 @@ +use std::{collections::VecDeque, sync::{Arc, Mutex}}; + +use crossbeam_queue::ArrayQueue; + +pub struct HybridChannel { + ring: ArrayQueue, + overflow: Mutex>, +} + +impl HybridChannel { + pub fn new(capacity: usize) -> Self { + Self { + ring: ArrayQueue::new(capacity), + overflow: Mutex::new(VecDeque::new()), + } + } + + pub fn push(&self, value: T) -> Result<(), T> { + match self.ring.push(value) { + Ok(()) => Ok(()), + Err(v) => { + self.overflow.lock().unwrap().push_back(v); + Ok(()) + } + } + } + + pub fn pop(&self) -> Option { + if let Some(value) = self.ring.pop() { + return Some(value); + } + + self.overflow.lock().unwrap().pop_front() + } + + pub fn len(&self) -> usize { + self.ring.len() + self.overflow.lock().unwrap().len() + } +} + +pub(crate) struct Receiver { + queue: Arc>, +} + +impl Receiver { + pub fn new(capacity: usize) -> Self { + let queue = Arc::new(HybridChannel::new(capacity)); + + Self { queue } + } + + pub fn len(&self) -> usize { + self.queue.len() + } + + pub fn try_recv(&self) -> Option { + return self.queue.pop(); + } + + pub fn new_sender(&self) -> Sender { + Sender { + queue: self.queue.clone(), + } + } +} + +pub(crate) struct Sender { + queue: Arc>, +} + +impl Sender { + pub fn try_send(&self, value: T) -> Result<(), T> { + return self.queue.push(value); + } +} + diff --git a/src/error.rs b/src/error.rs index 12183e2..9aa626a 100644 --- a/src/error.rs +++ b/src/error.rs @@ -2,7 +2,7 @@ /// # Usage /// ``` /// use swactor::Error; -/// +/// /// fn foo_if_even(num: u64) -> Result { /// if num % 2 == 0 { /// return Ok("foo".into()); diff --git a/src/lib.rs b/src/lib.rs index 9403a29..1d67009 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,9 +1,9 @@ pub mod actor; +mod channel; pub(crate) mod error; pub use error::Error; -mod ring_buffer; mod router; pub mod runtime; @@ -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::()]; } diff --git a/src/ring_buffer.rs b/src/ring_buffer.rs deleted file mode 100644 index b016b3e..0000000 --- a/src/ring_buffer.rs +++ /dev/null @@ -1,57 +0,0 @@ -//! Shallow wrapper around the `crossbeam_queue::ArrayQueue` implementation of a mpmc ring buffer. -use std::sync::Arc; -pub use crossbeam_queue::ArrayQueue; - -/// The receiving end of a `crossbeam_queue::ArrayQueue`, a lock-free mpmc queue. -/// The queue is constructed by the `Receiver::new()` method. -/// Responsible for creating the `Sender` ends of itself. -/// -/// Notably: The `Receiver` provides no guarentees that a sending end of the channel exists. -pub(crate) struct Receiver { - queue: Arc>, -} - -impl Receiver { - /// Constructs a new `ArrayQueue` with given capacity. - /// - /// # Panics - /// Will panic if capacity is passed as 0 - pub fn new(capacity: usize) -> Self { - Self { - queue: Arc::new(ArrayQueue::new(capacity)), - } - } - - /// Returns the number of elements in the inner queue - pub fn len(&self) -> usize { - self.queue.len() - } - - /// Attempt to retrieve a value from the queue. Returns `None` if empty - pub fn try_recv(&self) -> Option { - self.queue.pop() - } - - /// Construct a new `Sender` assosciated with this queue. - pub fn new_sender(&self) -> Sender { - Sender { - queue: self.queue.clone(), - } - } -} - -/// The sending end of a `crossbeam_queue::ArrayQueue`, a lock free mpsc queue. -/// The queue is initialized via calling the corresponding `Receiver::::new()` method, -/// and the sending end of the queue is constructed via calling `receiver.new_sender()`. -/// -/// Notably: The `Sender` provides no guarentees that a receiving end of the channel exists. -pub(crate) struct Sender { - queue: Arc>, -} - -impl Sender { - /// Attempt to push a value to the queue. Returns Err(value) if the queue is full. - pub fn try_send(&self, value: T) -> Result<(), T> { - self.queue.push(value) - } -} diff --git a/src/router.rs b/src/router.rs index 82fb10a..55296f0 100644 --- a/src/router.rs +++ b/src/router.rs @@ -2,12 +2,13 @@ use std::{collections::HashMap, sync::Arc}; use crate::{ actor::{ActorAddress, ActorInterface, Message}, - ring_buffer::Sender, runtime::Runtime, + channel::Sender, + runtime::Runtime, }; /// FIXME: Go over with a fine-toothed comb and reassure yourself this typing /// makes sense, that we are not doing loads of indirection on a hot path. -/// +/// /// A type erased `Message` to be routed between actor processes. pub(crate) type Envelope = Arc; @@ -60,13 +61,15 @@ impl ActorInterface for Router { match msg { RouterMessage::AddAddr(addr, sender) => { self.directory.insert(addr, sender); - }, - RouterMessage::RemoveAddr(addr) => { self.directory.remove(&addr); }, + } + RouterMessage::RemoveAddr(addr) => { + self.directory.remove(&addr); + } RouterMessage::SendToAddr { addr, msg } => { if let Some(sender) = self.directory.get(&addr) { sender.try_send(msg); } - }, + } } } } diff --git a/src/runtime.rs b/src/runtime.rs index ebf7b6b..7a9f768 100644 --- a/src/runtime.rs +++ b/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>, + actor_queue: HybridChannel>, router_interface: Sender, router: Option>, // `None` if single-threaded @@ -91,7 +90,7 @@ impl Runtime { /// Builds a new `Runtime` struct, but does not yet run anything. If multithreaded, call /// `run()`, if single threaded, needs to be driven by calls to the `tick()` method. pub fn new(config: RuntimeConfig) -> Self { - let actor_queue = ArrayQueue::new(config.max_actors); + let actor_queue = HybridChannel::new(config.max_actors); // router is a unique actor in that the runtime needs access to it's `Sender` handle let router_inner = Router::new(); diff --git a/tests/stress/concurrency.rs b/tests/stress/concurrency.rs index 654300e..2511745 100644 --- a/tests/stress/concurrency.rs +++ b/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] diff --git a/tests/stress/saturation.rs b/tests/stress/saturation.rs index 6301ae3..a8fc97e 100644 --- a/tests/stress/saturation.rs +++ b/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::(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::(sink, Msg).is_ok() { - burst_sent += 1; - } else { - burst_failed += 1; - } - } - total_sent += burst_sent; - total_failed += burst_failed; - - // Recovery: process until queue is drained - let recovery_start = std::time::Instant::now(); - for _ in 0..5000 { - runtime.tick(); - } - let recovery_time = recovery_start.elapsed(); - - println!( - " Cycle {}: sent={}, failed={}, recovery={:?}", - cycle + 1, - burst_sent, - burst_failed, - recovery_time - ); - } - - println!( - "\n Total sent: {}, Total failed: {}", - total_sent, total_failed - ); - println!(">>> Burst traffic test complete\n"); -} - -/// Find the message drop cliff - at what load factor do drops spike? -#[test] -#[cfg(feature = "stress")] -fn message_drop_curve() { - println!("\n>>> STRESS: Message Drop Curve"); - println!(" Testing drop rate at various load factors...\n"); - - // Test at different load factors (messages per tick) - for msgs_per_tick in [1, 5, 10, 20, 50, 100] { - let config = RuntimeConfig { - max_actors: 10, - router_max_messages: 1000, - actor_max_messages: 500, - num_threads: 1, - }; - let runtime = Runtime::new(config); - let sink = runtime.spawn(BlackHole).unwrap(); - - // Warmup - for _ in 0..10 { - runtime.tick(); - } - - let mut sent = 0u64; - let mut failed = 0u64; - - // Run for fixed iterations - for _ in 0..100 { - // Send burst - for _ in 0..msgs_per_tick { - if runtime.send_to::(sink, Msg).is_ok() { - sent += 1; - } else { - failed += 1; - } - } - // Process one tick - runtime.tick(); - } - - let drop_rate = if sent + failed > 0 { - (failed as f64 / (sent + failed) as f64) * 100.0 - } else { - 0.0 - }; - - println!( - " msgs/tick={:3} sent={:5} failed={:5} drop_rate={:.1}%", - msgs_per_tick, sent, failed, drop_rate - ); - } - - println!("\n>>> Message drop curve test complete\n"); -}