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;