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

refactor: major library changes #5

Merged
zacheryasc merged 9 commits from swactor-runtime-api into master 2026-02-06 11:25:38 +00:00
Conversation 0 Commits 9 Files changed 26 +185 -156
12 changed files with 185 additions and 156 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit 06566b5e0f - Show all commits

31
ARCHITECTURE.md
View file

@ -120,10 +120,11 @@ src/
├── 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
│ # - ActorInterface::handle(&mut self, ctx: &Ctx, msg)
│ # - actors depend ONLY on Ctx, nothing else
│
├── context.rs # Context trait — the "syscall interface" for actors
├── context.rs # Ctx wrapper — the "syscall interface" for actors
│ # - wraps &dyn ContextInner (solves object-safety)
│ # - send(), self_addr(), spawn()
│ # - this is ALL actors can see of the framework
│
@ -135,9 +136,10 @@ src/
│ # - shared read-heavy structure
│ # - written on spawn, read on every send
│
├── transfer.rs # Transfer queue — per-worker MPSC
├── channel/
│ └── mod.rs # HybridChannel — per-worker MPSC
│ # - the ONE concurrent data structure on the hot path
│ # - carries (ActorAddress, Envelope) pairs
│ # - carries Envelope (cross-worker) and spawn tuples
│
├── worker/
│ ├── mod.rs # Worker struct and worker loop
@ -150,22 +152,21 @@ src/
│ │ # - 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
│ # - local HashMap 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)
│ # - implements ContextInner (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.)
│ # - num_threads, max_actors, actor_max_messages
│ # - mailbox_waterlevel (drain threshold per actor)
│ # - BackoffPolicy (spin/yield/sleep thresholds)
│
└── placement.rs # Actor placement strategy
# - decides which worker a new actor goes to
# - round-robin, least-loaded, caller-affinity
# - currently: round-robin across workers
```
## Components
@ -175,12 +176,12 @@ src/
| **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) |
| **Transfer Queue** | `channel/mod.rs` | HybridChannel 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) |
| **Context** | `context.rs` | `Ctx` wrapper over `&dyn ContextInner`. `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) |
| **Placement** | `placement.rs` | Decides which worker gets a new actor. Currently round-robin. | No (called at spawn time) |
## Single-Threaded / WASM Mode

44
TODOs.md
View file

@ -1,44 +0,0 @@
### Profiling and Benchmarking
- Research as to modern art on benching and profiling
- Implement an MVP here.
- Bench/profile against a suite of tests selected for generality across actor framework usecases
- Identify hot paths and bottlenecks
- e.g. pretty sure the router is a major bottleneck, what else
- follow through the entire message cycle:
Parent process -> Convert to swactor::Message/Envelope -> Router -> Delivery -> Processing -> etc.
Identify every small detail on which you may be able to improve, any unneeded processing or branching
- (optional) Visualization tools:
- make some pretty stuff for tracing messages, actor activity, router activity, etc.
### Usage
- After benching and profiling, cleaning up the most egregious wrongdoings we will:
- actually implement our own projects in the framework, ones I actually find useful personally
- Optimization pipeline:
- Once we have well-established benches and profiles for general cases, build a set of tools that can auto-optimize
for given use cases. Tuning, for example, the channel buffers, router behavior, message consumption behavior, etc.
### Chores
- go over all the FIXMEs littered about. Add comments.
- add misc features as they come up. Prefer tools for understanding execution flows, visualizing flows, and adding
robustness, over ergonomics. Better to be slightly clunky but fast and optimized, than vice versa.
### Far future
- make language bindings. e.g., an npm package, python bindings, etc.
### Optimization
- Localize actors and inboxes:
- Because the entire runtime is message driven, the happy path must be fast. Even lock free, when we have to go through
several calls of an atomic ring buffer in order to process a single message, its unnecessary.
Parent process -> router -> actor -> router -> inbox -> parent process; every transfer going through an atomic buffer.
- to do this, design heavily around a localized worker thread. Actors on a working thread should have their inbox localized, they
should be 'sticky' to that thread (FILO queue?), and we should route messages based on core locality. Future optimizations can include
a tunable algorithm that puts actors that frequently communicate together on the same thread.

