From b422605f6bb6718653b83c6f293e49361379a876 Mon Sep 17 00:00:00 2001 From: zacheryasc Date: Sun, 25 Jan 2026 13:38:34 +0000 Subject: [PATCH] feat: Multithreaded runtime (#2) Implements a tunable configuration for a single or multi-threaded runtime. Co-authored-by: Zachery Aaron Shores-Chmielewski Reviewed-on: http://zachery.lol/code/code/zacheryasc/swactor/pulls/2 --- DESIGN.md | 196 ++++++++++++++++++++++++++++++- examples/hello.rs | 25 ++-- examples/ring.rs | 71 ++++++++++++ src/actor.rs | 109 ++++++++++++++++++ src/error.rs | 15 ++- src/lib.rs | 253 +++-------------------------------------- src/ring_buffer.rs | 5 +- src/router.rs | 72 ++++++++++++ src/runtime.rs | 252 ++++++++++++++++++++++++++++++++++++++++ tests/runtime_tests.rs | 130 +++++++++++++++++++++ 10 files changed, 875 insertions(+), 253 deletions(-) create mode 100644 examples/ring.rs create mode 100644 src/actor.rs create mode 100644 src/router.rs create mode 100644 src/runtime.rs create mode 100644 tests/runtime_tests.rs diff --git a/DESIGN.md b/DESIGN.md index 1a50043..7870a2e 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -74,4 +74,198 @@ impl HybridChannel { } } } -``` \ No newline at end of file +``` + +### Kimi Suggestions + +IMPROVEMENTS FITTING DESIGN GOALS +Here are improvements aligned with the stated goals: "maximum usability and speed while keeping line count low" and "no footguns." +Priority 1: Critical Bug Fixes & MVP Completion +1. Fix Runtime Constructor (~5 lines) + - Implement Runtime::new() + - Implement Builder::build() + - Fix examples to compile +2. Handle Full Inboxes Gracefully (~15 lines) + - Return Result<(), Error> from send_to on full inbox + - Provide backpressure signal instead of silent drop + - Add try_send() vs send() distinction +3. Implement Multithreaded Runtime (~30-40 lines) + - Complete threading infrastructure (already partially designed) + - Router runs in separate thread + - Actor processing pool with work-stealing (simple round-robin first) +Priority 2: Usability Enhancements (Low Line Count) +4. Ergonomic Macros (~20-30 lines procedural macro crate) + #[derive(Actor)] + struct MyActor { ... } + - Auto-impl ActorInterface for simple cases + - Reduces boilerplate significantly +5. Request/Response Helper (~15 lines) + let resp: Response = rt.request(addr, msg).await?; + - Common pattern many users need + - Maintains simplicity +6. Inbox Capacity Configuration (~5 lines) + - Per-actor capacity instead of global constant + - Allow spawn_with_capacity() +Priority 3: Performance Optimizations +7. Sharded Router (~30-40 lines) + - Multiple HashMaps based on address hash + - Reduces contention on messaging hot path + - Maintains O(1) lookups +8. Actor Work Stealing (~40-50 lines) + - Multiple actor queues instead of single global queue + - Threads steal work when idle + - Improves cache locality +9. Hybrid Channel (from DESIGN.md) (~25 lines) + - Implements the overflow mechanism described + - Ring buffer + Mutex for emergencies + - Prevent message loss under burst loads +10. Actor State Colocation (~15 lines) + - Group related actors by affinity + - Optional "actor system" or "node" concept + - Better cache locality +Priority 4: Observability (Minimal Overhead) +11. Lightweight Metrics (~15-20 lines) + - Message counts per actor (atomic counters) + - Overflow/drop tracking + - Optional, compile-time feature flag +12. Message Tracing (~10-15 lines opt-in) + - Optional trace ID in envelope + - Zero-cost when disabled (feature flag) +Priority 5: Reliability Patterns +13. Bounded Channels with Overflow (~20 lines) + - Implement HybridChannel from design doc + - Graceful degradation under load +14. Watchdog Timer Pattern (~15 lines example) + - Show pattern: actor checking heartbeats + - Keep library simple, document patterns +--- +SPECIFIC CODE IMPROVEMENTS +Fix Silent Failures (Priority: CRITICAL) +Current (src/runtime.rs:85-93): +pub fn send_to(&self, addr: ActorAddress, msg: M) -> Result<(), ()> { + let envelope: Envelope = Box::new(msg); + self.router_inbox + .try_send(RouterMessage::SendToAddr { addr, msg: envelope }) + .map_err(|_| ()) +} +Improved: +pub fn try_send(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { + let envelope: Envelope = Box::new(msg); + self.router_inbox + .try_send(RouterMessage::SendToAddr { addr, msg: envelope }) + .map_err(|_| Error::from("Router inbox full")) +} +// Add send that blocks/resizes +pub fn send(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { ... } +Implement HybridChannel (Priority: HIGH) +From design doc, add to ring_buffer.rs: +pub struct HybridChannel { + ring: ArrayQueue, + overflow: Mutex>, + overflow_count: AtomicUsize, +} +impl HybridChannel { + fn push(&self, value: T) -> Result<(), T> { + if self.ring.push(value).is_err() { + self.overflow.lock().push_back(value); + self.overflow_count.fetch_add(1, Relaxed); + // Optionally resize ring if overflow_count > threshold + } + Ok(()) + } +} +Fix Runtime Construction (Priority: CRITICAL) +Current: Runtime::new() doesn't exist but examples use it. +Add to runtime.rs: +impl Runtime { + pub fn new(capacity: usize, flavor: Option) -> Self { + let router = Router::new(capacity); + let router_inbox = router.new_sender(); + + Self { + flavor: flavor.unwrap_or_default(), + router, + router_inbox, + actor_queue: ArrayQueue::new(capacity), + thread_pool: Vec::new(), + } + } +} +Add Sharded Router (Priority: MEDIUM) +Current: Single HashMap for all addresses +Improved: N HashMaps based on address modulo +pub(crate) struct Router { + shards: Vec>>, + shard_mask: usize, // shards.len() - 1 (power of 2) + inbox: Receiver, +} +impl Router { + fn get_shard(&self, addr: ActorAddress) -> &HashMap<...> { + &self.shards[(addr as usize) & self.shard_mask] + } +} +--- +RECOMMENDED ROADMAP +Phase 1: Bug Fixes & MVP (1-2 days) +1. Implement Runtime::new() +2. Implement Builder::build() +3. Fix compilation errors +4. Add error handling for full inboxes +5. Document API +Phase 2: Single-Threaded Polish (1 week) +1. Ergonomic macros +2. Request/response helpers +3. Inbox capacity configuration +4. Example improvements +5. Basic tests +Phase 3: Multi-Threaded (2 weeks) +1. Implement threaded runtime +2. Worker thread pool +3. Router in separate thread +4. Work-stealing queues +5. Performance benchmarks +Phase 4: Production Hardening (2 weeks) +1. Sharded router +2. Hybrid channels +3. Metrics (opt-in) +4. Message tracing (opt-in) +5. Stress testing +Phase 5: Documentation & Patterns (1 week) +1. Actor patterns guide +2. Performance tuning guide +3. WASM integration examples +4. Best practices documentation +--- +ALTERNATIVE ARCHITECTURES TO CONSIDER +For Even Smaller Line Count +If the goal is absolutely minimal code, consider: +- Single-threaded only: Remove multi-threading complexity entirely +- No router: Direct mpsc channels between actors (more Erlang-like) +- Simpler scheduling: Round-robin over actors array +Tradeoff: Less flexible, but potentially <200 lines total. +For Better Performance +If performance outweighs minimalism: +- Lock-free HashMap: Use dashmap or equivalent for router +- SegQueue: Better for work-stealing than ArrayQueue +- Pre-allocated: Fixed-size actor pool with object pool pattern +- Lock-free message passing: Use crossbeam or tokio channels throughout +Tradeoff: More dependencies, larger binary size. +For Better Ergonomics +If usability is primary goal: +- Async/Await native: Integrate with tokio or async-std +- Actor supervision: Basic supervisor trees (small implementation) +- Message DSL: Macro for pattern-matching message handlers +Tradeoff: Increases complexity substantially, may conflict with "small" goal. +--- +CONCLUSION +swactor has a solid, minimalist architecture that delivers on its core promise: a small, WASM-compatible actor library. The design is clean, modular, and avoids unnecessary complexity. +Current Grade: C+ (Incomplete MVP) +- Architecture: B+ +- Ease of Use: D (examples don't compile, silent failures) +- Performance: B (good primitives but scalability concerns) +Potential Grade with improvements: A- +- Fixing critical bugs would make it immediately usable +- Sharded router + work-stealing would address scalability +- Ergonomic macros would dramatically improve UX +- Hybrid channels would solve burst-load scenarios +Recommendation: Focus on completing Phase 1 (bug fixes) and Phase 2 (usability). The architecture is sound—it's just incomplete. Avoid premature optimization; measure performance first, then implement sharding/work-stealing if benchmarks show contention. \ No newline at end of file diff --git a/examples/hello.rs b/examples/hello.rs index 3cf176a..d67c439 100644 --- a/examples/hello.rs +++ b/examples/hello.rs @@ -1,4 +1,7 @@ -use swactor::{ActorAddress, ActorInterface, Message, Runtime, RuntimeFlavor}; +use swactor::{ + actor::{ActorAddress, ActorInterface}, + runtime::{Runtime, RuntimeConfig}, +}; #[derive(Debug, Default)] struct Greeter { @@ -13,7 +16,9 @@ struct GreetMessage { /// who do we send out greeting back to? return_addr: ActorAddress, } -impl Message for GreetMessage {} + +#[derive(Debug, Default, Clone)] +struct GreetResponse(String); impl ActorInterface for Greeter { type Incoming = GreetMessage; @@ -29,17 +34,18 @@ impl ActorInterface for Greeter { } } -#[derive(Debug, Default, Clone)] -struct GreetResponse(String); -impl Message for GreetResponse {} - fn main() { - let mut rt = Runtime::new(100, Some(RuntimeFlavor::SingleThreaded)); + let rt = Runtime::new(RuntimeConfig::default()); + + // spawn a `Greeter` in the runtime, returning an address to contact it with let addr = rt .spawn(Greeter::default()) .expect("failed to spawn greeter"); - let inbox = rt.new_inbox::(); + // create an `Inbox` that allows us to receive messages from the runtime + let inbox = rt.new_inbox::().unwrap(); + + // send a message to the `Greeter` we spawned rt.send_to( addr, GreetMessage { @@ -48,10 +54,11 @@ fn main() { }, ) .unwrap(); + + // default runtime is single threaded, and requires the parent process to drive for _ in 0..3 { rt.tick(); } - let resp = inbox.try_recv().expect("greeter should have said hello"); println!("{}", resp.0); diff --git a/examples/ring.rs b/examples/ring.rs new file mode 100644 index 0000000..d3c6031 --- /dev/null +++ b/examples/ring.rs @@ -0,0 +1,71 @@ +use swactor::{ + actor::{ActorAddress, ActorInterface}, + runtime::{Inbox, Runtime, RuntimeConfig}, +}; + +#[derive(Debug, Default, Clone)] +pub struct RingMessage { + count: usize, +} + +impl RingMessage { + pub fn next(self) -> Self { + Self { + count: self.count + 1, + } + } +} + +#[derive(Debug, Default)] +struct RingActor { + next: ActorAddress, +} + +impl RingActor { + pub fn new(next: ActorAddress) -> Self { + Self { next } + } +} + +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()) { + // do nothing + } + } +} + +fn main() { + let config = RuntimeConfig::default(); + let rt = Runtime::new(config); + let inbox: Inbox = rt.new_inbox().unwrap(); + + let mut next = rt + .spawn(RingActor::new(*inbox.addr())) + .expect("failed to spawn"); + let num_passes = 500; + for _ in 0..num_passes { + let new = rt.spawn(RingActor::new(next)).expect("failed to spawn"); + next = new; + } + rt.send_to(next, RingMessage { count: 0 }) + .expect("failed to start message ring"); + + let msg: RingMessage; + loop { + match inbox.try_recv() { + Some(m) => { + msg = m; + break; + } + None => { + rt.tick(); + } + } + } + assert_eq!(msg.count, num_passes + 1); // count should equal the number of passes plus the return to main process inbox + + println!("{msg:?}"); +} diff --git a/src/actor.rs b/src/actor.rs new file mode 100644 index 0000000..2b80d40 --- /dev/null +++ b/src/actor.rs @@ -0,0 +1,109 @@ +use crate::{runtime::Runtime, WATERLEVEL, get_random, ring_buffer::Receiver}; + +/// The primary trait defining data that can be passed to and from actor processes +pub trait Message: 'static + Sized + Clone + Send + Sync {} +impl Message for T {} + +/// The trait that needs to be implemented in order to run a process as an `Actor` +/// +/// The `Incoming` type represents `Messages` that can be delivered to the `Actor`. +/// +/// The `Response` type represents possible `Messages` the actor may attempt to reply with. +/// +/// The `fn handle(..)` is where you implement the logic for handling `Incoming` messages +/// +/// # Example +/// ``` +/// use swactor::{actor::{ActorAddress, ActorInterface}, runtime::Runtime}; +/// +/// struct Greeter { +/// num_greeted: usize, +/// } +/// +/// #[derive(Clone)] // required to auto implement `Message` +/// struct GreetMessage { +/// who: String, +/// return_addr: ActorAddress, +/// } +/// +/// #[derive(Clone)] +/// struct GreetResponse(String); +/// +/// impl ActorInterface for Greeter { +/// type Incoming = GreetMessage; +/// type Response = GreetResponse; +/// +/// fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming) { +/// let response = GreetResponse(format!("Hello, {}!", msg.who).to_string()); +/// if let Ok(_) = ctx.send_to(msg.return_addr, response) { +/// self.num_greeted += 1; +/// } +/// } +/// } +/// ``` +pub trait ActorInterface: 'static + Send { + type Incoming: Message; + type Response: Message; + fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming); +} + +/// A unique address for this actor. 32 bytes is overkill for a small application, +/// but most systems are powerful, and this allows us to create a global map of +/// actor processes in the future, without worrying about collision. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash)] +pub struct ActorAddress(pub [u8; 32]); +impl ActorAddress { + pub fn new_random() -> Self { + let mut bytes = [0u8; 32]; + get_random(&mut bytes); + Self(bytes) + } +} + +/// The actor process as represented in the Runtime, with the actor state stored with it's inbox. +pub(crate) struct Actor +where + A: ActorInterface, +{ + inbox: Receiver, + inner: A, +} + +impl Actor { + pub(crate) fn new(inbox: Receiver, inner: A) -> Self { + Self { + inbox, + inner, + } + } +} + +/// Trait for type-erased actors +pub(crate) trait AnyActor: Send { + fn tick(&mut self, ctx: &Runtime); +} + +impl AnyActor for Actor +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. + 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" + ), + } + } + } +} diff --git a/src/error.rs b/src/error.rs index e00c61c..12183e2 100644 --- a/src/error.rs +++ b/src/error.rs @@ -1,6 +1,19 @@ +/// Simple, ergonomic, local `Error` type. +/// # Usage +/// ``` +/// use swactor::Error; +/// +/// fn foo_if_even(num: u64) -> Result { +/// if num % 2 == 0 { +/// return Ok("foo".into()); +/// } +/// else { +/// return Err(Error::from("baz")); +/// } +/// } +/// ``` #[derive(Debug)] pub struct Error(Box); -pub type Result = std::result::Result; pub(crate) fn convert_err(e: E) -> Error { Error(format!("{e:?}").into()) } diff --git a/src/lib.rs b/src/lib.rs index bd6913b..5ea2d6b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,251 +1,24 @@ +pub mod actor; + +pub(crate) mod error; +pub use error::Error; + mod ring_buffer; - -use std::collections::HashMap; - -use crossbeam_queue::ArrayQueue; -use ring_buffer::{Receiver, Sender}; - -pub mod error; -use error::Error; +mod router; +pub mod runtime; #[cfg(feature = "getrandom")] -pub fn get_random(buf: &mut [u8]) { +pub(crate) fn get_random(buf: &mut [u8]) { getrandom::getrandom(buf).unwrap() } +/// FIXME: remove hard coded defaults /// The strategy for message processing is such: +/// +/// ```ignore /// if total_messages < WATERLEVEL: /// process all /// else -/// process total_messages // 2 +/// process total_messages >> 1 +/// ``` const WATERLEVEL: usize = 10; - -const DEFAULT_INBOX_CAPACITY: usize = 100; - -pub trait Message: 'static + Sized + Clone + Send {} -pub type Envelope = Box; - -pub trait ActorInterface: 'static + Send { - type Incoming: Message; - type Response: Message; - fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming); -} - -pub type ActorAddress = u64; - -pub struct Actor -where - A: ActorInterface, -{ - _addr: ActorAddress, - inbox: Receiver, - inner: A, -} - -/// Trait for type-erased actors -trait AnyActor: Send { - fn tick(&mut self, ctx: &Runtime); -} - -impl AnyActor for Actor -where - A: ActorInterface, -{ - fn tick(&mut self, ctx: &Runtime) { - 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" - ), - } - } - } -} - -pub struct Inbox { - addr: ActorAddress, - inner: Receiver, -} - -impl Inbox { - pub fn addr(&self) -> &ActorAddress { - &self.addr - } - - pub fn try_recv(&self) -> Option { - self.inner.try_recv() - } -} - -#[derive(Debug, Default)] -pub enum RuntimeFlavor { - #[default] - SingleThreaded, - Multithreaded(usize), -} - -pub struct Runtime { - flavor: RuntimeFlavor, - router: Router, - router_inbox: Sender, - actor_queue: ArrayQueue>, -} - -impl Runtime { - pub fn new(capacity: usize, flavor: Option) -> Self { - let router = Router::new(DEFAULT_INBOX_CAPACITY); - let router_inbox = router.new_sender(); - Self { - flavor: flavor.unwrap_or_default(), - router, - router_inbox, - actor_queue: ArrayQueue::new(capacity), - } - } - - pub fn spawn(&self, actor: A) -> Result { - let addr = { - let mut bytes = u64::to_le_bytes(0); - get_random(&mut bytes); - u64::from_le_bytes(bytes) - }; - let inbox = Receiver::::new(DEFAULT_INBOX_CAPACITY); - let sender = inbox.new_sender(); - - // Register the sender with the router - let _ = self - .router_inbox - .try_send(RouterMessage::AddAddr(addr, Box::new(sender))); - - self.actor_queue - .push(Box::new(Actor { - _addr: addr, - inbox, - inner: actor, - })) - .map_err(|_| Error::from("Runtime error: Failed to spawn actor."))?; - - Ok(addr) - } - - pub fn send_to(&self, addr: ActorAddress, msg: M) -> Result<(), ()> { - let envelope: Envelope = Box::new(msg); - self.router_inbox - .try_send(RouterMessage::SendToAddr { - addr, - msg: envelope, - }) - .map_err(|_| ()) - } - - pub fn tick(&mut self) { - // Pop actor, tick it, push it back - if let Some(mut actor) = self.actor_queue.pop() { - actor.tick(self); - let _ = self.actor_queue.push(actor); - } - - match self.flavor { - RuntimeFlavor::Multithreaded(_) => (), // router has its own thread - RuntimeFlavor::SingleThreaded => self.router.tick(), - } - } - - pub fn new_inbox(&self) -> Inbox { - let addr = { - let mut bytes = u64::to_le_bytes(0); - get_random(&mut bytes); - u64::from_le_bytes(bytes) - }; - let receiver = Receiver::::new(DEFAULT_INBOX_CAPACITY); - let sender = receiver.new_sender(); - // Register the sender with the router - let _ = self - .router_inbox - .try_send(RouterMessage::AddAddr(addr, Box::new(sender))); - Inbox { - addr, - inner: receiver, - } - } -} - -pub trait SenderT: Send { - fn try_send(&self, envelope: Envelope); -} - -impl SenderT for Sender { - fn try_send(&self, envelope: Envelope) { - if let Ok(msg) = envelope.downcast::() { - let _ = Sender::try_send(self, *msg); - } - } -} - -/// Internal messages for the Router's own inbox -pub enum RouterMessage { - /// register addrs with sender - AddAddr(ActorAddress, Box), - /// remove an actor from the address book - RemoveAddr(ActorAddress), - /// send to - SendToAddr { addr: ActorAddress, msg: Envelope }, -} - -struct Router { - directory: HashMap>, - inbox: Receiver, -} - -impl Router { - pub fn new(cap: usize) -> Self { - Self { - directory: HashMap::new(), - inbox: Receiver::new(cap), - } - } - - pub fn tick(&mut self) { - 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.handle(msg), - None => unreachable!("We ran checks on total messages before processing."), - } - } - } - - pub fn new_sender(&self) -> Sender { - self.inbox.new_sender() - } - - fn handle(&mut self, msg: RouterMessage) { - 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); - } - } - } - } -} diff --git a/src/ring_buffer.rs b/src/ring_buffer.rs index fea5113..b016b3e 100644 --- a/src/ring_buffer.rs +++ b/src/ring_buffer.rs @@ -1,7 +1,8 @@ -pub use crossbeam_queue::ArrayQueue; +//! Shallow wrapper around the `crossbeam_queue::ArrayQueue` implementation of a mpmc ring buffer. use std::sync::Arc; +pub use crossbeam_queue::ArrayQueue; -/// The receiving end of a `crossbeam_queue::ArrayQueue`, a lock-free mpsc queue. +/// The receiving end of a `crossbeam_queue::ArrayQueue`, a lock-free mpmc queue. /// The queue is constructed by the `Receiver::new()` method. /// Responsible for creating the `Sender` ends of itself. /// diff --git a/src/router.rs b/src/router.rs new file mode 100644 index 0000000..82fb10a --- /dev/null +++ b/src/router.rs @@ -0,0 +1,72 @@ +use std::{collections::HashMap, sync::Arc}; + +use crate::{ + actor::{ActorAddress, ActorInterface, Message}, + ring_buffer::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; + +pub(crate) trait SenderT: Send + Sync { + fn try_send(&self, envelope: Envelope); +} + +impl SenderT for Sender { + fn try_send(&self, envelope: Envelope) { + if let Some(msg) = envelope.downcast_ref::() { + let _ = Sender::try_send(self, msg.clone()); + } + } +} + +/// Internal messages for the Router's own inbox +#[derive(Clone)] +pub(crate) enum RouterMessage { + /// register addrs with sender + AddAddr(ActorAddress, Arc), + + /// 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 to + 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>, +} + +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); + } + }, + } + } +} diff --git a/src/runtime.rs b/src/runtime.rs new file mode 100644 index 0000000..6a7a76a --- /dev/null +++ b/src/runtime.rs @@ -0,0 +1,252 @@ +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::thread::{self, JoinHandle}; + +use crossbeam_queue::ArrayQueue; + +use crate::{ + Error, + actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}, + ring_buffer::{Receiver, Sender}, + router::{Router, RouterMessage}, +}; + +/// Generic message inbox for receiving messages outside of the runtime. +pub struct Inbox { + addr: ActorAddress, + inner: Receiver, +} + +impl Inbox { + pub fn addr(&self) -> &ActorAddress { + &self.addr + } + + pub fn try_recv(&self) -> Option { + self.inner.try_recv() + } +} + +/// 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, + } + } +} + +/// The `Runtime` struct is the primary gateway for interacting with the framework. +pub struct Runtime { + config: RuntimeConfig, + actor_queue: ArrayQueue>, + router_interface: Sender, + router: Option>, // `None` if single-threaded + + // for multithreaded contexts + is_running: AtomicBool, +} + +/// Handle for dealing with a runtime that has started via the `Runtime::run()` method. +pub struct RuntimeHandle { + pub runtime: Arc, + threads: Vec>, +} + +impl RuntimeHandle { + pub fn join(self) { + for handle in self.threads { + let _ = handle.join(); + } + } + + /// Simple helper, calls the inner `Runtime::shutdown()` method + pub fn shutdown(&self) { + self.runtime.shutdown(); + } +} + +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 = ArrayQueue::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 = + Receiver::<::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) + .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, + router_interface: router_sender, + is_running: AtomicBool::new(false), + router: router_option, + } + } + + /// Spawn an actor, returns its address + pub fn spawn(&self, actor: A) -> Result { + // assign a stochastic + let addr = ActorAddress::new_random(); + let inbox = Receiver::::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") + })?; + + self.actor_queue + .push(Box::new(Actor::new(inbox, 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(&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(&self) -> Result, Error> { + let addr = ActorAddress::new_random(); + + let receiver = Receiver::::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 { + 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 rt = Arc::new(self); + let mut handles: Vec> = vec![]; + + // Router thread owns the router directly - no synchronization needed + let router_handle = { + let ctx = rt.clone(); + thread::spawn(move || { + while ctx.is_running.load(Ordering::Acquire) { + router.tick(&ctx); + thread::yield_now(); + } + }) + }; + handles.push(router_handle); + + // Spawn worker threads + let num_workers = rt.config.num_threads - 1; + for _ in 0..num_workers { + let ctx = rt.clone(); + let handle = thread::spawn(move || { + while ctx.is_running.load(Ordering::Acquire) { + if let Some(mut actor) = ctx.actor_queue.pop() { + actor.tick(&ctx); + if let Err(_) = ctx.actor_queue.push(actor) { + panic!( + "Runtime panic: attempted to return an actor to the queue, but queue was full." + ) + } + } else { + thread::yield_now(); + } + } + }); + 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); + } +} diff --git a/tests/runtime_tests.rs b/tests/runtime_tests.rs new file mode 100644 index 0000000..56ca321 --- /dev/null +++ b/tests/runtime_tests.rs @@ -0,0 +1,130 @@ +use swactor::{actor::{ActorAddress, ActorInterface}, runtime::{Inbox, Runtime, RuntimeConfig}}; + +#[derive(Clone)] +struct PingMessage { + reply_to: ActorAddress, +} + +#[derive(Clone)] +struct PongMessage; + +struct PongActor; + +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); + } +} + +/// An actor that forwards messages to another address +struct ForwarderActor { + target: ActorAddress, +} + +#[derive(Clone)] +struct ForwardMessage(usize); + +impl ActorInterface for ForwarderActor { + type Incoming = ForwardMessage; + type Response = (); + + fn handle(&mut self, ctx: &Runtime, msg: ForwardMessage) { + let _ = ctx.send_to(self.target, msg); + } +} + +#[test] +fn test_single_threaded_ping_pong() { + let rt = Runtime::new(RuntimeConfig::default()); + let inbox: Inbox = rt.new_inbox().unwrap(); + + let pong_addr = rt.spawn(PongActor).expect("spawn pong"); + + // Send ping + rt.send_to( + pong_addr, + PingMessage { + reply_to: *inbox.addr(), + }, + ) + .unwrap(); + + // Tick until we get a response + for _ in 0..10 { + rt.tick(); + if inbox.try_recv().is_some() { + return; // Success! + } + } + + panic!("Did not receive pong response"); +} + +#[test] +fn test_single_threaded_message_chain() { + let rt = Runtime::new(RuntimeConfig::default()); + let inbox: Inbox = rt.new_inbox().unwrap(); + + // Create a chain: A -> B -> C -> inbox + let c_addr = rt + .spawn(ForwarderActor { + target: *inbox.addr(), + }) + .unwrap(); + let b_addr = rt.spawn(ForwarderActor { target: c_addr }).unwrap(); + let a_addr = rt.spawn(ForwarderActor { target: b_addr }).unwrap(); + + // Send message to start of chain + rt.send_to(a_addr, ForwardMessage(42)).unwrap(); + + // Tick until message arrives + for _ in 0..20 { + rt.tick(); + if let Some(ForwardMessage(val)) = inbox.try_recv() { + assert_eq!(val, 42); + return; + } + } + + panic!("Message did not traverse the chain"); +} + +#[test] +fn test_multithreaded_message_passing() { + let config = RuntimeConfig { + num_threads: 4, + ..Default::default() + }; + let rt = Runtime::new(config); + let inbox: Inbox = rt.new_inbox().unwrap(); + + // Create a longer chain to exercise multi-threading + let mut target = *inbox.addr(); + for _ in 0..20 { + target = rt.spawn(ForwarderActor { target }).unwrap(); + } + let start_addr = target; + + // Send message + rt.send_to(start_addr, ForwardMessage(999)).unwrap(); + + // Spawn thread to check for result and shutdown + let ctx = rt.run().unwrap(); + let inbox_check = std::thread::spawn(move || { + for _ in 0..100 { + std::thread::sleep(std::time::Duration::from_millis(10)); + if let Some(ForwardMessage(val)) = inbox.try_recv() { + ctx.shutdown(); + return Some(val); + } + } + ctx.shutdown(); + None + }); + + let result = inbox_check.join().unwrap(); + assert_eq!(result, Some(999)); +}