From 018a86cee0af4aa0b023128a823e8e93c058c4c5 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Sat, 7 Feb 2026 18:15:12 +0700 Subject: [PATCH] feat: further improvements from spectrum-analysis --- src/delivery.rs | 89 +++++++++++++++++++++++++++++++++++ src/lib.rs | 2 + src/runtime.rs | 110 ++++---------------------------------------- src/stats.rs | 36 +++++++++++++++ src/worker/mod.rs | 78 +++---------------------------- src/worker/tests.rs | 5 +- 6 files changed, 146 insertions(+), 174 deletions(-) create mode 100644 src/delivery.rs create mode 100644 src/stats.rs diff --git a/src/delivery.rs b/src/delivery.rs new file mode 100644 index 0000000..d19eecc --- /dev/null +++ b/src/delivery.rs @@ -0,0 +1,89 @@ +use std::any::Any; +use std::collections::HashMap; +use std::sync::{Arc, RwLock}; + +use crate::actor::{ActorAddress, AnyActor, Message}; +use crate::address_map::{AddressMap, Placement}; +use crate::channel::Sender; +use crate::config::RuntimeConfig; +use crate::Error; + +/// A type-erased message envelope for cross-worker delivery. +/// +/// Uses `Box` (no atomic refcount) and move semantics (no clone). +pub(crate) struct Envelope { + dest: ActorAddress, + payload: Box, +} + +impl Envelope { + pub fn new(dest: ActorAddress, payload: Box) -> Self { + Self { dest, payload } + } + + pub fn dest(&self) -> ActorAddress { + self.dest + } + + pub fn downcast(self) -> Option { + self.payload.downcast::().ok().map(|b| *b) + } + + pub fn into_payload(self) -> Box { + self.payload + } +} + +/// Type-erased sender for external inboxes. +pub(crate) trait SenderT: Send + Sync { + fn try_send_any(&self, msg: Box); +} + +impl SenderT for Sender { + fn try_send_any(&self, msg: Box) { + if let Ok(typed) = msg.downcast::() { + let _ = Sender::try_send(self, *typed); + } + } +} + +/// Registry of external inboxes — replaces the Router's role for non-actor receivers. +pub(crate) struct InboxRegistry { + senders: RwLock>>, +} + +impl InboxRegistry { + pub fn new() -> Self { + Self { + senders: RwLock::new(HashMap::new()), + } + } + + pub fn register(&self, addr: ActorAddress, sender: Arc) { + self.senders.write().unwrap().insert(addr, sender); + } + + pub fn try_deliver( + &self, + addr: ActorAddress, + msg: Box, + ) -> Result<(), Error> { + let senders = self.senders.read().unwrap(); + if let Some(sender) = senders.get(&addr) { + sender.try_send_any(msg); + Ok(()) + } else { + Err(Error::from("Address not found")) + } + } +} + +/// 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<(ActorAddress, Box)>], + pub(crate) placement: &'a Placement, + pub(crate) inbox_registry: &'a InboxRegistry, + pub(crate) config: &'a RuntimeConfig, +} diff --git a/src/lib.rs b/src/lib.rs index 715af33..56fd68c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,6 +8,8 @@ pub use error::Error; pub(crate) mod address_map; pub mod config; +pub(crate) mod delivery; +pub mod stats; pub mod runtime; diff --git a/src/runtime.rs b/src/runtime.rs index abce263..e555524 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,8 +1,7 @@ use std::any::Any; use std::cell::RefCell; -use std::collections::HashMap; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, RwLock}; +use std::sync::Arc; use std::thread::{self, JoinHandle}; use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}; @@ -10,26 +9,13 @@ use crate::address_map::{AddressMap, Placement, WorkerId}; use crate::channel::{Receiver, Sender}; // Re-export config types so existing code using `runtime::RuntimeConfig` still works pub use crate::config::{BackoffPolicy, RuntimeConfig}; -use crate::worker::{TickContext, Worker, WorkerStats}; +use crate::delivery::{Envelope, InboxRegistry, TickContext}; +use crate::stats::WorkerStats; +// Re-export stats types so existing code using `runtime::*` still works +pub use crate::stats::{RuntimeStats, WorkerInfo}; +use crate::worker::Worker; use crate::Error; - -/// Snapshot of per-worker state. -pub struct WorkerInfo { - pub id: usize, - pub num_actors: usize, - pub mailbox_depth: usize, - pub messages_processed: u64, -} - -/// Snapshot of overall runtime state. -pub struct RuntimeStats { - pub num_workers: usize, - /// Each entry is (address, worker_id). - pub actors: Vec<(ActorAddress, usize)>, - pub workers: Vec, -} - /// Generic message inbox for receiving messages outside of the runtime. pub struct Inbox { addr: ActorAddress, @@ -65,22 +51,9 @@ impl RuntimeHandle { } } -// Re-export Ctx and ContextInner for backwards compatibility -pub use crate::actor::{ContextInner, Ctx}; - -/// Type-erased sender for external inboxes. -pub(crate) trait SenderT: Send + Sync { - fn try_send_any(&self, msg: Box); -} - -impl SenderT for Sender { - fn try_send_any(&self, msg: Box) { - if let Ok(typed) = msg.downcast::() { - let _ = Sender::try_send(self, *typed); - } - } -} - +// Re-export Ctx for backwards compatibility +pub use crate::actor::Ctx; +use crate::actor::ContextInner; // ─── Runtime ───────────────────────────────────────────────────────────────── @@ -300,71 +273,6 @@ impl Runtime { } } - - -/// A type-erased message envelope for cross-worker delivery. -/// -/// Uses `Box` (no atomic refcount) and move semantics (no clone). -pub(crate) struct Envelope { - dest: ActorAddress, - payload: Box, -} - -impl Envelope { - pub fn new(dest: ActorAddress, payload: Box) -> Self { - Self { dest, payload } - } - - pub fn dest(&self) -> ActorAddress { - self.dest - } - - pub fn downcast(self) -> Option { - self.payload.downcast::().ok().map(|b| *b) - } - - pub fn into_payload(self) -> Box { - self.payload - } -} - - -// ─── InboxRegistry ─────────────────────────────────────────────────────────── - -/// Registry of external inboxes — replaces the Router's role for non-actor receivers. -pub(crate) struct InboxRegistry { - senders: RwLock>>, -} - -impl InboxRegistry { - pub fn new() -> Self { - Self { - senders: RwLock::new(HashMap::new()), - } - } - - pub fn register(&self, addr: ActorAddress, sender: Arc) { - self.senders.write().unwrap().insert(addr, sender); - } - - pub fn try_deliver( - &self, - addr: ActorAddress, - msg: Box, - ) -> Result<(), Error> { - let senders = self.senders.read().unwrap(); - if let Some(sender) = senders.get(&addr) { - sender.try_send_any(msg); - Ok(()) - } else { - Err(Error::from("Address not found")) - } - } -} - - - - impl ContextInner for Runtime { fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error> { match self.address_map.lookup(&addr) { diff --git a/src/stats.rs b/src/stats.rs new file mode 100644 index 0000000..67a6a2f --- /dev/null +++ b/src/stats.rs @@ -0,0 +1,36 @@ +use std::sync::atomic::{AtomicU64, AtomicUsize}; + +use crate::actor::ActorAddress; + +/// Per-worker stats published via atomics. Readable from any thread. +pub struct WorkerStats { + pub num_actors: AtomicUsize, + pub total_mailbox_depth: AtomicUsize, + pub messages_processed: AtomicU64, +} + +impl WorkerStats { + pub fn new() -> Self { + Self { + num_actors: AtomicUsize::new(0), + total_mailbox_depth: AtomicUsize::new(0), + messages_processed: AtomicU64::new(0), + } + } +} + +/// Snapshot of per-worker state. +pub struct WorkerInfo { + pub id: usize, + pub num_actors: usize, + pub mailbox_depth: usize, + pub messages_processed: u64, +} + +/// Snapshot of overall runtime state. +pub struct RuntimeStats { + pub num_workers: usize, + /// Each entry is (address, worker_id). + pub actors: Vec<(ActorAddress, usize)>, + pub workers: Vec, +} diff --git a/src/worker/mod.rs b/src/worker/mod.rs index c2493ff..3208953 100644 --- a/src/worker/mod.rs +++ b/src/worker/mod.rs @@ -1,44 +1,17 @@ use std::any::Any; use std::cell::RefCell; use std::collections::{HashMap, VecDeque}; -use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::thread; -use crate::actor::{ActorAddress, AnyActor, ContextInner, Ctx, Message}; -use crate::address_map::{AddressMap, Placement, WorkerId}; -use crate::channel::{Receiver, Sender}; -use crate::config::RuntimeConfig; -use crate::runtime::{Envelope, InboxRegistry}; +use crate::actor::{ActorAddress, AnyActor, ContextInner, Ctx}; +use crate::address_map::WorkerId; +use crate::channel::Receiver; +use crate::delivery::{Envelope, TickContext}; +use crate::stats::WorkerStats; use crate::Error; -/// Per-worker stats published via atomics. Readable from any thread. -pub(crate) struct WorkerStats { - pub num_actors: AtomicUsize, - pub total_mailbox_depth: AtomicUsize, - pub messages_processed: AtomicU64, -} - -impl WorkerStats { - pub fn new() -> Self { - Self { - num_actors: AtomicUsize::new(0), - total_mailbox_depth: AtomicUsize::new(0), - messages_processed: AtomicU64::new(0), - } - } -} - -/// 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<(ActorAddress, Box)>], - pub(crate) placement: &'a Placement, - pub(crate) inbox_registry: &'a InboxRegistry, - pub(crate) config: &'a RuntimeConfig, -} - /// A worker owns a set of actors and runs them in a loop. pub(crate) struct Worker { id: WorkerId, @@ -188,7 +161,7 @@ impl ContextInner for WorkerContext<'_> { /// How many messages to process this tick: /// - `len < waterlevel` → process all (`len`) /// - `len >= waterlevel` → process half (`len >> 1`) -pub(crate) fn drain_count(len: usize, waterlevel: usize) -> usize { +pub fn drain_count(len: usize, waterlevel: usize) -> usize { if len < waterlevel { len } else { @@ -265,42 +238,5 @@ impl ActorPool { -pub(crate) struct Mailbox { - queue: VecDeque, - waterlevel: usize, -} - -impl Mailbox { - pub fn new(waterlevel: usize) -> Self { - Self { - queue: VecDeque::new(), - waterlevel, - } - } - - pub fn push(&mut self, msg: M) { - self.queue.push_back(msg); - } - - pub fn pop(&mut self) -> Option { - self.queue.pop_front() - } - - pub fn len(&self) -> usize { - self.queue.len() - } - - pub fn is_empty(&self) -> bool { - self.queue.is_empty() - } - - /// How many messages to process this tick: - /// - `len < waterlevel` → process all (`len`) - /// - `len >= waterlevel` → process half (`len >> 1`) - pub fn drain_count(&self) -> usize { - drain_count(self.queue.len(), self.waterlevel) - } -} - #[cfg(test)] mod tests; diff --git a/src/worker/tests.rs b/src/worker/tests.rs index 5cdca33..1546aa4 100644 --- a/src/worker/tests.rs +++ b/src/worker/tests.rs @@ -8,9 +8,10 @@ use crate::actor::{ActorAddress, AnyActor, Ctx}; use crate::address_map::{AddressMap, Placement, WorkerId}; use crate::channel::Receiver; use crate::config::RuntimeConfig; -use crate::runtime::{Envelope, InboxRegistry}; +use crate::delivery::{Envelope, InboxRegistry, TickContext}; +use crate::stats::WorkerStats; -use super::{TickContext, Worker, WorkerStats}; +use super::Worker; // ── Actors ─────────────────────────────────────────────────────────