diff --git a/benches/worker_benchmarks.rs b/benches/worker_benchmarks.rs index 0b3e57e..d6ed8cd 100644 --- a/benches/worker_benchmarks.rs +++ b/benches/worker_benchmarks.rs @@ -3,32 +3,6 @@ use std::collections::VecDeque; use criterion::{ criterion_group, criterion_main, BenchmarkId, Criterion, Throughput, }; -use swactor::worker::drain_count; - -// --------------------------------------------------------------------------- -// drain_count O(1) verification -// --------------------------------------------------------------------------- - -fn bench_drain_count(c: &mut Criterion) { - let mut group = c.benchmark_group("drain_count"); - - // Below waterlevel - group.bench_function("below", |b| { - b.iter(|| std::hint::black_box(drain_count(50, 100))); - }); - - // At waterlevel - group.bench_function("at", |b| { - b.iter(|| std::hint::black_box(drain_count(100, 100))); - }); - - // Above waterlevel - group.bench_function("above", |b| { - b.iter(|| std::hint::black_box(drain_count(500, 100))); - }); - - group.finish(); -} // --------------------------------------------------------------------------- // VecDeque push throughput (mirrors old mailbox_push) @@ -79,43 +53,9 @@ fn vecdeque_pop(c: &mut Criterion) { group.finish(); } -// --------------------------------------------------------------------------- -// Simulated actor tick: drain_count + pop N from VecDeque -// --------------------------------------------------------------------------- - -fn simulated_actor_tick(c: &mut Criterion) { - let mut group = c.benchmark_group("simulated_actor_tick"); - - for (wl, fill) in [(10, 5), (10, 10), (10, 50), (100, 200)] { - let param = format!("wl={wl},fill={fill}"); - group.bench_function(BenchmarkId::from_parameter(¶m), |b| { - b.iter_batched( - || { - let mut q: VecDeque = VecDeque::new(); - for i in 0..fill { - q.push_back(i as u64); - } - q - }, - |mut q| { - let n = drain_count(q.len(), wl); - for _ in 0..n { - std::hint::black_box(q.pop_front()); - } - }, - criterion::BatchSize::SmallInput, - ); - }); - } - - group.finish(); -} - criterion_group!( benches, - bench_drain_count, vecdeque_push, vecdeque_pop, - simulated_actor_tick, ); criterion_main!(benches); diff --git a/crates/swactor-python/src/lib.rs b/crates/swactor-python/src/lib.rs index d6a1924..f2c47e6 100644 --- a/crates/swactor-python/src/lib.rs +++ b/crates/swactor-python/src/lib.rs @@ -227,8 +227,6 @@ pub struct PyRuntimeConfig { #[pyo3(get, set)] actor_max_messages: usize, #[pyo3(get, set)] - mailbox_waterlevel: usize, - #[pyo3(get, set)] spin_threshold: u32, #[pyo3(get, set)] yield_threshold: u32, @@ -246,7 +244,6 @@ impl PyRuntimeConfig { num_threads = 1, max_actors = 1_000, actor_max_messages = 1_000, - mailbox_waterlevel = 10, spin_threshold = 64, yield_threshold = 256, sleep_increment_us = 50, @@ -256,7 +253,6 @@ impl PyRuntimeConfig { num_threads: usize, max_actors: usize, actor_max_messages: usize, - mailbox_waterlevel: usize, spin_threshold: u32, yield_threshold: u32, sleep_increment_us: u64, @@ -266,7 +262,6 @@ impl PyRuntimeConfig { num_threads, max_actors, actor_max_messages, - mailbox_waterlevel, spin_threshold, yield_threshold, sleep_increment_us, @@ -281,7 +276,6 @@ impl From for RuntimeConfig { num_threads: py.num_threads, max_actors: py.max_actors, actor_max_messages: py.actor_max_messages, - mailbox_waterlevel: py.mailbox_waterlevel, backoff_policy: BackoffPolicy { spin_threshold: py.spin_threshold, yield_threshold: py.yield_threshold, diff --git a/fuzz/fuzz_targets/fuzz_runtime.rs b/fuzz/fuzz_targets/fuzz_runtime.rs index 954e815..aea7a0a 100644 --- a/fuzz/fuzz_targets/fuzz_runtime.rs +++ b/fuzz/fuzz_targets/fuzz_runtime.rs @@ -210,7 +210,6 @@ enum RawAction { #[derive(Debug, Arbitrary)] struct FuzzInput { max_actors: u8, - mailbox_waterlevel: u8, scenarios: Vec, } @@ -773,7 +772,6 @@ impl fmt::Debug for FuzzState { fuzz_target!(|input: FuzzInput| { let max_actors = (input.max_actors as usize).max(1).min(200); - let mailbox_waterlevel = (input.mailbox_waterlevel as usize).max(1).min(50); let interval = log_interval(); let run = if interval > 0 { @@ -783,7 +781,6 @@ fuzz_target!(|input: FuzzInput| { let config = RuntimeConfig { max_actors, - mailbox_waterlevel, num_threads: 1, ..Default::default() }; @@ -806,7 +803,7 @@ fuzz_target!(|input: FuzzInput| { let depth: usize = stats.workers.iter().map(|w| w.mailbox_depth).sum(); let processed: u64 = stats.workers.iter().map(|w| w.messages_processed).sum(); eprintln!("\ -\n=== Run #{run} | cap={max_actors} waterlevel={mailbox_waterlevel} === +\n=== Run #{run} | cap={max_actors} === {trace}\ --- {spawned} spawned, {sent} sent, {recv} received, \ {alive} alive, {depth} queued, {processed} processed ---\n", diff --git a/src/actor.rs b/src/actor.rs index 773a30b..a083dac 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -64,7 +64,6 @@ where pub trait ContextInner { fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error>; fn spawn_any(&self, addr: ActorAddress, actor: Box) -> Result<(), Error>; - fn mailbox_waterlevel(&self) -> usize; } /// Actor syscall interface — passed to `ActorInterface::handle()`. diff --git a/src/config.rs b/src/config.rs index 9dec626..f0bf07a 100644 --- a/src/config.rs +++ b/src/config.rs @@ -28,7 +28,6 @@ pub struct RuntimeConfig { pub max_actors: usize, pub actor_max_messages: usize, pub num_threads: usize, - pub mailbox_waterlevel: usize, pub backoff_policy: BackoffPolicy, } @@ -40,16 +39,12 @@ const DEFAULT_MAX_ACTORS: usize = 1_000; /// 1_000 * 16kB = 16MB const DEFAULT_ACTOR_MAX_MESSAGES: usize = 1_000; -const DEFAULT_MAILBOX_WATERLEVEL: usize = 10; - impl Default for RuntimeConfig { fn default() -> Self { Self { max_actors: DEFAULT_MAX_ACTORS, actor_max_messages: DEFAULT_ACTOR_MAX_MESSAGES, num_threads: 1, - // Clamp to minimum 2: drain_count(1,1)==0 causes livelock - mailbox_waterlevel: DEFAULT_MAILBOX_WATERLEVEL.max(2), backoff_policy: BackoffPolicy::default(), } } diff --git a/src/crate_test/mod.rs b/src/crate_test/mod.rs index 627ce1e..c23af3c 100644 --- a/src/crate_test/mod.rs +++ b/src/crate_test/mod.rs @@ -253,13 +253,8 @@ fn wrong_type_silently_dropped() { } #[test] -fn backpressure_drains_half() { - let config = RuntimeConfig { - mailbox_waterlevel: 4, - ..Default::default() - }; - - run_with(config, &[ +fn all_messages_drain_in_one_tick() { + run(&[ Step::Spawn(1), Step::Tick, @@ -268,15 +263,7 @@ fn backpressure_drains_half() { Step::Send(1, 4), Step::Send(1, 5), Step::Send(1, 6), Step::Send(1, 7), Step::Send(1, 8), Step::Send(1, 9), - // Tick 1: 10 in mailbox, drain_count(10, 4) = 5 - Step::Tick, - Step::Expect { pool_len: 1, depth: 5, processed: 5 }, - - // Tick 2: 5 remaining, drain_count(5, 4) = 2 - Step::Tick, - Step::Expect { pool_len: 1, depth: 3, processed: 7 }, - - // Tick 3: 3 remaining, drain_count(3, 4) = 3 (below waterlevel → all) + // All 10 processed in a single tick Step::Tick, Step::Expect { pool_len: 1, depth: 0, processed: 10 }, Step::ExpectHandled(1, 10), diff --git a/src/runtime.rs b/src/runtime.rs index deb5554..01a8fc4 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -66,15 +66,12 @@ pub struct Runtime { placement: Placement, is_running: AtomicBool, worker_stats: Vec>, - /// Single-threaded mode: worker stored inline - single_worker: Option>, - /// Multi-threaded mode: workers waiting to be assigned to threads by run() - pending_workers: Option>, + /// Workers available for tick(). run() drains this and moves workers to threads. + tick_workers: RefCell>, } -// Safety: RefCell is only accessed from the thread that owns the Runtime -// in single-threaded mode. In multi-threaded mode, single_worker is None and -// pending_workers is consumed by run() before Arc sharing. +// Safety: RefCell> is only accessed from the owning thread via tick(). +// After run() the RefCell is empty and not accessed by worker threads. unsafe impl Sync for Runtime {} impl Runtime { @@ -111,35 +108,16 @@ impl Runtime { workers.push(Worker::new(WorkerId(i), transfer_rx, spawn_rx, stats)); } - if config.num_threads < 2 { - // Single-threaded: store one worker inline - let worker = workers.remove(0); - Self { - config, - address_map, - inbox_registry, - transfer_txs, - spawn_txs, - placement, - is_running: AtomicBool::new(false), - worker_stats, - single_worker: Some(RefCell::new(worker)), - pending_workers: None, - } - } else { - // Multi-threaded: stash workers for run() - Self { - config, - address_map, - inbox_registry, - transfer_txs, - spawn_txs, - placement, - is_running: AtomicBool::new(false), - worker_stats, - single_worker: None, - pending_workers: Some(workers), - } + Self { + config, + address_map, + inbox_registry, + transfer_txs, + spawn_txs, + placement, + is_running: AtomicBool::new(false), + worker_stats, + tick_workers: RefCell::new(workers), } } @@ -179,17 +157,23 @@ impl Runtime { } /// Drive one tick of the single-threaded worker. + /// + /// Panics if called on a multi-threaded runtime — use `run()` instead. pub fn tick(&self) { - if let Some(ref worker) = self.single_worker { - let tc = TickContext { - address_map: &self.address_map, - transfer_txs: &self.transfer_txs, - spawn_txs: &self.spawn_txs, - placement: &self.placement, - inbox_registry: &self.inbox_registry, - config: &self.config, - }; - worker.borrow_mut().tick_once(&tc); + assert!( + self.config.num_threads < 2, + "tick() is only valid for single-threaded runtimes; use run() for multi-threaded" + ); + let tc = TickContext { + address_map: &self.address_map, + transfer_txs: &self.transfer_txs, + spawn_txs: &self.spawn_txs, + placement: &self.placement, + inbox_registry: &self.inbox_registry, + config: &self.config, + }; + for worker in self.tick_workers.borrow_mut().iter_mut() { + worker.tick_once(&tc); } } @@ -198,17 +182,10 @@ impl Runtime { /// /// Works in both single-threaded and multi-threaded configurations. /// In single-threaded mode, one background thread is spawned. - pub fn run(mut self) -> Result { + pub fn run(self) -> Result { self.is_running.store(true, Ordering::Release); - let mut workers: Vec = Vec::new(); - - if let Some(w) = self.single_worker.take() { - workers.push(w.into_inner()); - } - if let Some(ws) = self.pending_workers.take() { - workers.extend(ws); - } + let workers: Vec = self.tick_workers.replace(Vec::new()); let rt = Arc::new(self); let mut handles: Vec> = Vec::with_capacity(workers.len()); @@ -294,8 +271,4 @@ impl ContextInner for Runtime { .try_send((addr, actor)) .map_err(|_| Error::from("Spawn queue full")) } - - fn mailbox_waterlevel(&self) -> usize { - self.config.mailbox_waterlevel - } } diff --git a/src/worker.rs b/src/worker.rs index 24c981c..5e0b749 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -159,20 +159,6 @@ impl ContextInner for WorkerContext<'_> { .map_err(|_| Error::from("Spawn queue full")) } - fn mailbox_waterlevel(&self) -> usize { - self.tc.config.mailbox_waterlevel - } -} - -/// How many messages to process this tick: -/// - `len < waterlevel` → process all (`len`) -/// - `len >= waterlevel` → process half (`len >> 1`) -pub fn drain_count(len: usize, waterlevel: usize) -> usize { - if len < waterlevel { - len - } else { - len >> 1 - } } struct ActorSlot { @@ -218,16 +204,15 @@ impl ActorPool { pub fn tick_all(&mut self, inner: &dyn ContextInner) -> usize { let mut count = 0; for (&addr, slot) in self.actors.iter_mut() { - let len = slot.mailbox.len(); - let n = drain_count(len, inner.mailbox_waterlevel()); - if n > 0 { - let ctx = Ctx::new(inner, addr); - for _ in 0..n { - if let Some(msg) = slot.mailbox.pop_front() { - slot.actor.handle_any(&ctx, msg); - count += 1; - } + 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); + })); + if result.is_err() { + eprintln!("swactor: actor {addr} panicked in handler"); } + count += 1; } } count diff --git a/tests/test_python.py b/tests/test_python.py index 3c8d691..9847673 100644 --- a/tests/test_python.py +++ b/tests/test_python.py @@ -38,7 +38,6 @@ class TestRuntimeConfig(unittest.TestCase): self.assertEqual(cfg.num_threads, 1) self.assertEqual(cfg.max_actors, 1000) self.assertEqual(cfg.actor_max_messages, 1000) - self.assertEqual(cfg.mailbox_waterlevel, 10) self.assertEqual(cfg.spin_threshold, 64) self.assertEqual(cfg.yield_threshold, 256) self.assertEqual(cfg.sleep_increment_us, 50)