zacheryasc/swactor
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(core): drop native threading for tick-driven execution

Browse source 
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<OnceLock<Thread>> 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
This commit is contained in:
Zachery Aaron Shores-Chmielewski 2026-08-09 13:50:33 +04:00
parent 1d0d4d66f0
commit 4a348ab4cd
14 changed files with 350 additions and 1519 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
README.md
View file

@ -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

37
apps/myelin/src/orchestration/provider_adapters/vastai/mod.rs
View file

@ -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<RuntimeHandle>,
runtime: Arc<Runtime>,
actor: ActorAddress,
tick_thread: Option<JoinHandle<()>>,
stop_flag: Arc<AtomicBool>,
}
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))
}

97
crates/bindings/python/src/lib.rs
View file

@ -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<PyRuntimeConfig> 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<PyActorAddress> {
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<PyInbox> {
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<PyMsg> = 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<PyRuntimeHandle> {
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<PyRuntimeStats> {
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<RuntimeHandle>,
}
#[pymethods]
impl PyRuntimeHandle {
fn spawn(&self, handler: PyObject) -> PyResult<PyActorAddress> {
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<PyInbox> {
let handle = self.inner.as_ref().ok_or_else(|| {
pyo3::exceptions::PyRuntimeError::new_err("RuntimeHandle consumed by join()")
})?;
let inbox: Inbox<PyMsg> = handle.runtime.new_inbox().map_err(to_py_err)?;
Ok(PyInbox { inner: inbox })
}
fn stats(&self) -> PyResult<PyRuntimeStats> {
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::<PyInbox>()?;
m.add_class::<PyRuntimeConfig>()?;
m.add_class::<PyRuntime>()?;
m.add_class::<PyRuntimeHandle>()?;
m.add_class::<PyActorInfo>()?;
m.add_class::<PyWorkerInfo>()?;
m.add_class::<PyRuntimeStats>()?;

5
crates/bindings/wasm-runtime/src/lib.rs
View file

@ -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 }
}

412
crates/dashboard/src/bin/swactor_dummy_node.rs
View file

@ -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<ActorAddress> },
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<ActorAddress>,
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<HashMap<ActorAddress, ActorDetail>>,
}
#[derive(Clone)]
struct ActorDetail {
worker_id: usize,
mailbox_depth: usize,
last_msg_type: Option<String>,
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<ActorAddress, String>,
) -> Vec<ActorDetailFrame> {
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<String>,
worker_id: usize,
mailbox_depth: usize,
last_msg_type: Option<String>,
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<ActorAddress, String>,
) {
let stats = runtime.stats();
let actor_details = stats_hook.snapshot(&stats.actors, names);
let actors: Vec<Value> = 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);
}

2
src/config.rs
View file

@ -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,
}
}

142
src/delivery.rs
View file

@ -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<V> = HashMap<ActorAddress, V, AddrBuildHasher>;
/// HashSet optimized for ActorAddress keys.
pub type AddrSet = HashSet<ActorAddress, AddrBuildHasher>;
// ─── 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<HashMap>` — zero contention for parallel reads, write-rare (only on spawn).
/// `RwLock<AddrSet>` — zero contention for parallel reads, write-rare (only on spawn).
pub(crate) struct AddressMap {
inner: RwLock<AddrMap<WorkerId>>,
inner: RwLock<AddrSet>,
}
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<WorkerId> {
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<Arc<WorkerStats>>,
}
impl Placement {
pub fn new(num_workers: usize, worker_stats: Vec<Arc<WorkerStats>>) -> 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<ActorAddress> {
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<dyn SenderT>) {
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<dyn Any + Send>) -> 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<Envelope>],
pub(crate) spawn_txs: &'a [Sender<SpawnRequest>],
pub(crate) placement: &'a Placement,
pub(crate) spawn_tx: &'a Sender<SpawnRequest>,
pub(crate) transfer_tx: &'a Sender<Envelope>,
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<dyn crate::process_observer::ProcessOutputObserver>>,
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<Thread>],
/// Per-worker stats for summing total_actors across workers.
pub(crate) worker_stats: &'a [Arc<WorkerStats>],
/// 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>,

479
src/runtime.rs
View file

