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 1/2] 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 ───────────────────────────────────────────────────────── -- 2.45.2 From e3bb230af2321a68a861837b635a4456415fb52d Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Sat, 7 Feb 2026 18:59:04 +0700 Subject: [PATCH 2/2] refactor: step --- benches/worker_benchmarks.rs | 140 ++++++++++++++--------------------- src/address_map.rs | 130 -------------------------------- src/delivery.rs | 131 +++++++++++++++++++++++++++++++- src/lib.rs | 1 - src/runtime.rs | 3 +- src/worker/mod.rs | 3 +- src/worker/tests.rs | 3 +- 7 files changed, 187 insertions(+), 224 deletions(-) delete mode 100644 src/address_map.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/address_map.rs deleted file mode 100644 index 6cef070..0000000 --- a/src/address_map.rs +++ /dev/null @@ -1,130 +0,0 @@ -use std::collections::HashMap; -use std::sync::RwLock; -use std::sync::atomic::{AtomicUsize, Ordering}; - -use crate::actor::ActorAddress; - -/// 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. -/// -/// `RwLock` — zero contention for parallel reads, write-rare (only on spawn). -pub(crate) struct AddressMap { - inner: RwLock>, -} - -impl AddressMap { - pub fn new() -> Self { - Self { - inner: RwLock::new(HashMap::new()), - } - } - - pub fn with_capacity(cap: usize) -> Self { - Self { - inner: RwLock::new(HashMap::with_capacity(cap)), - } - } - - pub fn insert(&self, addr: ActorAddress, worker: WorkerId) { - self.inner.write().unwrap().insert(addr, worker); - } - - pub fn remove(&self, addr: &ActorAddress) { - self.inner.write().unwrap().remove(addr); - } - - pub fn lookup(&self, addr: &ActorAddress) -> Option { - self.inner.read().unwrap().get(addr).copied() - } - - pub fn len(&self) -> usize { - self.inner.read().unwrap().len() - } - - /// 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() - } -} - -/// Round-robin actor placement strategy. -pub(crate) struct Placement { - next: AtomicUsize, - num_workers: usize, -} - -impl Placement { - pub fn new(num_workers: usize) -> Self { - Self { - next: AtomicUsize::new(0), - num_workers, - } - } - - pub fn next_worker(&self) -> WorkerId { - let id = self.next.fetch_add(1, Ordering::Relaxed) % self.num_workers; - WorkerId(id) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn insert_and_lookup() { - let map = AddressMap::new(); - let addr = ActorAddress::default(); - let wid = WorkerId(3); - map.insert(addr, wid); - assert_eq!(map.lookup(&addr), Some(wid)); - } - - #[test] - fn lookup_missing_returns_none() { - let map = AddressMap::new(); - let addr = ActorAddress::default(); - assert_eq!(map.lookup(&addr), None); - } - - #[test] - fn remove_works() { - let map = AddressMap::new(); - let addr = ActorAddress::default(); - map.insert(addr, WorkerId(0)); - map.remove(&addr); - assert_eq!(map.lookup(&addr), None); - } - - #[test] - fn len_tracks_entries() { - let map = AddressMap::with_capacity(10); - assert_eq!(map.len(), 0); - let addr1 = ActorAddress::default(); - map.insert(addr1, WorkerId(0)); - assert_eq!(map.len(), 1); - } - - #[test] - fn round_robin() { - let p = Placement::new(3); - assert_eq!(p.next_worker(), WorkerId(0)); - assert_eq!(p.next_worker(), WorkerId(1)); - assert_eq!(p.next_worker(), WorkerId(2)); - assert_eq!(p.next_worker(), WorkerId(0)); - } -} diff --git a/src/delivery.rs b/src/delivery.rs index d19eecc..ec520b1 100644 --- a/src/delivery.rs +++ b/src/delivery.rs @@ -1,13 +1,94 @@ use std::any::Any; use std::collections::HashMap; +use std::sync::atomic::{AtomicUsize, Ordering}; 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; +// ─── Address Map Types ─────────────────────────────────────────────────────── + +/// 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. +/// +/// `RwLock` — zero contention for parallel reads, write-rare (only on spawn). +pub(crate) struct AddressMap { + inner: RwLock>, +} + +impl AddressMap { + pub fn new() -> Self { + Self { + inner: RwLock::new(HashMap::new()), + } + } + + pub fn with_capacity(cap: usize) -> Self { + Self { + inner: RwLock::new(HashMap::with_capacity(cap)), + } + } + + pub fn insert(&self, addr: ActorAddress, worker: WorkerId) { + self.inner.write().unwrap().insert(addr, worker); + } + + pub fn remove(&self, addr: &ActorAddress) { + self.inner.write().unwrap().remove(addr); + } + + pub fn lookup(&self, addr: &ActorAddress) -> Option { + self.inner.read().unwrap().get(addr).copied() + } + + pub fn len(&self) -> usize { + self.inner.read().unwrap().len() + } + + /// 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() + } +} + +/// Round-robin actor placement strategy. +pub(crate) struct Placement { + next: AtomicUsize, + num_workers: usize, +} + +impl Placement { + pub fn new(num_workers: usize) -> Self { + Self { + next: AtomicUsize::new(0), + num_workers, + } + } + + pub fn next_worker(&self) -> WorkerId { + let id = self.next.fetch_add(1, Ordering::Relaxed) % self.num_workers; + WorkerId(id) + } +} + +// ─── Delivery Types ────────────────────────────────────────────────────────── + /// A type-erased message envelope for cross-worker delivery. /// /// Uses `Box` (no atomic refcount) and move semantics (no clone). @@ -87,3 +168,51 @@ pub(crate) struct TickContext<'a> { pub(crate) inbox_registry: &'a InboxRegistry, pub(crate) config: &'a RuntimeConfig, } + +#[cfg(test)] +mod address_map_tests { + use super::*; + + #[test] + fn insert_and_lookup() { + let map = AddressMap::new(); + let addr = ActorAddress::default(); + let wid = WorkerId(3); + map.insert(addr, wid); + assert_eq!(map.lookup(&addr), Some(wid)); + } + + #[test] + fn lookup_missing_returns_none() { + let map = AddressMap::new(); + let addr = ActorAddress::default(); + assert_eq!(map.lookup(&addr), None); + } + + #[test] + fn remove_works() { + let map = AddressMap::new(); + let addr = ActorAddress::default(); + map.insert(addr, WorkerId(0)); + map.remove(&addr); + assert_eq!(map.lookup(&addr), None); + } + + #[test] + fn len_tracks_entries() { + let map = AddressMap::with_capacity(10); + assert_eq!(map.len(), 0); + let addr1 = ActorAddress::default(); + map.insert(addr1, WorkerId(0)); + assert_eq!(map.len(), 1); + } + + #[test] + fn round_robin() { + let p = Placement::new(3); + assert_eq!(p.next_worker(), WorkerId(0)); + assert_eq!(p.next_worker(), WorkerId(1)); + assert_eq!(p.next_worker(), WorkerId(2)); + assert_eq!(p.next_worker(), WorkerId(0)); + } +} diff --git a/src/lib.rs b/src/lib.rs index 56fd68c..855556a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -6,7 +6,6 @@ pub(crate) mod error; pub use error::Error; -pub(crate) mod address_map; pub mod config; pub(crate) mod delivery; pub mod stats; diff --git a/src/runtime.rs b/src/runtime.rs index e555524..78ca383 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -5,11 +5,10 @@ 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::delivery::{Envelope, InboxRegistry, TickContext}; +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}; diff --git a/src/worker/mod.rs b/src/worker/mod.rs index 3208953..e45acee 100644 --- a/src/worker/mod.rs +++ b/src/worker/mod.rs @@ -6,9 +6,8 @@ use std::sync::Arc; use std::thread; use crate::actor::{ActorAddress, AnyActor, ContextInner, Ctx}; -use crate::address_map::WorkerId; use crate::channel::Receiver; -use crate::delivery::{Envelope, TickContext}; +use crate::delivery::{Envelope, TickContext, WorkerId}; use crate::stats::WorkerStats; use crate::Error; diff --git a/src/worker/tests.rs b/src/worker/tests.rs index 1546aa4..b9fa56e 100644 --- a/src/worker/tests.rs +++ b/src/worker/tests.rs @@ -5,10 +5,9 @@ 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::delivery::{Envelope, InboxRegistry, TickContext}; +use crate::delivery::{AddressMap, Envelope, InboxRegistry, Placement, TickContext, WorkerId}; use crate::stats::WorkerStats; use super::Worker; -- 2.45.2