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

Compare commits

base: zacheryasc:abedaf6307cc130faf1e7d0a9cf2c8a731f37b30
Branches Tags
zacheryasc:master
zacheryasc:development
zacheryasc:backup/pre-rewrite-msg
..
compare: zacheryasc:7c7947c823c77285a8275c11cbd426cf4c0a60e4
Branches Tags
zacheryasc:development
zacheryasc:master
zacheryasc:backup/pre-rewrite-msg

2 commits
abedaf6307 ... 7c7947c823

Author SHA1 Message Date
Zachery Aaron Shores-Chmielewski
7c7947c823 feat: actor addresses, Ctx abstraction, pools, and more 
Fleshing out the architecture before abstracting into components amenable to api-based test harnesses for fuzz and optimization working loops.
 2026-02-05 22:54:03 +07:00
Zachery Aaron Shores-Chmielewski
172e78a50b feat: mailbox 
In the middle of a vibe-coded infra change. Modularizing the components into APIs amenable to lots of testing and optimization.
 2026-02-05 21:22:36 +07:00
26 changed files with 1359 additions and 1577 deletions
Show stats 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

20
Cargo.lock generated
View file

@ -8,25 +8,6 @@ version = "1.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801"
[[package]]
name = "crossbeam-deque"
version = "0.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9dd111b7b7f7d55b72c0a6ae361660ee5853c9af73f70c3c2ef6858b950e2e51"
dependencies = [
"crossbeam-epoch",
"crossbeam-utils",
]
[[package]]
name = "crossbeam-epoch"
version = "0.9.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e"
dependencies = [
"crossbeam-utils",
]
[[package]]
name = "crossbeam-queue"
version = "0.3.12"
@ -69,7 +50,6 @@ checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03"
name = "swactor"
version = "0.1.0"
dependencies = [
"crossbeam-deque",
"crossbeam-queue",
"crossbeam-utils",
"getrandom",

1
Cargo.toml
View file

@ -15,7 +15,6 @@ stress = [] # Enable stress tests
[dependencies]
getrandom = { version = "0.2", optional = true }
crossbeam-deque = "0.8"
crossbeam-queue = "0.3.12"
crossbeam-utils = "0.8.21"
smallvec = "1.13"

59
README.md
View file

@ -1,2 +1,59 @@
# swactor
Small wasm-compatible actor library
(S)mall (W)ASM-compatible (actor) library
## Quick example
```rust
use swactor::{
Ctx,
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: &Ctx, msg: GreetMessage) {
let res = GreetResponse(format!("Hello, {}!", msg.who));
self.num_greeted += 1;
if let Err(_) = ctx.send(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
```

18
benches/scaling.rs
View file

@ -8,6 +8,7 @@
use crate::harness::{black_box, Bench, BenchSuite};
use std::thread;
use swactor::{
Ctx,
actor::ActorInterface,
runtime::{Runtime, RuntimeConfig},
};
@ -34,7 +35,7 @@ impl ActorInterface for CounterActor {
type Incoming = Increment;
type Response = ();
fn handle(&mut self, _ctx: &Runtime, _msg: Increment) {
fn handle(&mut self, _ctx: &Ctx, _msg: Increment) {
self.count += 1;
}
}
@ -47,7 +48,7 @@ impl ActorInterface for SharedCounter {
type Incoming = Increment;
type Response = ();
fn handle(&mut self, _ctx: &Runtime, _msg: Increment) {
fn handle(&mut self, _ctx: &Ctx, _msg: Increment) {
self.count
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
@ -71,7 +72,7 @@ impl ActorInterface for PayloadActor {
type Incoming = Payload;
type Response = ();
fn handle(&mut self, _ctx: &Runtime, msg: Payload) {
fn handle(&mut self, _ctx: &Ctx, msg: Payload) {
self.bytes_received += msg.0.len();
black_box(&msg.0);
}
@ -97,8 +98,7 @@ pub fn bench_actor_count_scaling(suite: &mut BenchSuite) {
|| {
let config = RuntimeConfig {
max_actors: (actor_count as usize) + 100,
router_max_messages: (total_messages as usize) * 3,
actor_max_messages: (messages_per_actor as usize) * 2,
actor_max_messages: (total_messages as usize) * 3,
num_threads: 1,
};
let runtime = Runtime::new(config);
@ -110,7 +110,7 @@ pub fn bench_actor_count_scaling(suite: &mut BenchSuite) {
actors.push(addr);
}
// Process registrations
// Process spawns
for _ in 0..(actor_count * 2) {
runtime.tick();
}
@ -156,8 +156,7 @@ pub fn bench_thread_count_scaling(suite: &mut BenchSuite) {
let counter = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
let config = RuntimeConfig {
max_actors: (actor_count as usize) + 100,
router_max_messages: (total_messages as usize) * 3,
actor_max_messages: (messages_per_actor as usize) * 2,
actor_max_messages: (total_messages as usize) * 3,
num_threads: thread_count,
};
let runtime = Runtime::new(config);
@ -233,14 +232,13 @@ pub fn bench_payload_size_scaling(suite: &mut BenchSuite) {
|| {
let config = RuntimeConfig {
max_actors: 10,
router_max_messages: (message_count as usize) * 2,
actor_max_messages: (message_count as usize) * 2,
num_threads: 1,
};
let runtime = Runtime::new(config);
let sink = runtime.spawn(PayloadActor::new()).unwrap();
// Process registration
// Process spawn
for _ in 0..10 {
runtime.tick();
}

32
benches/throughput.rs
View file

@ -8,6 +8,7 @@
use crate::harness::{black_box, Bench, BenchSuite};
use swactor::{
Ctx,
actor::{ActorAddress, ActorInterface},
runtime::{Runtime, RuntimeConfig},
};
@ -34,7 +35,7 @@ impl ActorInterface for SinkActor {
type Incoming = Ping;
type Response = ();
fn handle(&mut self, _ctx: &Runtime, _msg: Ping) {
fn handle(&mut self, _ctx: &Ctx, _msg: Ping) {
self.count += 1;
}
}
@ -58,9 +59,9 @@ impl ActorInterface for ForwardActor {
type Incoming = Ping;
type Response = Ping;
fn handle(&mut self, ctx: &Runtime, msg: Ping) {
fn handle(&mut self, ctx: &Ctx, msg: Ping) {
if let Some(next) = self.next {
let _ = ctx.send_to(next, msg);
let _ = ctx.send(next, msg);
}
}
}
@ -69,7 +70,7 @@ impl ActorInterface for ForwardActor {
// Benchmarks
// ============================================================================
/// Benchmark: Messages sent through the router to a single sink actor
/// Benchmark: Messages sent through the runtime to a single sink actor
pub fn bench_message_throughput(suite: &mut BenchSuite) {
for msg_count in [1_000u64, 10_000, 100_000] {
let name = format!("message_throughput_{}", msg_count);
@ -83,7 +84,6 @@ pub fn bench_message_throughput(suite: &mut BenchSuite) {
// Setup: create runtime and sink actor
let config = RuntimeConfig {
max_actors: 100,
router_max_messages: (msg_count as usize) * 2,
actor_max_messages: (msg_count as usize) * 2,
num_threads: 1,
};
@ -97,8 +97,6 @@ pub fn bench_message_throughput(suite: &mut BenchSuite) {
let _ = runtime.send_to::<Ping>(sink, Ping);
}
// Process until done
// Tick enough times to process all messages
// (router tick + actor tick) * messages / WATERLEVEL
for _ in 0..(count * 3) {
runtime.tick();
}
@ -123,7 +121,6 @@ pub fn bench_spawn_rate(suite: &mut BenchSuite) {
|| {
let config = RuntimeConfig {
max_actors: 1000,
router_max_messages: 10_000,
actor_max_messages: 100,
num_threads: 1,
};
@ -133,7 +130,7 @@ pub fn bench_spawn_rate(suite: &mut BenchSuite) {
for _ in 0..actor_count {
let _ = runtime.spawn(SinkActor::new());
}
// Process router messages to register all actors
// Process spawns
for _ in 0..(actor_count * 2) {
runtime.tick();
}
@ -159,10 +156,9 @@ pub fn bench_fanout(suite: &mut BenchSuite) {
|| {
let config = RuntimeConfig {
max_actors: (fan_count as usize) + 10,
router_max_messages: (fan_count as usize)
actor_max_messages: (fan_count as usize)
* (messages_per_receiver as usize)
* 2,
actor_max_messages: (messages_per_receiver as usize) * 2,
num_threads: 1,
};
let runtime = Runtime::new(config);
@ -174,7 +170,7 @@ pub fn bench_fanout(suite: &mut BenchSuite) {
sinks.push(addr);
}
// Process router registrations
// Process spawns
for _ in 0..(fan_count * 2) {
runtime.tick();
}
@ -217,8 +213,7 @@ pub fn bench_fanin(suite: &mut BenchSuite) {
let total_messages = (sender_count * messages_per_sender) as usize;
let config = RuntimeConfig {
max_actors: (sender_count as usize) + 10,
router_max_messages: total_messages * 3,
actor_max_messages: total_messages * 2,
actor_max_messages: total_messages * 3,
num_threads: 1,
};
let runtime = Runtime::new(config);
@ -233,7 +228,7 @@ pub fn bench_fanin(suite: &mut BenchSuite) {
senders.push(addr);
}
// Process router registrations
// Process spawns
for _ in 0..((sender_count + 1) * 2) {
runtime.tick();
}
@ -275,8 +270,7 @@ pub fn bench_ring(suite: &mut BenchSuite) {
|| {
let config = RuntimeConfig {
max_actors: (ring_size as usize) + 10,
router_max_messages: 10_000,
actor_max_messages: 1_000,
actor_max_messages: 10_000,
num_threads: 1,
};
let runtime = Runtime::new(config);
@ -292,7 +286,7 @@ pub fn bench_ring(suite: &mut BenchSuite) {
// so instead we'll use an inbox to receive the final message
// For now, we'll just measure message passing through a chain
// Process registrations
// Process spawns
for _ in 0..(ring_size * 2) {
runtime.tick();
}
@ -301,7 +295,7 @@ pub fn bench_ring(suite: &mut BenchSuite) {
},
|(runtime, actors, laps)| {
// Send to first actor (even though they don't forward, we're
// measuring the router + inbox overhead)
// measuring the transfer queue + inbox overhead)
for _ in 0..laps {
for actor in &actors {
let _ = runtime.send_to::<Ping>(*actor, Ping);

5
examples/hello.rs
View file

@ -1,4 +1,5 @@
use swactor::{
Ctx,
actor::{ActorAddress, ActorInterface},
runtime::{Runtime, RuntimeConfig},
};
@ -24,10 +25,10 @@ impl ActorInterface for Greeter {
type Incoming = GreetMessage;
type Response = GreetResponse;
fn handle(&mut self, ctx: &Runtime, msg: GreetMessage) {
fn handle(&mut self, ctx: &Ctx, msg: GreetMessage) {
let res = GreetResponse(format!("Hello, {}!", msg.who));
self.num_greeted += 1;
if let Err(_) = ctx.send_to(msg.return_addr, res) {
if let Err(_) = ctx.send(msg.return_addr, res) {
// no error handling
self.num_greeted -= 1;
}

5
examples/ring.rs
View file

@ -1,4 +1,5 @@
use swactor::{
Ctx,
actor::{ActorAddress, ActorInterface},
runtime::{Inbox, Runtime, RuntimeConfig},
};
@ -30,8 +31,8 @@ impl RingActor {
impl ActorInterface for RingActor {
type Incoming = RingMessage;
type Response = ();
fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming) {
if let Err(_) = ctx.send_to(self.next, msg.next()) {
fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming) {
if let Err(_) = ctx.send(self.next, msg.next()) {
// do nothing
}
}

69
src/actor.rs
View file

@ -1,4 +1,7 @@
use crate::{WATERLEVEL, channel::Receiver, get_random, runtime::Runtime};
use std::any::Any;
use crate::{get_random, worker::mailbox::Mailbox};
use crate::context::{ContextInner, Ctx};
/// The primary trait defining data that can be passed to and from actor processes
pub trait Message: 'static + Sized + Clone + Send + Sync {}
@ -14,7 +17,7 @@ impl<T: 'static + Sized + Clone + Send + Sync> Message for T {}
///
/// # Example
/// ```
/// use swactor::{actor::{ActorAddress, ActorInterface}, runtime::Runtime};
/// use swactor::{Ctx, actor::{ActorAddress, ActorInterface}};
///
/// struct Greeter {
/// num_greeted: usize,
@ -32,10 +35,10 @@ impl<T: 'static + Sized + Clone + Send + Sync> Message for T {}
/// impl ActorInterface for Greeter {
/// type Incoming = GreetMessage;
/// type Response = GreetResponse;
///
/// fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming) {
///
/// fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming) {
/// let response = GreetResponse(format!("Hello, {}!", msg.who).to_string());
/// if let Ok(_) = ctx.send_to(msg.return_addr, response) {
/// if let Ok(_) = ctx.send(msg.return_addr, response) {
/// self.num_greeted += 1;
/// }
/// }
@ -44,7 +47,7 @@ impl<T: 'static + Sized + Clone + Send + Sync> Message for T {}
pub trait ActorInterface: 'static + Send {
type Incoming: Message;
type Response: Message;
fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming);
fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming);
}
/// A unique address for this actor. 32 bytes is overkill for a small application,
@ -60,48 +63,58 @@ impl ActorAddress {
}
}
/// The actor process as represented in the Runtime, with the actor state stored with it's inbox.
/// The actor process as represented in the Runtime, with the actor state stored with its mailbox.
pub(crate) struct Actor<A>
where
A: ActorInterface,
{
inbox: Receiver<A::Incoming>,
addr: ActorAddress,
mailbox: Mailbox<A::Incoming>,
inner: A,
}
impl<A: ActorInterface> Actor<A> {
pub(crate) fn new(inbox: Receiver<A::Incoming>, inner: A) -> Self {
Self { inbox, inner }
pub(crate) fn new(addr: ActorAddress, mailbox: Mailbox<A::Incoming>, inner: A) -> Self {
Self {
addr,
mailbox,
inner,
}
}
}
/// Trait for type-erased actors
pub(crate) trait AnyActor: Send {
fn tick(&mut self, ctx: &Runtime);
/// Tick the actor, processing pending messages. Returns `true` if any work was done.
fn tick(&mut self, inner: &dyn ContextInner) -> bool;
/// Deliver a type-erased message into this actor's mailbox.
/// Returns `true` if the downcast succeeded.
fn deliver(&mut self, msg: Box<dyn Any + Send>) -> bool;
}
impl<A> AnyActor for Actor<A>
where
A: ActorInterface,
{
fn tick(&mut self, ctx: &Runtime) {
// TODO: WATERLEVEL is hard coded, and so is this message handling scheme. We should
// make it so both are more flexible, with sane defaults.
// FIXME: lots of indirection just to get length on a hot path
let total_messages = self.inbox.len();
let messages_to_process = if total_messages < WATERLEVEL {
total_messages
} else {
total_messages >> 1
};
for _ in 0..messages_to_process {
match self.inbox.try_recv() {
Some(msg) => self.inner.handle(ctx, msg),
None => unreachable!(
"We checked number of unprocessed messages in the queue ahead of processing"
),
fn tick(&mut self, inner: &dyn ContextInner) -> bool {
let n = self.mailbox.drain_count();
if n > 0 {
let ctx = Ctx::new(inner, self.addr);
for _ in 0..n {
if let Some(msg) = self.mailbox.pop() {
self.inner.handle(&ctx, msg);
}
}
}
n > 0
}
fn deliver(&mut self, msg: Box<dyn Any + Send>) -> bool {
if let Ok(typed) = msg.downcast::<A::Incoming>() {
self.mailbox.push(*typed);
true
} else {
false
}
}
}

90
src/address_map.rs Normal file
View file

@ -0,0 +1,90 @@
use std::collections::HashMap;
use std::sync::RwLock;
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()
}
}
#[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);
}
}

24
src/config.rs Normal file
View file

@ -0,0 +1,24 @@
/// The tunable settings for the runtime.
pub struct RuntimeConfig {
pub max_actors: usize,
pub actor_max_messages: usize,
pub num_threads: usize,
}
/// 8kB for the `Box<..>` before counting the rest of the memory
const DEFAULT_MAX_ACTORS: usize = 1_000;
/// 16kB PER ACTOR to alloc space for storing messages.
/// With default setting of [DEFAULT_MAX_ACTORS] this is:
/// 1_000 * 16kB = 16MB
const DEFAULT_ACTOR_MAX_MESSAGES: usize = 1_000;
impl Default for RuntimeConfig {
fn default() -> Self {
Self {
max_actors: DEFAULT_MAX_ACTORS,
actor_max_messages: DEFAULT_ACTOR_MAX_MESSAGES,
num_threads: 1,
}
}
}

45
src/context.rs Normal file
View file

@ -0,0 +1,45 @@
use std::any::Any;
use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message};
use crate::worker::mailbox::Mailbox;
use crate::Error;
/// Object-safe inner trait for sending type-erased messages.
pub(crate) trait ContextInner {
fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> Result<(), Error>;
fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>) -> Result<(), Error>;
}
/// Actor syscall interface — passed to `ActorInterface::handle()`.
///
/// Wraps a `&dyn ContextInner` to solve the object-safety problem while
/// providing a typed public API.
pub struct Ctx<'a> {
inner: &'a dyn ContextInner,
self_addr: ActorAddress,
}
impl<'a> Ctx<'a> {
pub(crate) fn new(inner: &'a dyn ContextInner, self_addr: ActorAddress) -> Self {
Self { inner, self_addr }
}
/// Returns the address of the actor currently being ticked.
pub fn self_addr(&self) -> ActorAddress {
self.self_addr
}
/// Send a typed message to an actor address.
pub fn send<M: Message>(&self, addr: ActorAddress, msg: M) -> Result<(), Error> {
self.inner.send_any(addr, Box::new(msg))
}
/// Spawn a new actor, returning its address.
pub fn spawn<A: ActorInterface>(&self, actor: A) -> Result<ActorAddress, Error> {
let addr = ActorAddress::new_random();
let actor = Actor::new(addr, Mailbox::new(), actor);
let boxed: Box<dyn AnyActor> = Box::new(actor);
self.inner.spawn_any(addr, boxed)?;
Ok(addr)
}
}

49
src/envelope.rs Normal file
View file

@ -0,0 +1,49 @@
use std::any::Any;
use crate::actor::ActorAddress;
/// 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
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn envelope_roundtrip() {
let addr = ActorAddress::default();
let env = Envelope::new(addr, Box::new(42u64));
assert_eq!(env.dest(), addr);
assert_eq!(env.downcast::<u64>(), Some(42u64));
}
#[test]
fn envelope_wrong_type_returns_none() {
let addr = ActorAddress::default();
let env = Envelope::new(addr, Box::new(42u64));
assert!(env.downcast::<String>().is_none());
}
}

23
src/lib.rs
View file

@ -1,10 +1,18 @@
pub mod actor;
pub mod worker;
mod channel;
pub(crate) mod channel;
pub(crate) mod error;
pub use error::Error;
mod router;
pub mod context;
pub use context::Ctx;
pub(crate) mod envelope;
pub(crate) mod address_map;
pub(crate) mod placement;
pub mod config;
pub mod runtime;
#[cfg(feature = "getrandom")]
@ -25,14 +33,3 @@ pub(crate) fn get_random(buf: &mut [u8]) {
*byte = bytes[i % core::mem::size_of::<usize>()];
}
}
/// FIXME: remove hard coded defaults
/// The strategy for message processing is such:
///
/// ```ignore
/// if total_messages < WATERLEVEL:
/// process all
/// else
/// process total_messages >> 1
/// ```
const WATERLEVEL: usize = 10;

37
src/placement.rs Normal file
View file

@ -0,0 +1,37 @@
use std::sync::atomic::{AtomicUsize, Ordering};
use crate::address_map::WorkerId;
/// 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 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));
}
}

83
src/router.rs
View file

@ -1,83 +0,0 @@
use std::{collections::HashMap, sync::Arc};
use crate::{
actor::{ActorAddress, ActorInterface, Message},
channel::Sender,
runtime::Runtime,
};
/// FIXME: Go over with a fine-toothed comb and reassure yourself this typing
/// makes sense, that we are not doing loads of indirection on a hot path.
///
/// A type erased `Message` to be routed between actor processes.
pub(crate) type Envelope = Arc<dyn std::any::Any + Send + Sync>;
pub(crate) trait SenderT: Send + Sync {
fn try_send(&self, envelope: Envelope);
}
impl<M: Message> SenderT for Sender<M> {
fn try_send(&self, envelope: Envelope) {
if let Some(msg) = envelope.downcast_ref::<M>() {
// FIXME: we are directly cloning the contents of the Arc pointer here
// Do we want to? Should we provide another way?
//
// The standard concept of an actor has message and state
// isolation, so we should leave this as is. However, we should
// make it clear and obvious this pathway is the heavy, contained
// pathway, and include a shared memory pathway for logic that
// may need it.
let _ = Sender::try_send(self, msg.clone());
}
}
}
/// Internal messages for the Router's own inbox
#[derive(Clone)]
pub(crate) enum RouterMessage {
/// register addrs <addr> with sender <sender>
AddAddr(ActorAddress, Arc<dyn SenderT>),
/// FIXME: this will be active when we allow actors to shut themselves
/// down. For now, disable the warning.
#[allow(dead_code)]
/// remove an actor from the address book
RemoveAddr(ActorAddress),
/// send <msg> to <addr>
SendToAddr { addr: ActorAddress, msg: Envelope },
}
/// The `Router` is responsible for taking in and delivering all messages in the runtime.
pub(crate) struct Router {
directory: HashMap<ActorAddress, Arc<dyn SenderT>>,
}
impl Router {
pub fn new() -> Self {
Self {
directory: HashMap::new(),
}
}
}
impl ActorInterface for Router {
type Incoming = RouterMessage;
type Response = ();
fn handle(&mut self, _ctx: &Runtime, msg: Self::Incoming) {
match msg {
RouterMessage::AddAddr(addr, sender) => {
self.directory.insert(addr, sender);
}
RouterMessage::RemoveAddr(addr) => {
self.directory.remove(&addr);
}
RouterMessage::SendToAddr { addr, msg } => {
if let Some(sender) = self.directory.get(&addr) {
sender.try_send(msg);
}
}
}
}
}

585
src/runtime.rs
View file

@ -1,17 +1,71 @@
use std::any::Any;
use std::cell::RefCell;
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::sync::{Arc, RwLock};
use std::thread::{self, JoinHandle};
use std::time::Duration;
use crossbeam_deque::{Injector, Steal, Stealer, Worker};
use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message};
use crate::address_map::{AddressMap, WorkerId};
use crate::channel::{Receiver, Sender};
// Re-export RuntimeConfig so existing code using `runtime::RuntimeConfig` still works
pub use crate::config::RuntimeConfig;
use crate::context::ContextInner;
use crate::envelope::Envelope;
use crate::placement::Placement;
use crate::worker::mailbox::Mailbox;
use crate::worker::Worker;
use crate::Error;
use crate::channel::HybridChannel;
use crate::{
actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message},
channel::{Receiver, Sender},
router::{Router, RouterMessage},
Error,
};
// ─── SenderT trait (moved from router.rs) ────────────────────────────────────
/// 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);
}
}
}
// ─── 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"))
}
}
}
// ─── Inbox ───────────────────────────────────────────────────────────────────
/// Generic message inbox for receiving messages outside of the runtime.
pub struct Inbox<M: Message> {
@ -29,59 +83,210 @@ impl<M: Message> Inbox<M> {
}
}
/// The tunable settings for the runtime.
pub struct RuntimeConfig {
pub max_actors: usize,
pub router_max_messages: usize,
pub actor_max_messages: usize,
pub num_threads: usize,
}
/// 8kB for the `Box<..>` before counting the rest of the memory
const DEFAULT_MAX_ACTORS: usize = 1_000;
/// 160kB for the `Arc<..>` before counting the rest of the memory
const DEFAULT_ROUTER_MAX_MESSAGES: usize = 10_000;
/// 16kB PER ACTOR to alloc space for storing the `Arc<..>` pointers
/// With default setting of [DEFAULT_MAX_ACTORS] this is:
/// 1_000 * 16kB = 16MB
const DEFAULT_ACTOR_MAX_MESSAGES: usize = 1_000;
impl Default for RuntimeConfig {
fn default() -> Self {
Self {
max_actors: DEFAULT_MAX_ACTORS,
router_max_messages: DEFAULT_ROUTER_MAX_MESSAGES,
actor_max_messages: DEFAULT_ACTOR_MAX_MESSAGES,
num_threads: 1,
}
}
}
/// Per-worker state for work-stealing scheduler
struct WorkerState {
local: Worker<Box<dyn AnyActor>>, // FIFO for fairness
id: usize,
}
/// Shared scheduler state for work-stealing
struct SchedulerState {
injector: Injector<Box<dyn AnyActor>>, // For spawn()
stealers: Vec<Stealer<Box<dyn AnyActor>>>, // For work-stealing
is_running: AtomicBool,
}
// ─── Runtime ─────────────────────────────────────────────────────────────────
/// The `Runtime` struct is the primary gateway for interacting with the framework.
pub struct Runtime {
config: RuntimeConfig,
actor_queue: HybridChannel<Box<dyn AnyActor>>, // Used for single-threaded mode
scheduler: Option<Arc<SchedulerState>>, // Used for multi-threaded mode
router_interface: Sender<RouterMessage>,
router: Option<Actor<Router>>, // `None` if single-threaded
// for multithreaded contexts
address_map: Arc<AddressMap>,
inbox_registry: Arc<InboxRegistry>,
transfer_txs: Vec<Sender<Envelope>>,
spawn_txs: Vec<Sender<(ActorAddress, Box<dyn AnyActor>)>>,
placement: Placement,
is_running: AtomicBool,
/// Single-threaded mode: worker stored inline
single_worker: Option<RefCell<Worker>>,
/// Multi-threaded mode: workers waiting to be assigned to threads by run()
pending_workers: Option<Vec<Worker>>,
}
// Safety: RefCell<Worker> is only accessed from the thread that owns the Runtime
// in single-threaded mode. In multi-threaded mode, single_worker is None and
// pending_workers is consumed by run() before Arc sharing.
unsafe impl Sync for Runtime {}
impl Runtime {
/// Builds a new `Runtime` struct, but does not yet run anything. If multithreaded, call
/// `run()`, if single threaded, needs to be driven by calls to the `tick()` method.
pub fn new(config: RuntimeConfig) -> Self {
let num_workers = if config.num_threads < 2 {
1
} else {
config.num_threads
};
let address_map = Arc::new(AddressMap::with_capacity(config.max_actors));
let inbox_registry = Arc::new(InboxRegistry::new());
let placement = Placement::new(num_workers);
let mut transfer_txs = Vec::with_capacity(num_workers);
let mut spawn_txs = Vec::with_capacity(num_workers);
let mut workers = Vec::with_capacity(num_workers);
for i in 0..num_workers {
let transfer_rx = Receiver::<Envelope>::new(config.actor_max_messages);
let transfer_tx = transfer_rx.new_sender();
transfer_txs.push(transfer_tx);
let spawn_rx =
Receiver::<(ActorAddress, Box<dyn AnyActor>)>::new(config.max_actors);
let spawn_tx = spawn_rx.new_sender();
spawn_txs.push(spawn_tx);
workers.push(Worker::new(WorkerId(i), transfer_rx, spawn_rx));
}
if config.num_threads < 2 {
// Single-threaded: store one worker inline
let worker = workers.remove(0);
Self {
config,
address_map,
inbox_registry,
transfer_txs,
spawn_txs,
placement,
is_running: AtomicBool::new(false),
single_worker: Some(RefCell::new(worker)),
pending_workers: None,
}
} else {
// Multi-threaded: stash workers for run()
Self {
config,
address_map,
inbox_registry,
transfer_txs,
spawn_txs,
placement,
is_running: AtomicBool::new(false),
single_worker: None,
pending_workers: Some(workers),
}
}
}
/// Spawn an actor, returns its address
pub fn spawn<A: ActorInterface>(&self, actor: A) -> Result<ActorAddress, Error> {
let addr = ActorAddress::new_random();
let worker_id = self.placement.next_worker();
self.address_map.insert(addr, worker_id);
let actor = Actor::new(addr, Mailbox::new(), actor);
let boxed: Box<dyn AnyActor> = Box::new(actor);
self.spawn_txs[worker_id.as_usize()]
.try_send((addr, boxed))
.map_err(|_| Error::from("Runtime error: spawn queue full"))?;
Ok(addr)
}
/// Send a message to an actor address
pub fn send_to<M: Message>(&self, addr: ActorAddress, msg: M) -> Result<(), Error> {
let msg_box: Box<dyn Any + Send> = Box::new(msg);
match self.address_map.lookup(&addr) {
Some(wid) => self.transfer_txs[wid.as_usize()]
.try_send(Envelope::new(addr, msg_box))
.map_err(|_| Error::from("Transfer queue full")),
None => self.inbox_registry.try_deliver(addr, msg_box),
}
}
/// Create an external inbox for receiving messages in the outer process containing the runtime
pub fn new_inbox<M: Message>(&self) -> Result<Inbox<M>, Error> {
let addr = ActorAddress::new_random();
let receiver = Receiver::<M>::new(self.config.actor_max_messages);
let sender = receiver.new_sender();
self.inbox_registry.register(addr, Arc::new(sender));
Ok(Inbox {
addr,
inner: receiver,
})
}
/// Drive one tick of the single-threaded worker.
pub fn tick(&self) {
if let Some(ref worker) = self.single_worker {
worker.borrow_mut().tick_once(
&self.address_map,
&self.transfer_txs,
&self.spawn_txs,
&self.placement,
&self.inbox_registry,
);
}
}
/// Spawn worker threads and start processing, returning a set of handles and
/// a Runtime object to interface with.
///
/// ### WARN:
/// ##### Returns an error if the configuration is set as single threaded
/// `config.num_threads == 1`
pub fn run(mut self) -> Result<RuntimeHandle, Error> {
if self.config.num_threads < 2 {
return Err(Error::from(
"Runtime error: cannot call `Runtime::run()` from a single-threaded context.",
));
}
self.is_running.store(true, Ordering::Release);
let workers = self
.pending_workers
.take()
.expect("Workers must be present for multi-threaded runtime");
let rt = Arc::new(self);
let mut handles: Vec<JoinHandle<()>> = Vec::with_capacity(workers.len());
for mut worker in workers {
let rt_clone = rt.clone();
let handle = thread::spawn(move || {
worker_loop(&mut worker, &rt_clone);
});
handles.push(handle);
}
Ok(RuntimeHandle {
runtime: rt,
threads: handles,
})
}
/// Signal all workers to stop
pub fn shutdown(&self) {
self.is_running.store(false, Ordering::Release);
}
}
/// Worker thread loop for multi-threaded runtime.
/// Uses spin → yield → park backoff to reduce CPU usage when idle.
fn worker_loop(worker: &mut Worker, rt: &Runtime) {
let mut idle_count: u32 = 0;
while rt.is_running.load(Ordering::Acquire) {
let did_work = worker.tick_once(
&rt.address_map,
&rt.transfer_txs,
&rt.spawn_txs,
&rt.placement,
&rt.inbox_registry,
);
if did_work {
idle_count = 0;
} else {
idle_count = idle_count.saturating_add(1);
if idle_count < 64 {
core::hint::spin_loop();
} else if idle_count < 256 {
thread::yield_now();
} else {
// Park: sleep briefly, cap at 1ms
let micros = std::cmp::min((idle_count - 256) as u64 * 50, 1000);
thread::sleep(std::time::Duration::from_micros(micros));
}
}
}
}
/// Handle for dealing with a runtime that has started via the `Runtime::run()` method.
@ -103,258 +308,22 @@ impl RuntimeHandle {
}
}
impl Runtime {
/// Builds a new `Runtime` struct, but does not yet run anything. If multithreaded, call
/// `run()`, if single threaded, needs to be driven by calls to the `tick()` method.
pub fn new(config: RuntimeConfig) -> Self {
let actor_queue = HybridChannel::new(config.max_actors);
// router is a unique actor in that the runtime needs access to it's `Sender` handle
let router_inner = Router::new();
let router_inbox: Receiver<RouterMessage> =
Receiver::<<Router as ActorInterface>::Incoming>::new(config.router_max_messages);
let router_sender = router_inbox.new_sender();
let router = Actor::new(router_inbox, router_inner);
// Single-threaded: router goes in queue. Multi-threaded: stays in Option
let router_option = if config.num_threads < 2 {
actor_queue
.push(Box::new(router) as Box<dyn AnyActor>)
.map_err(|_| "failed to add router to actor queue")
.expect("failed to spawn router at runtime initialization.");
None
} else {
Some(router)
};
Self {
config,
actor_queue,
scheduler: None, // Initialized in run() for multi-threaded mode
router_interface: router_sender,
is_running: AtomicBool::new(false),
router: router_option,
impl ContextInner for Runtime {
fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> Result<(), Error> {
match self.address_map.lookup(&addr) {
Some(wid) => {
let _ = self.transfer_txs[wid.as_usize()].try_send(Envelope::new(addr, msg));
Ok(())
}
None => self.inbox_registry.try_deliver(addr, msg),
}
}
/// Spawn an actor, returns its address
pub fn spawn<A: ActorInterface>(&self, actor: A) -> Result<ActorAddress, Error> {
// assign a stochastic
let addr = ActorAddress::new_random();
let inbox = Receiver::<A::Incoming>::new(self.config.actor_max_messages);
let sender = inbox.new_sender();
// Register the sender with the router
self.router_interface
.try_send(RouterMessage::AddAddr(addr, Arc::new(sender)))
.map_err(|_| {
Error::from("Runtime error: failed to add actor to router. Router inbox full")
})?;
let boxed_actor: Box<dyn AnyActor> = Box::new(Actor::new(inbox, actor));
// Multi-threaded: use injector, single-threaded: use shared queue
if let Some(ref scheduler) = self.scheduler {
scheduler.injector.push(boxed_actor);
} else {
self.actor_queue
.push(boxed_actor)
.map_err(|_| Error::from("Runtime error: Failed to spawn actor. Queue full."))?;
}
Ok(addr)
}
/// Send a message to an actor address
pub fn send_to<M: Message>(&self, addr: ActorAddress, msg: M) -> Result<(), Error> {
self.router_interface
.try_send(RouterMessage::SendToAddr {
addr,
msg: Arc::new(msg),
})
.map_err(|_| Error::from("Failed to send message to router."))
}
/// Create an external inbox for receiving messages in the outer process containing the runtime
pub fn new_inbox<M: Message>(&self) -> Result<Inbox<M>, Error> {
let addr = ActorAddress::new_random();
let receiver = Receiver::<M>::new(self.config.actor_max_messages);
let sender = receiver.new_sender();
// Register the sender with the router
self.router_interface
.try_send(RouterMessage::AddAddr(addr, Arc::new(sender)))
.map_err(|_| {
Error::from(
"Runtime error: failed to add a new inbox channel. Router inbox is full.",
)
})?;
Ok(Inbox {
addr,
inner: receiver,
})
}
/// Spawn worker threads and start processing, returning a set of handles and
/// a Runtime object to interface with.
///
/// ### WARN:
/// ##### This function panics if the configuration is set as single threaded
/// `config.num_threads == 1`
pub fn run(mut self) -> Result<RuntimeHandle, Error> {
if self.config.num_threads < 2 {
return Err(Error::from(
"Runtime error: cannot call `Runtime::run()` from a single-threaded context.",
));
}
self.is_running.store(true, Ordering::Release);
// Take router out before wrapping in Arc - it will be owned by router thread
let mut router = self
.router
.take()
.expect("Router must be present for multi-threaded runtime");
let num_workers = self.config.num_threads - 1;
// Create workers and collect stealers for work-stealing
let mut workers = Vec::with_capacity(num_workers);
let mut stealers = Vec::with_capacity(num_workers);
for id in 0..num_workers {
let worker = WorkerState {
local: Worker::new_fifo(), // FIFO for fairness
id,
};
stealers.push(worker.local.stealer());
workers.push(worker);
}
let scheduler = Arc::new(SchedulerState {
injector: Injector::new(),
stealers,
is_running: AtomicBool::new(true),
});
// Transfer any actors spawned before run() to the injector
while let Some(actor) = self.actor_queue.pop() {
scheduler.injector.push(actor);
}
self.scheduler = Some(scheduler.clone());
let rt = Arc::new(self);
let mut handles: Vec<JoinHandle<()>> = vec![];
// Router thread owns the router directly - no synchronization needed
// Process multiple ticks per cycle to maximize throughput
const ROUTER_BATCH_SIZE: usize = 64;
let router_handle = {
let ctx = rt.clone();
let sched = scheduler.clone();
thread::spawn(move || {
while sched.is_running.load(Ordering::Acquire) {
for _ in 0..ROUTER_BATCH_SIZE {
router.tick(&ctx);
}
}
})
};
handles.push(router_handle);
// Spawn worker threads, each owns its WorkerState
for worker in workers {
let ctx = rt.clone();
let sched = scheduler.clone();
let handle = thread::spawn(move || worker_loop(ctx, sched, worker));
handles.push(handle);
}
Ok(RuntimeHandle {
runtime: rt,
threads: handles,
})
}
/// Pop the actor off the top of the queue and process it's messages, returning it to the back of
/// the queue upon completion.
pub fn tick(&self) {
if let Some(mut actor) = self.actor_queue.pop() {
actor.tick(&self);
let _ = self.actor_queue.push(actor);
}
}
/// Signal all workers to stop
pub fn shutdown(&self) {
self.is_running.store(false, Ordering::Release);
// Also stop the scheduler if multi-threaded
if let Some(ref scheduler) = self.scheduler {
scheduler.is_running.store(false, Ordering::Release);
}
}
}
/// Work-stealing worker loop for multi-threaded runtime
fn worker_loop(ctx: Arc<Runtime>, scheduler: Arc<SchedulerState>, worker: WorkerState) {
const TICKS_PER_ACTOR: usize = 4;
const INJECTOR_CHECK_INTERVAL: usize = 64; // Check injector every N iterations
let mut spin_count: usize = 0;
let mut iteration: usize = 0;
while scheduler.is_running.load(Ordering::Acquire) {
iteration = iteration.wrapping_add(1);
// Priority 1: Local queue
let mut actor = worker.local.pop();
// Priority 2: Periodically check injector for new actors
// This ensures newly spawned actors get picked up even when workers have local work
if actor.is_none() || (iteration % INJECTOR_CHECK_INTERVAL == 0) {
if let Steal::Success(a) = scheduler.injector.steal_batch_and_pop(&worker.local) {
if actor.is_some() {
// We already had an actor, push the stolen one to local
worker.local.push(a);
} else {
actor = Some(a);
}
}
}
// Priority 3: Steal from peer workers when idle
if actor.is_none() {
for (i, stealer) in scheduler.stealers.iter().enumerate() {
if i == worker.id {
continue;
}
if let Steal::Success(a) = stealer.steal_batch_and_pop(&worker.local) {
actor = Some(a);
break;
}
}
}
// Process or backoff
match actor {
Some(mut actor) => {
spin_count = 0;
for _ in 0..TICKS_PER_ACTOR {
actor.tick(&ctx);
}
worker.local.push(actor);
}
None => {
// Exponential backoff
spin_count = spin_count.saturating_add(1);
if spin_count < 10 {
std::hint::spin_loop();
} else if spin_count < 100 {
thread::yield_now();
} else {
thread::sleep(Duration::from_micros(10));
}
}
}
fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>) -> Result<(), Error> {
let worker_id = self.placement.next_worker();
self.address_map.insert(addr, worker_id);
self.spawn_txs[worker_id.as_usize()]
.try_send((addr, actor))
.map_err(|_| Error::from("Spawn queue full"))
}
}

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

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

@ -0,0 +1,138 @@
pub mod mailbox;
pub(crate) mod pool;
use std::any::Any;
use std::cell::RefCell;
use crate::actor::{ActorAddress, AnyActor};
use crate::address_map::{AddressMap, WorkerId};
use crate::channel::{Receiver, Sender};
use crate::context::ContextInner;
use crate::envelope::Envelope;
use crate::placement::Placement;
use crate::runtime::InboxRegistry;
use crate::Error;
use pool::ActorPool;
/// A worker owns a set of actors and runs them in a loop.
pub(crate) struct Worker {
id: WorkerId,
pool: ActorPool,
transfer_rx: Receiver<Envelope>,
spawn_rx: Receiver<(ActorAddress, Box<dyn AnyActor>)>,
}
impl Worker {
pub fn new(
id: WorkerId,
transfer_rx: Receiver<Envelope>,
spawn_rx: Receiver<(ActorAddress, Box<dyn AnyActor>)>,
) -> Self {
Self {
id,
pool: ActorPool::new(),
transfer_rx,
spawn_rx,
}
}
/// Run one iteration of the worker loop. Returns `true` if any work was done.
pub fn tick_once(
&mut self,
address_map: &AddressMap,
transfer_txs: &[Sender<Envelope>],
spawn_txs: &[Sender<(ActorAddress, Box<dyn AnyActor>)>],
placement: &Placement,
inbox_registry: &InboxRegistry,
) -> bool {
let mut did_work = false;
// 1. Drain spawn queue → add actors to pool
while let Some((addr, actor)) = self.spawn_rx.try_recv() {
self.pool.insert(addr, actor);
did_work = true;
}
// 2. Drain transfer queue → deliver envelopes to actors
while let Some(envelope) = self.transfer_rx.try_recv() {
let dest = envelope.dest();
let payload = envelope.into_payload();
self.pool.deliver(&dest, payload);
did_work = true;
}
// 3. Tick all actors with WorkerContext
let pending_local: RefCell<Vec<(ActorAddress, Box<dyn Any + Send>)>> =
RefCell::new(Vec::new());
{
let worker_ctx = WorkerContext {
worker_id: self.id,
address_map,
transfer_txs,
spawn_txs,
placement,
inbox_registry,
pending_local: &pending_local,
};
if self.pool.tick_all(&worker_ctx) {
did_work = true;
}
}
// 4. Drain pending_local buffer → deliver to local actors
let pending = pending_local.into_inner();
if !pending.is_empty() {
did_work = true;
}
for (addr, msg) in pending {
self.pool.deliver(&addr, msg);
}
did_work
}
}
/// The `ContextInner` impl for worker threads.
///
/// Same-worker sends are buffered in `pending_local` (delivered after current tick round).
/// Cross-worker sends go through the transfer queue.
struct WorkerContext<'a> {
worker_id: WorkerId,
address_map: &'a AddressMap,
transfer_txs: &'a [Sender<Envelope>],
spawn_txs: &'a [Sender<(ActorAddress, Box<dyn AnyActor>)>],
placement: &'a Placement,
inbox_registry: &'a InboxRegistry,
pending_local: &'a RefCell<Vec<(ActorAddress, Box<dyn Any + Send>)>>,
}
impl ContextInner for WorkerContext<'_> {
fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> Result<(), Error> {
match self.address_map.lookup(&addr) {
Some(wid) if wid == self.worker_id => {
// Same worker: buffer for local delivery (after current tick round)
self.pending_local.borrow_mut().push((addr, msg));
Ok(())
}
Some(wid) => {
// Cross worker: envelope through transfer queue
let envelope = Envelope::new(addr, msg);
let _ = self.transfer_txs[wid.as_usize()].try_send(envelope);
Ok(())
}
None => {
// Try inbox registry (external inboxes)
self.inbox_registry.try_deliver(addr, msg)
}
}
}
fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>) -> Result<(), Error> {
let worker_id = self.placement.next_worker();
self.address_map.insert(addr, worker_id);
self.spawn_txs[worker_id.as_usize()]
.try_send((addr, actor))
.map_err(|_| Error::from("Spawn queue full"))
}
}

51
src/worker/pool.rs Normal file
View file

@ -0,0 +1,51 @@
use std::any::Any;
use std::collections::HashMap;
use crate::actor::{ActorAddress, AnyActor};
use crate::context::ContextInner;
/// Per-worker actor storage.
pub(crate) struct ActorPool {
actors: HashMap<ActorAddress, Box<dyn AnyActor>>,
}
impl ActorPool {
pub fn new() -> Self {
Self {
actors: HashMap::new(),
}
}
pub fn insert(&mut self, addr: ActorAddress, actor: Box<dyn AnyActor>) {
self.actors.insert(addr, actor);
}
pub fn remove(&mut self, addr: &ActorAddress) -> Option<Box<dyn AnyActor>> {
self.actors.remove(addr)
}
/// Deliver a type-erased message to the actor at `addr`.
/// Returns `true` if the actor was found and the message type matched.
pub fn deliver(&mut self, addr: &ActorAddress, msg: Box<dyn Any + Send>) -> bool {
if let Some(actor) = self.actors.get_mut(addr) {
actor.deliver(msg)
} else {
false
}
}
/// Tick all actors in the pool. Returns `true` if any actor processed messages.
pub fn tick_all(&mut self, inner: &dyn ContextInner) -> bool {
let mut did_work = false;
for actor in self.actors.values_mut() {
if actor.tick(inner) {
did_work = true;
}
}
did_work
}
pub fn len(&self) -> usize {
self.actors.len()
}
}

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

10
tests/runtime_tests.rs
View file

@ -1,4 +1,4 @@
use swactor::{actor::{ActorAddress, ActorInterface}, runtime::{Inbox, Runtime, RuntimeConfig}};
use swactor::{Ctx, actor::{ActorAddress, ActorInterface}, runtime::{Inbox, Runtime, RuntimeConfig}};
#[derive(Clone)]
struct PingMessage {
@ -14,8 +14,8 @@ impl ActorInterface for PongActor {
type Incoming = PingMessage;
type Response = PongMessage;
fn handle(&mut self, ctx: &Runtime, msg: PingMessage) {
let _ = ctx.send_to(msg.reply_to, PongMessage);
fn handle(&mut self, ctx: &Ctx, msg: PingMessage) {
let _ = ctx.send(msg.reply_to, PongMessage);
}
}
@ -31,8 +31,8 @@ impl ActorInterface for ForwarderActor {
type Incoming = ForwardMessage;
type Response = ();
fn handle(&mut self, ctx: &Runtime, msg: ForwardMessage) {
let _ = ctx.send_to(self.target, msg);
fn handle(&mut self, ctx: &Ctx, msg: ForwardMessage) {
let _ = ctx.send(self.target, msg);
}
}

7
tests/stress/concurrency.rs
View file

@ -25,7 +25,6 @@ fn shutdown_under_load() {
let result = std::panic::catch_unwind(|| {
let config = RuntimeConfig {
max_actors: 100,
router_max_messages: 10_000,
actor_max_messages: 1000,
num_threads: 4,
};
@ -100,7 +99,6 @@ fn send_to_newborn() {
for _ in 0..100 {
let config = RuntimeConfig {
max_actors: 1000,
router_max_messages: 10_000,
actor_max_messages: 100,
num_threads: 4,
};
@ -149,7 +147,6 @@ fn rapid_spawn_churn() {
let config = RuntimeConfig {
max_actors: 100,
router_max_messages: 10_000,
actor_max_messages: 100,
num_threads: 4,
};
@ -210,8 +207,7 @@ fn inbox_contention() {
let config = RuntimeConfig {
max_actors: 10,
router_max_messages: 100_000,
actor_max_messages: 10_000,
actor_max_messages: 100_000,
num_threads: 4,
};
let runtime = Runtime::new(config);
@ -284,7 +280,6 @@ fn shutdown_timing_fuzz() {
let result = std::panic::catch_unwind(|| {
let config = RuntimeConfig {
max_actors: 50,
router_max_messages: 1000,
actor_max_messages: 100,
num_threads: 4,
};

6
tests/stress/mod.rs
View file

@ -165,7 +165,7 @@ impl Stress {
}
// Test actors used across stress tests
use swactor::{actor::ActorInterface, runtime::Runtime};
use swactor::{Ctx, actor::ActorInterface};
/// An actor that just absorbs messages
pub struct BlackHole;
@ -176,7 +176,7 @@ pub struct Msg;
impl ActorInterface for BlackHole {
type Incoming = Msg;
type Response = ();
fn handle(&mut self, _ctx: &Runtime, _msg: Msg) {}
fn handle(&mut self, _ctx: &Ctx, _msg: Msg) {}
}
/// An actor that counts messages received
@ -193,7 +193,7 @@ impl Counter {
impl ActorInterface for Counter {
type Incoming = Msg;
type Response = ();
fn handle(&mut self, _ctx: &Runtime, _msg: Msg) {
fn handle(&mut self, _ctx: &Ctx, _msg: Msg) {
self.count += 1;
}
}

51
tests/stress/saturation.rs
View file

@ -6,23 +6,25 @@ use super::{BlackHole, Counter, Msg, Stress, StressResult};
use std::time::Duration;
use swactor::runtime::{Runtime, RuntimeConfig};
/// Blast the router inbox
/// Blast the transfer queue (replaces router_inbox_overflow)
#[test]
#[cfg(feature = "stress")]
fn router_inbox_overflow() {
println!("\n>>> STRESS: Router Inbox Overflow");
fn transfer_queue_overflow() {
println!("\n>>> STRESS: Transfer Queue Overflow");
let config = RuntimeConfig {
max_actors: 10,
router_max_messages: 100, // Tiny buffer
actor_max_messages: 1000,
actor_max_messages: 100, // Tiny buffer
num_threads: 1,
};
let runtime = Runtime::new(config);
let sink = runtime.spawn(BlackHole).unwrap();
// Process spawn
runtime.tick();
// Blast messages without processing
let mut result = StressResult::new("router_inbox_overflow");
let mut result = StressResult::new("transfer_queue_overflow");
let start = std::time::Instant::now();
for _ in 0..10_000 {
@ -35,15 +37,15 @@ fn router_inbox_overflow() {
}
result.duration = start.elapsed();
result.note(format!("Router buffer: 100, Messages sent: 10,000"));
result.note(format!("Transfer buffer: 100, Messages sent: 10,000"));
// With hybrid, no failures expected
// With hybrid channel, no failures expected
assert_eq!(result.failures, 0, "Hybrid channel should not reject");
result.print();
println!(">>> PASS: Hybrid channel prevented router overflow\n");
println!(">>> PASS: Hybrid channel prevented transfer queue overflow\n");
}
/// Blast a single actor's inbox
/// Blast a single actor's mailbox via transfer queue
#[test]
#[cfg(feature = "stress")]
fn actor_inbox_overflow() {
@ -51,26 +53,25 @@ fn actor_inbox_overflow() {
let config = RuntimeConfig {
max_actors: 10,
router_max_messages: 100_000, // Large router buffer
actor_max_messages: 100, // Tiny actor inbox
actor_max_messages: 100_000, // Large transfer buffer
num_threads: 1,
};
let runtime = Runtime::new(config);
let sink = runtime.spawn(Counter::new()).unwrap();
// Process router registration
// Process spawn
runtime.tick();
// Now blast messages - router will accept them but actor inbox will fill
// Now blast messages
let mut sent = 0u64;
let mut router_failed = 0u64;
let mut failed = 0u64;
for _ in 0..10_000 {
if runtime.send_to::<Msg>(sink, Msg).is_ok() {
sent += 1;
} else {
router_failed += 1;
failed += 1;
}
// Tick occasionally to let router deliver
// Tick occasionally to let worker deliver
if sent % 100 == 0 {
runtime.tick();
}
@ -81,10 +82,10 @@ fn actor_inbox_overflow() {
runtime.tick();
}
println!(" Router accepted: {}", sent);
println!(" Router rejected: {}", router_failed);
println!(" Sent: {}", sent);
println!(" Failed: {}", failed);
assert_eq!(router_failed, 0, "Router rejected message under load");
assert_eq!(failed, 0, "Transfer queue rejected message under load");
println!(">>> PASS: No message loss with hybrid channel\n");
}
@ -96,7 +97,6 @@ fn actor_queue_overflow() {
let config = RuntimeConfig {
max_actors: 100, // Small actor queue
router_max_messages: 10_000,
actor_max_messages: 100,
num_threads: 1,
};
@ -118,12 +118,12 @@ fn actor_queue_overflow() {
result.note(format!("Queue capacity: 100, Spawn attempts: 500"));
result.print();
// Note: Router also takes a slot, so we expect ~99 actors max
// Note: With hybrid channel (overflow to SegQueue), we expect no failures
assert_eq!(
result.failures, 0,
"Spawned more actors than queue capacity"
);
println!(">>> PASS: Actor queue correctly rejects when full\n");
println!(">>> PASS: Actor queue correctly handles overflow\n");
}
/// FIXME: IS this actually testing what it should be?
@ -136,8 +136,7 @@ fn sustained_overload() {
let config = RuntimeConfig {
max_actors: 100,
router_max_messages: 1000,
actor_max_messages: 100,
actor_max_messages: 1000,
num_threads: 1,
};
let runtime = Runtime::new(config);
@ -150,7 +149,7 @@ fn sustained_overload() {
}
}
// Process registrations
// Process spawn registrations
for _ in 0..200 {
runtime.tick();
}