From 635dfc12a39ac071a896b9999c6e1ca745f5da97 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Tue, 10 Feb 2026 14:33:17 +0700 Subject: [PATCH] fix: stats datatypes and refactor channel signature --- Cargo.toml | 4 - benches/mt_benchmarks.rs | 2 +- benches/runtime_benchmarks.rs | 2 +- benches/worker_benchmarks.rs | 61 --- .../examples/bench_dashboard.rs | 2 +- .../examples/dashboard_demo.rs | 2 +- .../runtime-dashboard/examples/record_demo.rs | 2 +- crates/swactor-gossip/src/sim.rs | 4 +- crates/swactor-python/src/lib.rs | 10 +- src/actor.rs | 13 +- src/channel.rs | 14 +- src/config.rs | 11 +- src/delivery.rs | 2 +- src/error.rs | 4 +- src/runtime.rs | 48 +- src/stats.rs | 23 +- src/worker.rs | 59 ++- tests/runtime_api.rs | 492 +++++++++++++++++- tests/test_python.py | 2 +- 19 files changed, 591 insertions(+), 166 deletions(-) delete mode 100644 benches/worker_benchmarks.rs diff --git a/Cargo.toml b/Cargo.toml index 9d83c9d..0200dd9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -37,10 +37,6 @@ criterion = { version = "0.5", features = ["html_reports"] } name = "runtime_benchmarks" harness = false -[[bench]] -name = "worker_benchmarks" -harness = false - [[bench]] name = "mt_benchmarks" harness = false diff --git a/benches/mt_benchmarks.rs b/benches/mt_benchmarks.rs index 9fe2e47..2d7979a 100644 --- a/benches/mt_benchmarks.rs +++ b/benches/mt_benchmarks.rs @@ -15,7 +15,7 @@ fn mt_config(threads: usize, max_actors: usize, max_messages: usize) -> RuntimeC RuntimeConfig { num_threads: threads, max_actors, - actor_max_messages: max_messages, + channel_buffer_size: max_messages, backoff_policy: BackoffPolicy { spin_threshold: 32, yield_threshold: 64, diff --git a/benches/runtime_benchmarks.rs b/benches/runtime_benchmarks.rs index 64eff02..06ad48b 100644 --- a/benches/runtime_benchmarks.rs +++ b/benches/runtime_benchmarks.rs @@ -14,7 +14,7 @@ use swactor::{ fn make_config(max_actors: usize, max_messages: usize) -> RuntimeConfig { RuntimeConfig { max_actors, - actor_max_messages: max_messages, + channel_buffer_size: max_messages, num_threads: 1, ..Default::default() } diff --git a/benches/worker_benchmarks.rs b/benches/worker_benchmarks.rs deleted file mode 100644 index d6ed8cd..0000000 --- a/benches/worker_benchmarks.rs +++ /dev/null @@ -1,61 +0,0 @@ -use std::collections::VecDeque; - -use criterion::{ - criterion_group, criterion_main, BenchmarkId, Criterion, Throughput, -}; - -// --------------------------------------------------------------------------- -// VecDeque push throughput (mirrors old mailbox_push) -// --------------------------------------------------------------------------- - -fn vecdeque_push(c: &mut Criterion) { - let mut group = c.benchmark_group("vecdeque_push"); - for n in [100, 1_000, 10_000] { - group.throughput(Throughput::Elements(n as u64)); - group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| { - b.iter(|| { - let mut q: VecDeque = VecDeque::new(); - for i in 0..n { - q.push_back(i as u64); - } - }); - }); - } - group.finish(); -} - -// --------------------------------------------------------------------------- -// VecDeque pop throughput (mirrors old mailbox_pop) -// --------------------------------------------------------------------------- - -fn vecdeque_pop(c: &mut Criterion) { - let mut group = c.benchmark_group("vecdeque_pop"); - for n in [100, 1_000, 10_000] { - group.throughput(Throughput::Elements(n as u64)); - group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| { - b.iter_batched( - || { - let mut q: VecDeque = VecDeque::new(); - for i in 0..n { - q.push_back(i as u64); - } - q - }, - |mut q| { - for _ in 0..n { - std::hint::black_box(q.pop_front()); - } - }, - criterion::BatchSize::SmallInput, - ); - }); - } - group.finish(); -} - -criterion_group!( - benches, - vecdeque_push, - vecdeque_pop, -); -criterion_main!(benches); diff --git a/crates/runtime-dashboard/examples/bench_dashboard.rs b/crates/runtime-dashboard/examples/bench_dashboard.rs index e398f2a..e249241 100644 --- a/crates/runtime-dashboard/examples/bench_dashboard.rs +++ b/crates/runtime-dashboard/examples/bench_dashboard.rs @@ -79,7 +79,7 @@ fn bench_config(threads: usize, max_actors: usize, max_messages: usize) -> Runti RuntimeConfig { num_threads: threads, max_actors, - actor_max_messages: max_messages, + channel_buffer_size: max_messages, backoff_policy: BackoffPolicy { spin_threshold: 32, yield_threshold: 64, diff --git a/crates/runtime-dashboard/examples/dashboard_demo.rs b/crates/runtime-dashboard/examples/dashboard_demo.rs index 3e35a60..649853a 100644 --- a/crates/runtime-dashboard/examples/dashboard_demo.rs +++ b/crates/runtime-dashboard/examples/dashboard_demo.rs @@ -84,7 +84,7 @@ fn main() { let rt = Runtime::new(RuntimeConfig { num_threads: 4, max_actors: 1024, - actor_max_messages: 2000, + channel_buffer_size: 2000, ..Default::default() }); diff --git a/crates/runtime-dashboard/examples/record_demo.rs b/crates/runtime-dashboard/examples/record_demo.rs index 3d424fb..99cd5b4 100644 --- a/crates/runtime-dashboard/examples/record_demo.rs +++ b/crates/runtime-dashboard/examples/record_demo.rs @@ -59,7 +59,7 @@ fn main() { let rt = Runtime::new(RuntimeConfig { num_threads: 4, max_actors: 512, - actor_max_messages: 1000, + channel_buffer_size: 1000, ..Default::default() }); diff --git a/crates/swactor-gossip/src/sim.rs b/crates/swactor-gossip/src/sim.rs index 9e1846f..74c0b77 100644 --- a/crates/swactor-gossip/src/sim.rs +++ b/crates/swactor-gossip/src/sim.rs @@ -64,7 +64,7 @@ fn run_simulation_single_threaded(config: SimConfig) -> SimulationTrace { let rt = Runtime::new(RuntimeConfig { num_threads: 1, max_actors: (config.num_nodes + 64).next_power_of_two(), - actor_max_messages: (config.num_nodes * 4).max(1_000), + channel_buffer_size: (config.num_nodes * 4).max(1_000), ..Default::default() }); @@ -177,7 +177,7 @@ fn run_simulation_multi_threaded(config: SimConfig) -> SimulationTrace { let rt = Runtime::new(RuntimeConfig { num_threads: config.num_threads, max_actors: (config.num_nodes + 64).next_power_of_two(), - actor_max_messages: (config.num_nodes * 4).max(1_000), + channel_buffer_size: (config.num_nodes * 4).max(1_000), ..Default::default() }); diff --git a/crates/swactor-python/src/lib.rs b/crates/swactor-python/src/lib.rs index f2c47e6..f978d44 100644 --- a/crates/swactor-python/src/lib.rs +++ b/crates/swactor-python/src/lib.rs @@ -225,7 +225,7 @@ pub struct PyRuntimeConfig { #[pyo3(get, set)] max_actors: usize, #[pyo3(get, set)] - actor_max_messages: usize, + channel_buffer_size: usize, #[pyo3(get, set)] spin_threshold: u32, #[pyo3(get, set)] @@ -243,7 +243,7 @@ impl PyRuntimeConfig { *, num_threads = 1, max_actors = 1_000, - actor_max_messages = 1_000, + channel_buffer_size = 1_000, spin_threshold = 64, yield_threshold = 256, sleep_increment_us = 50, @@ -252,7 +252,7 @@ impl PyRuntimeConfig { fn new( num_threads: usize, max_actors: usize, - actor_max_messages: usize, + channel_buffer_size: usize, spin_threshold: u32, yield_threshold: u32, sleep_increment_us: u64, @@ -261,7 +261,7 @@ impl PyRuntimeConfig { Self { num_threads, max_actors, - actor_max_messages, + channel_buffer_size, spin_threshold, yield_threshold, sleep_increment_us, @@ -275,7 +275,7 @@ impl From for RuntimeConfig { RuntimeConfig { num_threads: py.num_threads, max_actors: py.max_actors, - actor_max_messages: py.actor_max_messages, + channel_buffer_size: py.channel_buffer_size, backoff_policy: BackoffPolicy { spin_threshold: py.spin_threshold, yield_threshold: py.yield_threshold, diff --git a/src/actor.rs b/src/actor.rs index a083dac..2c3955f 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -45,17 +45,22 @@ impl Actor { } /// Trait for type-erased actors — single-message handler. +/// +/// Returns `true` if the message was handled, `false` on type mismatch. pub trait AnyActor: Send { - fn handle_any(&mut self, ctx: &Ctx, msg: Box); + fn handle_any(&mut self, ctx: &Ctx, msg: Box) -> bool; } impl AnyActor for Actor where A: ActorInterface, { - fn handle_any(&mut self, ctx: &Ctx, msg: Box) { + fn handle_any(&mut self, ctx: &Ctx, msg: Box) -> bool { if let Ok(typed) = msg.downcast::() { self.0.handle(ctx, *typed); + true + } else { + false } } } @@ -63,7 +68,7 @@ where /// Object-safe inner trait for sending type-erased messages. pub trait ContextInner { fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error>; - fn spawn_any(&self, addr: ActorAddress, actor: Box) -> Result<(), Error>; + fn spawn_any(&self, addr: ActorAddress, actor: Box); } /// Actor syscall interface — passed to `ActorInterface::handle()`. @@ -98,7 +103,7 @@ impl<'a> Ctx<'a> { pub fn spawn(&self, actor: A) -> Result { let addr = ActorAddress::new_random(); let boxed: Box = Box::new(Actor::new(actor)); - self.inner.spawn_any(addr, boxed)?; + self.inner.spawn_any(addr, boxed); Ok(addr) } } diff --git a/src/channel.rs b/src/channel.rs index 76da28c..4860704 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -16,17 +16,13 @@ impl HybridChannel { } } - pub fn push(&self, value: T) -> Result<(), T> { + pub fn push(&self, value: T) { if !self.overflow.is_empty() { self.overflow.push(value); - return Ok(()); + return; } - match self.ring.push(value) { - Ok(()) => Ok(()), - Err(v) => { - self.overflow.push(v); - Ok(()) - } + if let Err(v) = self.ring.push(value) { + self.overflow.push(v); } } @@ -61,7 +57,7 @@ pub(crate) struct Sender { } impl Sender { - pub fn try_send(&self, value: T) -> Result<(), T> { + pub fn send(&self, value: T) { self.queue.push(value) } } diff --git a/src/config.rs b/src/config.rs index f0bf07a..ef8c9b7 100644 --- a/src/config.rs +++ b/src/config.rs @@ -26,7 +26,7 @@ impl Default for BackoffPolicy { /// The tunable settings for the runtime. pub struct RuntimeConfig { pub max_actors: usize, - pub actor_max_messages: usize, + pub channel_buffer_size: usize, pub num_threads: usize, pub backoff_policy: BackoffPolicy, } @@ -34,16 +34,15 @@ pub struct RuntimeConfig { /// 8kB for the `Box<..>` before counting the rest of the memory const DEFAULT_MAX_ACTORS: usize = 1_000; -/// 16kB PER ACTOR to alloc space for storing messages. -/// With default setting of [DEFAULT_MAX_ACTORS] this is: -/// 1_000 * 16kB = 16MB -const DEFAULT_ACTOR_MAX_MESSAGES: usize = 1_000; +/// Pre-allocated ring buffer capacity for each channel (transfer, spawn, inbox). +/// When the ring is full, messages overflow into an unbounded backup queue. +const DEFAULT_CHANNEL_BUFFER_SIZE: usize = 1_000; impl Default for RuntimeConfig { fn default() -> Self { Self { max_actors: DEFAULT_MAX_ACTORS, - actor_max_messages: DEFAULT_ACTOR_MAX_MESSAGES, + channel_buffer_size: DEFAULT_CHANNEL_BUFFER_SIZE, num_threads: 1, backoff_policy: BackoffPolicy::default(), } diff --git a/src/delivery.rs b/src/delivery.rs index 384916a..f6ca3a8 100644 --- a/src/delivery.rs +++ b/src/delivery.rs @@ -105,7 +105,7 @@ pub(crate) trait SenderT: Send + Sync { impl SenderT for Sender { fn try_send_any(&self, msg: Box) { if let Ok(typed) = msg.downcast::() { - let _ = Sender::try_send(self, *typed); + Sender::send(self, *typed); } } } diff --git a/src/error.rs b/src/error.rs index 39a444b..3294a58 100644 --- a/src/error.rs +++ b/src/error.rs @@ -17,12 +17,12 @@ pub struct Error(Box); impl> From for Error { fn from(value: T) -> Self { - Error(format!("{:?}", value.as_ref()).into()) + Error(value.as_ref().to_string().into()) } } impl std::fmt::Display for Error { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{:?}", self.0) + write!(f, "{}", self.0) } } diff --git a/src/runtime.rs b/src/runtime.rs index 0fc4f94..202295a 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -101,7 +101,7 @@ impl Runtime { let mut workers = Vec::with_capacity(num_workers); for i in 0..num_workers { - let transfer_rx = Receiver::::new(config.actor_max_messages); + let transfer_rx = Receiver::::new(config.channel_buffer_size); let transfer_tx = transfer_rx.new_sender(); transfer_txs.push(transfer_tx); @@ -151,8 +151,7 @@ impl Runtime { self.address_map.insert(addr, worker_id); let boxed: Box = Box::new(Actor::new(actor)); self.spawn_txs[worker_id.as_usize()] - .try_send((addr, boxed)) - .map_err(|_| Error::from("Runtime error: spawn queue full"))?; + .send((addr, boxed)); #[cfg(feature = "tracing")] tracing::info!( @@ -177,7 +176,7 @@ impl Runtime { /// Create an external inbox for receiving messages in the outer process containing the runtime pub fn new_inbox(&self) -> Result, Error> { let addr = ActorAddress::new_random(); - let receiver = Receiver::::new(self.config.actor_max_messages); + let receiver = Receiver::::new(self.config.channel_buffer_size); let sender = receiver.new_sender(); self.inbox_registry.register(addr, Arc::new(sender)); Ok(Inbox { @@ -296,24 +295,6 @@ impl Runtime { self.transport_router = Some(router); } - /// Route a message whose destination is not in the local address map. - fn route_nonlocal( - &self, - addr: ActorAddress, - msg: Box, - ) -> Result<(), Error> { - #[cfg(feature = "transport")] - { - if self.inbox_registry.contains(&addr) { - return self.inbox_registry.try_deliver(addr, msg); - } - if let (Some(cr), Some(tr)) = (&self.codec_registry, &self.transport_router) { - return crate::transport::send_via_transport(addr, msg, cr, tr); - } - } - self.inbox_registry.try_deliver(addr, msg) - } - /// Deliver a raw deserialized message into the runtime. /// /// Used by [`CodecRegistry::receive`](crate::transport::CodecRegistry::receive) @@ -325,9 +306,11 @@ impl Runtime { msg: Box, ) -> Result<(), Error> { match self.address_map.lookup(&addr) { - Some(wid) => self.transfer_txs[wid.as_usize()] - .try_send(Envelope::new(addr, msg)) - .map_err(|_| Error::from("Transfer queue full")), + Some(wid) => { + self.transfer_txs[wid.as_usize()] + .send(Envelope::new(addr, msg)); + Ok(()) + } None => self.inbox_registry.try_deliver(addr, msg), } } @@ -336,18 +319,19 @@ impl Runtime { impl ContextInner for Runtime { fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error> { match self.address_map.lookup(&addr) { - Some(wid) => self.transfer_txs[wid.as_usize()] - .try_send(Envelope::new(addr, msg)) - .map_err(|_| Error::from("Transfer queue full")), - None => self.route_nonlocal(addr, msg), + Some(wid) => { + self.transfer_txs[wid.as_usize()] + .send(Envelope::new(addr, msg)); + Ok(()) + } + None => self.make_tick_context().route_nonlocal(addr, msg), } } - fn spawn_any(&self, addr: ActorAddress, actor: Box) -> Result<(), Error> { + fn spawn_any(&self, addr: ActorAddress, actor: Box) { let worker_id = self.placement.next_worker(); self.address_map.insert(addr, worker_id); self.spawn_txs[worker_id.as_usize()] - .try_send((addr, actor)) - .map_err(|_| Error::from("Spawn queue full")) + .send((addr, actor)); } } diff --git a/src/stats.rs b/src/stats.rs index 1d542d1..d0a1764 100644 --- a/src/stats.rs +++ b/src/stats.rs @@ -1,6 +1,7 @@ -use std::collections::VecDeque; use std::sync::atomic::{AtomicU64, AtomicUsize}; +use crossbeam_queue::ArrayQueue; + use crate::actor::ActorAddress; const TICK_BUFFER_CAP: usize = 1024; @@ -29,8 +30,8 @@ pub struct WorkerStats { // Error counters pub type_mismatches: AtomicU64, pub panics: AtomicU64, - // Tick timing buffer (last N ticks) - tick_timings: std::sync::Mutex>, + // Tick timing ring buffer (last N ticks, lock-free) + tick_timings: ArrayQueue, } impl WorkerStats { @@ -44,21 +45,25 @@ impl WorkerStats { inbox_sends: AtomicU64::new(0), type_mismatches: AtomicU64::new(0), panics: AtomicU64::new(0), - tick_timings: std::sync::Mutex::new(VecDeque::with_capacity(TICK_BUFFER_CAP)), + tick_timings: ArrayQueue::new(TICK_BUFFER_CAP), } } pub fn push_tick_timing(&self, timing: TickTiming) { - let mut buf = self.tick_timings.lock().unwrap(); - if buf.len() >= TICK_BUFFER_CAP { - buf.pop_front(); + if let Err(rejected) = self.tick_timings.push(timing) { + // Ring full — drop oldest, then retry (best-effort for stats) + let _ = self.tick_timings.pop(); + let _ = self.tick_timings.push(rejected); } - buf.push_back(timing); } /// Returns a snapshot of recent tick timings (drains the buffer). pub fn drain_tick_timings(&self) -> Vec { - self.tick_timings.lock().unwrap().drain(..).collect() + let mut out = Vec::new(); + while let Some(t) = self.tick_timings.pop() { + out.push(t); + } + out } /// Create a point-in-time snapshot as a [`WorkerInfo`]. diff --git a/src/worker.rs b/src/worker.rs index af665ef..12561b8 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -119,15 +119,14 @@ impl Worker { } let t5 = Instant::now(); - // 6. Publish stats - self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed); - self.stats.total_mailbox_depth.store(self.pool.total_mailbox_depth(), Ordering::Relaxed); - self.stats.messages_processed.fetch_add(processed as u64, Ordering::Relaxed); + // 6. Publish stats (skip entirely when idle to avoid allocation + mutex) + if did_work { + self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed); + self.stats.total_mailbox_depth.store(self.pool.total_mailbox_depth(), Ordering::Relaxed); + self.stats.messages_processed.fetch_add(processed as u64, Ordering::Relaxed); - // Publish per-actor mailbox depths - { - let depths: Vec<(ActorAddress, usize)> = self.pool.mailbox_depths(); - *self.mailbox_snapshot.lock().unwrap() = depths; + let mut snap = self.mailbox_snapshot.lock().unwrap(); + self.pool.mailbox_depths_into(&mut snap); } let t6 = Instant::now(); @@ -210,7 +209,7 @@ impl ContextInner for WorkerContext<'_> { } Some(wid) => { self.stats.cross_sends.fetch_add(1, Ordering::Relaxed); - let _ = self.tc.transfer_txs[wid.as_usize()].try_send(Envelope::new(addr, msg)); + self.tc.transfer_txs[wid.as_usize()].send(Envelope::new(addr, msg)); Ok(()) } None => { @@ -220,18 +219,18 @@ impl ContextInner for WorkerContext<'_> { } } - fn spawn_any(&self, addr: ActorAddress, actor: Box) -> Result<(), Error> { + fn spawn_any(&self, addr: ActorAddress, actor: Box) { let worker_id = self.tc.placement.next_worker(); self.tc.address_map.insert(addr, worker_id); self.tc.spawn_txs[worker_id.as_usize()] - .try_send((addr, actor)) - .map_err(|_| Error::from("Spawn queue full")) + .send((addr, actor)) } } struct ActorSlot { mailbox: VecDeque>, actor: Box, + poisoned: bool, } /// Per-worker actor storage. Owns per-actor mailboxes. @@ -248,8 +247,9 @@ impl ActorPool { pub fn insert(&mut self, addr: ActorAddress, actor: Box) { self.actors.insert(addr, ActorSlot { - mailbox: VecDeque::new(), + mailbox: VecDeque::with_capacity(16), actor, + poisoned: false, }); } @@ -268,16 +268,30 @@ impl ActorPool { pub fn tick_all(&mut self, inner: &dyn ContextInner, stats: &WorkerStats) -> usize { let mut count = 0; for (&addr, slot) in self.actors.iter_mut() { + if slot.poisoned { + // Discard all messages for poisoned actors + slot.mailbox.clear(); + continue; + } let ctx = Ctx::new(inner, addr); while let Some(msg) = slot.mailbox.pop_front() { let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - slot.actor.handle_any(&ctx, msg); + slot.actor.handle_any(&ctx, msg) })); - if result.is_err() { - stats.panics.fetch_add(1, Ordering::Relaxed); - eprintln!("swactor: actor {addr} panicked in handler"); - #[cfg(feature = "tracing")] - tracing::error!(actor_addr = %addr, "actor.panicked"); + match result { + Ok(false) => { + stats.type_mismatches.fetch_add(1, Ordering::Relaxed); + } + Err(_) => { + stats.panics.fetch_add(1, Ordering::Relaxed); + eprintln!("swactor: actor {addr} panicked — poisoned, future messages will be discarded"); + #[cfg(feature = "tracing")] + tracing::error!(actor_addr = %addr, "actor.panicked"); + slot.poisoned = true; + slot.mailbox.clear(); + break; + } + Ok(true) => {} } count += 1; } @@ -293,8 +307,9 @@ impl ActorPool { self.actors.values().map(|slot| slot.mailbox.len()).sum() } - /// Returns per-actor mailbox depths for dashboard reporting. - pub fn mailbox_depths(&self) -> Vec<(ActorAddress, usize)> { - self.actors.iter().map(|(&addr, slot)| (addr, slot.mailbox.len())).collect() + /// Fill `out` with per-actor mailbox depths, reusing the existing allocation. + pub fn mailbox_depths_into(&self, out: &mut Vec<(ActorAddress, usize)>) { + out.clear(); + out.extend(self.actors.iter().map(|(&addr, slot)| (addr, slot.mailbox.len()))); } } diff --git a/tests/runtime_api.rs b/tests/runtime_api.rs index f58bbca..33273f7 100644 --- a/tests/runtime_api.rs +++ b/tests/runtime_api.rs @@ -590,6 +590,41 @@ fn panic_does_not_corrupt_subsequent_messages() { assert_eq!(replies, vec![Count(1), Count(2)], "counter should be unaffected by peer panics"); } +#[test] +fn panicked_actor_is_poisoned_and_discards_future_messages() { + // Given a CounterActor that receives 3 messages: Increment, PanicMsg, Increment + // We need an actor that can handle both — so we use PanicActor for the panic + // and a separate CounterActor that continues working. + // + // Specifically: a PanicActor receives one PanicMsg, panics, then future + // PanicMsgs should be silently discarded (actor is poisoned). + let rt = Runtime::new(RuntimeConfig::default()); + let panic_addr = rt.spawn(PanicActor).unwrap(); + let good_addr = rt.spawn(CounterActor { count: 0 }).unwrap(); + let inbox = rt.new_inbox::().unwrap(); + + // Send a panic message, then more panic messages — they should be discarded + rt.send_to(panic_addr, PanicMsg).unwrap(); + rt.send_to(panic_addr, PanicMsg).unwrap(); + rt.send_to(panic_addr, PanicMsg).unwrap(); + + // Also send to a healthy actor to prove the system still works + rt.send_to(good_addr, Increment { reply_to: *inbox.addr() }).unwrap(); + + // When messages are processed + for _ in 0..20 { + rt.tick(); + } + + // Then: healthy actor still works, and only 1 panic recorded (not 3) + let reply = inbox.try_recv(); + assert!(reply.is_some(), "healthy actor should still reply after peer is poisoned"); + + let s = rt.stats(); + let total_panics: u64 = s.workers.iter().map(|w| w.panics).sum(); + assert_eq!(total_panics, 1, "only the first panic should be recorded; rest are discarded"); +} + // ═══════════════════════════════════════════════════════════════════════════ // Observability // ═══════════════════════════════════════════════════════════════════════════ @@ -662,16 +697,467 @@ fn stats_record_panics() { rt.tick(); } - // Then stats record the panics + // Then stats record the panic (second message is discarded — actor is poisoned) let s = rt.stats(); let total_panics: u64 = s.workers.iter().map(|w| w.panics).sum(); assert!( - total_panics >= 2, - "stats should record at least 2 panics, got {}", + total_panics >= 1, + "stats should record at least 1 panic, got {}", total_panics ); } +// ═══════════════════════════════════════════════════════════════════════════ +// Edge Cases & Adversarial Tests +// ═══════════════════════════════════════════════════════════════════════════ + +// ── Additional actors for edge-case tests ──────────────────────────────── + +/// Sends a countdown message to itself, then replies Done(0) when remaining hits zero. +/// Tests pending_local self-delivery path. +struct SelfSendActor; + +#[derive(Clone)] +struct Countdown { + remaining: usize, + reply_to: ActorAddress, +} + +impl ActorInterface for SelfSendActor { + type Incoming = Countdown; + type Response = Done; + fn handle(&mut self, ctx: &Ctx, msg: Countdown) { + if msg.remaining == 0 { + let _ = ctx.send(msg.reply_to, Done(0)); + } else { + let _ = ctx.send( + ctx.self_addr(), + Countdown { remaining: msg.remaining - 1, reply_to: msg.reply_to }, + ); + } + } +} + +/// Spawns a DoubleActor child, sends it work, then panics. +/// The child should still process the forwarded message. +struct SpawnThenPanicActor; + +impl ActorInterface for SpawnThenPanicActor { + type Incoming = Forward; + type Response = (); + fn handle(&mut self, ctx: &Ctx, msg: Forward) { + let child = ctx.spawn(DoubleActor).unwrap(); + let _ = ctx.send(child, Forward { value: msg.value, reply_to: msg.reply_to }); + panic!("intentional panic after spawn+send"); + } +} + +/// Processes `remaining_good` messages, then panics on the next one. +/// Uses a shared counter so the test can observe how many were processed. +struct PanicAfterNActor { + remaining_good: usize, + counter: Arc, +} + +impl ActorInterface for PanicAfterNActor { + type Incoming = Ping; + type Response = (); + fn handle(&mut self, _ctx: &Ctx, _msg: Ping) { + if self.remaining_good == 0 { + panic!("intentional delayed panic"); + } + self.remaining_good -= 1; + self.counter.fetch_add(1, Ordering::SeqCst); + } +} + +/// Sends a reply, then panics. Tests that messages sent before the panic +/// are still delivered (they're already in the queue). +struct SendThenPanicActor; + +impl ActorInterface for SendThenPanicActor { + type Incoming = Ping; + type Response = Pong; + fn handle(&mut self, ctx: &Ctx, msg: Ping) { + let _ = ctx.send(msg.reply_to, Pong); + panic!("intentional panic after send"); + } +} + +// ── Tests ──────────────────────────────────────────────────────────────── + + +#[test] +fn wrong_type_to_actor_increments_type_mismatch_counter() { + // Given a PingPongActor that expects Ping + let rt = Runtime::new(RuntimeConfig::default()); + let addr = rt.spawn(PingPongActor).unwrap(); + + // When I send it a Count message (wrong type) + rt.send_to(addr, Count(42)).unwrap(); + for _ in 0..10 { + rt.tick(); + } + + // Then stats record the type mismatch + let s = rt.stats(); + let mismatches: u64 = s.workers.iter().map(|w| w.type_mismatches).sum(); + assert_eq!(mismatches, 1, "sending wrong type should increment type_mismatches"); +} + +// FIXME dont count dropped messages +#[test] +fn type_mismatch_still_counted_as_processed() { + // Given a PingPongActor + let rt = Runtime::new(RuntimeConfig::default()); + let addr = rt.spawn(PingPongActor).unwrap(); + + // When I send it 3 wrong-type messages + for _ in 0..3 { + rt.send_to(addr, Count(0)).unwrap(); + } + for _ in 0..10 { + rt.tick(); + } + + // Then all 3 are counted in both type_mismatches AND messages_processed + // (the message was dequeued and attempted — it "went through" the system) + let s = rt.stats(); + let mismatches: u64 = s.workers.iter().map(|w| w.type_mismatches).sum(); + let processed: u64 = s.workers.iter().map(|w| w.messages_processed).sum(); + assert_eq!(mismatches, 3); + assert!( + processed >= 3, + "type-mismatched messages count as processed (dequeued+attempted), got {}", + processed + ); +} + +#[test] +fn self_send_chain_completes() { + // Given a SelfSendActor that will bounce a message to itself 10 times + let rt = Runtime::new(RuntimeConfig::default()); + let addr = rt.spawn(SelfSendActor).unwrap(); + let inbox = rt.new_inbox::().unwrap(); + + // When triggered with remaining=10 + rt.send_to(addr, Countdown { remaining: 10, reply_to: *inbox.addr() }).unwrap(); + + // Then after enough ticks the chain completes. + // Each self-send goes through pending_local → next tick's mailbox, + // so it needs at least 11 ticks (1 initial + 10 bounces). + let reply = tick_until_recv(&rt, &inbox, 50); + assert_eq!(reply, Some(Done(0)), "self-send chain should complete"); +} + +#[test] +fn panic_mid_batch_discards_remaining_messages() { + // Given an actor that processes 2 messages then panics on the 3rd + let counter = Arc::new(AtomicUsize::new(0)); + let rt = Runtime::new(RuntimeConfig::default()); + let dummy = rt.new_inbox::().unwrap(); + let addr = rt.spawn(PanicAfterNActor { + remaining_good: 2, + counter: counter.clone(), + }).unwrap(); + + // When I queue 5 messages and tick (all arrive before first tick_all) + for _ in 0..5 { + rt.send_to(addr, Ping { reply_to: *dummy.addr() }).unwrap(); + } + for _ in 0..20 { + rt.tick(); + } + + // Then only 2 messages were processed — the 3rd panicked, 4th+5th discarded + assert_eq!( + counter.load(Ordering::SeqCst), + 2, + "only messages before the panic should be processed" + ); + let s = rt.stats(); + let panics: u64 = s.workers.iter().map(|w| w.panics).sum(); + assert_eq!(panics, 1, "exactly one panic should be recorded"); +} + +#[test] +fn spawn_then_panic_child_survives() { + // Given a SpawnThenPanicActor + let rt = Runtime::new(RuntimeConfig::default()); + let addr = rt.spawn(SpawnThenPanicActor).unwrap(); + let inbox = rt.new_inbox::().unwrap(); + + // When the parent spawns a child, sends it work, then panics + rt.send_to(addr, Forward { value: 5, reply_to: *inbox.addr() }).unwrap(); + + // Then the child still processes the forwarded message and replies Done(10) + let reply = tick_until_recv(&rt, &inbox, 30); + assert_eq!( + reply, + Some(Done(10)), + "child spawned before parent panic should still work" + ); +} + +#[test] +fn panic_after_send_still_delivers_sent_messages() { + // Given a SendThenPanicActor (sends Pong, then panics) + let rt = Runtime::new(RuntimeConfig::default()); + let addr = rt.spawn(SendThenPanicActor).unwrap(); + let inbox = rt.new_inbox::().unwrap(); + + // When it processes a Ping (sends reply, then panics) + rt.send_to(addr, Ping { reply_to: *inbox.addr() }).unwrap(); + + // Then the Pong reply still arrives — sends happen before the panic unwinds + let reply = tick_until_recv(&rt, &inbox, 20); + assert!( + reply.is_some(), + "message sent before panic should still be delivered" + ); +} + +// FIXME: document somewhere this behavior. No test is needed. It is not obvious what to do +// about failed messages. Because this is going to be distributed, we cannot rely on delivery always +// succeeeding. +#[test] +fn send_to_poisoned_actor_is_a_silent_black_hole() { + // Given a poisoned actor (panicked on first message) + let rt = Runtime::new(RuntimeConfig::default()); + let panic_addr = rt.spawn(PanicActor).unwrap(); + rt.send_to(panic_addr, PanicMsg).unwrap(); + for _ in 0..5 { + rt.tick(); + } + + // When I send more messages to it + let result = rt.send_to(panic_addr, PanicMsg); + + // Then send_to succeeds (address is still in address_map) + assert!( + result.is_ok(), + "send_to poisoned actor should succeed from sender's POV" + ); + + // And ticking doesn't produce new panics — messages are discarded in tick_all + for _ in 0..10 { + rt.tick(); + } + let s = rt.stats(); + let panics: u64 = s.workers.iter().map(|w| w.panics).sum(); + assert_eq!(panics, 1, "poisoned actor should not produce new panics"); +} + +#[test] +fn tiny_buffer_delivers_all_messages_in_order() { + // Given a runtime with channel_buffer_size=1 (overflow on every 2nd message) + let rt = Runtime::new(RuntimeConfig { + channel_buffer_size: 1, + ..Default::default() + }); + let addr = rt.spawn(CounterActor { count: 0 }).unwrap(); + let inbox = rt.new_inbox::().unwrap(); + + // When I send 50 messages (almost all hit the overflow queue) + for _ in 0..50 { + rt.send_to(addr, Increment { reply_to: *inbox.addr() }).unwrap(); + } + + // Then all 50 arrive and in FIFO order + let replies = tick_and_drain(&rt, &inbox, 100); + assert_eq!(replies.len(), 50, "all messages should arrive despite tiny buffer"); + assert_eq!( + replies.last(), + Some(&Count(50)), + "messages should maintain FIFO order through overflow queue" + ); +} + +#[test] +fn empty_runtime_tick_and_stats_are_safe() { + // Given a runtime with no actors at all + let rt = Runtime::new(RuntimeConfig::default()); + + // When I tick and check stats + for _ in 0..10 { + rt.tick(); + } + let s = rt.stats(); + + // Then everything reports zeros without panicking + assert_eq!(s.actors.len(), 0); + assert_eq!(s.num_workers, 1); + let total: u64 = s.workers.iter().map(|w| w.messages_processed).sum(); + assert_eq!(total, 0); +} + +#[test] +fn stats_stable_after_idle_ticks() { + // Given an actor that processes a message + let rt = Runtime::new(RuntimeConfig::default()); + let addr = rt.spawn(CounterActor { count: 0 }).unwrap(); + let inbox = rt.new_inbox::().unwrap(); + rt.send_to(addr, Increment { reply_to: *inbox.addr() }).unwrap(); + for _ in 0..5 { + rt.tick(); + } + let _ = inbox.try_recv(); + let s1 = rt.stats(); + + // When I tick 100 more times with no messages + for _ in 0..100 { + rt.tick(); + } + let s2 = rt.stats(); + + // Then messages_processed doesn't grow during idle ticks + let total1: u64 = s1.workers.iter().map(|w| w.messages_processed).sum(); + let total2: u64 = s2.workers.iter().map(|w| w.messages_processed).sum(); + assert_eq!( + total1, total2, + "idle ticks must not inflate messages_processed" + ); +} + +#[test] +fn deep_spawn_chain_completes() { + // Given a 100-level chain (tests no stack overflow from recursive tick_all) + let rt = Runtime::new(RuntimeConfig { + max_actors: 2000, + ..Default::default() + }); + let addr = rt.spawn(ChainActor).unwrap(); + let inbox = rt.new_inbox::().unwrap(); + + // When chain of depth 100 is triggered + rt.send_to( + addr, + ChainMsg { remaining: 100, depth: 0, reply_to: *inbox.addr() }, + ).unwrap(); + + // Then the leaf at depth 100 replies + let reply = tick_until_recv(&rt, &inbox, 500); + assert_eq!( + reply, + Some(Done(100)), + "100-level chain should complete" + ); +} + +#[test] +fn all_spawned_addresses_are_unique() { + let rt = Runtime::new(RuntimeConfig { + max_actors: 10_000, + ..Default::default() + }); + let mut addrs: Vec = (0..1000) + .map(|_| rt.spawn(PingPongActor).unwrap()) + .collect(); + + addrs.sort_by_key(|a| a.0); + let before = addrs.len(); + addrs.dedup_by_key(|a| a.0); + assert_eq!(addrs.len(), before, "all 1000 addresses should be unique"); +} + +#[test] +fn inbox_empty_before_any_tick() { + // Given a sent message that hasn't been ticked + let rt = Runtime::new(RuntimeConfig::default()); + let addr = rt.spawn(PingPongActor).unwrap(); + let inbox = rt.new_inbox::().unwrap(); + rt.send_to(addr, Ping { reply_to: *inbox.addr() }).unwrap(); + + // Then inbox is empty — no processing without tick + assert!(inbox.try_recv().is_none()); +} + +#[test] +fn interleaved_spawn_and_send_in_handler_all_complete() { + // Given a FanOutActor that spawns 20 children with interleaved spawn+send + let rt = Runtime::new(RuntimeConfig::default()); + let addr = rt.spawn(FanOutActor).unwrap(); + let inbox = rt.new_inbox::().unwrap(); + + rt.send_to(addr, FanOut { count: 20, reply_to: *inbox.addr() }).unwrap(); + + let replies = tick_and_drain(&rt, &inbox, 50); + assert_eq!( + replies.len(), + 20, + "all 20 children spawned+messaged in same handler should reply" + ); +} + +#[test] +fn multiple_inbox_types_coexist() { + // Given two inboxes of different types on the same runtime + let rt = Runtime::new(RuntimeConfig::default()); + let counter = rt.spawn(CounterActor { count: 0 }).unwrap(); + let pinger = rt.spawn(PingPongActor).unwrap(); + let count_inbox = rt.new_inbox::().unwrap(); + let pong_inbox = rt.new_inbox::().unwrap(); + + // When both actors reply to their respective inboxes + rt.send_to(counter, Increment { reply_to: *count_inbox.addr() }).unwrap(); + rt.send_to(pinger, Ping { reply_to: *pong_inbox.addr() }).unwrap(); + for _ in 0..10 { + rt.tick(); + } + + // Then each inbox gets its correct type — no cross-contamination + assert_eq!(count_inbox.try_recv(), Some(Count(1))); + assert_eq!(pong_inbox.try_recv(), Some(Pong)); +} + +#[test] +fn poisoned_actor_messages_not_counted_as_processed() { + // Given a poisoned actor that then receives 10 more messages + let rt = Runtime::new(RuntimeConfig::default()); + let panic_addr = rt.spawn(PanicActor).unwrap(); + rt.send_to(panic_addr, PanicMsg).unwrap(); + for _ in 0..5 { + rt.tick(); + } + let s1 = rt.stats(); + let processed_before: u64 = s1.workers.iter().map(|w| w.messages_processed).sum(); + + // When I send 10 messages to the poisoned actor and tick + for _ in 0..10 { + rt.send_to(panic_addr, PanicMsg).unwrap(); + } + for _ in 0..20 { + rt.tick(); + } + let s2 = rt.stats(); + let processed_after: u64 = s2.workers.iter().map(|w| w.messages_processed).sum(); + + // Then the 10 discarded messages should NOT increase the processed count + assert_eq!( + processed_before, processed_after, + "messages discarded by poisoned actors should not be counted as processed" + ); +} + +#[test] +fn rapid_spawn_and_immediate_send() { + // Given a runtime, spawn an actor and immediately send before any tick + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + + // When I spawn + send in rapid succession, 50 times + let mut addrs = Vec::new(); + for _ in 0..50 { + let addr = rt.spawn(PingPongActor).unwrap(); + rt.send_to(addr, Ping { reply_to: *inbox.addr() }).unwrap(); + addrs.push(addr); + } + + // Then all 50 replies eventually arrive (spawn queue drained before transfer) + let replies = tick_and_drain(&rt, &inbox, 50); + assert_eq!(replies.len(), 50, "all spawn+send pairs should complete"); +} + // ═══════════════════════════════════════════════════════════════════════════ // Configuration // ═══════════════════════════════════════════════════════════════════════════ diff --git a/tests/test_python.py b/tests/test_python.py index 9847673..9e8a773 100644 --- a/tests/test_python.py +++ b/tests/test_python.py @@ -37,7 +37,7 @@ class TestRuntimeConfig(unittest.TestCase): cfg = RuntimeConfig() self.assertEqual(cfg.num_threads, 1) self.assertEqual(cfg.max_actors, 1000) - self.assertEqual(cfg.actor_max_messages, 1000) + self.assertEqual(cfg.channel_buffer_size, 1000) self.assertEqual(cfg.spin_threshold, 64) self.assertEqual(cfg.yield_threshold, 256) self.assertEqual(cfg.sleep_increment_us, 50)