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

feat: actor addresses, Ctx abstraction, pools, and more

Browse source 
Fleshing out the architecture before abstracting into components amenable to api-based test harnesses for fuzz and optimization working loops.
This commit is contained in:
Zachery Aaron Shores-Chmielewski 2026-02-05 22:54:03 +07:00
parent 172e78a50b
commit 7c7947c823
22 changed files with 824 additions and 528 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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"

5
README.md
View file

@ -5,6 +5,7 @@
```rust
use swactor::{
Ctx,
actor::{ActorAddress, ActorInterface},
runtime::{Runtime, RuntimeConfig},
};
@ -22,10 +23,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) {
self.num_greeted -= 1;
}
}

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

67
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,
@ -33,9 +36,9 @@ impl<T: 'static + Sized + Clone + Send + Sync> Message for T {}
/// 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());
}
}

22
src/lib.rs
View file

@ -1,11 +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")]
@ -26,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"))
}
}

137
src/worker/mod.rs
View file

@ -1 +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()
}
}

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