@ -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<M: Message> Inbox<M> {
pub fn try_recv(&self) -> Option<M> {
self.inner.try_recv()
}
pub fn recv_ticking(&self, rt: &Runtime, max_ticks: usize) -> Option<M> {
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<R: Message> {
}
impl<R: Message> Ask<R> {
/// Try to receive the response without ticking.
pub fn try_recv(&self) -> Option<R> {
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<R, Error> {
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<Runtime>,
threads: Vec<JoinHandle<()>>,
}
#[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<R> {
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<AddressMap>,
inbox_registry: Arc<InboxRegistry>,
extension: Option<Arc<dyn RuntimeExtension>>,
transfer_txs: Vec<Sender<Envelope>>,
spawn_txs: Vec<Sender<SpawnRequest>>,
admin_txs: Vec<Sender<AdminCommand>>,
placement: Placement,
is_running: AtomicBool,
worker_stats: Vec<Arc<WorkerStats>>,
transfer_tx: Sender<Envelope>,
spawn_tx: Sender<SpawnRequest>,
admin_tx: Sender<AdminCommand>,
worker_stats: Arc<WorkerStats>,
stats_hook: Option<Arc<dyn StatsHook>>,
process_output_observer: OnceLock<Arc<dyn crate::process_observer::ProcessOutputObserver>>,
/// Workers available for tick(). run() drains this and moves workers to threads.
tick_workers: RefCell<Vec<Worker>>,
/// Thread handles for waking parked workers. Set by workers on startup via OnceLock.
worker_threads: Arc<Vec<OnceLock<Thread>>>,
worker: RefCell<Worker>,
created_at: Instant,
#[cfg(feature = "transport")]
remote_sink: Option<Arc<dyn RemoteSink>>,
}
// Safety: RefCell<Vec<Worker>> is only accessed from the owning thread via tick().
// After run() the RefCell is empty and not accessed by worker threads.
// Safety: `RefCell<Worker>` is only borrowed from the owning thread in
// `tick()` / `try_tick()` / `has_work()` / `with_extension()`. All `&self`
// methods callable through `Arc<Runtime>` 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<dyn Any + Send>) -> Result<(), Error>;
@ -172,27 +148,20 @@ impl RuntimeAddress {
/// actor system.
pub struct ExternalSender {
address_map: Arc<AddressMap>,
transfer_txs: Vec<Sender<Envelope>>,
worker_threads: Arc<Vec<OnceLock<Thread>>>,
transfer_tx: Sender<Envelope>,
}
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<AddressMap> uses RwLock,
// Sender<Envelope> wraps Arc<HybridChannel>, 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<M: Message>(&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::<Envelope>::new(config.channel_buffer_size);
let transfer_tx = transfer_rx.new_sender();
for i in 0..num_workers {
let transfer_rx = Receiver::<Envelope>::new(config.channel_buffer_size);
let transfer_tx = transfer_rx.new_sender();
transfer_txs.push(transfer_tx);
let spawn_rx = Receiver::<SpawnRequest>::new(config.max_actors);
let spawn_tx = spawn_rx.new_sender();
let spawn_rx = Receiver::<SpawnRequest>::new(config.max_actors);
let spawn_tx = spawn_rx.new_sender();
spawn_txs.push(spawn_tx);
let admin_rx = Receiver::<AdminCommand>::new(config.channel_buffer_size);
let admin_tx = admin_rx.new_sender();
let admin_rx = Receiver::<AdminCommand>::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<Vec<OnceLock<Thread>>> =
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<A: ActorInterface>(&self, actor: A) -> Result<ActorAddress, Error> {
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<dyn AnyActor> = 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<ActorAddress, Error> {
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<dyn AnyActor> = 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<dyn RuntimeExtension>) -> 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<RuntimeHandle, Error> {
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<Worker> = self.tick_workers.replace(Vec::new());
let rt = Arc::new(self);
let mut handles: Vec<JoinHandle<()>> = 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<dyn StatsHook>) {
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<dyn Any + Send>) -> 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<Admin<ListActorsResponse>, Error> {
let (admin, reply_to) = self.new_admin::<ListActorsResponse>()?;
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<Admin<InspectActorResponse>, Error> {
let Some(wid) = self.runtime.address_map.lookup(&actor) else {
if !self.runtime.address_map.contains(&actor) {
return self.ready::<InspectActorResponse>(Err(AdminError::ActorNotFound { actor }));
};
}
let (admin, reply_to) = self.new_admin::<InspectActorResponse>()?;
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::<GetActorStateResponse<A>>(Err(AdminError::ActorNotFound { actor }));
};
}
let (admin, reply_to) = self.new_admin::<GetActorStateResponse<A>>()?;
let get = Box::new(
@ -728,14 +558,12 @@ impl RuntimeAdmin<'_> {
)) as Box<dyn Any + Send>
});
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::<OperationResult>(Err(AdminError::ActorNotFound { actor }));
};
}
let (admin, reply_to) = self.new_admin::<OperationResult>()?;
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<Admin<OperationResult>, Error> {
let Some(wid) = self.runtime.address_map.lookup(&actor) else {
if !self.runtime.address_map.contains(&actor) {
return self.ready::<OperationResult>(Err(AdminError::ActorNotFound { actor }));
};
}
let (admin, reply_to) = self.new_admin::<OperationResult>()?;
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<Admin<OperationResult>, Error> {
let Some(wid) = self.runtime.address_map.lookup(&actor) else {
if !self.runtime.address_map.contains(&actor) {
return self.ready::<OperationResult>(Err(AdminError::ActorNotFound { actor }));
};
}
let (admin, reply_to) = self.new_admin::<OperationResult>()?;
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<Admin<OperationResult>, Error> {
let Some(wid) = self.runtime.address_map.lookup(&actor) else {
if !self.runtime.address_map.contains(&actor) {
return self.ready::<OperationResult>(Err(AdminError::ActorNotFound { actor }));
};
}
let (admin, reply_to) = self.new_admin::<OperationResult>()?;
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<Thread>], 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<dyn Any + Send>) -> 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,
}
}