26
benches/scaling.rs
View file

@ -84,15 +84,18 @@ impl ActorInterface for PayloadActor {
/// Benchmark: How throughput scales with actor count
pub fn bench_actor_count_scaling(suite: &mut BenchSuite) {
let messages_per_actor = 100u64;
for actor_count in [10u64, 100, 500, 1000] {
for (actor_count, messages_per_actor, warmup_n, iters_n) in [
(10u64, 200u64, 5usize, 50usize),
(100, 200, 5, 30),
(500, 200, 5, 15),
(1000, 200, 5, 12),
] {
let name = format!("scaling_{}_actors", actor_count);
let total_messages = actor_count * messages_per_actor;
let result = Bench::new(&name)
.warmup(2)
.iters(10)
.warmup(warmup_n)
.iters(iters_n)
.elements(total_messages)
.run_with_setup(
|| {
@ -100,6 +103,7 @@ pub fn bench_actor_count_scaling(suite: &mut BenchSuite) {
max_actors: (actor_count as usize) + 100,
actor_max_messages: (total_messages as usize) * 3,
num_threads: 1,
..Default::default()
};
let runtime = Runtime::new(config);
@ -148,8 +152,8 @@ pub fn bench_thread_count_scaling(suite: &mut BenchSuite) {
let name = format!("scaling_{}_threads", thread_count);
let result = Bench::new(&name)
.warmup(1)
.iters(5)
.warmup(5)
.iters(30)
.elements(total_messages)
.run_with_setup(
|| {
@ -158,6 +162,7 @@ pub fn bench_thread_count_scaling(suite: &mut BenchSuite) {
max_actors: (actor_count as usize) + 100,
actor_max_messages: (total_messages as usize) * 3,
num_threads: thread_count,
..Default::default()
};
let runtime = Runtime::new(config);
@ -218,15 +223,15 @@ pub fn bench_thread_count_scaling(suite: &mut BenchSuite) {
/// Benchmark: How throughput scales with message payload size
pub fn bench_payload_size_scaling(suite: &mut BenchSuite) {
let message_count = 1_000u64;
let message_count = 3_000u64;
for payload_size in [64usize, 1024, 16384, 65536] {
let name = format!("payload_{}B", payload_size);
let payload = vec![0u8; payload_size];
let result = Bench::new(&name)
.warmup(2)
.iters(20)
.warmup(5)
.iters(40)
.elements(message_count)
.run_with_setup(
|| {
@ -234,6 +239,7 @@ pub fn bench_payload_size_scaling(suite: &mut BenchSuite) {
max_actors: 10,
actor_max_messages: (message_count as usize) * 2,
num_threads: 1,
..Default::default()
};
let runtime = Runtime::new(config);
let sink = runtime.spawn(PayloadActor::new()).unwrap();

46
benches/throughput.rs
View file

@ -76,8 +76,8 @@ pub fn bench_message_throughput(suite: &mut BenchSuite) {
let name = format!("message_throughput_{}", msg_count);
let result = Bench::new(&name)
.warmup(3)
.iters(20)
.warmup(5)
.iters(100)
.elements(msg_count)
.run_with_setup(
|| {
@ -86,6 +86,7 @@ pub fn bench_message_throughput(suite: &mut BenchSuite) {
max_actors: 100,
actor_max_messages: (msg_count as usize) * 2,
num_threads: 1,
..Default::default()
};
let runtime = Runtime::new(config);
let sink = runtime.spawn(SinkActor::new()).unwrap();
@ -114,8 +115,8 @@ pub fn bench_spawn_rate(suite: &mut BenchSuite) {
let name = format!("spawn_rate_{}_actors", actor_count);
let result = Bench::new(&name)
.warmup(3)
.iters(50)
.warmup(5)
.iters(100)
.elements(actor_count)
.run_with_setup(
|| {
@ -123,6 +124,7 @@ pub fn bench_spawn_rate(suite: &mut BenchSuite) {
max_actors: 1000,
actor_max_messages: 100,
num_threads: 1,
..Default::default()
};
Runtime::new(config)
},
@ -144,13 +146,16 @@ pub fn bench_spawn_rate(suite: &mut BenchSuite) {
/// Benchmark: Fan-out (1 sender to N receivers)
pub fn bench_fanout(suite: &mut BenchSuite) {
for fan_count in [10u64, 100, 500] {
for (fan_count, messages_per_receiver, warmup_n, iters_n) in [
(10u64, 500u64, 5usize, 50usize),
(100, 200, 5, 40),
(500, 100, 5, 30),
] {
let name = format!("fanout_1_to_{}", fan_count);
let messages_per_receiver = 100u64;
let result = Bench::new(&name)
.warmup(2)
.iters(20)
.warmup(warmup_n)
.iters(iters_n)
.elements(fan_count * messages_per_receiver)
.run_with_setup(
|| {
@ -160,6 +165,7 @@ pub fn bench_fanout(suite: &mut BenchSuite) {
* (messages_per_receiver as usize)
* 2,
num_threads: 1,
..Default::default()
};
let runtime = Runtime::new(config);
@ -200,13 +206,16 @@ pub fn bench_fanout(suite: &mut BenchSuite) {
/// Benchmark: Fan-in (N senders to 1 receiver)
pub fn bench_fanin(suite: &mut BenchSuite) {
for sender_count in [10u64, 100, 500] {
for (sender_count, messages_per_sender, warmup_n, iters_n) in [
(10u64, 300u64, 5usize, 50usize),
(100, 100, 5, 30),
(500, 100, 5, 20),
] {
let name = format!("fanin_{}_to_1", sender_count);
let messages_per_sender = 100u64;
let result = Bench::new(&name)
.warmup(2)
.iters(20)
.warmup(warmup_n)
.iters(iters_n)
.elements(sender_count * messages_per_sender)
.run_with_setup(
|| {
@ -215,6 +224,7 @@ pub fn bench_fanin(suite: &mut BenchSuite) {
max_actors: (sender_count as usize) + 10,
actor_max_messages: total_messages * 3,
num_threads: 1,
..Default::default()
};
let runtime = Runtime::new(config);
@ -258,13 +268,16 @@ pub fn bench_fanin(suite: &mut BenchSuite) {
/// Benchmark: Ring topology (message passed around N actors in a circle)
pub fn bench_ring(suite: &mut BenchSuite) {
for ring_size in [10u64, 100, 500] {
for (ring_size, laps, warmup_n, iters_n) in [
(10u64, 100u64, 5usize, 100usize),
(100, 20, 5, 50),
(500, 10, 5, 40),
] {
let name = format!("ring_{}_actors", ring_size);
let laps = 10u64; // How many times around the ring
let result = Bench::new(&name)
.warmup(2)
.iters(20)
.warmup(warmup_n)
.iters(iters_n)
.elements(ring_size * laps)
.run_with_setup(
|| {
@ -272,6 +285,7 @@ pub fn bench_ring(suite: &mut BenchSuite) {
max_actors: (ring_size as usize) + 10,
actor_max_messages: 10_000,
num_threads: 1,
..Default::default()
};
let runtime = Runtime::new(config);

31
src/config.rs
View file

@ -1,8 +1,35 @@
/// Backoff policy for worker threads when idle.
///
/// Workers spin → yield → sleep with increasing delay when no work is available.
pub struct BackoffPolicy {
/// Number of idle ticks before switching from spin to yield.
pub spin_threshold: u32,
/// Number of idle ticks before switching from yield to sleep.
pub yield_threshold: u32,
/// Microseconds added per tick beyond the yield threshold.
pub sleep_increment_us: u64,
/// Maximum sleep duration in microseconds.
pub sleep_max_us: u64,
}
impl Default for BackoffPolicy {
fn default() -> Self {
Self {
spin_threshold: 64,
yield_threshold: 256,
sleep_increment_us: 50,
sleep_max_us: 1000,
}
}
}
/// The tunable settings for the runtime.
pub struct RuntimeConfig {
pub max_actors: usize,
pub actor_max_messages: usize,
pub num_threads: usize,
pub mailbox_waterlevel: usize,
pub backoff_policy: BackoffPolicy,
}
/// 8kB for the `Box<..>` before counting the rest of the memory
@ -13,12 +40,16 @@ const DEFAULT_MAX_ACTORS: usize = 1_000;
/// 1_000 * 16kB = 16MB
const DEFAULT_ACTOR_MAX_MESSAGES: usize = 1_000;
const DEFAULT_MAILBOX_WATERLEVEL: usize = 10;
impl Default for RuntimeConfig {
fn default() -> Self {
Self {
max_actors: DEFAULT_MAX_ACTORS,
actor_max_messages: DEFAULT_ACTOR_MAX_MESSAGES,
num_threads: 1,
mailbox_waterlevel: DEFAULT_MAILBOX_WATERLEVEL,
backoff_policy: BackoffPolicy::default(),
}
}
}

4
src/context.rs
View file

@ -8,6 +8,7 @@ use crate::Error;
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>;
fn mailbox_waterlevel(&self) -> usize;
}
/// Actor syscall interface — passed to `ActorInterface::handle()`.
@ -37,7 +38,8 @@ impl<'a> Ctx<'a> {
/// 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 waterlevel = self.inner.mailbox_waterlevel();
let actor = Actor::new(addr, Mailbox::new(waterlevel), actor);
let boxed: Box<dyn AnyActor> = Box::new(actor);
self.inner.spawn_any(addr, boxed)?;
Ok(addr)

58
src/runtime.rs
View file

@ -8,13 +8,13 @@ use std::thread::{self, JoinHandle};
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;
// Re-export config types so existing code using `runtime::RuntimeConfig` still works
pub use crate::config::{BackoffPolicy, RuntimeConfig};
use crate::context::ContextInner;
use crate::envelope::Envelope;
use crate::placement::Placement;
use crate::worker::mailbox::Mailbox;
use crate::worker::Worker;
use crate::worker::{TickContext, Worker};
use crate::Error;
// ─── SenderT trait (moved from router.rs) ────────────────────────────────────
@ -171,7 +171,7 @@ impl Runtime {
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 actor = Actor::new(addr, Mailbox::new(self.config.mailbox_waterlevel), actor);
let boxed: Box<dyn AnyActor> = Box::new(actor);
self.spawn_txs[worker_id.as_usize()]
.try_send((addr, boxed))
@ -205,13 +205,15 @@ impl Runtime {
/// 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,
);
let tc = TickContext {
address_map: &self.address_map,
transfer_txs: &self.transfer_txs,
spawn_txs: &self.spawn_txs,
placement: &self.placement,
inbox_registry: &self.inbox_registry,
config: &self.config,
};
worker.borrow_mut().tick_once(&tc);
}
}
@ -261,28 +263,34 @@ impl Runtime {
/// 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 tc = TickContext {
address_map: &rt.address_map,
transfer_txs: &rt.transfer_txs,
spawn_txs: &rt.spawn_txs,
placement: &rt.placement,
inbox_registry: &rt.inbox_registry,
config: &rt.config,
};
let bp = &rt.config.backoff_policy;
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,
);
let did_work = worker.tick_once(&tc);
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 {
if idle_count < bp.spin_threshold {
// Hot spin — no hint, keep polling fast
} else if idle_count < bp.yield_threshold {
thread::yield_now();
} else {
// Park: sleep briefly, cap at 1ms
let micros = std::cmp::min((idle_count - 256) as u64 * 50, 1000);
// Park: sleep briefly, cap at configured max
let micros = std::cmp::min(
(idle_count - bp.yield_threshold) as u64 * bp.sleep_increment_us,
bp.sleep_max_us,
);
thread::sleep(std::time::Duration::from_micros(micros));
}
}
@ -326,4 +334,8 @@ impl ContextInner for Runtime {
.try_send((addr, actor))
.map_err(|_| Error::from("Spawn queue full"))
}
fn mailbox_waterlevel(&self) -> usize {
self.config.mailbox_waterlevel
}
}

11
src/worker/mailbox.rs
View file

@ -2,22 +2,13 @@ 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 {
pub fn new(waterlevel: usize) -> Self {
Self {
queue: VecDeque::new(),
waterlevel,

36
src/worker/mod.rs
View file

@ -7,6 +7,7 @@ use std::cell::RefCell;
use crate::actor::{ActorAddress, AnyActor};
use crate::address_map::{AddressMap, WorkerId};
use crate::channel::{Receiver, Sender};
use crate::config::RuntimeConfig;
use crate::context::ContextInner;
use crate::envelope::Envelope;
use crate::placement::Placement;
@ -14,6 +15,16 @@ use crate::runtime::InboxRegistry;
use crate::Error;
use pool::ActorPool;
/// Shared state passed to tick_once — single thin pointer avoids register spill.
pub(crate) struct TickContext<'a> {
pub address_map: &'a AddressMap,
pub transfer_txs: &'a [Sender<Envelope>],
pub spawn_txs: &'a [Sender<(ActorAddress, Box<dyn AnyActor>)>],
pub placement: &'a Placement,
pub inbox_registry: &'a InboxRegistry,
pub config: &'a RuntimeConfig,
}
/// A worker owns a set of actors and runs them in a loop.
pub(crate) struct Worker {
id: WorkerId,
@ -37,14 +48,7 @@ impl Worker {
}
/// 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 {
pub fn tick_once(&mut self, tc: &TickContext) -> bool {
let mut did_work = false;
// 1. Drain spawn queue → add actors to pool
@ -68,11 +72,12 @@ impl Worker {
{
let worker_ctx = WorkerContext {
worker_id: self.id,
address_map,
transfer_txs,
spawn_txs,
placement,
inbox_registry,
address_map: tc.address_map,
transfer_txs: tc.transfer_txs,
spawn_txs: tc.spawn_txs,
placement: tc.placement,
inbox_registry: tc.inbox_registry,
config: tc.config,
pending_local: &pending_local,
};
if self.pool.tick_all(&worker_ctx) {
@ -104,6 +109,7 @@ struct WorkerContext<'a> {
spawn_txs: &'a [Sender<(ActorAddress, Box<dyn AnyActor>)>],
placement: &'a Placement,
inbox_registry: &'a InboxRegistry,
config: &'a RuntimeConfig,
pending_local: &'a RefCell<Vec<(ActorAddress, Box<dyn Any + Send>)>>,
}
@ -135,4 +141,8 @@ impl ContextInner for WorkerContext<'_> {
.try_send((addr, actor))
.map_err(|_| Error::from("Spawn queue full"))
}
fn mailbox_waterlevel(&self) -> usize {
self.config.mailbox_waterlevel
}
}

42
tests/mailbox_tests.rs
View file

@ -4,14 +4,14 @@ use swactor::worker::mailbox::Mailbox;
#[test]
fn push_and_pop() {
let mut mb = Mailbox::new();
let mut mb = Mailbox::new(10);
mb.push(42i32);
assert_eq!(mb.pop(), Some(42));
}
#[test]
fn fifo_ordering() {
let mut mb = Mailbox::new();
let mut mb = Mailbox::new(10);
mb.push(1);
mb.push(2);
mb.push(3);
@ -22,13 +22,13 @@ fn fifo_ordering() {
#[test]
fn pop_empty() {
let mut mb: Mailbox<i32> = Mailbox::new();
let mut mb: Mailbox<i32> = Mailbox::new(10);
assert_eq!(mb.pop(), None);
}
#[test]
fn multiple_messages() {
let mut mb = Mailbox::new();
let mut mb = Mailbox::new(10);
for i in 0..100 {
mb.push(i);
}
@ -40,7 +40,7 @@ fn multiple_messages() {
#[test]
fn interleaved_push_pop() {
let mut mb = Mailbox::new();
let mut mb = Mailbox::new(10);
mb.push(1);
mb.push(2);
assert_eq!(mb.pop(), Some(1));
@ -54,13 +54,13 @@ fn interleaved_push_pop() {
#[test]
fn drain_count_empty() {
let mb: Mailbox<i32> = Mailbox::new();
let mb: Mailbox<i32> = Mailbox::new(10);
assert_eq!(mb.drain_count(), 0);
}
#[test]
fn drain_count_below_waterlevel() {
let mut mb = Mailbox::new();
let mut mb = Mailbox::new(10);
for i in 0..5 {
mb.push(i);
}
@ -70,7 +70,7 @@ fn drain_count_below_waterlevel() {
#[test]
fn drain_count_at_waterlevel() {
let mut mb = Mailbox::new();
let mut mb = Mailbox::new(10);
for i in 0..10 {
mb.push(i);
}
@ -80,7 +80,7 @@ fn drain_count_at_waterlevel() {
#[test]
fn drain_count_above_waterlevel() {
let mut mb = Mailbox::new();
let mut mb = Mailbox::new(10);
for i in 0..20 {
mb.push(i);
}
@ -90,7 +90,7 @@ fn drain_count_above_waterlevel() {
#[test]
fn drain_count_one_message() {
let mut mb = Mailbox::new();
let mut mb = Mailbox::new(10);
mb.push(1i32);
// 1 < 10 → process all → 1
assert_eq!(mb.drain_count(), 1);
@ -98,7 +98,7 @@ fn drain_count_one_message() {
#[test]
fn drain_count_just_below_waterlevel() {
let mut mb = Mailbox::new();
let mut mb = Mailbox::new(10);
for i in 0..9 {
mb.push(i);
}
@ -108,7 +108,7 @@ fn drain_count_just_below_waterlevel() {
#[test]
fn drain_count_large() {
let mut mb = Mailbox::new();
let mut mb = Mailbox::new(10);
for i in 0..1000 {
mb.push(i);
}
@ -118,7 +118,7 @@ fn drain_count_large() {
#[test]
fn drain_count_custom_waterlevel() {
let mut mb = Mailbox::with_waterlevel(4);
let mut mb = Mailbox::new(4);
for i in 0..3 {
mb.push(i);
}
@ -132,7 +132,7 @@ fn drain_count_custom_waterlevel() {
#[test]
fn drain_count_updates_after_pop() {
let mut mb = Mailbox::new();
let mut mb = Mailbox::new(10);
for i in 0..20 {
mb.push(i);
}
@ -151,7 +151,7 @@ fn drain_count_updates_after_pop() {
#[test]
fn len_tracks_pushes() {
let mut mb = Mailbox::new();
let mut mb = Mailbox::new(10);
assert_eq!(mb.len(), 0);
mb.push(1);
assert_eq!(mb.len(), 1);
@ -163,7 +163,7 @@ fn len_tracks_pushes() {
#[test]
fn len_tracks_pops() {
let mut mb = Mailbox::new();
let mut mb = Mailbox::new(10);
mb.push(1);
mb.push(2);
mb.push(3);
@ -178,13 +178,13 @@ fn len_tracks_pops() {
#[test]
fn is_empty_on_new() {
let mb: Mailbox<i32> = Mailbox::new();
let mb: Mailbox<i32> = Mailbox::new(10);
assert!(mb.is_empty());
}
#[test]
fn is_empty_after_drain() {
let mut mb = Mailbox::new();
let mut mb = Mailbox::new(10);
mb.push(1);
mb.push(2);
mb.push(3);
@ -199,11 +199,11 @@ fn is_empty_after_drain() {
#[test]
fn works_with_primitive_types() {
let mut mb_i32 = Mailbox::new();
let mut mb_i32 = Mailbox::new(10);
mb_i32.push(42i32);
assert_eq!(mb_i32.pop(), Some(42));
let mut mb_string = Mailbox::new();
let mut mb_string = Mailbox::new(10);
mb_string.push(String::from("hello"));
assert_eq!(mb_string.pop(), Some(String::from("hello")));
}
@ -216,7 +216,7 @@ fn works_with_custom_structs() {
payload: String,
}
let mut mb = Mailbox::new();
let mut mb = Mailbox::new(10);
let msg = MyMsg {
id: 1,
payload: "test".into(),

7
tests/stress/concurrency.rs
View file

@ -27,6 +27,7 @@ fn shutdown_under_load() {
max_actors: 100,
actor_max_messages: 1000,
num_threads: 4,
..Default::default()
};
let runtime = Runtime::new(config);
@ -101,6 +102,7 @@ fn send_to_newborn() {
max_actors: 1000,
actor_max_messages: 100,
num_threads: 4,
..Default::default()
};
let runtime = Runtime::new(config);
let handle = runtime.run().unwrap();
@ -137,7 +139,6 @@ fn send_to_newborn() {
println!(">>> Test complete\n");
}
/// FIXME: This test means nothing until we allow killing off actor processes
/// Rapid spawn/despawn cycles.
/// Target: Queue management under churn.
#[test]
@ -149,6 +150,7 @@ fn rapid_spawn_churn() {
max_actors: 100,
actor_max_messages: 100,
num_threads: 4,
..Default::default()
};
let runtime = Runtime::new(config);
let handle = runtime.run().unwrap();
@ -209,6 +211,7 @@ fn inbox_contention() {
max_actors: 10,
actor_max_messages: 100_000,
num_threads: 4,
..Default::default()
};
let runtime = Runtime::new(config);
let target = runtime.spawn(BlackHole).unwrap();
@ -262,7 +265,6 @@ fn inbox_contention() {
println!(">>> Test complete - no panics\n");
}
/// FIXME: Not sure this test is meaningful.
/// Shutdown timing fuzz - randomize when shutdown is called.
/// Target: Edge cases in shutdown state machine.
#[test]
@ -282,6 +284,7 @@ fn shutdown_timing_fuzz() {
max_actors: 50,
actor_max_messages: 100,
num_threads: 4,
..Default::default()
};
let runtime = Runtime::new(config);

5
tests/stress/saturation.rs
View file

@ -16,6 +16,7 @@ fn transfer_queue_overflow() {
max_actors: 10,
actor_max_messages: 100, // Tiny buffer
num_threads: 1,
..Default::default()
};
let runtime = Runtime::new(config);
let sink = runtime.spawn(BlackHole).unwrap();
@ -55,6 +56,7 @@ fn actor_inbox_overflow() {
max_actors: 10,
actor_max_messages: 100_000, // Large transfer buffer
num_threads: 1,
..Default::default()
};
let runtime = Runtime::new(config);
let sink = runtime.spawn(Counter::new()).unwrap();
@ -99,6 +101,7 @@ fn actor_queue_overflow() {
max_actors: 100, // Small actor queue
actor_max_messages: 100,
num_threads: 1,
..Default::default()
};
let runtime = Runtime::new(config);
@ -126,7 +129,6 @@ fn actor_queue_overflow() {
println!(">>> PASS: Actor queue correctly handles overflow\n");
}
/// FIXME: IS this actually testing what it should be?
/// Sustained overload - run at 2x capacity for extended period.
/// Documents: Does the system degrade gracefully or crash?
#[test]
@ -138,6 +140,7 @@ fn sustained_overload() {
max_actors: 100,
actor_max_messages: 1000,
num_threads: 1,
..Default::default()
};
let runtime = Runtime::new(config);