From f02700c9c600570dd83f6e5111f41f4226428007 Mon Sep 17 00:00:00 2001 From: zacheryasc Date: Sat, 7 Feb 2026 16:51:40 +0000 Subject: [PATCH] refactor: graph-optimization (#17) Use spectral analysis tool to find spurious edges in the code DAG, refactoring to prune. Signed-off-by: Zachery Aaron Shores-Chmielewski --- benches/worker_benchmarks.rs | 140 +++++++++++----------------- src/{address_map.rs => delivery.rs} | 94 ++++++++++++++++++- src/lib.rs | 3 +- src/runtime.rs | 111 ++-------------------- src/stats.rs | 36 +++++++ src/worker/mod.rs | 77 ++------------- src/worker/tests.rs | 6 +- 7 files changed, 201 insertions(+), 266 deletions(-) rename src/{address_map.rs => delivery.rs} (52%) create mode 100644 src/stats.rs diff --git a/benches/worker_benchmarks.rs b/benches/worker_benchmarks.rs index 7db7129..0b3e57e 100644 --- a/benches/worker_benchmarks.rs +++ b/benches/worker_benchmarks.rs @@ -1,21 +1,48 @@ +use std::collections::VecDeque; + use criterion::{ criterion_group, criterion_main, BenchmarkId, Criterion, Throughput, }; -use swactor::worker::Mailbox; +use swactor::worker::drain_count; // --------------------------------------------------------------------------- -// Push throughput +// drain_count O(1) verification // --------------------------------------------------------------------------- -fn mailbox_push(c: &mut Criterion) { - let mut group = c.benchmark_group("mailbox_push"); +fn bench_drain_count(c: &mut Criterion) { + let mut group = c.benchmark_group("drain_count"); + + // Below waterlevel + group.bench_function("below", |b| { + b.iter(|| std::hint::black_box(drain_count(50, 100))); + }); + + // At waterlevel + group.bench_function("at", |b| { + b.iter(|| std::hint::black_box(drain_count(100, 100))); + }); + + // Above waterlevel + group.bench_function("above", |b| { + b.iter(|| std::hint::black_box(drain_count(500, 100))); + }); + + group.finish(); +} + +// --------------------------------------------------------------------------- +// VecDeque push throughput (mirrors old mailbox_push) +// --------------------------------------------------------------------------- + +fn vecdeque_push(c: &mut Criterion) { + let mut group = c.benchmark_group("vecdeque_push"); for n in [100, 1_000, 10_000] { group.throughput(Throughput::Elements(n as u64)); group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| { b.iter(|| { - let mut mb: Mailbox = Mailbox::new(n); + let mut q: VecDeque = VecDeque::new(); for i in 0..n { - mb.push(i as u64); + q.push_back(i as u64); } }); }); @@ -24,25 +51,25 @@ fn mailbox_push(c: &mut Criterion) { } // --------------------------------------------------------------------------- -// Pop throughput +// VecDeque pop throughput (mirrors old mailbox_pop) // --------------------------------------------------------------------------- -fn mailbox_pop(c: &mut Criterion) { - let mut group = c.benchmark_group("mailbox_pop"); +fn vecdeque_pop(c: &mut Criterion) { + let mut group = c.benchmark_group("vecdeque_pop"); for n in [100, 1_000, 10_000] { group.throughput(Throughput::Elements(n as u64)); group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| { b.iter_batched( || { - let mut mb: Mailbox = Mailbox::new(n); + let mut q: VecDeque = VecDeque::new(); for i in 0..n { - mb.push(i as u64); + q.push_back(i as u64); } - mb + q }, - |mut mb| { + |mut q| { for _ in 0..n { - std::hint::black_box(mb.pop()); + std::hint::black_box(q.pop_front()); } }, criterion::BatchSize::SmallInput, @@ -53,85 +80,27 @@ fn mailbox_pop(c: &mut Criterion) { } // --------------------------------------------------------------------------- -// Interleaved push+pop +// Simulated actor tick: drain_count + pop N from VecDeque // --------------------------------------------------------------------------- -fn mailbox_interleaved(c: &mut Criterion) { - let mut group = c.benchmark_group("mailbox_interleaved"); - for n in [100, 1_000, 10_000] { - group.throughput(Throughput::Elements(n as u64 * 2)); - group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| { - b.iter(|| { - let mut mb: Mailbox = Mailbox::new(n); - for i in 0..n { - mb.push(i as u64); - std::hint::black_box(mb.pop()); - } - }); - }); - } - group.finish(); -} - -// --------------------------------------------------------------------------- -// drain_count O(1) verification -// --------------------------------------------------------------------------- - -fn mailbox_drain_count(c: &mut Criterion) { - let mut group = c.benchmark_group("mailbox_drain_count"); - - // Below waterlevel - group.bench_function("below", |b| { - let mut mb: Mailbox = Mailbox::new(100); - for i in 0..50 { - mb.push(i); - } - b.iter(|| std::hint::black_box(mb.drain_count())); - }); - - // At waterlevel - group.bench_function("at", |b| { - let mut mb: Mailbox = Mailbox::new(100); - for i in 0..100 { - mb.push(i); - } - b.iter(|| std::hint::black_box(mb.drain_count())); - }); - - // Above waterlevel - group.bench_function("above", |b| { - let mut mb: Mailbox = Mailbox::new(100); - for i in 0..500 { - mb.push(i); - } - b.iter(|| std::hint::black_box(mb.drain_count())); - }); - - group.finish(); -} - -// --------------------------------------------------------------------------- -// Simulated actor tick: drain_count + pop N -// --------------------------------------------------------------------------- - -fn mailbox_actor_tick(c: &mut Criterion) { - let mut group = c.benchmark_group("mailbox_actor_tick"); +fn simulated_actor_tick(c: &mut Criterion) { + let mut group = c.benchmark_group("simulated_actor_tick"); for (wl, fill) in [(10, 5), (10, 10), (10, 50), (100, 200)] { let param = format!("wl={wl},fill={fill}"); group.bench_function(BenchmarkId::from_parameter(¶m), |b| { b.iter_batched( || { - let mut mb: Mailbox = Mailbox::new(wl); + let mut q: VecDeque = VecDeque::new(); for i in 0..fill { - mb.push(i as u64); + q.push_back(i as u64); } - mb + q }, - |mut mb| { - let n = mb.drain_count(); + |mut q| { + let n = drain_count(q.len(), wl); for _ in 0..n { - std::hint::black_box(mb.pop()); + std::hint::black_box(q.pop_front()); } }, criterion::BatchSize::SmallInput, @@ -144,10 +113,9 @@ fn mailbox_actor_tick(c: &mut Criterion) { criterion_group!( benches, - mailbox_push, - mailbox_pop, - mailbox_interleaved, - mailbox_drain_count, - mailbox_actor_tick, + bench_drain_count, + vecdeque_push, + vecdeque_pop, + simulated_actor_tick, ); criterion_main!(benches); diff --git a/src/address_map.rs b/src/delivery.rs similarity index 52% rename from src/address_map.rs rename to src/delivery.rs index 6cef070..ec520b1 100644 --- a/src/address_map.rs +++ b/src/delivery.rs @@ -1,8 +1,14 @@ +use std::any::Any; use std::collections::HashMap; -use std::sync::RwLock; use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, RwLock}; -use crate::actor::ActorAddress; +use crate::actor::{ActorAddress, AnyActor, Message}; +use crate::channel::Sender; +use crate::config::RuntimeConfig; +use crate::Error; + +// ─── Address Map Types ─────────────────────────────────────────────────────── /// Identifies a worker thread. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] @@ -81,8 +87,90 @@ impl Placement { } } +// ─── Delivery Types ────────────────────────────────────────────────────────── + +/// 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, +} + #[cfg(test)] -mod tests { +mod address_map_tests { use super::*; #[test] diff --git a/src/lib.rs b/src/lib.rs index 715af33..855556a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -6,8 +6,9 @@ pub(crate) mod error; 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..78ca383 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,35 +1,20 @@ 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}; -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::{AddressMap, Envelope, InboxRegistry, Placement, TickContext, WorkerId}; +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 +50,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 +272,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..e45acee 100644 --- a/src/worker/mod.rs +++ b/src/worker/mod.rs @@ -1,44 +1,16 @@ 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::channel::Receiver; +use crate::delivery::{Envelope, TickContext, WorkerId}; +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 +160,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 +237,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..b9fa56e 100644 --- a/src/worker/tests.rs +++ b/src/worker/tests.rs @@ -5,12 +5,12 @@ use std::sync::Arc; use std::thread; 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::{AddressMap, Envelope, InboxRegistry, Placement, TickContext, WorkerId}; +use crate::stats::WorkerStats; -use super::{TickContext, Worker, WorkerStats}; +use super::Worker; // ── Actors ─────────────────────────────────────────────────────────