14
src/stats.rs
View file

@ -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,
}

146
src/worker.rs
View file

@ -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<Envelope>,
spawn_rx: Receiver<SpawnRequest>,
@ -86,14 +79,12 @@ pub(crate) struct Worker {
impl Worker {
pub(crate) fn new(
id: WorkerId,
transfer_rx: Receiver<Envelope>,
spawn_rx: Receiver<SpawnRequest>,
admin_rx: Receiver<AdminCommand>,
stats: Arc<WorkerStats>,
) -> 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<Vec<Box<dyn Any + Send>>> = 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<Vec<(ActorAddress, Box<dyn Any + Send>)>>,
stop_requests: &'a RefCell<Vec<ActorAddress>>,
@ -478,30 +451,19 @@ struct WorkerContext<'a> {
impl ContextInner for WorkerContext<'_> {
fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> 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<ActorSummary> {
fn actor_summary(&self, addr: ActorAddress) -> AdminResult<ActorSummary> {
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<ActorSummary>) {
fn actor_summaries_into(&self, out: &mut Vec<ActorSummary>) {
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<Vec<ActorAddress>>,
stop_with_values: &RefCell<Vec<(ActorAddress, ExitValue)>>,
suspend_requests: &RefCell<Vec<ActorAddress>>,
) -> 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;

2
tests/message_delivery.rs
View file

@ -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::<NumberedReply>().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::<RingDone>().unwrap();

139
tests/runtime_admin.rs
View file

@ -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<T: swactor::actor::Message>(
admin: &swactor::admin::Admin<T>,
timeout: Duration,
) -> Option<swactor::admin::AdminResult<T>> {
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<M: swactor::actor::Message>(inbox: &Inbox<M>, timeout: Duration) -> Option<M> {
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::<Pong>().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:?}"
);
}

349
tests/runtime_stress.rs
View file

@ -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<M: swactor::actor::Message>(inbox: &Inbox<M>, timeout: Duration) -> Option<M> {
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::<Pong>().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::<Pong>().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::<Done>().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::<Done>().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::<Done>().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::<Pong>().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::<Pong>().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::<Vec<_>>()
);
// ── 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);
}

41
tests/test_python.py
View file

@ -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()