fix: stats datatypes and refactor channel signature #28

Merged
zacheryasc merged 1 commit from cfuzz into master 2026-02-10 07:35:59 +00:00
19 changed files with 591 additions and 166 deletions

View file

@ -37,10 +37,6 @@ criterion = { version = "0.5", features = ["html_reports"] }
name = "runtime_benchmarks" name = "runtime_benchmarks"
harness = false harness = false
[[bench]]
name = "worker_benchmarks"
harness = false
[[bench]] [[bench]]
name = "mt_benchmarks" name = "mt_benchmarks"
harness = false harness = false

View file

@ -15,7 +15,7 @@ fn mt_config(threads: usize, max_actors: usize, max_messages: usize) -> RuntimeC
RuntimeConfig { RuntimeConfig {
num_threads: threads, num_threads: threads,
max_actors, max_actors,
actor_max_messages: max_messages, channel_buffer_size: max_messages,
backoff_policy: BackoffPolicy { backoff_policy: BackoffPolicy {
spin_threshold: 32, spin_threshold: 32,
yield_threshold: 64, yield_threshold: 64,

View file

@ -14,7 +14,7 @@ use swactor::{
fn make_config(max_actors: usize, max_messages: usize) -> RuntimeConfig { fn make_config(max_actors: usize, max_messages: usize) -> RuntimeConfig {
RuntimeConfig { RuntimeConfig {
max_actors, max_actors,
actor_max_messages: max_messages, channel_buffer_size: max_messages,
num_threads: 1, num_threads: 1,
..Default::default() ..Default::default()
} }

View file

@ -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<u64> = 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<u64> = 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);

View file

@ -79,7 +79,7 @@ fn bench_config(threads: usize, max_actors: usize, max_messages: usize) -> Runti
RuntimeConfig { RuntimeConfig {
num_threads: threads, num_threads: threads,
max_actors, max_actors,
actor_max_messages: max_messages, channel_buffer_size: max_messages,
backoff_policy: BackoffPolicy { backoff_policy: BackoffPolicy {
spin_threshold: 32, spin_threshold: 32,
yield_threshold: 64, yield_threshold: 64,

View file

@ -84,7 +84,7 @@ fn main() {
let rt = Runtime::new(RuntimeConfig { let rt = Runtime::new(RuntimeConfig {
num_threads: 4, num_threads: 4,
max_actors: 1024, max_actors: 1024,
actor_max_messages: 2000, channel_buffer_size: 2000,
..Default::default() ..Default::default()
}); });

View file

@ -59,7 +59,7 @@ fn main() {
let rt = Runtime::new(RuntimeConfig { let rt = Runtime::new(RuntimeConfig {
num_threads: 4, num_threads: 4,
max_actors: 512, max_actors: 512,
actor_max_messages: 1000, channel_buffer_size: 1000,
..Default::default() ..Default::default()
}); });

View file

@ -64,7 +64,7 @@ fn run_simulation_single_threaded(config: SimConfig) -> SimulationTrace {
let rt = Runtime::new(RuntimeConfig { let rt = Runtime::new(RuntimeConfig {
num_threads: 1, num_threads: 1,
max_actors: (config.num_nodes + 64).next_power_of_two(), 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() ..Default::default()
}); });
@ -177,7 +177,7 @@ fn run_simulation_multi_threaded(config: SimConfig) -> SimulationTrace {
let rt = Runtime::new(RuntimeConfig { let rt = Runtime::new(RuntimeConfig {
num_threads: config.num_threads, num_threads: config.num_threads,
max_actors: (config.num_nodes + 64).next_power_of_two(), 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() ..Default::default()
}); });

View file

@ -225,7 +225,7 @@ pub struct PyRuntimeConfig {
#[pyo3(get, set)] #[pyo3(get, set)]
max_actors: usize, max_actors: usize,
#[pyo3(get, set)] #[pyo3(get, set)]
actor_max_messages: usize, channel_buffer_size: usize,
#[pyo3(get, set)] #[pyo3(get, set)]
spin_threshold: u32, spin_threshold: u32,
#[pyo3(get, set)] #[pyo3(get, set)]
@ -243,7 +243,7 @@ impl PyRuntimeConfig {
*, *,
num_threads = 1, num_threads = 1,
max_actors = 1_000, max_actors = 1_000,
actor_max_messages = 1_000, channel_buffer_size = 1_000,
spin_threshold = 64, spin_threshold = 64,
yield_threshold = 256, yield_threshold = 256,
sleep_increment_us = 50, sleep_increment_us = 50,
@ -252,7 +252,7 @@ impl PyRuntimeConfig {
fn new( fn new(
num_threads: usize, num_threads: usize,
max_actors: usize, max_actors: usize,
actor_max_messages: usize, channel_buffer_size: usize,
spin_threshold: u32, spin_threshold: u32,
yield_threshold: u32, yield_threshold: u32,
sleep_increment_us: u64, sleep_increment_us: u64,
@ -261,7 +261,7 @@ impl PyRuntimeConfig {
Self { Self {
num_threads, num_threads,
max_actors, max_actors,
actor_max_messages, channel_buffer_size,
spin_threshold, spin_threshold,
yield_threshold, yield_threshold,
sleep_increment_us, sleep_increment_us,
@ -275,7 +275,7 @@ impl From<PyRuntimeConfig> for RuntimeConfig {
RuntimeConfig { RuntimeConfig {
num_threads: py.num_threads, num_threads: py.num_threads,
max_actors: py.max_actors, max_actors: py.max_actors,
actor_max_messages: py.actor_max_messages, channel_buffer_size: py.channel_buffer_size,
backoff_policy: BackoffPolicy { backoff_policy: BackoffPolicy {
spin_threshold: py.spin_threshold, spin_threshold: py.spin_threshold,
yield_threshold: py.yield_threshold, yield_threshold: py.yield_threshold,

View file

@ -45,17 +45,22 @@ impl<A: ActorInterface> Actor<A> {
} }
/// Trait for type-erased actors — single-message handler. /// Trait for type-erased actors — single-message handler.
///
/// Returns `true` if the message was handled, `false` on type mismatch.
pub trait AnyActor: Send { pub trait AnyActor: Send {
fn handle_any(&mut self, ctx: &Ctx, msg: Box<dyn Any + Send>); fn handle_any(&mut self, ctx: &Ctx, msg: Box<dyn Any + Send>) -> bool;
} }
impl<A> AnyActor for Actor<A> impl<A> AnyActor for Actor<A>
where where
A: ActorInterface, A: ActorInterface,
{ {
fn handle_any(&mut self, ctx: &Ctx, msg: Box<dyn Any + Send>) { fn handle_any(&mut self, ctx: &Ctx, msg: Box<dyn Any + Send>) -> bool {
if let Ok(typed) = msg.downcast::<A::Incoming>() { if let Ok(typed) = msg.downcast::<A::Incoming>() {
self.0.handle(ctx, *typed); self.0.handle(ctx, *typed);
true
} else {
false
} }
} }
} }
@ -63,7 +68,7 @@ where
/// Object-safe inner trait for sending type-erased messages. /// Object-safe inner trait for sending type-erased messages.
pub trait ContextInner { pub trait ContextInner {
fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> Result<(), Error>; fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> Result<(), Error>;
fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>) -> Result<(), Error>; fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>);
} }
/// Actor syscall interface — passed to `ActorInterface::handle()`. /// Actor syscall interface — passed to `ActorInterface::handle()`.
@ -98,7 +103,7 @@ impl<'a> Ctx<'a> {
pub fn spawn<A: ActorInterface>(&self, actor: A) -> Result<ActorAddress, Error> { pub fn spawn<A: ActorInterface>(&self, actor: A) -> Result<ActorAddress, Error> {
let addr = ActorAddress::new_random(); let addr = ActorAddress::new_random();
let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor)); let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor));
self.inner.spawn_any(addr, boxed)?; self.inner.spawn_any(addr, boxed);
Ok(addr) Ok(addr)
} }
} }

View file

@ -16,17 +16,13 @@ impl<T> HybridChannel<T> {
} }
} }
pub fn push(&self, value: T) -> Result<(), T> { pub fn push(&self, value: T) {
if !self.overflow.is_empty() { if !self.overflow.is_empty() {
self.overflow.push(value); self.overflow.push(value);
return Ok(()); return;
} }
match self.ring.push(value) { if let Err(v) = self.ring.push(value) {
Ok(()) => Ok(()),
Err(v) => {
self.overflow.push(v); self.overflow.push(v);
Ok(())
}
} }
} }
@ -61,7 +57,7 @@ pub(crate) struct Sender<T> {
} }
impl<T> Sender<T> { impl<T> Sender<T> {
pub fn try_send(&self, value: T) -> Result<(), T> { pub fn send(&self, value: T) {
self.queue.push(value) self.queue.push(value)
} }
} }

View file

@ -26,7 +26,7 @@ impl Default for BackoffPolicy {
/// The tunable settings for the runtime. /// The tunable settings for the runtime.
pub struct RuntimeConfig { pub struct RuntimeConfig {
pub max_actors: usize, pub max_actors: usize,
pub actor_max_messages: usize, pub channel_buffer_size: usize,
pub num_threads: usize, pub num_threads: usize,
pub backoff_policy: BackoffPolicy, pub backoff_policy: BackoffPolicy,
} }
@ -34,16 +34,15 @@ pub struct RuntimeConfig {
/// 8kB for the `Box<..>` before counting the rest of the memory /// 8kB for the `Box<..>` before counting the rest of the memory
const DEFAULT_MAX_ACTORS: usize = 1_000; const DEFAULT_MAX_ACTORS: usize = 1_000;
/// 16kB PER ACTOR to alloc space for storing messages. /// Pre-allocated ring buffer capacity for each channel (transfer, spawn, inbox).
/// With default setting of [DEFAULT_MAX_ACTORS] this is: /// When the ring is full, messages overflow into an unbounded backup queue.
/// 1_000 * 16kB = 16MB const DEFAULT_CHANNEL_BUFFER_SIZE: usize = 1_000;
const DEFAULT_ACTOR_MAX_MESSAGES: usize = 1_000;
impl Default for RuntimeConfig { impl Default for RuntimeConfig {
fn default() -> Self { fn default() -> Self {
Self { Self {
max_actors: DEFAULT_MAX_ACTORS, max_actors: DEFAULT_MAX_ACTORS,
actor_max_messages: DEFAULT_ACTOR_MAX_MESSAGES, channel_buffer_size: DEFAULT_CHANNEL_BUFFER_SIZE,
num_threads: 1, num_threads: 1,
backoff_policy: BackoffPolicy::default(), backoff_policy: BackoffPolicy::default(),
} }

View file

@ -105,7 +105,7 @@ pub(crate) trait SenderT: Send + Sync {
impl<M: Message> SenderT for Sender<M> { impl<M: Message> SenderT for Sender<M> {
fn try_send_any(&self, msg: Box<dyn Any + Send>) { fn try_send_any(&self, msg: Box<dyn Any + Send>) {
if let Ok(typed) = msg.downcast::<M>() { if let Ok(typed) = msg.downcast::<M>() {
let _ = Sender::try_send(self, *typed); Sender::send(self, *typed);
} }
} }
} }

View file

@ -17,12 +17,12 @@ pub struct Error(Box<dyn std::error::Error + Send + Sync + 'static>);
impl<T: AsRef<str>> From<T> for Error { impl<T: AsRef<str>> From<T> for Error {
fn from(value: T) -> Self { fn from(value: T) -> Self {
Error(format!("{:?}", value.as_ref()).into()) Error(value.as_ref().to_string().into())
} }
} }
impl std::fmt::Display for Error { impl std::fmt::Display for Error {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{:?}", self.0) write!(f, "{}", self.0)
} }
} }

View file

@ -101,7 +101,7 @@ impl Runtime {
let mut workers = Vec::with_capacity(num_workers); let mut workers = Vec::with_capacity(num_workers);
for i in 0..num_workers { for i in 0..num_workers {
let transfer_rx = Receiver::<Envelope>::new(config.actor_max_messages); let transfer_rx = Receiver::<Envelope>::new(config.channel_buffer_size);
let transfer_tx = transfer_rx.new_sender(); let transfer_tx = transfer_rx.new_sender();
transfer_txs.push(transfer_tx); transfer_txs.push(transfer_tx);
@ -151,8 +151,7 @@ impl Runtime {
self.address_map.insert(addr, worker_id); self.address_map.insert(addr, worker_id);
let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor)); let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor));
self.spawn_txs[worker_id.as_usize()] self.spawn_txs[worker_id.as_usize()]
.try_send((addr, boxed)) .send((addr, boxed));
.map_err(|_| Error::from("Runtime error: spawn queue full"))?;
#[cfg(feature = "tracing")] #[cfg(feature = "tracing")]
tracing::info!( tracing::info!(
@ -177,7 +176,7 @@ impl Runtime {
/// Create an external inbox for receiving messages in the outer process containing the runtime /// 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> { pub fn new_inbox<M: Message>(&self) -> Result<Inbox<M>, Error> {
let addr = ActorAddress::new_random(); let addr = ActorAddress::new_random();
let receiver = Receiver::<M>::new(self.config.actor_max_messages); let receiver = Receiver::<M>::new(self.config.channel_buffer_size);
let sender = receiver.new_sender(); let sender = receiver.new_sender();
self.inbox_registry.register(addr, Arc::new(sender)); self.inbox_registry.register(addr, Arc::new(sender));
Ok(Inbox { Ok(Inbox {
@ -296,24 +295,6 @@ impl Runtime {
self.transport_router = Some(router); 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<dyn Any + Send>,
) -> 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. /// Deliver a raw deserialized message into the runtime.
/// ///
/// Used by [`CodecRegistry::receive`](crate::transport::CodecRegistry::receive) /// Used by [`CodecRegistry::receive`](crate::transport::CodecRegistry::receive)
@ -325,9 +306,11 @@ impl Runtime {
msg: Box<dyn Any + Send>, msg: Box<dyn Any + Send>,
) -> Result<(), Error> { ) -> Result<(), Error> {
match self.address_map.lookup(&addr) { match self.address_map.lookup(&addr) {
Some(wid) => self.transfer_txs[wid.as_usize()] Some(wid) => {
.try_send(Envelope::new(addr, msg)) self.transfer_txs[wid.as_usize()]
.map_err(|_| Error::from("Transfer queue full")), .send(Envelope::new(addr, msg));
Ok(())
}
None => self.inbox_registry.try_deliver(addr, msg), None => self.inbox_registry.try_deliver(addr, msg),
} }
} }
@ -336,18 +319,19 @@ impl Runtime {
impl ContextInner for Runtime { impl ContextInner for Runtime {
fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> Result<(), Error> { fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> Result<(), Error> {
match self.address_map.lookup(&addr) { match self.address_map.lookup(&addr) {
Some(wid) => self.transfer_txs[wid.as_usize()] Some(wid) => {
.try_send(Envelope::new(addr, msg)) self.transfer_txs[wid.as_usize()]
.map_err(|_| Error::from("Transfer queue full")), .send(Envelope::new(addr, msg));
None => self.route_nonlocal(addr, msg), Ok(())
}
None => self.make_tick_context().route_nonlocal(addr, msg),
} }
} }
fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>) -> Result<(), Error> { fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>) {
let worker_id = self.placement.next_worker(); let worker_id = self.placement.next_worker();
self.address_map.insert(addr, worker_id); self.address_map.insert(addr, worker_id);
self.spawn_txs[worker_id.as_usize()] self.spawn_txs[worker_id.as_usize()]
.try_send((addr, actor)) .send((addr, actor));
.map_err(|_| Error::from("Spawn queue full"))
} }
} }

View file

@ -1,6 +1,7 @@
use std::collections::VecDeque;
use std::sync::atomic::{AtomicU64, AtomicUsize}; use std::sync::atomic::{AtomicU64, AtomicUsize};
use crossbeam_queue::ArrayQueue;
use crate::actor::ActorAddress; use crate::actor::ActorAddress;
const TICK_BUFFER_CAP: usize = 1024; const TICK_BUFFER_CAP: usize = 1024;
@ -29,8 +30,8 @@ pub struct WorkerStats {
// Error counters // Error counters
pub type_mismatches: AtomicU64, pub type_mismatches: AtomicU64,
pub panics: AtomicU64, pub panics: AtomicU64,
// Tick timing buffer (last N ticks) // Tick timing ring buffer (last N ticks, lock-free)
tick_timings: std::sync::Mutex<VecDeque<TickTiming>>, tick_timings: ArrayQueue<TickTiming>,
} }
impl WorkerStats { impl WorkerStats {
@ -44,21 +45,25 @@ impl WorkerStats {
inbox_sends: AtomicU64::new(0), inbox_sends: AtomicU64::new(0),
type_mismatches: AtomicU64::new(0), type_mismatches: AtomicU64::new(0),
panics: 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) { pub fn push_tick_timing(&self, timing: TickTiming) {
let mut buf = self.tick_timings.lock().unwrap(); if let Err(rejected) = self.tick_timings.push(timing) {
if buf.len() >= TICK_BUFFER_CAP { // Ring full — drop oldest, then retry (best-effort for stats)
buf.pop_front(); 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). /// Returns a snapshot of recent tick timings (drains the buffer).
pub fn drain_tick_timings(&self) -> Vec<TickTiming> { pub fn drain_tick_timings(&self) -> Vec<TickTiming> {
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`]. /// Create a point-in-time snapshot as a [`WorkerInfo`].

View file

@ -119,15 +119,14 @@ impl Worker {
} }
let t5 = Instant::now(); let t5 = Instant::now();
// 6. Publish stats // 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.num_actors.store(self.pool.len(), Ordering::Relaxed);
self.stats.total_mailbox_depth.store(self.pool.total_mailbox_depth(), 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); self.stats.messages_processed.fetch_add(processed as u64, Ordering::Relaxed);
// Publish per-actor mailbox depths let mut snap = self.mailbox_snapshot.lock().unwrap();
{ self.pool.mailbox_depths_into(&mut snap);
let depths: Vec<(ActorAddress, usize)> = self.pool.mailbox_depths();
*self.mailbox_snapshot.lock().unwrap() = depths;
} }
let t6 = Instant::now(); let t6 = Instant::now();
@ -210,7 +209,7 @@ impl ContextInner for WorkerContext<'_> {
} }
Some(wid) => { Some(wid) => {
self.stats.cross_sends.fetch_add(1, Ordering::Relaxed); 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(()) Ok(())
} }
None => { None => {
@ -220,18 +219,18 @@ impl ContextInner for WorkerContext<'_> {
} }
} }
fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>) -> Result<(), Error> { fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>) {
let worker_id = self.tc.placement.next_worker(); let worker_id = self.tc.placement.next_worker();
self.tc.address_map.insert(addr, worker_id); self.tc.address_map.insert(addr, worker_id);
self.tc.spawn_txs[worker_id.as_usize()] self.tc.spawn_txs[worker_id.as_usize()]
.try_send((addr, actor)) .send((addr, actor))
.map_err(|_| Error::from("Spawn queue full"))
} }
} }
struct ActorSlot { struct ActorSlot {
mailbox: VecDeque<Box<dyn Any + Send>>, mailbox: VecDeque<Box<dyn Any + Send>>,
actor: Box<dyn AnyActor>, actor: Box<dyn AnyActor>,
poisoned: bool,
} }
/// Per-worker actor storage. Owns per-actor mailboxes. /// Per-worker actor storage. Owns per-actor mailboxes.
@ -248,8 +247,9 @@ impl ActorPool {
pub fn insert(&mut self, addr: ActorAddress, actor: Box<dyn AnyActor>) { pub fn insert(&mut self, addr: ActorAddress, actor: Box<dyn AnyActor>) {
self.actors.insert(addr, ActorSlot { self.actors.insert(addr, ActorSlot {
mailbox: VecDeque::new(), mailbox: VecDeque::with_capacity(16),
actor, actor,
poisoned: false,
}); });
} }
@ -268,16 +268,30 @@ impl ActorPool {
pub fn tick_all(&mut self, inner: &dyn ContextInner, stats: &WorkerStats) -> usize { pub fn tick_all(&mut self, inner: &dyn ContextInner, stats: &WorkerStats) -> usize {
let mut count = 0; let mut count = 0;
for (&addr, slot) in self.actors.iter_mut() { 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); let ctx = Ctx::new(inner, addr);
while let Some(msg) = slot.mailbox.pop_front() { while let Some(msg) = slot.mailbox.pop_front() {
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { 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() { match result {
Ok(false) => {
stats.type_mismatches.fetch_add(1, Ordering::Relaxed);
}
Err(_) => {
stats.panics.fetch_add(1, Ordering::Relaxed); stats.panics.fetch_add(1, Ordering::Relaxed);
eprintln!("swactor: actor {addr} panicked in handler"); eprintln!("swactor: actor {addr} panicked — poisoned, future messages will be discarded");
#[cfg(feature = "tracing")] #[cfg(feature = "tracing")]
tracing::error!(actor_addr = %addr, "actor.panicked"); tracing::error!(actor_addr = %addr, "actor.panicked");
slot.poisoned = true;
slot.mailbox.clear();
break;
}
Ok(true) => {}
} }
count += 1; count += 1;
} }
@ -293,8 +307,9 @@ impl ActorPool {
self.actors.values().map(|slot| slot.mailbox.len()).sum() self.actors.values().map(|slot| slot.mailbox.len()).sum()
} }
/// Returns per-actor mailbox depths for dashboard reporting. /// Fill `out` with per-actor mailbox depths, reusing the existing allocation.
pub fn mailbox_depths(&self) -> Vec<(ActorAddress, usize)> { pub fn mailbox_depths_into(&self, out: &mut Vec<(ActorAddress, usize)>) {
self.actors.iter().map(|(&addr, slot)| (addr, slot.mailbox.len())).collect() out.clear();
out.extend(self.actors.iter().map(|(&addr, slot)| (addr, slot.mailbox.len())));
} }
} }

View file

@ -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"); 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::<Count>().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 // Observability
// ═══════════════════════════════════════════════════════════════════════════ // ═══════════════════════════════════════════════════════════════════════════
@ -662,16 +697,467 @@ fn stats_record_panics() {
rt.tick(); rt.tick();
} }
// Then stats record the panics // Then stats record the panic (second message is discarded — actor is poisoned)
let s = rt.stats(); let s = rt.stats();
let total_panics: u64 = s.workers.iter().map(|w| w.panics).sum(); let total_panics: u64 = s.workers.iter().map(|w| w.panics).sum();
assert!( assert!(
total_panics >= 2, total_panics >= 1,
"stats should record at least 2 panics, got {}", "stats should record at least 1 panic, got {}",
total_panics 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<AtomicUsize>,
}
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::<Done>().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::<Pong>().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::<Done>().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::<Pong>().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::<Count>().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::<Count>().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::<Done>().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<ActorAddress> = (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::<Pong>().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::<Done>().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::<Count>().unwrap();
let pong_inbox = rt.new_inbox::<Pong>().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::<Pong>().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 // Configuration
// ═══════════════════════════════════════════════════════════════════════════ // ═══════════════════════════════════════════════════════════════════════════

View file

@ -37,7 +37,7 @@ class TestRuntimeConfig(unittest.TestCase):
cfg = RuntimeConfig() cfg = RuntimeConfig()
self.assertEqual(cfg.num_threads, 1) self.assertEqual(cfg.num_threads, 1)
self.assertEqual(cfg.max_actors, 1000) 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.spin_threshold, 64)
self.assertEqual(cfg.yield_threshold, 256) self.assertEqual(cfg.yield_threshold, 256)
self.assertEqual(cfg.sleep_increment_us, 50) self.assertEqual(cfg.sleep_increment_us, 50)