From 4a348ab4cd4d6695423d23aa0ae66c416532ea2a Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Sun, 9 Aug 2026 13:50:33 +0400 Subject: [PATCH] refactor(core): drop native threading for tick-driven execution Core no longer owns or drives OS threads. The runtime is now a single tick-driven worker whose loop an external engine hosts and advances. This is the cutover required before the engine seam is introduced. Removed from core: - Runtime::run() and its owned thread pool (spawn, park/unpark, join) - notify_worker() and worker_threads: Vec> plumbing - Placement load-aware selector and its WorkerStats-driven next_worker() - WorkerId newtype and the address->worker routing map; AddressMap is now a plain AddrSet membership set - num_threads from RuntimeConfig Rewired for the single-worker tick API: - python/wasm bindings, dashboard dummy node (deleted), myelin vastai adapter, and the runtime/test suites Cleanup folded in during review: - prune three never-written WorkerStats counters (cross_sends, messages_dropped, restarts) - collapse the redundant tick_all params onto the WorkerContext handle --- README.md | 4 +- .../provider_adapters/vastai/mod.rs | 37 +- crates/bindings/python/src/lib.rs | 97 +--- crates/bindings/wasm-runtime/src/lib.rs | 5 +- .../dashboard/src/bin/swactor_dummy_node.rs | 412 --------------- src/config.rs | 2 - src/delivery.rs | 142 ++---- src/runtime.rs | 479 ++++++------------ src/stats.rs | 14 - src/worker.rs | 146 ++---- tests/message_delivery.rs | 2 - tests/runtime_admin.rs | 139 ++--- tests/runtime_stress.rs | 349 ++----------- tests/test_python.py | 41 +- 14 files changed, 350 insertions(+), 1519 deletions(-) delete mode 100644 crates/dashboard/src/bin/swactor_dummy_node.rs diff --git a/README.md b/README.md index 09eb8eb..0fc549f 100644 --- a/README.md +++ b/README.md @@ -50,8 +50,8 @@ fn main() -> Result<(), swactor::Error> { } ``` -`RuntimeConfig` is four knobs — `max_actors`, `channel_buffer_size`, -`num_threads`, and a per-tick `actor_message_budget` (inspired by BEAM's +`RuntimeConfig` is three knobs — `max_actors`, `channel_buffer_size`, +and a per-tick `actor_message_budget` (inspired by BEAM's reduction count, so one chatty mailbox can't starve the rest). ## Features diff --git a/apps/myelin/src/orchestration/provider_adapters/vastai/mod.rs b/apps/myelin/src/orchestration/provider_adapters/vastai/mod.rs index 2d23bbb..6f900cb 100644 --- a/apps/myelin/src/orchestration/provider_adapters/vastai/mod.rs +++ b/apps/myelin/src/orchestration/provider_adapters/vastai/mod.rs @@ -4,12 +4,13 @@ use std::io::{BufRead, BufReader, Read}; use std::path::{Path, PathBuf}; use std::process::{Child, Command, Stdio}; use std::sync::{Arc, mpsc}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::thread::{self, JoinHandle}; use std::time::{Duration, Instant}; use datastream::DatastreamProducer; use swactor::actor::{ActorAddress, ActorInterface}; -use swactor::runtime::{Ctx, ExternalSender, Runtime, RuntimeConfig, RuntimeHandle}; +use swactor::runtime::{Ctx, ExternalSender, Runtime, RuntimeConfig}; use swactor_vastai::{ CreateInstanceRequest, LifecyclePolicy, Offer, ProvisionRequest, ProvisionedInstance, SelectionPolicy, classify_vastai_error, create_instance, @@ -55,27 +56,46 @@ pub(crate) struct VastAiSshEndpoint { } pub(crate) struct VastAiProviderMonitor { - runtime: Option, + runtime: Arc, actor: ActorAddress, + tick_thread: Option>, + stop_flag: Arc, } impl VastAiProviderMonitor { - fn new(runtime: RuntimeHandle, actor: ActorAddress) -> Self { + fn new(runtime: Runtime, actor: ActorAddress) -> Self { + let runtime = Arc::new(runtime); + let stop_flag = Arc::new(AtomicBool::new(false)); + + let rt = Arc::clone(&runtime); + let flag = Arc::clone(&stop_flag); + let tick_thread = thread::spawn(move || { + while !flag.load(Ordering::Relaxed) || rt.has_work() { + if rt.has_work() { + rt.tick(); + } else { + thread::sleep(Duration::from_millis(10)); + } + } + }); + Self { - runtime: Some(runtime), + runtime, actor, + tick_thread: Some(tick_thread), + stop_flag, } } fn stop(&mut self) { - let Some(runtime) = self.runtime.take() else { + let Some(tick_thread) = self.tick_thread.take() else { return; }; - let _ = runtime + let _ = self .runtime .send_to(self.actor, VastAiProviderMonitorMsg::Stop); - runtime.shutdown(); - runtime.join(); + self.stop_flag.store(true, Ordering::Relaxed); + let _ = tick_thread.join(); } } @@ -552,7 +572,6 @@ impl VastAiLeaseClient for ToolsVastAiLeaseClient { sender, )) .ok()?; - let runtime = runtime.run().ok()?; Some(VastAiProviderMonitor::new(runtime, actor)) } diff --git a/crates/bindings/python/src/lib.rs b/crates/bindings/python/src/lib.rs index 0a828e0..1a86831 100644 --- a/crates/bindings/python/src/lib.rs +++ b/crates/bindings/python/src/lib.rs @@ -8,7 +8,7 @@ use ::swactor::actor::{ Actor, ActorAddress, ActorInterface, AnyActor, Ctx, Environment, SpawnRequest, }; use ::swactor::config::RuntimeConfig; -use ::swactor::runtime::{Inbox, Runtime, RuntimeHandle}; +use ::swactor::runtime::{Inbox, Runtime}; // ─── PyMsg newtype ─────────────────────────────────────────────────────────── @@ -220,8 +220,6 @@ impl PyInbox { #[pyclass(name = "RuntimeConfig")] #[derive(Clone)] pub struct PyRuntimeConfig { - #[pyo3(get, set)] - num_threads: usize, #[pyo3(get, set)] max_actors: usize, #[pyo3(get, set)] @@ -233,13 +231,11 @@ impl PyRuntimeConfig { #[new] #[pyo3(signature = ( *, - num_threads = 1, max_actors = 1_000, channel_buffer_size = 1_000, ))] - fn new(num_threads: usize, max_actors: usize, channel_buffer_size: usize) -> Self { + fn new(max_actors: usize, channel_buffer_size: usize) -> Self { Self { - num_threads, max_actors, channel_buffer_size, } @@ -249,7 +245,6 @@ impl PyRuntimeConfig { impl From for RuntimeConfig { fn from(py: PyRuntimeConfig) -> Self { RuntimeConfig { - num_threads: py.num_threads, max_actors: py.max_actors, channel_buffer_size: py.channel_buffer_size, ..Default::default() @@ -280,7 +275,7 @@ impl PyRuntime { fn spawn(&self, handler: PyObject) -> PyResult { let rt = self.inner.as_ref().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err("Runtime consumed by run()") + pyo3::exceptions::PyRuntimeError::new_err("Runtime not available") })?; let actor = PyActor::new(handler); let addr = rt.spawn(actor).map_err(to_py_err)?; @@ -289,14 +284,14 @@ impl PyRuntime { fn send(&self, addr: &PyActorAddress, msg: PyObject) -> PyResult<()> { let rt = self.inner.as_ref().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err("Runtime consumed by run()") + pyo3::exceptions::PyRuntimeError::new_err("Runtime not available") })?; rt.send_to(addr.inner, PyMsg(msg)).map_err(to_py_err) } fn inbox(&self) -> PyResult { let rt = self.inner.as_ref().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err("Runtime consumed by run()") + pyo3::exceptions::PyRuntimeError::new_err("Runtime not available") })?; let inbox: Inbox = rt.new_inbox().map_err(to_py_err)?; Ok(PyInbox { inner: inbox }) @@ -304,97 +299,20 @@ impl PyRuntime { fn tick(&self) -> PyResult<()> { let rt = self.inner.as_ref().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err("Runtime consumed by run()") + pyo3::exceptions::PyRuntimeError::new_err("Runtime not available") })?; rt.tick(); Ok(()) } - fn run(&mut self, py: Python<'_>) -> PyResult { - let rt = self.inner.take().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err("Runtime consumed by run()") - })?; - let handle = py.allow_threads(|| rt.run().map_err(to_py_err))?; - Ok(PyRuntimeHandle { - inner: Some(handle), - }) - } - fn stats(&self) -> PyResult { let rt = self.inner.as_ref().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err("Runtime consumed by run()") + pyo3::exceptions::PyRuntimeError::new_err("Runtime not available") })?; Ok(build_stats(rt)) } - - fn shutdown(&self) -> PyResult<()> { - let rt = self.inner.as_ref().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err("Runtime consumed by run()") - })?; - rt.shutdown(); - Ok(()) - } } -// ─── PyRuntimeHandle ───────────────────────────────────────────────────────── - -#[pyclass(name = "RuntimeHandle")] -pub struct PyRuntimeHandle { - inner: Option, -} - -#[pymethods] -impl PyRuntimeHandle { - fn spawn(&self, handler: PyObject) -> PyResult { - let handle = self.inner.as_ref().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err("RuntimeHandle consumed by join()") - })?; - let actor = PyActor::new(handler); - let addr = handle.runtime.spawn(actor).map_err(to_py_err)?; - Ok(PyActorAddress::from(addr)) - } - - fn send(&self, addr: &PyActorAddress, msg: PyObject) -> PyResult<()> { - let handle = self.inner.as_ref().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err("RuntimeHandle consumed by join()") - })?; - handle - .runtime - .send_to(addr.inner, PyMsg(msg)) - .map_err(to_py_err) - } - - fn inbox(&self) -> PyResult { - let handle = self.inner.as_ref().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err("RuntimeHandle consumed by join()") - })?; - let inbox: Inbox = handle.runtime.new_inbox().map_err(to_py_err)?; - Ok(PyInbox { inner: inbox }) - } - - fn stats(&self) -> PyResult { - let handle = self.inner.as_ref().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err("RuntimeHandle consumed by join()") - })?; - Ok(build_stats(&handle.runtime)) - } - - fn shutdown(&self) -> PyResult<()> { - let handle = self.inner.as_ref().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err("RuntimeHandle consumed by join()") - })?; - handle.shutdown(); - Ok(()) - } - - fn join(&mut self, py: Python<'_>) -> PyResult<()> { - let handle = self.inner.take().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err("RuntimeHandle consumed by join()") - })?; - py.allow_threads(|| handle.join()); - Ok(()) - } -} // ─── ActorInfo / RuntimeStats ──────────────────────────────────────────────── @@ -518,7 +436,6 @@ fn register(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; - m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/crates/bindings/wasm-runtime/src/lib.rs b/crates/bindings/wasm-runtime/src/lib.rs index 6b43ff8..e08d34e 100644 --- a/crates/bindings/wasm-runtime/src/lib.rs +++ b/crates/bindings/wasm-runtime/src/lib.rs @@ -103,10 +103,7 @@ pub struct WasmRuntime { impl WasmRuntime { #[wasm_bindgen(constructor)] pub fn new() -> Self { - let rt = Runtime::new(RuntimeConfig { - num_threads: 1, - ..RuntimeConfig::default() - }) + let rt = Runtime::new(RuntimeConfig::default()) .with_extension(Arc::new(StdExtension::new())); Self { rt } } diff --git a/crates/dashboard/src/bin/swactor_dummy_node.rs b/crates/dashboard/src/bin/swactor_dummy_node.rs deleted file mode 100644 index 7b07666..0000000 --- a/crates/dashboard/src/bin/swactor_dummy_node.rs +++ /dev/null @@ -1,412 +0,0 @@ -use std::collections::HashMap; -use std::sync::Arc; -use std::thread; -use std::time::Duration; - -use dashboard::swactor::{RUNTIME_ACTORS, RUNTIME_STATS, RUNTIME_WORKERS}; -use dashboard::{DashboardConfig, DashboardHandle, FrameEvent, StreamEvent, start_dashboard}; -use datastream::frame::{ChannelId, Frame, Lifetime, NodeId, Position, StreamId}; -use parking_lot::Mutex; -use serde::Serialize; -use serde_json::{Value, json}; -use swactor::actor::{ActorAddress, ActorInterface, Ctx}; -use swactor::config::RuntimeConfig; -use swactor::runtime::Runtime; -use swactor::stats::{ActorSnapshot, StatsHook}; - -const NODE_ID: &str = "dashboard-swactor-dummy"; -const WORKER_ACTORS: usize = 12; -const PUBLISH_INTERVAL: Duration = Duration::from_millis(250); -const PULSE_INTERVAL: Duration = Duration::from_millis(25); -const WORK_ITEM_DELAY: Duration = Duration::from_micros(200); - -#[derive(Clone)] -struct PulseTick { - seq: u64, -} - -#[derive(Clone)] -struct WorkItem { - seq: u64, - route: u32, - hops_left: u8, -} - -#[derive(Clone)] -enum RouterMsg { - Configure { workers: Vec }, - Beat { seq: u64 }, - Complete { worker: u32, seq: u64, route: u32 }, -} - -struct PulseActor { - router: ActorAddress, -} - -impl ActorInterface for PulseActor { - type Incoming = PulseTick; - type Response = (); - - fn handle(&mut self, ctx: &Ctx, msg: PulseTick) { - let _ = ctx.send(self.router, RouterMsg::Beat { seq: msg.seq }); - } -} - -struct RouterActor { - workers: Vec, - next: usize, - completed: u64, -} - -impl RouterActor { - fn new() -> Self { - Self { - workers: Vec::new(), - next: 0, - completed: 0, - } - } -} - -impl ActorInterface for RouterActor { - type Incoming = RouterMsg; - type Response = (); - - fn handle(&mut self, ctx: &Ctx, msg: RouterMsg) { - match msg { - RouterMsg::Configure { workers } => { - self.workers = workers; - self.next = 0; - } - RouterMsg::Beat { seq } => { - if self.workers.is_empty() { - return; - } - - let burst = 48 + (seq as usize % 32); - for route in 0..burst { - let target = self.workers[self.next % self.workers.len()]; - self.next = self.next.wrapping_add(1); - let _ = ctx.send( - target, - WorkItem { - seq, - route: route as u32, - hops_left: 1 + ((seq + route as u64) % 3) as u8, - }, - ); - } - } - RouterMsg::Complete { worker, seq, route } => { - self.completed = self.completed.wrapping_add(1); - - if self.completed % 7 == 0 && !self.workers.is_empty() { - let target = - self.workers[(worker as usize + route as usize) % self.workers.len()]; - let _ = ctx.send( - target, - WorkItem { - seq, - route: route.wrapping_add(1000), - hops_left: 1, - }, - ); - } - } - } - } -} - -struct WorkerActor { - id: u32, - router: ActorAddress, -} - -impl ActorInterface for WorkerActor { - type Incoming = WorkItem; - type Response = (); - - fn handle(&mut self, ctx: &Ctx, msg: WorkItem) { - if msg.hops_left > 0 { - let _ = ctx.send( - ctx.self_addr(), - WorkItem { - seq: msg.seq, - route: msg.route, - hops_left: msg.hops_left - 1, - }, - ); - return; - } - - thread::sleep(WORK_ITEM_DELAY); - let _ = ctx.send( - self.router, - RouterMsg::Complete { - worker: self.id, - seq: msg.seq, - route: msg.route, - }, - ); - } -} - -struct QueuedSinkActor; - -impl ActorInterface for QueuedSinkActor { - type Incoming = WorkItem; - type Response = (); - - fn handle(&mut self, ctx: &Ctx, _msg: WorkItem) { - ctx.suspend_self(); - } -} - -#[derive(Default)] -struct DashboardStatsHook { - actors: Mutex>, -} - -#[derive(Clone)] -struct ActorDetail { - worker_id: usize, - mailbox_depth: usize, - last_msg_type: Option, - messages_processed: u64, - poisoned: bool, - message_type_counts: Vec<(String, u64)>, -} - -impl StatsHook for DashboardStatsHook { - fn on_tick(&self, worker_id: usize, snapshots: &[ActorSnapshot]) { - let mut actors = self.actors.lock(); - for snapshot in snapshots { - actors.insert( - snapshot.address, - ActorDetail { - worker_id, - mailbox_depth: snapshot.mailbox_depth, - last_msg_type: snapshot.last_msg_type.map(str::to_owned), - messages_processed: snapshot.messages_processed, - poisoned: snapshot.poisoned, - message_type_counts: snapshot - .message_type_counts - .iter() - .map(|(name, count)| ((*name).to_owned(), *count)) - .collect(), - }, - ); - } - } -} - -impl DashboardStatsHook { - fn snapshot( - &self, - live_workers: &[(ActorAddress, usize)], - names: &HashMap, - ) -> Vec { - let actors = self.actors.lock(); - live_workers - .iter() - .map(|(address, worker_id)| { - let detail = actors.get(address); - ActorDetailFrame { - address: address.to_string(), - name: names.get(address).cloned(), - worker_id: detail.map_or(*worker_id, |detail| detail.worker_id), - mailbox_depth: detail.map_or(0, |detail| detail.mailbox_depth), - last_msg_type: detail.and_then(|detail| detail.last_msg_type.clone()), - messages_processed: detail.map_or(0, |detail| detail.messages_processed), - poisoned: detail.is_some_and(|detail| detail.poisoned), - message_type_counts: detail - .map(|detail| detail.message_type_counts.clone()) - .unwrap_or_default(), - } - }) - .collect() - } -} - -#[derive(Serialize)] -struct ActorDetailFrame { - address: String, - name: Option, - worker_id: usize, - mailbox_depth: usize, - last_msg_type: Option, - messages_processed: u64, - poisoned: bool, - message_type_counts: Vec<(String, u64)>, -} - -fn main() { - let dashboard = start_dashboard(DashboardConfig::default()); - dashboard.start_http_standalone(); - - let mut runtime = Runtime::new(RuntimeConfig { - num_threads: 4, - max_actors: 128, - channel_buffer_size: 4096, - actor_message_budget: 8, - }); - let stats_hook = Arc::new(DashboardStatsHook::default()); - runtime.set_stats_hook(stats_hook.clone()); - - let router = runtime.spawn(RouterActor::new()).expect("spawn router"); - let mut names = HashMap::new(); - names.insert(router, "router".to_owned()); - - let mut workers = Vec::with_capacity(WORKER_ACTORS); - for id in 0..WORKER_ACTORS { - let address = runtime - .spawn(WorkerActor { - id: id as u32, - router, - }) - .expect("spawn worker actor"); - names.insert(address, format!("worker-{id}")); - workers.push(address); - } - - let pulse = runtime.spawn(PulseActor { router }).expect("spawn pulse"); - names.insert(pulse, "pulse".to_owned()); - let queued_sink = runtime.spawn(QueuedSinkActor).expect("spawn queued sink"); - names.insert(queued_sink, "queued-sink".to_owned()); - runtime - .send_to( - router, - RouterMsg::Configure { - workers: workers.clone(), - }, - ) - .expect("configure router"); - - let runtime = runtime.run().expect("start swactor runtime"); - let stream = StreamId::new(NodeId::new(NODE_ID), Lifetime(1)); - let mut position = 0_u64; - let mut seq = 0_u64; - let mut ticks_until_publish = 0_u8; - - println!( - "dashboard listening at http://127.0.0.1:{}/view/swactor/workers", - DashboardConfig::default().port - ); - println!( - "dummy node {NODE_ID} running {} swactor actors", - names.len() - ); - - loop { - let _ = runtime.runtime.send_to(pulse, PulseTick { seq }); - if seq % 2 == 0 { - let _ = runtime.runtime.send_to( - queued_sink, - WorkItem { - seq, - route: u32::MAX, - hops_left: 0, - }, - ); - } - seq = seq.wrapping_add(1); - - if ticks_until_publish == 0 { - publish_runtime_snapshot( - &dashboard, - &stream, - &mut position, - &runtime.runtime, - &stats_hook, - &names, - ); - ticks_until_publish = (PUBLISH_INTERVAL.as_millis() / PULSE_INTERVAL.as_millis()) as u8; - } - ticks_until_publish = ticks_until_publish.saturating_sub(1); - - thread::sleep(PULSE_INTERVAL); - } -} - -fn publish_runtime_snapshot( - dashboard: &DashboardHandle, - stream: &StreamId, - position: &mut u64, - runtime: &Runtime, - stats_hook: &DashboardStatsHook, - names: &HashMap, -) { - let stats = runtime.stats(); - let actor_details = stats_hook.snapshot(&stats.actors, names); - let actors: Vec = stats - .actors - .iter() - .map(|(address, worker_id)| json!([address.to_string(), worker_id])) - .collect(); - let workers = serde_json::to_value(&stats.workers).expect("serialize worker stats"); - let actor_details = serde_json::to_value(actor_details).expect("serialize actor stats"); - let tick_timings = serde_json::to_value(&stats.tick_timings).expect("serialize tick timings"); - let total_mailbox_depth: usize = stats - .workers - .iter() - .map(|worker| worker.mailbox_depth) - .sum(); - - publish_json( - dashboard, - stream, - position, - RUNTIME_STATS, - json!({ - "num_workers": stats.num_workers, - "uptime_ms": stats.uptime_ms, - "actors_live": stats.actors.len(), - "mailbox_depth": total_mailbox_depth, - "actors": actors, - "workers": workers, - "actor_details": actor_details, - "tick_timings": tick_timings, - }), - ); - - publish_json( - dashboard, - stream, - position, - RUNTIME_WORKERS, - json!({ "workers": stats.workers }), - ); - - publish_json( - dashboard, - stream, - position, - RUNTIME_ACTORS, - json!({ "actors": actor_details }), - ); -} - -fn publish_json( - dashboard: &DashboardHandle, - stream: &StreamId, - position: &mut u64, - channel: &str, - value: Value, -) { - let payload = serde_json::to_vec(&value).expect("serialize dashboard frame"); - let frame = Frame::new( - ChannelId(*position as u32 + 1), - Position(*position), - payload, - ); - dashboard.publish(FrameEvent { - stream: StreamEvent { - node: stream.node.as_str().to_string(), - life: stream.life.0, - }, - channel: channel.to_owned(), - position: *position, - payload: frame.payload.clone(), - }); - *position = position.wrapping_add(1); -} diff --git a/src/config.rs b/src/config.rs index 4c116da..251a8f0 100644 --- a/src/config.rs +++ b/src/config.rs @@ -2,7 +2,6 @@ pub struct RuntimeConfig { pub max_actors: usize, pub channel_buffer_size: usize, - pub num_threads: usize, /// Maximum messages processed per actor per tick. /// Prevents a single actor with a large mailbox from starving others. /// `0` means unlimited (drain entire mailbox). @@ -26,7 +25,6 @@ impl Default for RuntimeConfig { Self { max_actors: DEFAULT_MAX_ACTORS, channel_buffer_size: DEFAULT_CHANNEL_BUFFER_SIZE, - num_threads: 1, actor_message_budget: DEFAULT_ACTOR_MESSAGE_BUDGET, } } diff --git a/src/delivery.rs b/src/delivery.rs index 44a56e0..622dc60 100644 --- a/src/delivery.rs +++ b/src/delivery.rs @@ -1,10 +1,8 @@ use std::any::Any; use std::collections::{HashMap, HashSet}; use std::hash::{BuildHasher, Hasher}; -use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::{Arc, OnceLock, RwLock}; -use std::thread::Thread; - +use parking_lot::RwLock; +use std::sync::Arc; use crate::Error; use crate::actor::{ActorAddress, Message, SpawnRequest}; use crate::channel::Sender; @@ -24,20 +22,17 @@ use crate::stats::WorkerStats; pub struct AddrHasher(u64); impl Hasher for AddrHasher { - #[inline] - fn finish(&self) -> u64 { - self.0 - } - - #[inline] - fn write(&mut self, _bytes: &[u8]) { - // Unused — ActorAddress::hash calls write_u64 directly. - } - - #[inline] fn write_u64(&mut self, i: u64) { self.0 = i; } + + fn write(&mut self, _: &[u8]) { + // unreachable for ActorAddress (uses write_u64 via custom Hash) + } + + fn finish(&self) -> u64 { + self.0 + } } /// BuildHasher for creating AddrHasher instances. @@ -47,8 +42,7 @@ pub struct AddrBuildHasher; impl BuildHasher for AddrBuildHasher { type Hasher = AddrHasher; - #[inline] - fn build_hasher(&self) -> AddrHasher { + fn build_hasher(&self) -> Self::Hasher { AddrHasher(0) } } @@ -60,108 +54,45 @@ pub type AddrMap = HashMap; /// HashSet optimized for ActorAddress keys. pub type AddrSet = HashSet; -// ─── Address Map Types ─────────────────────────────────────────────────────── +// ─── Address Registry ─────────────────────────────────────────────────────── -/// Identifies a worker thread. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub(crate) struct WorkerId(pub(crate) usize); - -impl WorkerId { - pub fn as_usize(self) -> usize { - self.0 - } -} - -/// Maps actor addresses to the worker that owns them. +/// Tracks which actor addresses belong to this runtime. /// -/// `RwLock` — zero contention for parallel reads, write-rare (only on spawn). +/// `RwLock` — zero contention for parallel reads, write-rare (only on spawn). pub(crate) struct AddressMap { - inner: RwLock>, + inner: RwLock, } impl AddressMap { - pub fn with_capacity(cap: usize) -> Self { + pub fn with_capacity(capacity: usize) -> Self { Self { - inner: RwLock::new(HashMap::with_capacity_and_hasher(cap, AddrBuildHasher)), + inner: RwLock::new(HashSet::with_capacity_and_hasher( + capacity, + AddrBuildHasher, + )), } } - pub fn insert(&self, addr: ActorAddress, worker: WorkerId) { - self.inner.write().unwrap().insert(addr, worker); + pub fn insert(&self, addr: ActorAddress) { + self.inner.write().insert(addr); } - pub fn lookup(&self, addr: &ActorAddress) -> Option { - self.inner.read().unwrap().get(addr).copied() + pub fn contains(&self, addr: &ActorAddress) -> bool { + self.inner.read().contains(addr) } - /// Remove an actor address from the map (e.g., after permanent poisoning). pub fn remove(&self, addr: &ActorAddress) { - self.inner.write().unwrap().remove(addr); + self.inner.write().remove(addr); } - /// Returns a snapshot of all (address, worker) pairs. - pub fn snapshot(&self) -> Vec<(ActorAddress, WorkerId)> { - self.inner - .read() - .unwrap() - .iter() - .map(|(addr, wid)| (*addr, *wid)) - .collect() - } -} - -/// Load-aware actor placement strategy. -/// -/// Picks the worker with the lowest load score (actor count + mailbox depth). -/// When all workers have equal load (e.g., before any ticks), falls back to -/// round-robin via a rotating start position for the scan. -pub(crate) struct Placement { - next: AtomicUsize, - num_workers: usize, - worker_stats: Vec>, -} - -impl Placement { - pub fn new(num_workers: usize, worker_stats: Vec>) -> Self { - Self { - next: AtomicUsize::new(0), - num_workers, - worker_stats, - } - } - - pub fn next_worker(&self) -> WorkerId { - let n = self.num_workers; - if n == 1 { - return WorkerId(0); - } - - // Rotate the scan start for round-robin tie-breaking - let rr = self.next.fetch_add(1, Ordering::Relaxed); - - let mut best_id = rr % n; - let mut best_score = usize::MAX; - - for offset in 0..n { - let i = (rr + offset) % n; - let actors = self.worker_stats[i].num_actors.load(Ordering::Relaxed); - let depth = self.worker_stats[i] - .total_mailbox_depth - .load(Ordering::Relaxed); - let score = actors + depth; - if score < best_score { - best_score = score; - best_id = i; - } - } - - WorkerId(best_id) + pub fn addresses(&self) -> Vec { + self.inner.read().iter().copied().collect() } } // ─── Delivery Types ────────────────────────────────────────────────────────── -/// A type-erased message envelope for cross-worker delivery. +/// A type-erased message envelope for depositing into the worker's inbox. /// /// Uses `Box` (no atomic refcount) and move semantics (no clone). pub(crate) struct Envelope { @@ -209,17 +140,17 @@ impl InboxRegistry { } pub fn register(&self, addr: ActorAddress, sender: Arc) { - self.senders.write().unwrap().insert(addr, sender); + self.senders.write().insert(addr, sender); } /// Check if an address is registered without consuming a message. #[cfg(feature = "transport")] pub fn contains(&self, addr: &ActorAddress) -> bool { - self.senders.read().unwrap().contains_key(addr) + self.senders.read().contains_key(addr) } pub fn try_deliver(&self, addr: ActorAddress, msg: Box) -> Result<(), Error> { - let senders = self.senders.read().unwrap(); + let senders = self.senders.read(); if let Some(sender) = senders.get(&addr) { sender.try_send_any(msg); Ok(()) @@ -232,20 +163,15 @@ impl InboxRegistry { /// Shared state passed to tick_once — single thin pointer avoids register spill. pub(crate) struct TickContext<'a> { pub(crate) address_map: &'a AddressMap, - pub(crate) transfer_txs: &'a [Sender], - pub(crate) spawn_txs: &'a [Sender], - pub(crate) placement: &'a Placement, + pub(crate) spawn_tx: &'a Sender, + pub(crate) transfer_tx: &'a Sender, pub(crate) inbox_registry: &'a InboxRegistry, pub(crate) config: &'a RuntimeConfig, pub(crate) extension: Option<&'a dyn crate::extension::RuntimeExtension>, pub(crate) process_output_observer: Option<&'a Arc>, pub(crate) stats_hook: Option<&'a dyn crate::stats::StatsHook>, - /// Thread handles for waking parked workers on cross-worker sends. - pub(crate) worker_threads: &'a [OnceLock], - /// Per-worker stats for summing total_actors across workers. - pub(crate) worker_stats: &'a [Arc], - /// Runtime creation time for computing uptime_ms. + pub(crate) worker_stats: &'a WorkerStats, pub(crate) created_at: crate::Instant, #[cfg(feature = "transport")] pub(crate) remote_sink: Option<&'a dyn crate::runtime::RemoteSink>, diff --git a/src/runtime.rs b/src/runtime.rs index a984966..9f37504 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,11 +1,8 @@ use crate::Instant; use std::any::{Any, TypeId}; use std::cell::RefCell; -use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, OnceLock}; -use std::thread::Thread; -#[cfg(not(target_arch = "wasm32"))] -use std::thread::{self, JoinHandle}; use crate::actor::{ Actor, ActorAddress, ActorInterface, ActorTypeMetadata, AnyActor, Environment, ExitValue, @@ -18,7 +15,7 @@ use crate::admin::{ use crate::channel::{Receiver, Sender}; // Re-export config types so existing code using `runtime::RuntimeConfig` still works pub use crate::config::RuntimeConfig; -use crate::delivery::{AddressMap, Envelope, InboxRegistry, Placement, TickContext, WorkerId}; +use crate::delivery::{AddressMap, Envelope, InboxRegistry, TickContext}; use crate::extension::RuntimeExtension; use crate::stats::{StatsHook, WorkerStats}; // Re-export stats types so existing code using `runtime::*` still works @@ -40,6 +37,16 @@ impl Inbox { pub fn try_recv(&self) -> Option { self.inner.try_recv() } + + pub fn recv_ticking(&self, rt: &Runtime, max_ticks: usize) -> Option { + for _ in 0..max_ticks { + rt.tick(); + if let Some(msg) = self.inner.try_recv() { + return Some(msg); + } + } + None + } } /// Pending ask response — wraps an inbox with convenience recv methods. @@ -51,48 +58,12 @@ pub struct Ask { } impl Ask { - /// Try to receive the response without ticking. pub fn try_recv(&self) -> Option { self.inbox.try_recv() } - /// Tick the runtime until a response arrives or `max_ticks` is exhausted. - /// - /// Only valid for single-threaded runtimes (panics if `num_threads >= 2`). - pub fn recv_ticking(&self, rt: &Runtime, max_ticks: usize) -> Result { - for _ in 0..max_ticks { - rt.tick(); - if let Some(resp) = self.inbox.try_recv() { - return Ok(resp); - } - } - Err(Error::from("ask timeout: no response within max_ticks")) - } - - /// Get the reply address (for manual message construction). - pub fn reply_addr(&self) -> &ActorAddress { - self.inbox.addr() - } -} - -/// Handle for dealing with a runtime that has started via the `Runtime::run()` method. -#[cfg(not(target_arch = "wasm32"))] -pub struct RuntimeHandle { - pub runtime: Arc, - threads: Vec>, -} - -#[cfg(not(target_arch = "wasm32"))] -impl RuntimeHandle { - pub fn join(self) { - for handle in self.threads { - let _ = handle.join(); - } - } - - /// Simple helper, calls the inner `Runtime::shutdown()` method - pub fn shutdown(&self) { - self.runtime.shutdown(); + pub fn recv_ticking(&self, rt: &Runtime, max_ticks: usize) -> Option { + self.inbox.recv_ticking(rt, max_ticks) } } @@ -103,38 +74,43 @@ pub use crate::actor::Ctx; // ─── Runtime ───────────────────────────────────────────────────────────────── /// The `Runtime` struct is the primary gateway for interacting with the framework. +/// +/// Owns a single `Worker` advanced by the caller via `tick()` / `try_tick()`. +/// +/// Core is a transition-only state machine. Each call mutates state and +/// returns immediately, holding no control flow between calls. When to take +/// the next step is the engine's decision, not core's. Any driver that can +/// call `tick` (tokio, std-thread, a test stepper) can host it. pub struct Runtime { config: RuntimeConfig, address_map: Arc, inbox_registry: Arc, extension: Option>, - transfer_txs: Vec>, - spawn_txs: Vec>, - admin_txs: Vec>, - placement: Placement, - is_running: AtomicBool, - worker_stats: Vec>, + transfer_tx: Sender, + spawn_tx: Sender, + admin_tx: Sender, + worker_stats: Arc, stats_hook: Option>, process_output_observer: OnceLock>, - /// Workers available for tick(). run() drains this and moves workers to threads. - tick_workers: RefCell>, - /// Thread handles for waking parked workers. Set by workers on startup via OnceLock. - worker_threads: Arc>>, + worker: RefCell, created_at: Instant, #[cfg(feature = "transport")] remote_sink: Option>, } -// Safety: RefCell> is only accessed from the owning thread via tick(). -// After run() the RefCell is empty and not accessed by worker threads. +// Safety: `RefCell` is only borrowed from the owning thread in +// `tick()` / `try_tick()` / `has_work()` / `with_extension()`. All `&self` +// methods callable through `Arc` from other threads (`send_to`, +// `spawn`, `deliver_raw`, `stats`, `create_sender`) access only `Sync` fields +// (Arcs, atomics, channels) — never the `RefCell`. No worker threads exist. unsafe impl Sync for Runtime {} /// Core's only hook for delivering a message to a **non-local** address. /// -/// Implemented outside core (e.g. `swactor-transport`'s `CodecRemoteSink`), +/// Implemented outside core (e.g., `swactor-transport`'s `CodecRemoteSink`), /// which owns all codec/transport concerns. Core stays codec-free: it hands the /// sink a type-erased message and an address, and nothing more. `Send + Sync` -/// because the sink is stored in an `Arc` and shared across worker threads. +/// because the sink is stored in an `Arc` and shared across threads. #[cfg(feature = "transport")] pub trait RemoteSink: Send + Sync { fn send(&self, addr: ActorAddress, msg: Box) -> Result<(), Error>; @@ -172,27 +148,20 @@ impl RuntimeAddress { /// actor system. pub struct ExternalSender { address_map: Arc, - transfer_txs: Vec>, - worker_threads: Arc>>, + transfer_tx: Sender, } impl Clone for ExternalSender { fn clone(&self) -> Self { Self { address_map: self.address_map.clone(), - transfer_txs: self.transfer_txs.clone(), - worker_threads: self.worker_threads.clone(), + transfer_tx: self.transfer_tx.clone(), } } } -// Safety: All fields are Send+Sync (Arc uses RwLock, -// Sender wraps Arc, Thread is Send+Sync). -unsafe impl Send for ExternalSender {} -unsafe impl Sync for ExternalSender {} - impl ExternalSender { - /// Send a typed message to an actor address, waking the owning worker thread. + /// Send a typed message to an actor address. /// /// Returns `Ok(())` if the message was accepted for routing. This does **not** /// guarantee delivery — the recipient may stop before processing it. If @@ -200,80 +169,47 @@ impl ExternalSender { /// /// Returns `Err` if the address is not found in the runtime's address map. pub fn send_to(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { - match self.address_map.lookup(&addr) { - Some(wid) => { - self.transfer_txs[wid.as_usize()].send(Envelope::new(addr, Box::new(msg))); - notify_worker(&self.worker_threads, wid.as_usize()); - Ok(()) - } - None => Err(Error::from("Address not found")), + if self.address_map.contains(&addr) { + self.transfer_tx.send(Envelope::new(addr, Box::new(msg))); + Ok(()) + } else { + Err(Error::from("Address not found")) } } } 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. + /// Builds a new `Runtime` struct, but does not yet run anything. + /// Drive via `tick()` / `try_tick()`. pub fn new(config: RuntimeConfig) -> Self { - let num_workers = if config.num_threads < 2 { - 1 - } else { - config.num_threads - }; - let address_map = Arc::new(AddressMap::with_capacity(config.max_actors)); let inbox_registry = Arc::new(InboxRegistry::new()); - let mut transfer_txs = Vec::with_capacity(num_workers); - let mut spawn_txs = Vec::with_capacity(num_workers); - let mut admin_txs = Vec::with_capacity(num_workers); - let mut worker_stats = Vec::with_capacity(num_workers); - let mut workers = Vec::with_capacity(num_workers); + let transfer_rx = Receiver::::new(config.channel_buffer_size); + let transfer_tx = transfer_rx.new_sender(); - for i in 0..num_workers { - let transfer_rx = Receiver::::new(config.channel_buffer_size); - let transfer_tx = transfer_rx.new_sender(); - transfer_txs.push(transfer_tx); + let spawn_rx = Receiver::::new(config.max_actors); + let spawn_tx = spawn_rx.new_sender(); - let spawn_rx = Receiver::::new(config.max_actors); - let spawn_tx = spawn_rx.new_sender(); - spawn_txs.push(spawn_tx); + let admin_rx = Receiver::::new(config.channel_buffer_size); + let admin_tx = admin_rx.new_sender(); - let admin_rx = Receiver::::new(config.channel_buffer_size); - let admin_tx = admin_rx.new_sender(); - admin_txs.push(admin_tx); + let worker_stats = Arc::new(WorkerStats::new()); - let stats = Arc::new(WorkerStats::new()); - worker_stats.push(stats.clone()); - workers.push(Worker::new( - WorkerId(i), - transfer_rx, - spawn_rx, - admin_rx, - stats, - )); - } - - let placement = Placement::new(num_workers, worker_stats.clone()); - - let worker_threads: Arc>> = - Arc::new((0..num_workers).map(|_| OnceLock::new()).collect()); + let worker = Worker::new(transfer_rx, spawn_rx, admin_rx, worker_stats.clone()); let rt = Self { config, address_map, inbox_registry, extension: None, - transfer_txs, - spawn_txs, - admin_txs, - placement, - is_running: AtomicBool::new(false), + transfer_tx, + spawn_tx, + admin_tx, worker_stats, stats_hook: None, process_output_observer: OnceLock::new(), - tick_workers: RefCell::new(workers), - worker_threads, + worker: RefCell::new(worker), created_at: Instant::now(), #[cfg(feature = "transport")] remote_sink: None, @@ -281,7 +217,6 @@ impl Runtime { #[cfg(feature = "tracing")] tracing::info!( - num_workers, max_actors = rt.config.max_actors, "runtime.created" ); @@ -292,10 +227,9 @@ impl Runtime { /// Spawn an actor, returns its address pub fn spawn(&self, actor: A) -> Result { let addr = ActorAddress::new_random(); - let worker_id = self.placement.next_worker(); - self.address_map.insert(addr, worker_id); + self.address_map.insert(addr); let boxed: Box = Box::new(Actor::new(actor)); - self.spawn_txs[worker_id.as_usize()].send(SpawnRequest { + self.spawn_tx.send(SpawnRequest { addr, actor: boxed, parent: None, @@ -305,7 +239,6 @@ impl Runtime { #[cfg(feature = "tracing")] tracing::info!( actor_addr = %addr, - worker_id = worker_id.as_usize(), "actor.spawned" ); @@ -319,10 +252,9 @@ impl Runtime { env: Environment, ) -> Result { let addr = ActorAddress::new_random(); - let worker_id = self.placement.next_worker(); - self.address_map.insert(addr, worker_id); + self.address_map.insert(addr); let boxed: Box = Box::new(Actor::new(actor)); - self.spawn_txs[worker_id.as_usize()].send(SpawnRequest { + self.spawn_tx.send(SpawnRequest { addr, actor: boxed, parent: None, @@ -332,7 +264,6 @@ impl Runtime { #[cfg(feature = "tracing")] tracing::info!( actor_addr = %addr, - worker_id = worker_id.as_usize(), "actor.spawned" ); @@ -342,13 +273,10 @@ impl Runtime { /// Install a runtime extension. Extensions provide higher-level features /// (naming, monitoring, groups) via lifecycle hooks. /// - /// Must be called before `run()` or `tick()`. + /// Must be called before `tick()`. pub fn with_extension(mut self, ext: Arc) -> Self { - // Create per-worker extensions (e.g., timer wheels) - for worker in self.tick_workers.get_mut().iter_mut() { - if let Some(wext) = ext.create_worker_extension() { - worker.worker_ext = Some(wext); - } + if let Some(wext) = ext.create_worker_extension() { + self.worker.get_mut().worker_ext = Some(wext); } self.extension = Some(ext); self @@ -414,23 +342,20 @@ impl Runtime { pub fn create_sender(&self) -> ExternalSender { ExternalSender { address_map: self.address_map.clone(), - transfer_txs: self.transfer_txs.iter().cloned().collect(), - worker_threads: self.worker_threads.clone(), + transfer_tx: self.transfer_tx.clone(), } } fn make_tick_context(&self) -> TickContext<'_> { TickContext { address_map: &self.address_map, - transfer_txs: &self.transfer_txs, - spawn_txs: &self.spawn_txs, - placement: &self.placement, + spawn_tx: &self.spawn_tx, + transfer_tx: &self.transfer_tx, inbox_registry: &self.inbox_registry, config: &self.config, extension: self.extension.as_deref(), process_output_observer: self.process_output_observer.get(), stats_hook: self.stats_hook.as_deref(), - worker_threads: &self.worker_threads, worker_stats: &self.worker_stats, created_at: self.created_at, #[cfg(feature = "transport")] @@ -438,120 +363,42 @@ impl Runtime { } } - /// Return whether the single-threaded runtime currently has schedulable work. - /// - /// Panics if called on a multi-threaded runtime — use `run()` instead. + /// Return whether the runtime currently has schedulable work. pub fn has_work(&self) -> bool { - assert!( - self.config.num_threads < 2, - "has_work() is only valid for single-threaded runtimes; use run() for multi-threaded" - ); - - self.tick_workers.borrow().iter().any(Worker::has_work) + self.worker.borrow().has_work() } - /// Try to drive one tick of the single-threaded runtime. + /// Try to drive one tick of the runtime. /// - /// Returns `false` if no worker performed work. - /// Returns `true` if at least one worker performed work. - /// - /// Panics if called on a multi-threaded runtime — use `run()` instead. + /// Returns `false` if no work was performed. + /// Returns `true` if at least one actor was processed. pub fn try_tick(&self) -> bool { - assert!( - self.config.num_threads < 2, - "try_tick() is only valid for single-threaded runtimes; use run() for multi-threaded" - ); - let tc = self.make_tick_context(); - self.tick_workers - .borrow_mut() - .iter_mut() - .fold(false, |did_work, worker| worker.tick_once(&tc) || did_work) + self.worker.borrow_mut().tick_once(&tc) } - /// Drive one tick of the single-threaded worker. - /// - /// Panics if called on a multi-threaded runtime — use `run()` instead. + /// Drive one tick of the runtime. pub fn tick(&self) { let _ = self.try_tick(); } - /// Spawn worker threads and start processing, returning a handle - /// to interact with the runtime and join the threads later. - /// - /// Works in both single-threaded and multi-threaded configurations. - /// In single-threaded mode, one background thread is spawned. - /// - /// Not available on wasm32 — use the browser crate's Web Worker-based run instead. - #[cfg(not(target_arch = "wasm32"))] - pub fn run(self) -> Result { - self.is_running.store(true, Ordering::Release); - - #[cfg(feature = "tracing")] - tracing::info!( - num_workers = self.config.num_threads.max(1), - "runtime.started" - ); - - let workers: Vec = self.tick_workers.replace(Vec::new()); - - let rt = Arc::new(self); - let mut handles: Vec> = Vec::with_capacity(workers.len()); - - for mut worker in workers { - let rt_clone = rt.clone(); - let worker_id = worker.id.0; - let name = format!("swactor-worker-{}", worker_id); - let handle = thread::Builder::new() - .name(name) - .spawn(move || { - // Register this thread so send_to/spawn can unpark us - let _ = rt_clone.worker_threads[worker_id].set(thread::current()); - let tc = rt_clone.make_tick_context(); - worker.run(&tc, &rt_clone.is_running); - }) - .expect("failed to spawn worker thread"); - handles.push(handle); - } - - Ok(RuntimeHandle { - runtime: rt, - threads: handles, - }) - } - /// Returns a snapshot of runtime stats: actor placements and per-worker info. pub fn stats(&self) -> RuntimeStats { - let num_workers = if self.config.num_threads < 2 { - 1 - } else { - self.config.num_threads - }; - - let workers = self - .worker_stats - .iter() - .enumerate() - .map(|(i, ws)| ws.snapshot(i)) - .collect(); + let workers = vec![self.worker_stats.snapshot(0)]; let actors = self .address_map - .snapshot() + .addresses() .into_iter() - .map(|(addr, wid)| (addr, wid.as_usize())) + .map(|addr| (addr, 0)) .collect(); - let tick_timings = self - .worker_stats - .iter() - .map(|ws| ws.drain_tick_timings()) - .collect(); + let tick_timings = vec![self.worker_stats.drain_tick_timings()]; let uptime_ms = self.created_at.elapsed().as_millis() as u64; RuntimeStats { - num_workers, + num_workers: 1, uptime_ms, actors, workers, @@ -567,33 +414,18 @@ impl Runtime { /// /// Returns `Err` if the actor address is not found in the runtime. pub fn stop_actor(&self, addr: ActorAddress) -> Result<(), Error> { - match self.address_map.lookup(&addr) { - Some(wid) => { - self.transfer_txs[wid.as_usize()].send(Envelope::new(addr, Box::new(StopSignal))); - notify_worker(&self.worker_threads, wid.as_usize()); - Ok(()) - } - None => Err(Error::from("Actor not found")), - } - } - - /// Signal all workers to stop and wake any that are parked. - pub fn shutdown(&self) { - #[cfg(feature = "tracing")] - tracing::info!("runtime.shutdown"); - - self.is_running.store(false, Ordering::Release); - // Wake all parked workers so they see the shutdown flag immediately - for thread in self.worker_threads.iter() { - if let Some(t) = thread.get() { - t.unpark(); - } + if self.address_map.contains(&addr) { + self.transfer_tx + .send(Envelope::new(addr, Box::new(StopSignal))); + Ok(()) + } else { + Err(Error::from("Actor not found")) } } /// Set a stats hook to receive per-actor snapshots from workers. /// - /// Must be called before [`run()`](Self::run) or [`tick()`](Self::tick). + /// Must be called before [`tick()`](Self::tick). pub fn set_stats_hook(&mut self, hook: Arc) { self.stats_hook = Some(hook); } @@ -613,12 +445,11 @@ impl Runtime { /// this to inject the resulting message for a local actor or inbox. #[cfg(feature = "transport")] pub fn deliver_raw(&self, addr: ActorAddress, msg: Box) -> Result<(), Error> { - match self.address_map.lookup(&addr) { - Some(wid) => { - self.transfer_txs[wid.as_usize()].send(Envelope::new(addr, msg)); - Ok(()) - } - None => self.inbox_registry.try_deliver(addr, msg), + if self.address_map.contains(&addr) { + self.transfer_tx.send(Envelope::new(addr, msg)); + Ok(()) + } else { + self.inbox_registry.try_deliver(addr, msg) } } } @@ -642,27 +473,26 @@ impl RuntimeAdmin<'_> { pub fn list_actors(&self) -> Result, Error> { let (admin, reply_to) = self.new_admin::()?; let acc = Arc::new(ListActorsAccumulator { - remaining: AtomicUsize::new(self.runtime.admin_txs.len()), + remaining: AtomicUsize::new(1), summaries: parking_lot::Mutex::new(Vec::new()), reply_to, }); - for (idx, tx) in self.runtime.admin_txs.iter().enumerate() { - tx.send(AdminCommand::ListActors { acc: acc.clone() }); - notify_worker(&self.runtime.worker_threads, idx); - } + self.runtime + .admin_tx + .send(AdminCommand::ListActors { acc: acc.clone() }); Ok(admin) } pub fn inspect_actor(&self, actor: ActorAddress) -> Result, Error> { - let Some(wid) = self.runtime.address_map.lookup(&actor) else { + if !self.runtime.address_map.contains(&actor) { return self.ready::(Err(AdminError::ActorNotFound { actor })); - }; + } let (admin, reply_to) = self.new_admin::()?; - let worker_idx = wid.as_usize(); - self.runtime.admin_txs[worker_idx].send(AdminCommand::InspectActor { actor, reply_to }); - notify_worker(&self.runtime.worker_threads, worker_idx); + self.runtime + .admin_tx + .send(AdminCommand::InspectActor { actor, reply_to }); Ok(admin) } @@ -673,10 +503,10 @@ impl RuntimeAdmin<'_> { where A: ActorInterface + Clone + Sync, { - let Some(wid) = self.runtime.address_map.lookup(&actor) else { + if !self.runtime.address_map.contains(&actor) { return self .ready::>(Err(AdminError::ActorNotFound { actor })); - }; + } let (admin, reply_to) = self.new_admin::>()?; let get = Box::new( @@ -728,14 +558,12 @@ impl RuntimeAdmin<'_> { )) as Box }); - let worker_idx = wid.as_usize(); - self.runtime.admin_txs[worker_idx].send(AdminCommand::GetActorState { + self.runtime.admin_tx.send(AdminCommand::GetActorState { actor, reply_to, get, not_found, }); - notify_worker(&self.runtime.worker_threads, worker_idx); Ok(admin) } @@ -754,9 +582,9 @@ impl RuntimeAdmin<'_> { })); } - let Some(wid) = self.runtime.address_map.lookup(&actor) else { + if !self.runtime.address_map.contains(&actor) { return self.ready::(Err(AdminError::ActorNotFound { actor })); - }; + } let (admin, reply_to) = self.new_admin::()?; let actor_instance = state.actor_instance; @@ -791,104 +619,89 @@ impl RuntimeAdmin<'_> { }, ); - let worker_idx = wid.as_usize(); - self.runtime.admin_txs[worker_idx].send(AdminCommand::ReplaceActorState { - actor, - reply_to, - replace, - }); - notify_worker(&self.runtime.worker_threads, worker_idx); + self.runtime + .admin_tx + .send(AdminCommand::ReplaceActorState { + actor, + reply_to, + replace, + }); Ok(admin) } pub fn stop_actor(&self, actor: ActorAddress) -> Result, Error> { - let Some(wid) = self.runtime.address_map.lookup(&actor) else { + if !self.runtime.address_map.contains(&actor) { return self.ready::(Err(AdminError::ActorNotFound { actor })); - }; + } let (admin, reply_to) = self.new_admin::()?; - let worker_idx = wid.as_usize(); - self.runtime.admin_txs[worker_idx].send(AdminCommand::StopActor { actor, reply_to }); - notify_worker(&self.runtime.worker_threads, worker_idx); + self.runtime + .admin_tx + .send(AdminCommand::StopActor { actor, reply_to }); Ok(admin) } pub fn suspend_actor(&self, actor: ActorAddress) -> Result, Error> { - let Some(wid) = self.runtime.address_map.lookup(&actor) else { + if !self.runtime.address_map.contains(&actor) { return self.ready::(Err(AdminError::ActorNotFound { actor })); - }; + } let (admin, reply_to) = self.new_admin::()?; - let worker_idx = wid.as_usize(); - self.runtime.admin_txs[worker_idx].send(AdminCommand::SuspendActor { actor, reply_to }); - notify_worker(&self.runtime.worker_threads, worker_idx); + self.runtime + .admin_tx + .send(AdminCommand::SuspendActor { actor, reply_to }); Ok(admin) } pub fn resume_actor(&self, actor: ActorAddress) -> Result, Error> { - let Some(wid) = self.runtime.address_map.lookup(&actor) else { + if !self.runtime.address_map.contains(&actor) { return self.ready::(Err(AdminError::ActorNotFound { actor })); - }; + } let (admin, reply_to) = self.new_admin::()?; - let worker_idx = wid.as_usize(); - self.runtime.admin_txs[worker_idx].send(AdminCommand::ResumeActor { actor, reply_to }); - notify_worker(&self.runtime.worker_threads, worker_idx); + self.runtime + .admin_tx + .send(AdminCommand::ResumeActor { actor, reply_to }); Ok(admin) } } -/// Wake a parked worker thread so it can process new work. -/// No-op if the thread handle hasn't been registered yet (single-threaded tick mode). -#[inline] -pub(crate) fn notify_worker(threads: &[OnceLock], wid: usize) { - if let Some(t) = threads.get(wid).and_then(|o| o.get()) { - t.unpark(); - } -} - #[allow(private_interfaces)] 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()].send(Envelope::new(addr, msg)); - notify_worker(&self.worker_threads, wid.as_usize()); - Ok(()) - } - None => self.make_tick_context().route_nonlocal(addr, msg), + if self.address_map.contains(&addr) { + self.transfer_tx.send(Envelope::new(addr, msg)); + Ok(()) + } else { + self.make_tick_context().route_nonlocal(addr, msg) } } fn spawn_any(&self, request: SpawnRequest) { - let worker_id = self.placement.next_worker(); - self.address_map.insert(request.addr, worker_id); - self.spawn_txs[worker_id.as_usize()].send(request); - notify_worker(&self.worker_threads, worker_id.as_usize()); + self.address_map.insert(request.addr); + self.spawn_tx.send(request); } fn request_stop(&self, addr: ActorAddress) { - // From spawn context (outside worker), send StopSignal through transfer queue - if let Some(wid) = self.address_map.lookup(&addr) { - self.transfer_txs[wid.as_usize()].send(Envelope::new(addr, Box::new(StopSignal))); - notify_worker(&self.worker_threads, wid.as_usize()); + if self.address_map.contains(&addr) { + self.transfer_tx + .send(Envelope::new(addr, Box::new(StopSignal))); } } fn request_stop_with(&self, addr: ActorAddress, value: ExitValue) { - if let Some(wid) = self.address_map.lookup(&addr) { - self.transfer_txs[wid.as_usize()] + if self.address_map.contains(&addr) { + self.transfer_tx .send(Envelope::new(addr, Box::new(StopWithSignal(value)))); - notify_worker(&self.worker_threads, wid.as_usize()); } } fn request_suspend(&self, addr: ActorAddress) { - // Outside worker context — not supported (suspend is per-actor, from handler) + // From spawn context (outside worker), not supported (suspend is per-actor, from handler) eprintln!("swactor: request_suspend called outside worker context for {addr} — ignored"); } fn request_resume(&self, addr: ActorAddress) { - if let Some(wid) = self.address_map.lookup(&addr) { - self.transfer_txs[wid.as_usize()].send(Envelope::new(addr, Box::new(ResumeSignal))); - notify_worker(&self.worker_threads, wid.as_usize()); + if self.address_map.contains(&addr) { + self.transfer_tx + .send(Envelope::new(addr, Box::new(ResumeSignal))); } } @@ -909,16 +722,10 @@ impl ContextInner for Runtime { } fn system_info(&self) -> SystemInfo { - let num_workers = self.config.num_threads.max(1); - let total_actors: usize = self - .worker_stats - .iter() - .map(|ws| ws.num_actors.load(Ordering::Relaxed)) - .sum(); SystemInfo { worker_id: 0, - num_workers, - total_actors, + num_workers: 1, + total_actors: self.worker_stats.num_actors.load(Ordering::Relaxed), uptime_ms: self.created_at.elapsed().as_millis() as u64, } } diff --git a/src/stats.rs b/src/stats.rs index 962183d..800d50f 100644 --- a/src/stats.rs +++ b/src/stats.rs @@ -25,15 +25,10 @@ pub struct WorkerStats { pub messages_processed: AtomicU64, // Message routing counters pub local_sends: AtomicU64, - pub cross_sends: AtomicU64, pub inbox_sends: AtomicU64, // Error counters pub type_mismatches: AtomicU64, pub panics: AtomicU64, - /// Messages dropped due to mailbox overflow (bounded mailbox policy). - pub messages_dropped: AtomicU64, - /// Number of actor restarts after panic (restartable actors only). - pub restarts: AtomicU64, /// Number of actors gracefully stopped via `ctx.stop_self()` or `Runtime::stop_actor()`. pub stops: AtomicU64, // Tick timing ring buffer (last N ticks, lock-free) @@ -53,12 +48,9 @@ impl WorkerStats { total_mailbox_depth: AtomicUsize::new(0), messages_processed: AtomicU64::new(0), local_sends: AtomicU64::new(0), - cross_sends: AtomicU64::new(0), inbox_sends: AtomicU64::new(0), type_mismatches: AtomicU64::new(0), panics: AtomicU64::new(0), - messages_dropped: AtomicU64::new(0), - restarts: AtomicU64::new(0), stops: AtomicU64::new(0), tick_timings: ArrayQueue::new(TICK_BUFFER_CAP), } @@ -90,12 +82,9 @@ impl WorkerStats { mailbox_depth: self.total_mailbox_depth.load(Relaxed), messages_processed: self.messages_processed.load(Relaxed), local_sends: self.local_sends.load(Relaxed), - cross_sends: self.cross_sends.load(Relaxed), inbox_sends: self.inbox_sends.load(Relaxed), type_mismatches: self.type_mismatches.load(Relaxed), panics: self.panics.load(Relaxed), - messages_dropped: self.messages_dropped.load(Relaxed), - restarts: self.restarts.load(Relaxed), stops: self.stops.load(Relaxed), } } @@ -110,12 +99,9 @@ pub struct WorkerInfo { pub mailbox_depth: usize, pub messages_processed: u64, pub local_sends: u64, - pub cross_sends: u64, pub inbox_sends: u64, pub type_mismatches: u64, pub panics: u64, - pub messages_dropped: u64, - pub restarts: u64, pub stops: u64, } diff --git a/src/worker.rs b/src/worker.rs index 86e2d0e..a9f65c9 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -3,8 +3,7 @@ use std::any::Any; use std::cell::RefCell; use std::collections::{HashMap, VecDeque}; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, Ordering}; -use std::thread; +use std::sync::atomic::Ordering; use crate::Error; use crate::actor::{ @@ -16,7 +15,7 @@ use crate::admin::{ ListActorsResponse, OperationResult, }; use crate::channel::Receiver; -use crate::delivery::{AddrBuildHasher, AddrMap, Envelope, TickContext, WorkerId}; +use crate::delivery::{AddrBuildHasher, AddrMap, Envelope, TickContext}; use crate::stats::{ActorSnapshot, TickTiming, WorkerStats}; use crate::extension::WorkerExtension; @@ -42,7 +41,7 @@ pub(crate) fn determine_stop_reason(poisoned: bool, has_exit_value: bool) -> Sto } } -/// Route a message: try local pool first, then address_map for cross-worker, +/// Route a message: try local pool first, then deposit for pending-spawn actors, /// then inbox_registry for external receivers. fn route_to_pool_or_remote( pool: &mut ActorPool, @@ -52,24 +51,18 @@ fn route_to_pool_or_remote( ) { if pool.contains(&dest) { pool.deliver(&dest, msg); + } else if tc.address_map.contains(&dest) { + // Actor exists but not yet in pool (pending spawn) — deposit for next tick + tc.transfer_tx.send(Envelope::new(dest, msg)); } else { - match tc.address_map.lookup(&dest) { - Some(wid) => { - tc.transfer_txs[wid.as_usize()].send(Envelope::new(dest, msg)); - crate::runtime::notify_worker(tc.worker_threads, wid.as_usize()); - } - None => { - let _ = tc.inbox_registry.try_deliver(dest, msg); - } - } + let _ = tc.route_nonlocal(dest, msg); } } // ─── Worker ───────────────────────────────────────────────────────────────── -/// A worker owns a set of actors and runs them in a loop. +/// A worker owns a set of actors and processes them via tick_once. pub(crate) struct Worker { - pub(crate) id: WorkerId, pub(crate) pool: ActorPool, transfer_rx: Receiver, spawn_rx: Receiver, @@ -86,14 +79,12 @@ pub(crate) struct Worker { impl Worker { pub(crate) fn new( - id: WorkerId, transfer_rx: Receiver, spawn_rx: Receiver, admin_rx: Receiver, stats: Arc, ) -> Self { Self { - id, pool: ActorPool::new(), transfer_rx, spawn_rx, @@ -142,7 +133,7 @@ impl Worker { #[cfg(feature = "tracing")] if spawn_count > 0 { tracing::debug!( - worker_id = self.id.0, + worker_id = 0, count = spawn_count, "worker.spawns_drained" ); @@ -171,7 +162,7 @@ impl Worker { match cmd { AdminCommand::ListActors { acc } => { let mut local = Vec::new(); - self.pool.actor_summaries_into(self.id, &mut local); + self.pool.actor_summaries_into(&mut local); { let mut summaries = acc.summaries.lock(); summaries.extend(local); @@ -187,7 +178,7 @@ impl Worker { AdminCommand::InspectActor { actor, reply_to } => { let result = self .pool - .actor_summary(self.id, actor) + .actor_summary(actor) .map(|summary| InspectActorResponse { summary }); Self::send_admin_reply(tc, reply_to, result); } @@ -242,7 +233,6 @@ impl Worker { let cleanup_requests: RefCell>> = RefCell::new(Vec::new()); let dead = { let cleanup_ctx = WorkerContext { - worker_id: self.id, tc, pending_local: &cleanup_pending, stop_requests: &cleanup_stops, @@ -290,7 +280,7 @@ impl Worker { pub(crate) fn tick_once(&mut self, tc: &TickContext) -> bool { #[cfg(feature = "tracing")] - let _span = tracing::trace_span!("worker.tick", worker_id = self.id.0).entered(); + let _span = tracing::trace_span!("worker.tick", worker_id = 0).entered(); // Fast idle path: skip the entire tick when nothing could have changed. // Cost: ~3 atomic loads, zero syscalls, zero actor iteration. @@ -339,7 +329,6 @@ impl Worker { let processed; { let worker_ctx = WorkerContext { - worker_id: self.id, tc, pending_local: &pending_local, stop_requests: &stop_requests, @@ -352,9 +341,6 @@ impl Worker { &worker_ctx, &self.stats, tc.config.actor_message_budget, - &stop_requests, - &stop_with_values, - &suspend_requests, ); if processed > 0 { did_work = true; @@ -365,7 +351,7 @@ impl Worker { #[cfg(feature = "tracing")] if processed > 0 { tracing::debug!( - worker_id = self.id.0, + worker_id = 0, messages_processed = processed, "worker.tick_all" ); @@ -408,7 +394,7 @@ impl Worker { if let Some(hook) = tc.stats_hook { self.pool.mailbox_depths_into(&mut self.snapshot_buf); - hook.on_tick(self.id.0, &self.snapshot_buf); + hook.on_tick(0, &self.snapshot_buf); } } @@ -432,7 +418,7 @@ impl Worker { #[cfg(feature = "tracing")] if did_work { tracing::debug!( - worker_id = self.id.0, + worker_id = 0, num_actors = self.pool.len(), mailbox_depth = self.pool.total_mailbox_depth(), messages_processed = processed, @@ -447,26 +433,13 @@ impl Worker { did_work } - pub(crate) fn run(&mut self, tc: &TickContext, is_running: &AtomicBool) { - #[cfg(feature = "tracing")] - let _span = tracing::info_span!("worker.run", worker_id = self.id.0).entered(); - - while is_running.load(Ordering::Acquire) { - if !self.tick_once(tc) { - // Park indefinitely — woken by unpark() from send_to/spawn/stop/shutdown. - // Spurious wakes hit the fast idle path (~3 atomic loads) and park again. - thread::park(); - } - } - } } -/// The `ContextInner` impl for worker threads. +/// The `ContextInner` impl for in-worker sends. /// -/// Same-worker sends are buffered in `pending_local` (delivered after current tick round). -/// Cross-worker sends go through the transfer queue. +/// All sends to local actors are buffered in `pending_local` (delivered after +/// the current tick round). Non-local addresses route to inbox_registry or remote. struct WorkerContext<'a> { - worker_id: WorkerId, tc: &'a TickContext<'a>, pending_local: &'a RefCell)>>, stop_requests: &'a RefCell>, @@ -478,30 +451,19 @@ struct WorkerContext<'a> { impl ContextInner for WorkerContext<'_> { fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error> { - match self.tc.address_map.lookup(&addr) { - Some(wid) if wid == self.worker_id => { - self.stats.local_sends.fetch_add(1, Ordering::Relaxed); - self.pending_local.borrow_mut().push((addr, msg)); - Ok(()) - } - Some(wid) => { - self.stats.cross_sends.fetch_add(1, Ordering::Relaxed); - self.tc.transfer_txs[wid.as_usize()].send(Envelope::new(addr, msg)); - crate::runtime::notify_worker(self.tc.worker_threads, wid.as_usize()); - Ok(()) - } - None => { - self.stats.inbox_sends.fetch_add(1, Ordering::Relaxed); - self.tc.route_nonlocal(addr, msg) - } + if self.tc.address_map.contains(&addr) { + self.stats.local_sends.fetch_add(1, Ordering::Relaxed); + self.pending_local.borrow_mut().push((addr, msg)); + Ok(()) + } else { + self.stats.inbox_sends.fetch_add(1, Ordering::Relaxed); + self.tc.route_nonlocal(addr, msg) } } fn spawn_any(&self, request: SpawnRequest) { - let worker_id = self.tc.placement.next_worker(); - self.tc.address_map.insert(request.addr, worker_id); - self.tc.spawn_txs[worker_id.as_usize()].send(request); - crate::runtime::notify_worker(self.tc.worker_threads, worker_id.as_usize()); + self.tc.address_map.insert(request.addr); + self.tc.spawn_tx.send(request); } fn request_stop(&self, addr: ActorAddress) { @@ -517,8 +479,6 @@ impl ContextInner for WorkerContext<'_> { } fn request_resume(&self, addr: ActorAddress) { - // Same-worker: buffer as pending_local ResumeSignal - // Cross-worker: would go through transfer queue (handled by Runtime impl) self.pending_local .borrow_mut() .push((addr, Box::new(ResumeSignal))); @@ -539,17 +499,10 @@ impl ContextInner for WorkerContext<'_> { } fn system_info(&self) -> SystemInfo { - let num_workers = self.tc.config.num_threads.max(1); - let total_actors: usize = self - .tc - .worker_stats - .iter() - .map(|ws| ws.num_actors.load(Ordering::Relaxed)) - .sum(); SystemInfo { - worker_id: self.worker_id.0, - num_workers, - total_actors, + worker_id: 0, + num_workers: 1, + total_actors: self.tc.worker_stats.num_actors.load(Ordering::Relaxed), uptime_ms: self.tc.created_at.elapsed().as_millis() as u64, } } @@ -652,17 +605,13 @@ impl ActorPool { self.actors.get_mut(&addr).map(|slot| slot.actor.as_mut()) } - fn actor_summary_from_slot( - worker_id: WorkerId, - address: ActorAddress, - slot: &ActorSlot, - ) -> ActorSummary { + fn actor_summary_from_slot(address: ActorAddress, slot: &ActorSlot) -> ActorSummary { let metadata = slot.actor.metadata(); ActorSummary { address, actor_type: metadata.actor_type_name, message_type: metadata.message_type_name, - worker_id: worker_id.as_usize(), + worker_id: 0, parent: slot.parent_addr, mailbox_depth: slot.mailbox.len(), status: ActorStatus { @@ -676,19 +625,19 @@ impl ActorPool { } } - fn actor_summary(&self, worker_id: WorkerId, addr: ActorAddress) -> AdminResult { + fn actor_summary(&self, addr: ActorAddress) -> AdminResult { self.actors .get(&addr) - .map(|slot| Self::actor_summary_from_slot(worker_id, addr, slot)) + .map(|slot| Self::actor_summary_from_slot(addr, slot)) .ok_or(AdminError::ActorNotFound { actor: addr }) } - fn actor_summaries_into(&self, worker_id: WorkerId, out: &mut Vec) { + fn actor_summaries_into(&self, out: &mut Vec) { out.clear(); out.extend( self.actors .iter() - .map(|(&addr, slot)| Self::actor_summary_from_slot(worker_id, addr, slot)), + .map(|(&addr, slot)| Self::actor_summary_from_slot(addr, slot)), ); } @@ -734,14 +683,11 @@ impl ActorPool { /// /// Each actor processes up to `budget` messages per tick (0 = unlimited). /// This prevents a single hot actor from starving others on the same worker. - pub fn tick_all( + fn tick_all( &mut self, - inner: &dyn ContextInner, + wctx: &WorkerContext<'_>, stats: &WorkerStats, budget: usize, - stop_requests: &RefCell>, - stop_with_values: &RefCell>, - suspend_requests: &RefCell>, ) -> usize { let mut count = 0; for (&addr, slot) in self.actors.iter_mut() { @@ -769,7 +715,7 @@ impl ActorPool { snap_type_counts.sort_by(|a, b| b.1.cmp(&a.1)); let ctx = Ctx::new( - inner, + wctx, addr, slot.parent_addr, slot.env.clone(), @@ -795,14 +741,14 @@ impl ActorPool { } // Check if on_start requested stop or stop_with { - let stops = stop_requests.borrow(); + let stops = wctx.stop_requests.borrow(); if !stops.is_empty() && stops.contains(&addr) { drop(stops); slot.stopping = true; stats.stops.fetch_add(1, Ordering::Relaxed); slot.mailbox.clear(); // Check for stop_with value - let mut sws = stop_with_values.borrow_mut(); + let mut sws = wctx.stop_with_values.borrow_mut(); if let Some(pos) = sws.iter().position(|(a, _)| *a == addr) { let (_, val) = sws.swap_remove(pos); slot.exit_value = Some(val); @@ -812,7 +758,7 @@ impl ActorPool { } // Check if on_start requested stop_with (without plain stop) { - let mut sws = stop_with_values.borrow_mut(); + let mut sws = wctx.stop_with_values.borrow_mut(); if let Some(pos) = sws.iter().position(|(a, _)| *a == addr) { let (_, val) = sws.swap_remove(pos); slot.exit_value = Some(val); @@ -824,7 +770,7 @@ impl ActorPool { } // Check if on_start requested suspend { - let suspends = suspend_requests.borrow(); + let suspends = wctx.suspend_requests.borrow(); if !suspends.is_empty() && suspends.contains(&addr) { drop(suspends); slot.suspended = true; @@ -891,11 +837,11 @@ impl ActorPool { // Check if handler requested self-stop or stop_with { - let stops = stop_requests.borrow(); + let stops = wctx.stop_requests.borrow(); let has_stop = !stops.is_empty() && stops.contains(&addr); drop(stops); - let mut sws = stop_with_values.borrow_mut(); + let mut sws = wctx.stop_with_values.borrow_mut(); let sw_pos = sws.iter().position(|(a, _)| *a == addr); if has_stop || sw_pos.is_some() { @@ -914,7 +860,7 @@ impl ActorPool { // Check if handler requested suspend { - let suspends = suspend_requests.borrow(); + let suspends = wctx.suspend_requests.borrow(); if !suspends.is_empty() && suspends.contains(&addr) { drop(suspends); slot.suspended = true; diff --git a/tests/message_delivery.rs b/tests/message_delivery.rs index abbc610..a551a01 100644 --- a/tests/message_delivery.rs +++ b/tests/message_delivery.rs @@ -111,7 +111,6 @@ fn message_routing_at_scale() { let rt = std_runtime(RuntimeConfig { max_actors: 300, channel_buffer_size: 1024, - num_threads: 1, ..Default::default() }); let inbox = rt.new_inbox::().unwrap(); @@ -148,7 +147,6 @@ fn message_routing_at_scale() { let rt = std_runtime(RuntimeConfig { max_actors: 200, channel_buffer_size: 1024, - num_threads: 1, ..Default::default() }); let inbox = rt.new_inbox::().unwrap(); diff --git a/tests/runtime_admin.rs b/tests/runtime_admin.rs index 28ec3f7..b42c167 100644 --- a/tests/runtime_admin.rs +++ b/tests/runtime_admin.rs @@ -6,7 +6,6 @@ use common::*; use std::collections::HashSet; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::time::{Duration, Instant}; use swactor::admin::{ActorStateSnapshot, AdminError, OperationResult}; use swactor::config::RuntimeConfig; @@ -75,30 +74,6 @@ impl ActorInterface for StopProbe { } } -fn poll_admin( - admin: &swactor::admin::Admin, - timeout: Duration, -) -> Option> { - let start = Instant::now(); - while start.elapsed() < timeout { - if let Some(value) = admin.try_recv() { - return Some(value); - } - std::thread::sleep(Duration::from_millis(5)); - } - None -} - -fn poll_inbox(inbox: &Inbox, timeout: Duration) -> Option { - let start = Instant::now(); - while start.elapsed() < timeout { - if let Some(value) = inbox.try_recv() { - return Some(value); - } - std::thread::sleep(Duration::from_millis(5)); - } - None -} fn operation_applied() -> OperationResult { OperationResult { applied: true } @@ -500,11 +475,8 @@ fn admin_stop_clears_pending_mailbox_without_calling_handle() { } #[test] -fn threaded_admin_suspend_resume_wakes_parked_worker() { - let rt = std_runtime(RuntimeConfig { - num_threads: 2, - ..Default::default() - }); +fn admin_suspend_resume() { + let rt = std_runtime(RuntimeConfig::default()); let counter = Arc::new(AtomicUsize::new(0)); let addr = rt .spawn(CountingPingActor { @@ -512,45 +484,46 @@ fn threaded_admin_suspend_resume_wakes_parked_worker() { }) .unwrap(); let pong_inbox = rt.new_inbox::().unwrap(); + rt.tick(); // process on_start - let handle = rt.run().unwrap(); - std::thread::sleep(Duration::from_millis(50)); + // Suspend + let suspended = rt.admin().suspend_actor(addr).unwrap(); + let suspended = suspended.recv_ticking(&rt, 5); + assert_eq!(suspended, Ok(operation_applied())); - let suspended = handle.runtime.admin().suspend_actor(addr).unwrap(); - let suspended = poll_admin(&suspended, Duration::from_secs(1)); + // Send while suspended — should not process + rt.send_to( + addr, + Ping { + reply_to: *pong_inbox.addr(), + }, + ) + .unwrap(); + tick_n(&rt, 3); + assert!( + pong_inbox.try_recv().is_none(), + "no pong while suspended" + ); + assert_eq!(counter.load(Ordering::SeqCst), 0); - handle - .runtime - .send_to( - addr, - Ping { - reply_to: *pong_inbox.addr(), - }, - ) - .unwrap(); - let pong_while_suspended = poll_inbox(&pong_inbox, Duration::from_millis(100)); - let count_while_suspended = counter.load(Ordering::SeqCst); + // Resume + let resumed = rt.admin().resume_actor(addr).unwrap(); + let resumed = resumed.recv_ticking(&rt, 5); + assert_eq!(resumed, Ok(operation_applied())); - let resumed = handle.runtime.admin().resume_actor(addr).unwrap(); - let resumed = poll_admin(&resumed, Duration::from_secs(1)); - let pong_after_resume = poll_inbox(&pong_inbox, Duration::from_secs(1)); - let final_count = counter.load(Ordering::SeqCst); - - handle.shutdown(); - handle.join(); - - assert_eq!(suspended, Some(Ok(operation_applied()))); - assert_eq!(pong_while_suspended, None); - assert_eq!(count_while_suspended, 0); - assert_eq!(resumed, Some(Ok(operation_applied()))); - assert_eq!(pong_after_resume, Some(Pong)); - assert_eq!(final_count, 1); + // Tick — message should now be processed + tick_n(&rt, 3); + assert_eq!( + pong_inbox.try_recv(), + Some(Pong), + "pong delivered after resume" + ); + assert_eq!(counter.load(Ordering::SeqCst), 1); } #[test] -fn admin_list_actors_aggregates_all_workers() { +fn admin_list_actors() { let rt = std_runtime(RuntimeConfig { - num_threads: 4, max_actors: 100, ..Default::default() }); @@ -558,51 +531,25 @@ fn admin_list_actors_aggregates_all_workers() { for _ in 0..16 { addrs.push(rt.spawn(CounterActor { count: 0 }).unwrap()); } + rt.tick(); // process spawns - let handle = rt.run().unwrap(); - let start = Instant::now(); - let mut response = None; - while start.elapsed() < Duration::from_secs(1) { - let admin = handle.runtime.admin().list_actors().unwrap(); - if let Some(Ok(list)) = poll_admin(&admin, Duration::from_millis(100)) { - if list.actors.len() == addrs.len() { - response = Some(list); - break; - } - } - std::thread::sleep(Duration::from_millis(5)); - } + let admin = rt.admin().list_actors().unwrap(); + let list = admin + .recv_ticking(&rt, 5) + .expect("list_actors timed out"); - handle.shutdown(); - handle.join(); - - let response = response.expect("admin list did not observe all spawned actors within timeout"); let expected: HashSet<_> = addrs.iter().copied().collect(); - let actual: HashSet<_> = response - .actors - .iter() - .map(|summary| summary.address) - .collect(); + let actual: HashSet<_> = list.actors.iter().map(|s| s.address).collect(); assert_eq!(actual, expected); + for addr in &addrs { assert_eq!( - response - .actors + list.actors .iter() .filter(|summary| summary.address == *addr) .count(), 1, - "actor {addr} appears exactly once in aggregated list" + "actor {addr} appears exactly once" ); } - - let worker_ids: HashSet<_> = response - .actors - .iter() - .map(|summary| summary.worker_id) - .collect(); - assert!( - worker_ids.len() >= 2, - "aggregation should include actors from at least two workers, got {worker_ids:?}" - ); } diff --git a/tests/runtime_stress.rs b/tests/runtime_stress.rs index c193e92..68b9ffb 100644 --- a/tests/runtime_stress.rs +++ b/tests/runtime_stress.rs @@ -1,58 +1,22 @@ -//! Runtime Stress Tests — multi-threaded execution, parking, placement, and scale. +//! Runtime Stress Tests — delivery at scale, panic isolation, and sustained throughput. //! -//! Covers: single vs multi-threaded processing, high-volume MT delivery, -//! panic isolation under load, worker parking/shutdown, sustained throughput, -//! and load-aware actor placement. +//! Covers: high-volume delivery, panic isolation under load, and sustained +//! throughput with no message loss. All tests are tick-driven (single worker). mod common; use common::*; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::time::{Duration, Instant}; - -// ── Helpers ────────────────────────────────────────────────────────────────── - -/// Poll `inbox` until a message arrives or `timeout` elapses. -fn poll_inbox(inbox: &Inbox, timeout: Duration) -> Option { - let deadline = Instant::now() + timeout; - loop { - if let Some(msg) = inbox.try_recv() { - return Some(msg); - } - if Instant::now() > deadline { - return None; - } - std::thread::sleep(Duration::from_millis(1)); - } -} - -/// Wait until `counter` reaches `target` or `timeout` elapses. -fn wait_for_count(counter: &AtomicUsize, target: usize, timeout: Duration) -> usize { - let deadline = Instant::now() + timeout; - loop { - let n = counter.load(Ordering::SeqCst); - if n >= target { - return n; - } - if Instant::now() > deadline { - return n; - } - std::thread::sleep(Duration::from_millis(5)); - } -} // ── Tests ──────────────────────────────────────────────────────────────────── -/// Single-threaded vs multi-threaded runtime basics. +/// Single-worker runtime basics. /// -/// Story: We start with a single-threaded runtime driven by tick(), confirm -/// nothing happens without ticking, then graduate to a multi-threaded runtime -/// with run() and verify background processing, cross-worker delegation, -/// custom thread counts, and clean shutdown. +/// Story: A runtime driven by tick() does nothing before the first tick, +/// then processes messages correctly. #[test] -fn single_vs_multi_threaded_basics() { - // ── Part A: Single-threaded requires tick() ── +fn runtime_basics() { let rt = std_runtime(RuntimeConfig::default()); let addr = rt.spawn(PingPongActor).unwrap(); let inbox = rt.new_inbox::().unwrap(); @@ -68,75 +32,37 @@ fn single_vs_multi_threaded_basics() { tick_n(&rt, 2); assert!( inbox.try_recv().is_some(), - "tick() drives single-threaded processing" + "tick() drives processing" ); - // ── Part B: Multi-threaded processes without ticking ── - let rt_mt = std_runtime(RuntimeConfig { - num_threads: 4, - ..Default::default() - }); - let addr = rt_mt.spawn(PingPongActor).unwrap(); - let inbox = rt_mt.new_inbox::().unwrap(); - rt_mt - .send_to( - addr, - Ping { - reply_to: *inbox.addr(), - }, - ) - .unwrap(); + // Delegation (spawn child from handler) works on a single worker. + let addr2 = rt.spawn(DelegatorActor).unwrap(); + let done_inbox = rt.new_inbox::().unwrap(); + rt.send_to( + addr2, + Forward { + value: 3, + reply_to: *done_inbox.addr(), + }, + ) + .unwrap(); - let handle = rt_mt.run().unwrap(); - let reply = poll_inbox(&inbox, Duration::from_secs(5)); - assert!( - reply.is_some(), - "background workers process without manual ticking" - ); - - // ── Part C: Cross-worker delegation (2 threads, spawn child from handler) ── - let addr2 = handle.runtime.spawn(DelegatorActor).unwrap(); - let done_inbox = handle.runtime.new_inbox::().unwrap(); - handle - .runtime - .send_to( - addr2, - Forward { - value: 3, - reply_to: *done_inbox.addr(), - }, - ) - .unwrap(); - - let reply = poll_inbox(&done_inbox, Duration::from_secs(5)); + tick_n(&rt, 3); assert_eq!( - reply, + done_inbox.try_recv(), Some(Done(6)), - "cross-worker delegation delivers reply" + "delegation delivers reply" ); - - // ── Part D: Custom thread count reflected in stats ── - let stats = handle.runtime.stats(); - assert_eq!( - stats.num_workers, 4, - "runtime respects requested thread count" - ); - - // ── Part E: Clean shutdown ── - handle.shutdown(); - handle.join(); - // Test passes by not hanging. } -/// High-volume multi-threaded delivery. +/// High-volume delivery. /// -/// Story: We throw large workloads at a 4-thread runtime — 50 senders -/// each firing 100 messages at one receiver, 200 concurrent spawn+send -/// pairs, and a 50-level chain that must hop across workers. +/// Story: We throw large workloads at the runtime — 50 senders each firing +/// 100 messages at one receiver, 200 concurrent spawn+send pairs, and a +/// 50-level chain. All messages must be accounted for. #[test] -fn mt_high_volume_delivery() { +fn high_volume_delivery() { let cfg = || RuntimeConfig { - num_threads: 4, max_actors: 5_000, channel_buffer_size: 10_000, ..Default::default() @@ -166,10 +92,8 @@ fn mt_high_volume_delivery() { } } - let handle = rt.run().unwrap(); - let processed = wait_for_count(&counter, total_expected, Duration::from_secs(5)); - handle.shutdown(); - handle.join(); + tick_n(&rt, 200); + let processed = counter.load(Ordering::SeqCst); assert_eq!( processed, total_expected, "all 5000 messages delivered to single receiver" @@ -196,20 +120,14 @@ fn mt_high_volume_delivery() { .unwrap(); } - let handle = rt.run().unwrap(); - let received = wait_for_count(&counter, 200, Duration::from_secs(5)); - handle.shutdown(); - handle.join(); + tick_n(&rt, 50); + let received = counter.load(Ordering::SeqCst); assert_eq!(received, 200, "all 200 spawn+send pairs complete"); } - // ── Part C: 50-level chain across workers ── + // ── Part C: 50-level chain ── { - let rt = std_runtime(RuntimeConfig { - num_threads: 2, - max_actors: 5_000, - ..Default::default() - }); + let rt = std_runtime(cfg()); let addr = rt.spawn(ChainActor).unwrap(); let inbox = rt.new_inbox::().unwrap(); rt.send_to( @@ -222,26 +140,22 @@ fn mt_high_volume_delivery() { ) .unwrap(); - let handle = rt.run().unwrap(); - let reply = poll_inbox(&inbox, Duration::from_secs(5)); - handle.shutdown(); - handle.join(); + tick_n(&rt, 100); assert_eq!( - reply, + inbox.try_recv(), Some(Done(50)), - "50-level chain completes across workers" + "50-level chain completes" ); } } -/// Panic isolation under multi-threaded load. +/// Panic isolation under load. /// -/// Story: 10 panicking actors and 10 healthy actors on 4 threads — every -/// panic is isolated and all 1000 healthy messages are still processed. +/// Story: 10 panicking actors and 10 healthy actors — every panic is isolated +/// and all 1000 healthy messages are still processed. #[test] -fn mt_panic_isolation_under_load() { +fn panic_isolation_under_load() { let rt = std_runtime(RuntimeConfig { - num_threads: 4, max_actors: 5_000, channel_buffer_size: 10_000, ..Default::default() @@ -278,11 +192,9 @@ fn mt_panic_isolation_under_load() { } } - let handle = rt.run().unwrap(); + tick_n(&rt, 200); let expected = 10 * 100; - let processed = wait_for_count(&counter, expected, Duration::from_secs(5)); - handle.shutdown(); - handle.join(); + let processed = counter.load(Ordering::SeqCst); assert_eq!( processed, expected, @@ -290,92 +202,11 @@ fn mt_panic_isolation_under_load() { ); } -/// Worker parking and shutdown latency. -/// -/// Story: Workers park after idle time. We verify they wake quickly on new -/// messages, that messages sent after run() are delivered, and that shutdown -/// wakes all parked workers promptly. -#[test] -fn worker_parking_and_shutdown() { - // ── Part A: Parked workers wake on send ── - let rt = std_runtime(RuntimeConfig { - num_threads: 2, - ..Default::default() - }); - let addr = rt.spawn(PingPongActor).unwrap(); - let inbox = rt.new_inbox::().unwrap(); - - let handle = rt.run().unwrap(); - std::thread::sleep(Duration::from_millis(50)); // Let workers park. - - let before = Instant::now(); - handle - .runtime - .send_to( - addr, - Ping { - reply_to: *inbox.addr(), - }, - ) - .unwrap(); - let reply = poll_inbox(&inbox, Duration::from_secs(1)); - let latency = before.elapsed(); - - assert!(reply.is_some(), "parked worker should wake and process"); - assert!( - latency.as_millis() < 100, - "wake latency should be <100ms, was {:?}", - latency - ); - - // ── Part B: Send after run() delivers ── - let addr2 = handle.runtime.spawn(PingPongActor).unwrap(); - let inbox2 = handle.runtime.new_inbox::().unwrap(); - std::thread::sleep(Duration::from_millis(10)); - handle - .runtime - .send_to( - addr2, - Ping { - reply_to: *inbox2.addr(), - }, - ) - .unwrap(); - - let reply2 = poll_inbox(&inbox2, Duration::from_secs(5)); - assert!( - reply2.is_some(), - "message sent after run() must be delivered" - ); - - handle.shutdown(); - handle.join(); - - // ── Part C: Shutdown wakes parked workers quickly ── - let rt2 = std_runtime(RuntimeConfig { - num_threads: 4, - ..Default::default() - }); - let h2 = rt2.run().unwrap(); - std::thread::sleep(Duration::from_millis(50)); // Let workers park. - - let before = Instant::now(); - h2.shutdown(); - h2.join(); - let shutdown_time = before.elapsed(); - - assert!( - shutdown_time.as_millis() < 500, - "shutdown should complete quickly with parked workers, took {:?}", - shutdown_time - ); -} - /// Sustained throughput with no message loss. /// -/// Story: We send 10 batches of 100 messages, ticking between batches on a -/// single-threaded runtime. Each batch must make forward progress, and after -/// draining, all 1000 messages are accounted for. +/// Story: We send 10 batches of 100 messages, ticking between batches. +/// Each batch must make forward progress, and after draining, all 1000 +/// messages are accounted for. #[test] fn sustained_throughput_no_message_loss() { let rt = std_runtime(RuntimeConfig::default()); @@ -410,93 +241,3 @@ fn sustained_throughput_no_message_loss() { let total = counter.load(Ordering::SeqCst); assert_eq!(total, 1000, "sustained load should not drop any messages"); } - -/// Load-aware actor placement. -/// -/// Story: A fresh runtime falls back to round-robin (even distribution). -/// Under imbalanced load, new actors bias toward the lighter worker. -/// A single-worker runtime degrades gracefully. -#[test] -fn load_aware_actor_placement() { - // ── Part A: Round-robin fallback on fresh runtime (4 workers, 100 actors) ── - let rt = std_runtime(RuntimeConfig { - num_threads: 4, - ..Default::default() - }); - for _ in 0..100 { - rt.spawn(CounterActor { count: 0 }).unwrap(); - } - - let handle = rt.run().unwrap(); - std::thread::sleep(Duration::from_millis(20)); - - let stats = handle.runtime.stats(); - handle.shutdown(); - handle.join(); - - for w in &stats.workers { - assert!( - w.num_actors >= 20 && w.num_actors <= 30, - "worker {} has {} actors, expected ~25 (round-robin)", - w.id, - w.num_actors - ); - } - - // ── Part B: Imbalanced load biases toward lighter worker ── - let rt2 = std_runtime(RuntimeConfig { - num_threads: 2, - ..Default::default() - }); - let mut addrs = Vec::new(); - for _ in 0..20 { - addrs.push(rt2.spawn(CounterActor { count: 0 }).unwrap()); - } - - let h2 = rt2.run().unwrap(); - std::thread::sleep(Duration::from_millis(10)); - - // Bombard the first 10 actors (likely worker 0) with messages. - for addr in &addrs[..10] { - for _ in 0..50 { - let _ = h2.runtime.send_to(*addr, Increment { reply_to: *addr }); - } - } - std::thread::sleep(Duration::from_millis(20)); - - // Spawn 10 more — should bias toward lighter worker. - for _ in 0..10 { - h2.runtime.spawn(CounterActor { count: 0 }).unwrap(); - } - std::thread::sleep(Duration::from_millis(20)); - - let stats2 = h2.runtime.stats(); - h2.shutdown(); - h2.join(); - - let total_actors: usize = stats2.workers.iter().map(|w| w.num_actors).sum(); - assert!( - total_actors >= 20, - "expected at least 20 actors, got {total_actors}" - ); - assert!( - stats2.workers.iter().all(|w| w.num_actors > 0), - "both workers should have actors: {:?}", - stats2 - .workers - .iter() - .map(|w| w.num_actors) - .collect::>() - ); - - // ── Part C: Single-worker degrades gracefully ── - let rt3 = std_runtime(RuntimeConfig::default()); - for _ in 0..50 { - rt3.spawn(CounterActor { count: 0 }).unwrap(); - } - tick_n(&rt3, 10); - - let stats3 = rt3.stats(); - assert_eq!(stats3.workers.len(), 1); - assert_eq!(stats3.workers[0].num_actors, 50); -} diff --git a/tests/test_python.py b/tests/test_python.py index 4ad7b34..c064c54 100644 --- a/tests/test_python.py +++ b/tests/test_python.py @@ -35,13 +35,11 @@ class TestActorAddress(unittest.TestCase): class TestRuntimeConfig(unittest.TestCase): def test_defaults(self): cfg = RuntimeConfig() - self.assertEqual(cfg.num_threads, 1) self.assertEqual(cfg.max_actors, 1000) self.assertEqual(cfg.channel_buffer_size, 1000) def test_custom(self): - cfg = RuntimeConfig(num_threads=4, max_actors=500) - self.assertEqual(cfg.num_threads, 4) + cfg = RuntimeConfig(max_actors=500) self.assertEqual(cfg.max_actors, 500) @@ -107,42 +105,5 @@ class TestSingleThreaded(unittest.TestCase): self.assertIsNone(inbox.try_recv()) -class TestMultiThreaded(unittest.TestCase): - def test_run_shutdown_join(self): - """Multi-threaded runtime can spawn, send, and receive.""" - import time - - rt = Runtime(RuntimeConfig(num_threads=2)) - - def echo(ctx, msg): - ctx.send(msg["reply_to"], msg["payload"]) - - addr = rt.spawn(echo) - inbox = rt.inbox() - handle = rt.run() - handle.send(addr, {"payload": "mt_hello", "reply_to": inbox.addr}) - - # Poll for result - result = None - for _ in range(100): - result = inbox.try_recv() - if result is not None: - break - time.sleep(0.01) - self.assertEqual(result, "mt_hello") - - handle.shutdown() - handle.join() - - def test_run_consumes_runtime(self): - """After run(), tick() should raise.""" - rt = Runtime(RuntimeConfig(num_threads=2)) - handle = rt.run() - with self.assertRaises(RuntimeError): - rt.tick() - handle.shutdown() - handle.join() - - if __name__ == "__main__": unittest.main()