zacheryasc/swactor
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: mailbox

Browse source 
In the middle of a vibe-coded infra change. Modularizing the components into APIs amenable to lots of testing and optimization.
This commit is contained in:
Zachery Aaron Shores-Chmielewski 2026-02-05 21:22:36 +07:00
parent abedaf6307
commit 172e78a50b
7 changed files with 537 additions and 1051 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

198
ARCHITECTURE.md Normal file
View file

@ -0,0 +1,198 @@
# Swactor Architecture
## System Diagram
```
┌─────────────────────────────────────────────────────────────────────────────┐
│ Runtime │
│ (composes everything) │
│ │
│ ┌───────────────────────────────────────────────────────────────────────┐ │
│ │ Address Map │ │
│ │ ActorAddress → WorkerId │ │
│ │ (shared across all workers, read-heavy) │ │
│ └──────┬──────────────────┬──────────────────────┬─────────────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Worker 0 │ │ Worker 1 │ ... │ Worker N │ │
│ │ (thread) │ │ (thread) │ │ (thread) │ │
│ │ │ │ │ │ │ │
│ │ ┌─────────┐ │ │ ┌─────────┐ │ │ ┌─────────┐ │ │
│ │ │ Actor A │ │ │ │ Actor C │ │ │ │ Actor E │ │ │
│ │ │ [═════] │ │ │ │ [═════] │ │ │ │ [═════] │ │ │
│ │ │ mailbox │ │ │ │ mailbox │ │ │ │ mailbox │ │ │
│ │ └─────────┘ │ │ └─────────┘ │ │ └─────────┘ │ │
│ │ ┌─────────┐ │ │ ┌─────────┐ │ │ ┌─────────┐ │ │
│ │ │ Actor B │ │ │ │ Actor D │ │ │ │ Actor F │ │ │
│ │ │ [═════] │ │ │ │ [═════] │ │ │ │ [═════] │ │ │
│ │ │ mailbox │ │ │ │ mailbox │ │ │ │ mailbox │ │ │
│ │ └─────────┘ │ │ └─────────┘ │ │ └─────────┘ │ │
│ │ │ │ │ │ │ │
│ │ ┌─────────┐ │ │ ┌─────────┐ │ │ ┌─────────┐ │ │
│ │ │Transfer │◄├────├─┤Transfer │◄├───────├─┤Transfer │ │ │
│ │ │ Queue │ │ │ │ Queue │ │ │ │ Queue │ │ │
│ │ │ (MPSC) │─├────├►│ (MPSC) │─├───────├►│ (MPSC) │ │ │
│ │ └─────────┘ │ │ └─────────┘ │ │ └─────────┘ │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
══ = VecDeque (no atomics)
```
## Message Flow
```
SAME WORKER (fast path — zero atomics)
═══════════════════════════════════════
Actor A Actor B
handle() { mailbox (VecDeque)
ctx.send(addr_B, msg) ▲
│ │
├─ address_map[addr_B] │
│ → Worker 0 (that's me!) │
│ │
└─ mailbox_B.push(msg) ──────┘
} no atomics, no envelope
CROSS-WORKER (one atomic hop)
═══════════════════════════════
Actor A (Worker 0) Worker 1 Actor C (Worker 1)
handle() { transfer queue mailbox (VecDeque)
ctx.send(addr_C, msg) ▲ ▲
│ │ │
├─ address_map[addr_C] │ │
│ → Worker 1 (not me) │ │
│ │ │
└─ envelope(addr_C, msg) ──────┘ │
(atomic push) │ │
└── worker 1 pops ─────┘
and distributes
(local, no atomic)
}
```
## Worker Loop
```
┌─────────────────────────────────────────────┐
│ Worker Thread │
│ │
│ loop { │
│ ┌──────────────────────────────────────┐ │
│ │ 1. DRAIN TRANSFER QUEUE │ │
│ │ while let Some((addr, env)) = │ │
│ │ transfer_queue.pop() │ │
│ │ { │ │
│ │ local_actors[addr].mailbox │ │
│ │ .push(env.unpack()) │ │
│ │ } │ │
│ └──────────────────────────────────────┘ │
│ ┌──────────────────────────────────────┐ │
│ │ 2. TICK ACTORS │ │
│ │ for actor in &mut actor_pool { │ │
│ │ let n = drain_count(actor); │ │
│ │ for _ in 0..n { │ │
│ │ let msg = actor.mailbox.pop();│ │
│ │ actor.handle(&ctx, msg); │ │
│ │ } │ │
│ │ } │ │
│ └──────────────────────────────────────┘ │
│ ┌──────────────────────────────────────┐ │
│ │ 3. IDLE? │ │
│ │ if no messages processed: │ │
│ │ spin → yield → park │ │
│ └──────────────────────────────────────┘ │
│ } │
└───────────────────────────────────────────────┘
```
## File Tree
```
src/
├── lib.rs # crate root, feature flags, public exports
├── error.rs # Error type
│
├── actor.rs # Message trait, ActorInterface trait, ActorAddress
│ # - ActorInterface::handle(&mut self, ctx: &dyn Context, msg)
│ # - actors depend ONLY on Context, nothing else
│
├── context.rs # Context trait — the "syscall interface" for actors
│ # - send(), self_addr(), spawn()
│ # - this is ALL actors can see of the framework
│
├── envelope.rs # Envelope type — type erasure for cross-thread messages
│ # - wraps typed messages for the transfer queue
│ # - unwraps back to concrete type at destination
│
├── address_map.rs # ActorAddress → WorkerId mapping
│ # - shared read-heavy structure
│ # - written on spawn, read on every send
│
├── transfer.rs # Transfer queue — per-worker MPSC
│ # - the ONE concurrent data structure on the hot path
│ # - carries (ActorAddress, Envelope) pairs
│
├── worker/
│ ├── mod.rs # Worker struct and worker loop
│ │ # - owns actor pool + transfer queue
│ │ # - the thread boundary: concurrent outside, local inside
│ │ # - drain transfer queue → tick actors → backoff
│ │
│ ├── mailbox.rs # VecDeque-based local mailbox
│ │ # - NO atomics, NO Arc, NO crossbeam
│ │ # - only touched by the owning worker thread
│ │
│ └── pool.rs # Actor pool — stores actors assigned to this worker
│ # - local HashMap or Vec for ActorAddress → Actor lookup
│ # - insert on spawn, remove on shutdown
│
├── runtime.rs # Runtime — the composition point
│ # - creates workers, address map
│ # - implements Context (delegates to address map + transfer queues)
│ # - public API: new(), spawn(), send_to(), run(), tick(), shutdown()
│
├── config.rs # RuntimeConfig — tuning knobs
│ # - num_threads, max_actors, mailbox capacity
│ # - drain strategy, backoff policy
│ # - placement strategy (round-robin, caller-affinity, etc.)
│
└── placement.rs # Actor placement strategy
# - decides which worker a new actor goes to
# - round-robin, least-loaded, caller-affinity
```
## Components
| Component | File(s) | What It Does | Concurrent? |
|---|---|---|---|
| **Worker** | `worker/mod.rs` | Owns a thread, a pool of actors, their mailboxes, and a transfer queue. Runs the tick loop. Everything inside is single-threaded. | No (that's the point) |
| **Mailbox** | `worker/mailbox.rs` | `VecDeque<M>` per actor. Zero atomics. Only the owning worker reads/writes. | No |
| **Actor Pool** | `worker/pool.rs` | Stores actors on this worker. Local lookup by address. | No |
| **Transfer Queue** | `transfer.rs` | MPSC queue per worker. The only atomic boundary. Other workers push, this worker pops. | Yes (the ONE place) |
| **Address Map** | `address_map.rs` | Maps ActorAddress → WorkerId. Read on every cross-thread send, written on spawn. | Yes (read-heavy) |
| **Envelope** | `envelope.rs` | Type-erases messages for the transfer queue. Unwrapped at destination. | No (data format) |
| **Context** | `context.rs` | Trait that actors see. `send()`, `self_addr()`, `spawn()`. Hides all framework internals. | N/A (trait) |
| **Runtime** | `runtime.rs` | Wires it all together. Creates workers, holds address map, exposes public API. | Minimal (delegates) |
| **Placement** | `placement.rs` | Decides which worker gets a new actor. | No (called at spawn time) |
## Single-Threaded / WASM Mode
One worker. No transfer queue needed. No address map needed (everything is local). The system collapses to:
```
Worker 0
┌───────────────────────┐
│ Actor A [mailbox] │
│ Actor B [mailbox] │ All sends are local.
│ Actor C [mailbox] │ All mailboxes are VecDeque.
│ │ Zero atomics anywhere.
│ tick() drives loop │
└───────────────────────┘
```

1050
CACHE.md
View file

File diff suppressed because it is too large Load diff

58
README.md
View file

@ -1,2 +1,58 @@
# swactor
Small wasm-compatible actor library
(S)mall (W)ASM-compatible (actor) library
## Quick example
```rust
use swactor::{
actor::{ActorAddress, ActorInterface},
runtime::{Runtime, RuntimeConfig},
};
#[derive(Debug, Default)]
struct Greeter { num_greeted: usize }
#[derive(Debug, Default, Clone)]
struct GreetMessage { who: String, return_addr: ActorAddress }
#[derive(Debug, Default, Clone)]
struct GreetResponse(String);
impl ActorInterface for Greeter {
type Incoming = GreetMessage;
type Response = GreetResponse;
fn handle(&mut self, ctx: &Runtime, msg: GreetMessage) {
let res = GreetResponse(format!("Hello, {}!", msg.who));
self.num_greeted += 1;
if let Err(_) = ctx.send_to(msg.return_addr, res) {
self.num_greeted -= 1;
}
}
}
fn main() {
let rt = Runtime::new(RuntimeConfig::default());
let addr = rt.spawn(Greeter::default()).expect("failed to spawn");
let inbox = rt.new_inbox::<GreetResponse>().unwrap();
rt.send_to(addr, GreetMessage {
who: "world".into(),
return_addr: *inbox.addr(),
}).unwrap();
for _ in 0..3 { rt.tick(); }
let resp = inbox.try_recv().expect("should have response");
println!("{}", resp.0); // "Hello, world!"
}
```
## Build & test
```sh
cargo build
cargo test
cargo test --features stress # stress tests
cargo run --bin bench --release # benchmarks
cargo run --example hello
```

1
src/lib.rs
View file

@ -1,4 +1,5 @@
pub mod actor;
pub mod worker;
mod channel;
pub(crate) mod error;

54
src/worker/mailbox.rs Normal file
View file

@ -0,0 +1,54 @@
use std::collections::VecDeque;
use crate::actor::Message;
const DEFAULT_WATERLEVEL: usize = 10;
pub struct Mailbox<M: Message> {
queue: VecDeque<M>,
waterlevel: usize,
}
impl<M: Message> Mailbox<M> {
pub fn new() -> Self {
Self {
queue: VecDeque::new(),
waterlevel: DEFAULT_WATERLEVEL,
}
}
pub fn with_waterlevel(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 {
let len = self.queue.len();
if len < self.waterlevel {
len
} else {
len >> 1
}
}
}

1
src/worker/mod.rs Normal file
View file

@ -0,0 +1 @@
pub mod mailbox;

226
tests/mailbox_tests.rs Normal file
View file

@ -0,0 +1,226 @@
use swactor::worker::mailbox::Mailbox;
// ── Basic operations ──
#[test]
fn push_and_pop() {
let mut mb = Mailbox::new();
mb.push(42i32);
assert_eq!(mb.pop(), Some(42));
}
#[test]
fn fifo_ordering() {
let mut mb = Mailbox::new();
mb.push(1);
mb.push(2);
mb.push(3);
assert_eq!(mb.pop(), Some(1));
assert_eq!(mb.pop(), Some(2));
assert_eq!(mb.pop(), Some(3));
}
#[test]
fn pop_empty() {
let mut mb: Mailbox<i32> = Mailbox::new();
assert_eq!(mb.pop(), None);
}
#[test]
fn multiple_messages() {
let mut mb = Mailbox::new();
for i in 0..100 {
mb.push(i);
}
for i in 0..100 {
assert_eq!(mb.pop(), Some(i));
}
assert_eq!(mb.pop(), None);
}
#[test]
fn interleaved_push_pop() {
let mut mb = Mailbox::new();
mb.push(1);
mb.push(2);
assert_eq!(mb.pop(), Some(1));
mb.push(3);
assert_eq!(mb.pop(), Some(2));
assert_eq!(mb.pop(), Some(3));
assert_eq!(mb.pop(), None);
}
// ── Drain count / watermark logic ──
#[test]
fn drain_count_empty() {
let mb: Mailbox<i32> = Mailbox::new();
assert_eq!(mb.drain_count(), 0);
}
#[test]
fn drain_count_below_waterlevel() {
let mut mb = Mailbox::new();
for i in 0..5 {
mb.push(i);
}
// 5 < 10 (default waterlevel) → process all
assert_eq!(mb.drain_count(), 5);
}
#[test]
fn drain_count_at_waterlevel() {
let mut mb = Mailbox::new();
for i in 0..10 {
mb.push(i);
}
// 10 >= 10 → process half → 5
assert_eq!(mb.drain_count(), 5);
}
#[test]
fn drain_count_above_waterlevel() {
let mut mb = Mailbox::new();
for i in 0..20 {
mb.push(i);
}
// 20 >= 10 → 20 >> 1 = 10
assert_eq!(mb.drain_count(), 10);
}
#[test]
fn drain_count_one_message() {
let mut mb = Mailbox::new();
mb.push(1i32);
// 1 < 10 → process all → 1
assert_eq!(mb.drain_count(), 1);
}
#[test]
fn drain_count_just_below_waterlevel() {
let mut mb = Mailbox::new();
for i in 0..9 {
mb.push(i);
}
// 9 < 10 → process all → 9
assert_eq!(mb.drain_count(), 9);
}
#[test]
fn drain_count_large() {
let mut mb = Mailbox::new();
for i in 0..1000 {
mb.push(i);
}
// 1000 >= 10 → 1000 >> 1 = 500
assert_eq!(mb.drain_count(), 500);
}
#[test]
fn drain_count_custom_waterlevel() {
let mut mb = Mailbox::with_waterlevel(4);
for i in 0..3 {
mb.push(i);
}
// 3 < 4 → process all → 3
assert_eq!(mb.drain_count(), 3);
mb.push(99);
// 4 >= 4 → 4 >> 1 = 2
assert_eq!(mb.drain_count(), 2);
}
#[test]
fn drain_count_updates_after_pop() {
let mut mb = Mailbox::new();
for i in 0..20 {
mb.push(i);
}
// 20 >= 10 → 10
assert_eq!(mb.drain_count(), 10);
// pop 15, leaving 5
for _ in 0..15 {
mb.pop();
}
// 5 < 10 → process all → 5
assert_eq!(mb.drain_count(), 5);
}
// ── Properties ──
#[test]
fn len_tracks_pushes() {
let mut mb = Mailbox::new();
assert_eq!(mb.len(), 0);
mb.push(1);
assert_eq!(mb.len(), 1);
mb.push(2);
assert_eq!(mb.len(), 2);
mb.push(3);
assert_eq!(mb.len(), 3);
}
#[test]
fn len_tracks_pops() {
let mut mb = Mailbox::new();
mb.push(1);
mb.push(2);
mb.push(3);
assert_eq!(mb.len(), 3);
mb.pop();
assert_eq!(mb.len(), 2);
mb.pop();
assert_eq!(mb.len(), 1);
mb.pop();
assert_eq!(mb.len(), 0);
}
#[test]
fn is_empty_on_new() {
let mb: Mailbox<i32> = Mailbox::new();
assert!(mb.is_empty());
}
#[test]
fn is_empty_after_drain() {
let mut mb = Mailbox::new();
mb.push(1);
mb.push(2);
mb.push(3);
assert!(!mb.is_empty());
mb.pop();
mb.pop();
mb.pop();
assert!(mb.is_empty());
}
// ── Type tests ──
#[test]
fn works_with_primitive_types() {
let mut mb_i32 = Mailbox::new();
mb_i32.push(42i32);
assert_eq!(mb_i32.pop(), Some(42));
let mut mb_string = Mailbox::new();
mb_string.push(String::from("hello"));
assert_eq!(mb_string.pop(), Some(String::from("hello")));
}
#[test]
fn works_with_custom_structs() {
#[derive(Debug, Clone, PartialEq)]
struct MyMsg {
id: u64,
payload: String,
}
let mut mb = Mailbox::new();
let msg = MyMsg {
id: 1,
payload: "test".into(),
};
mb.push(msg.clone());
assert_eq!(mb.pop(), Some(msg));
}