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<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) {
-    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<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) {
-    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<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");
+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(&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| {
-                    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);

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<dyn Any + Send>,
+}
+
+impl Envelope {
+    pub fn new(dest: ActorAddress, payload: Box<dyn Any + Send>) -> Self {
+        Self { dest, payload }
+    }
+
+    pub fn dest(&self) -> ActorAddress {
+        self.dest
+    }
+
+    pub fn downcast<M: 'static>(self) -> Option<M> {
+        self.payload.downcast::<M>().ok().map(|b| *b)
+    }
+
+    pub fn into_payload(self) -> Box<dyn Any + Send> {
+        self.payload
+    }
+}
+
+/// Type-erased sender for external inboxes.
+pub(crate) trait SenderT: Send + Sync {
+    fn try_send_any(&self, msg: Box<dyn Any + Send>);
+}
+
+impl<M: Message> SenderT for Sender<M> {
+    fn try_send_any(&self, msg: Box<dyn Any + Send>) {
+        if let Ok(typed) = msg.downcast::<M>() {
+            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<HashMap<ActorAddress, Arc<dyn SenderT>>>,
+}
+
+impl InboxRegistry {
+    pub fn new() -> Self {
+        Self {
+            senders: RwLock::new(HashMap::new()),
+        }
+    }
+
+    pub fn register(&self, addr: ActorAddress, sender: Arc<dyn SenderT>) {
+        self.senders.write().unwrap().insert(addr, sender);
+    }
+
+    pub fn try_deliver(
+        &self,
+        addr: ActorAddress,
+        msg: Box<dyn Any + Send>,
+    ) -> 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<Envelope>],
+    pub(crate) spawn_txs: &'a [Sender<(ActorAddress, Box<dyn AnyActor>)>],
+    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]

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;
 

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<WorkerInfo>,
-}
-
 /// Generic message inbox for receiving messages outside of the runtime.
 pub struct Inbox<M: Message> {
     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<dyn Any + Send>);
-}
-
-impl<M: Message> SenderT for Sender<M> {
-    fn try_send_any(&self, msg: Box<dyn Any + Send>) {
-        if let Ok(typed) = msg.downcast::<M>() {
-            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<dyn Any + Send>,
-}
-
-impl Envelope {
-    pub fn new(dest: ActorAddress, payload: Box<dyn Any + Send>) -> Self {
-        Self { dest, payload }
-    }
-
-    pub fn dest(&self) -> ActorAddress {
-        self.dest
-    }
-
-    pub fn downcast<M: 'static>(self) -> Option<M> {
-        self.payload.downcast::<M>().ok().map(|b| *b)
-    }
-
-    pub fn into_payload(self) -> Box<dyn Any + Send> {
-        self.payload
-    }
-}
-
-
-// ─── InboxRegistry ───────────────────────────────────────────────────────────
-
-/// Registry of external inboxes — replaces the Router's role for non-actor receivers.
-pub(crate) struct InboxRegistry {
-    senders: RwLock<HashMap<ActorAddress, Arc<dyn SenderT>>>,
-}
-
-impl InboxRegistry {
-    pub fn new() -> Self {
-        Self {
-            senders: RwLock::new(HashMap::new()),
-        }
-    }
-
-    pub fn register(&self, addr: ActorAddress, sender: Arc<dyn SenderT>) {
-        self.senders.write().unwrap().insert(addr, sender);
-    }
-
-    pub fn try_deliver(
-        &self,
-        addr: ActorAddress,
-        msg: Box<dyn Any + Send>,
-    ) -> 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<dyn Any + Send>) -> Result<(), Error> {
         match self.address_map.lookup(&addr) {

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<WorkerInfo>,
+}

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<Envelope>],
-    pub(crate) spawn_txs: &'a [Sender<(ActorAddress, Box<dyn AnyActor>)>],
-    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<M: Message> {
-    queue: VecDeque<M>,
-    waterlevel: usize,
-}
-
-impl<M: Message> Mailbox<M> {
-    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<M> {
-        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;

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 ─────────────────────────────────────────────────────────