refactor: step

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) {
+fn bench_drain_count(c: &mut Criterion) {
-    let mut group = c.benchmark_group("mailbox_push");
+    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<u64> = Mailbox::new(n);
+                let mut q: VecDeque<u64> = 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) {
+fn vecdeque_pop(c: &mut Criterion) {
-    let mut group = c.benchmark_group("mailbox_pop");
+    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<u64> = Mailbox::new(n);
+                    let mut q: VecDeque<u64> = 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) {
+fn simulated_actor_tick(c: &mut Criterion) {
-    let mut group = c.benchmark_group("mailbox_interleaved");
+    let mut group = c.benchmark_group("simulated_actor_tick");
-    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<u64> = 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<u64> = 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<u64> = 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<u64> = 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");
 
     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(&param), |b| {
             b.iter_batched(
                 || {
-                    let mut mb: Mailbox<u64> = Mailbox::new(wl);
+                    let mut q: VecDeque<u64> = VecDeque::new();
                     for i in 0..fill {
-                        mb.push(i as u64);
+                        q.push_back(i as u64);
                     }
-                    mb
+                    q
                 },
-                |mut mb| {
+                |mut q| {
-                    let n = mb.drain_count();
+                    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,
+    bench_drain_count,
-    mailbox_pop,
+    vecdeque_push,
-    mailbox_interleaved,
+    vecdeque_pop,
-    mailbox_drain_count,
+    simulated_actor_tick,
-    mailbox_actor_tick,
 );
 criterion_main!(benches);

src/address_map.rs

@@ -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<HashMap>` — zero contention for parallel reads, write-rare (only on spawn).
-pub(crate) struct AddressMap {
-    inner: RwLock<HashMap<ActorAddress, WorkerId>>,
-}
-
-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<WorkerId> {
-        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));
-    }
-}

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<HashMap>` — zero contention for parallel reads, write-rare (only on spawn).
+pub(crate) struct AddressMap {
+    inner: RwLock<HashMap<ActorAddress, WorkerId>>,
+}
+
+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<WorkerId> {
+        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));
+    }
+}

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;

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

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;
 

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;