From a9f7abba25653b3fc46e4e1f7957c53522f3d28c Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Fri, 23 Jan 2026 13:48:26 +0700 Subject: [PATCH 1/6] feat: mvp actor ring test --- examples/hello.rs | 7 +++-- examples/ring.rs | 65 +++++++++++++++++++++++++++++++++++++++++++++++ src/lib.rs | 4 ++- 3 files changed, 71 insertions(+), 5 deletions(-) create mode 100644 examples/ring.rs diff --git a/examples/hello.rs b/examples/hello.rs index 3cf176a..d83fbc0 100644 --- a/examples/hello.rs +++ b/examples/hello.rs @@ -13,7 +13,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,9 +31,6 @@ 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)); diff --git a/examples/ring.rs b/examples/ring.rs new file mode 100644 index 0000000..4e854f5 --- /dev/null +++ b/examples/ring.rs @@ -0,0 +1,65 @@ +use swactor::{ActorAddress, ActorInterface, Inbox, Runtime, RuntimeFlavor}; + +#[derive(Debug, Default, Clone)] +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 mut rt = Runtime::new(10_000, Some(RuntimeFlavor::SingleThreaded)); + let inbox: Inbox = rt.new_inbox(); + + let mut next = rt + .spawn(RingActor::new(*inbox.addr())) + .expect("failed to spawn"); + for _ in 0..500 { + 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(); + } + } + } + + println!("{msg:?}") +} diff --git a/src/lib.rs b/src/lib.rs index bd6913b..7da4647 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -20,9 +20,11 @@ pub fn get_random(buf: &mut [u8]) { /// process total_messages // 2 const WATERLEVEL: usize = 10; -const DEFAULT_INBOX_CAPACITY: usize = 100; +const DEFAULT_INBOX_CAPACITY: usize = 1_000; pub trait Message: 'static + Sized + Clone + Send {} +impl Message for T {} + pub type Envelope = Box; pub trait ActorInterface: 'static + Send { -- 2.45.2 From 2450442566a57a8c44d45083a6777a773ced74e9 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Fri, 23 Jan 2026 14:09:31 +0700 Subject: [PATCH 2/6] feat: refactor: split out components into modules --- examples/hello.rs | 3 +- examples/ring.rs | 4 +- src/actor.rs | 60 +++++++++++ src/error.rs | 1 - src/lib.rs | 248 ++-------------------------------------------- src/router.rs | 77 ++++++++++++++ src/runtime.rs | 113 +++++++++++++++++++++ 7 files changed, 261 insertions(+), 245 deletions(-) create mode 100644 src/actor.rs create mode 100644 src/router.rs create mode 100644 src/runtime.rs diff --git a/examples/hello.rs b/examples/hello.rs index d83fbc0..7005d24 100644 --- a/examples/hello.rs +++ b/examples/hello.rs @@ -1,4 +1,4 @@ -use swactor::{ActorAddress, ActorInterface, Message, Runtime, RuntimeFlavor}; +use swactor::{actor::{ActorAddress, ActorInterface}, runtime::{Runtime, RuntimeFlavor}}; #[derive(Debug, Default)] struct Greeter { @@ -31,7 +31,6 @@ impl ActorInterface for Greeter { } } - fn main() { let mut rt = Runtime::new(100, Some(RuntimeFlavor::SingleThreaded)); let addr = rt diff --git a/examples/ring.rs b/examples/ring.rs index 4e854f5..bcfc07e 100644 --- a/examples/ring.rs +++ b/examples/ring.rs @@ -1,7 +1,7 @@ -use swactor::{ActorAddress, ActorInterface, Inbox, Runtime, RuntimeFlavor}; +use swactor::{actor::{ActorAddress, ActorInterface}, runtime::{Inbox, Runtime, RuntimeFlavor}}; #[derive(Debug, Default, Clone)] -struct RingMessage { +pub struct RingMessage { count: usize, } diff --git a/src/actor.rs b/src/actor.rs new file mode 100644 index 0000000..4f896c1 --- /dev/null +++ b/src/actor.rs @@ -0,0 +1,60 @@ +use crate::{runtime::Runtime, WATERLEVEL, ring_buffer::Receiver}; + + +pub trait Message: 'static + Sized + Clone + Send {} +impl Message for T {} + +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, +} + +impl Actor { + pub(crate) fn new(addr: ActorAddress, inbox: Receiver, inner: A) -> Self { + Self { + _addr: addr, + 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) { + 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" + ), + } + } + } +} \ No newline at end of file diff --git a/src/error.rs b/src/error.rs index e00c61c..91a260a 100644 --- a/src/error.rs +++ b/src/error.rs @@ -1,6 +1,5 @@ #[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 7da4647..85d51ae 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,15 +1,14 @@ +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() } @@ -19,235 +18,4 @@ pub fn get_random(buf: &mut [u8]) { /// else /// process total_messages // 2 const WATERLEVEL: usize = 10; - const DEFAULT_INBOX_CAPACITY: usize = 1_000; - -pub trait Message: 'static + Sized + Clone + Send {} -impl Message for T {} - -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/router.rs b/src/router.rs new file mode 100644 index 0000000..77b73d3 --- /dev/null +++ b/src/router.rs @@ -0,0 +1,77 @@ +use std::collections::HashMap; + +use crate::{WATERLEVEL, actor::{ActorAddress, Message}, ring_buffer::{Receiver, Sender}}; + +pub(crate) type Envelope = Box; + +pub(crate) 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(crate) 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 }, +} + +pub(crate) 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/runtime.rs b/src/runtime.rs new file mode 100644 index 0000000..28f8e4e --- /dev/null +++ b/src/runtime.rs @@ -0,0 +1,113 @@ +use crossbeam_queue::ArrayQueue; + +use crate::{ + DEFAULT_INBOX_CAPACITY, Error, + actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}, + get_random, + ring_buffer::{Receiver, Sender}, + router::{Envelope, Router, RouterMessage}, +}; + +#[derive(Debug, Default)] +pub enum RuntimeFlavor { + #[default] + SingleThreaded, + Multithreaded(usize), +} + +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() + } +} + +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::new(addr, inbox, 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, + } + } +} -- 2.45.2 From d504377ba984c5a6dc4c5d0e29e09eecd2c93544 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Sun, 25 Jan 2026 10:39:44 +0700 Subject: [PATCH 3/6] feat(WIP): multithreaded runtime Basic framework for a multithreaded runtime has been put in place. Needs plenty of fixes to keep the logic straightforward and performant. --- examples/hello.rs | 9 +- examples/ring.rs | 9 +- src/actor.rs | 16 +-- src/lib.rs | 9 +- src/ring_buffer.rs | 11 ++ src/router.rs | 11 +- src/runtime.rs | 269 ++++++++++++++++++++++++++++++++++++----- tests/runtime_tests.rs | 155 ++++++++++++++++++++++++ 8 files changed, 441 insertions(+), 48 deletions(-) create mode 100644 tests/runtime_tests.rs diff --git a/examples/hello.rs b/examples/hello.rs index 7005d24..db46a0e 100644 --- a/examples/hello.rs +++ b/examples/hello.rs @@ -1,4 +1,7 @@ -use swactor::{actor::{ActorAddress, ActorInterface}, runtime::{Runtime, RuntimeFlavor}}; +use swactor::{ + actor::{ActorAddress, ActorInterface}, + runtime::{Context, Runtime, RuntimeFlavor}, +}; #[derive(Debug, Default)] struct Greeter { @@ -21,7 +24,7 @@ impl ActorInterface for Greeter { type Incoming = GreetMessage; type Response = GreetResponse; - fn handle(&mut self, ctx: &Runtime, msg: GreetMessage) { + fn handle(&mut self, ctx: &Context, msg: GreetMessage) { let res = GreetResponse(format!("Hello, {}!", msg.who)); self.num_greeted += 1; if let Err(_) = ctx.send_to(msg.return_addr, res) { @@ -32,7 +35,7 @@ impl ActorInterface for Greeter { } fn main() { - let mut rt = Runtime::new(100, Some(RuntimeFlavor::SingleThreaded)); + let rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); let addr = rt .spawn(Greeter::default()) .expect("failed to spawn greeter"); diff --git a/examples/ring.rs b/examples/ring.rs index bcfc07e..313b779 100644 --- a/examples/ring.rs +++ b/examples/ring.rs @@ -1,4 +1,7 @@ -use swactor::{actor::{ActorAddress, ActorInterface}, runtime::{Inbox, Runtime, RuntimeFlavor}}; +use swactor::{ + actor::{ActorAddress, ActorInterface}, + runtime::{Context, Inbox, Runtime, RuntimeFlavor}, +}; #[derive(Debug, Default, Clone)] pub struct RingMessage { @@ -27,7 +30,7 @@ impl RingActor { impl ActorInterface for RingActor { type Incoming = RingMessage; type Response = (); - fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming) { + fn handle(&mut self, ctx: &Context, msg: Self::Incoming) { if let Err(_) = ctx.send_to(self.next, msg.next()) { // do nothing } @@ -35,7 +38,7 @@ impl ActorInterface for RingActor { } fn main() { - let mut rt = Runtime::new(10_000, Some(RuntimeFlavor::SingleThreaded)); + let rt = Runtime::new(10_000, RuntimeFlavor::SingleThreaded); let inbox: Inbox = rt.new_inbox(); let mut next = rt diff --git a/src/actor.rs b/src/actor.rs index 4f896c1..890243c 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -1,5 +1,4 @@ -use crate::{runtime::Runtime, WATERLEVEL, ring_buffer::Receiver}; - +use crate::{ring_buffer::Receiver, runtime::Context, WATERLEVEL}; pub trait Message: 'static + Sized + Clone + Send {} impl Message for T {} @@ -7,11 +6,14 @@ impl 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: &Context, msg: Self::Incoming); } pub type ActorAddress = u64; +/// FIXME: If we never have the actor struct reference its own address, should +/// we even include it as a variable here? We could instead grab this information +/// from the runtime or router. pub struct Actor where A: ActorInterface, @@ -26,21 +28,21 @@ impl Actor { Self { _addr: addr, inbox, - inner + inner, } } } /// Trait for type-erased actors pub(crate) trait AnyActor: Send { - fn tick(&mut self, ctx: &Runtime); + fn tick(&mut self, ctx: &Context); } impl AnyActor for Actor where A: ActorInterface, { - fn tick(&mut self, ctx: &Runtime) { + fn tick(&mut self, ctx: &Context) { let total_messages = self.inbox.len(); let messages_to_process = if total_messages < WATERLEVEL { total_messages @@ -57,4 +59,4 @@ where } } } -} \ No newline at end of file +} diff --git a/src/lib.rs b/src/lib.rs index 85d51ae..849606a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,15 +7,22 @@ mod ring_buffer; mod router; pub mod runtime; +// Re-export commonly used types +pub use actor::{ActorAddress, ActorInterface, Message}; +pub use runtime::{Context, Inbox, Runtime, RuntimeFlavor}; + #[cfg(feature = "getrandom")] pub(crate) fn get_random(buf: &mut [u8]) { getrandom::getrandom(buf).unwrap() } /// 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 = 1_000; diff --git a/src/ring_buffer.rs b/src/ring_buffer.rs index fea5113..5c78bf8 100644 --- a/src/ring_buffer.rs +++ b/src/ring_buffer.rs @@ -48,6 +48,17 @@ pub(crate) struct Sender { queue: Arc>, } +/// FIXME: I don't like this. Why do we need to clone the Sender +/// Because there are no guarentees on the existence of the Receiver +/// we need to be very careful about passing around access to the buffer. +impl Clone for Sender { + fn clone(&self) -> Self { + Self { + queue: Arc::clone(&self.queue), + } + } +} + impl Sender { /// Attempt to push a value to the queue. Returns Err(value) if the queue is full. pub fn try_send(&self, value: T) -> Result<(), T> { diff --git a/src/router.rs b/src/router.rs index 77b73d3..bde52f0 100644 --- a/src/router.rs +++ b/src/router.rs @@ -1,6 +1,10 @@ use std::collections::HashMap; -use crate::{WATERLEVEL, actor::{ActorAddress, Message}, ring_buffer::{Receiver, Sender}}; +use crate::{ + actor::{ActorAddress, Message}, + ring_buffer::{Receiver, Sender}, + WATERLEVEL, +}; pub(crate) type Envelope = Box; @@ -20,8 +24,13 @@ impl SenderT for Sender { pub(crate) enum RouterMessage { /// register addrs with sender AddAddr(ActorAddress, Box), + + /// 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 }, } diff --git a/src/runtime.rs b/src/runtime.rs index 28f8e4e..7acff9d 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,18 +1,24 @@ +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread::{self, JoinHandle}; + use crossbeam_queue::ArrayQueue; use crate::{ - DEFAULT_INBOX_CAPACITY, Error, actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}, get_random, ring_buffer::{Receiver, Sender}, router::{Envelope, Router, RouterMessage}, + Error, DEFAULT_INBOX_CAPACITY, }; -#[derive(Debug, Default)] +#[derive(Debug, Clone, Default)] pub enum RuntimeFlavor { #[default] SingleThreaded, - Multithreaded(usize), + Multithreaded { + workers: usize, + }, } pub struct Inbox { @@ -30,26 +36,88 @@ impl Inbox { } } -pub struct Runtime { - flavor: RuntimeFlavor, - router: Router, +/// A lightweight handle for sending messages to actors +/// This is what actors receive in their handle() method +#[derive(Clone)] +pub struct Context { + router_inbox: Sender, +} + +impl Context { + /// Send a message to an actor address + pub fn send_to(&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("Failed to send message: router inbox full")) + } +} + +/// Shared runtime state, wrapped in Arc for thread sharing +/// FIXME: We check the `running` variable too often. Should +/// push it somewhere where it is infreqently checked, and instead +/// have shutdown logic for everything else. +struct RuntimeInner { + running: AtomicBool, router_inbox: Sender, actor_queue: ArrayQueue>, } +/// Thread handles for the multithreaded runtime +struct RuntimeHandles { + workers: Vec>, + router_thread: Option>, +} + +/// The main runtime for executing actors +pub struct Runtime { + inner: Arc, + flavor: RuntimeFlavor, + /// Router is only accessed from a single thread (either main or dedicated router thread) + /// + /// FIXME: If single threaded, why do we have a mutex + router: Mutex, + /// Thread handles, created lazily when run() is called + handles: Mutex>, +} + impl Runtime { - pub fn new(capacity: usize, flavor: Option) -> Self { + /// Create a new runtime with given actor queue capacity and flavor + /// + /// FIXME: I don't like this interface. Maybe a builder or config pattern. I shouldnt + /// have to read code to understand what these variable names are. + pub fn new(capacity: usize, flavor: RuntimeFlavor) -> Self { + // FIXME: Avoid hard coded defaults, or at least put them all in one place 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), + inner: Arc::new(RuntimeInner { + running: AtomicBool::new(false), + router_inbox, + actor_queue: ArrayQueue::new(capacity), + }), + flavor, + router: Mutex::new(router), + handles: Mutex::new(None), } } + /// Get a context handle for sending messages + /// + /// FIXME: Do we need all this indirection? + pub fn context(&self) -> Context { + Context { + router_inbox: self.inner.router_inbox.clone(), + } + } + + /// Spawn an actor, returns its address pub fn spawn(&self, actor: A) -> Result { + // FIXME: Figure out what to do with this. + // Its distracting to include here, but not used elsewhere for now. let addr = { let mut bytes = u64::to_le_bytes(0); get_random(&mut bytes); @@ -60,39 +128,24 @@ impl Runtime { // Register the sender with the router let _ = self + .inner .router_inbox .try_send(RouterMessage::AddAddr(addr, Box::new(sender))); - self.actor_queue + self.inner + .actor_queue .push(Box::new(Actor::new(addr, inbox, 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(), - } + /// Send a message to an actor address + pub fn send_to(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { + self.context().send_to(addr, msg) } + /// Create an external inbox for receiving messages outside actors pub fn new_inbox(&self) -> Inbox { let addr = { let mut bytes = u64::to_le_bytes(0); @@ -103,6 +156,7 @@ impl Runtime { let sender = receiver.new_sender(); // Register the sender with the router let _ = self + .inner .router_inbox .try_send(RouterMessage::AddAddr(addr, Box::new(sender))); Inbox { @@ -110,4 +164,153 @@ impl Runtime { inner: receiver, } } + + /// Process one actor tick + router messages + /// Works in both single/multi mode (useful for testing and fine-grained control) + /// + /// FIXME: `tick` does not make sense in multithreaded context. Add a check to ensure single threaded + pub fn tick(&self) { + let ctx = self.context(); + + if let Some(mut actor) = self.inner.actor_queue.pop() { + actor.tick(&ctx); + let _ = self.inner.actor_queue.push(actor); + } + + // In single-threaded mode, also tick the router + if matches!(self.flavor, RuntimeFlavor::SingleThreaded) + && let Ok(mut router) = self.router.lock() + { + router.tick(); + } + } + + /// Run the runtime (blocking) + /// - SingleThreaded: runs in current thread until shutdown + /// - Multithreaded: spawns workers + router thread, blocks until shutdown + pub fn run(&self) { + self.inner.running.store(true, Ordering::Release); + + match &self.flavor { + RuntimeFlavor::SingleThreaded => { + self.run_single_threaded(); + } + RuntimeFlavor::Multithreaded { workers } => { + self.run_multi_threaded(*workers); + } + } + } + + /// FIXME: I don't like this loop. We can send shutdown signals from the process + /// that calls the actor runtime instead. It also does not make sense to have this + /// around for single threaded runtimes (people can instead loop over `runtime.tick()`). + fn run_single_threaded(&self) { + let ctx = self.context(); + + while self.inner.running.load(Ordering::Relaxed) { + // Pop actor, tick it, push it back + if let Some(mut actor) = self.inner.actor_queue.pop() { + actor.tick(&ctx); + let _ = self.inner.actor_queue.push(actor); + } + + // Tick the router + if let Ok(mut router) = self.router.lock() { + router.tick(); + } + + thread::yield_now(); + } + } + + fn run_multi_threaded(&self, num_workers: usize) { + // Take ownership of router for the dedicated router thread + // FIXME: this is weird and concerning. Introduces a class of logic errors + // wherein we try and call a dummy router. + let router = { + let mut guard = self.router.lock().unwrap(); + std::mem::replace(&mut *guard, Router::new(1)) // placeholder + }; + + // Spawn dedicated router thread + // FIXME: what is this, these variables are named terribly + let router_running = Arc::clone(&self.inner); + let router_handle = { + let running = router_running; + thread::spawn(move || { + router_loop(router, running); + }) + }; + + // Spawn worker threads + let worker_handles: Vec<_> = (0..num_workers) + .map(|_| { + let inner = Arc::clone(&self.inner); + thread::spawn(move || { + worker_loop(inner); + }) + }) + .collect(); + + // Store handles + *self.handles.lock().unwrap() = Some(RuntimeHandles { + workers: worker_handles, + router_thread: Some(router_handle), + }); + + // Block until shutdown - wait for all threads to complete + self.wait_for_shutdown(); + } + + fn wait_for_shutdown(&self) { + // Wait for the running flag to be set to false, then join threads + while self.inner.running.load(Ordering::Relaxed) { + thread::yield_now(); + } + + // Join all threads + let handles = self.handles.lock().unwrap().take(); + if let Some(h) = handles { + for worker in h.workers { + let _ = worker.join(); + } + if let Some(rt) = h.router_thread { + let _ = rt.join(); + } + } + } + + /// Signal all workers to stop + pub fn shutdown(&self) { + self.inner.running.store(false, Ordering::Release); + } + + /// Check if runtime is still active + pub fn is_running(&self) -> bool { + self.inner.running.load(Ordering::Acquire) + } +} + +/// Worker thread loop - processes actors from the shared queue +fn worker_loop(inner: Arc) { + let ctx = Context { + router_inbox: inner.router_inbox.clone(), + }; + + while inner.running.load(Ordering::Relaxed) { + if let Some(mut actor) = inner.actor_queue.pop() { + actor.tick(&ctx); + let _ = inner.actor_queue.push(actor); + } else { + thread::yield_now(); + } + } +} + +/// Router thread loop - processes router messages +fn router_loop(mut router: Router, inner: Arc) { + while inner.running.load(Ordering::Relaxed) { + router.tick(); + thread::yield_now(); + } } diff --git a/tests/runtime_tests.rs b/tests/runtime_tests.rs new file mode 100644 index 0000000..57e9023 --- /dev/null +++ b/tests/runtime_tests.rs @@ -0,0 +1,155 @@ +use std::sync::Arc; +use std::thread; +use std::time::Duration; + +use swactor::{ + actor::{ActorAddress, ActorInterface}, + runtime::{Context, Inbox, Runtime, RuntimeFlavor}, +}; + +// ============================================================================ +// Test Helpers +// ============================================================================ + +#[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: &Context, 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: &Context, msg: ForwardMessage) { + let _ = ctx.send_to(self.target, msg); + } +} + +#[test] +fn test_single_threaded_ping_pong() { + let rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); + let inbox: Inbox = rt.new_inbox(); + + 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(100, RuntimeFlavor::SingleThreaded); + let inbox: Inbox = rt.new_inbox(); + + // 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 rt = Arc::new(Runtime::new( + 1000, + RuntimeFlavor::Multithreaded { workers: 4 }, + )); + let inbox: Inbox = rt.new_inbox(); + + // 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 rt_clone = Arc::clone(&rt); + let inbox_check = thread::spawn(move || { + for _ in 0..100 { + thread::sleep(Duration::from_millis(10)); + if let Some(ForwardMessage(val)) = inbox.try_recv() { + rt_clone.shutdown(); + return Some(val); + } + } + rt_clone.shutdown(); + None + }); + + rt.run(); + + let result = inbox_check.join().unwrap(); + assert_eq!(result, Some(999)); +} + +#[test] +fn test_is_running_flag() { + let rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); + + // Before run(), is_running should be false + assert!(!rt.is_running()); + + // After shutdown before run, still false + rt.shutdown(); + assert!(!rt.is_running()); +} -- 2.45.2 From 2f91c7d1bdff21392b2c1ac5c34f7b1a2ef12690 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Sun, 25 Jan 2026 11:45:34 +0700 Subject: [PATCH 4/6] feat(WIP): runtime/router refactor Simplify the implementation of the multithreaded runtime and router. --- DESIGN.md | 196 +++++++++++++++++++++++++++++++++++- examples/hello.rs | 6 +- examples/ring.rs | 6 +- src/actor.rs | 15 +-- src/lib.rs | 4 +- src/ring_buffer.rs | 11 -- src/runtime.rs | 223 +++++++++++------------------------------ tests/runtime_tests.rs | 24 ++--- 8 files changed, 277 insertions(+), 208 deletions(-) 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 db46a0e..ef306e5 100644 --- a/examples/hello.rs +++ b/examples/hello.rs @@ -1,6 +1,6 @@ use swactor::{ actor::{ActorAddress, ActorInterface}, - runtime::{Context, Runtime, RuntimeFlavor}, + runtime::{Runtime, RuntimeFlavor}, }; #[derive(Debug, Default)] @@ -24,7 +24,7 @@ impl ActorInterface for Greeter { type Incoming = GreetMessage; type Response = GreetResponse; - fn handle(&mut self, ctx: &Context, msg: GreetMessage) { + fn handle(&mut self, ctx: &Runtime, msg: GreetMessage) { let res = GreetResponse(format!("Hello, {}!", msg.who)); self.num_greeted += 1; if let Err(_) = ctx.send_to(msg.return_addr, res) { @@ -35,7 +35,7 @@ impl ActorInterface for Greeter { } fn main() { - let rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); + let mut rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); let addr = rt .spawn(Greeter::default()) .expect("failed to spawn greeter"); diff --git a/examples/ring.rs b/examples/ring.rs index 313b779..cc3af6e 100644 --- a/examples/ring.rs +++ b/examples/ring.rs @@ -1,6 +1,6 @@ use swactor::{ actor::{ActorAddress, ActorInterface}, - runtime::{Context, Inbox, Runtime, RuntimeFlavor}, + runtime::{Inbox, Runtime, RuntimeFlavor}, }; #[derive(Debug, Default, Clone)] @@ -30,7 +30,7 @@ impl RingActor { impl ActorInterface for RingActor { type Incoming = RingMessage; type Response = (); - fn handle(&mut self, ctx: &Context, msg: Self::Incoming) { + fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming) { if let Err(_) = ctx.send_to(self.next, msg.next()) { // do nothing } @@ -38,7 +38,7 @@ impl ActorInterface for RingActor { } fn main() { - let rt = Runtime::new(10_000, RuntimeFlavor::SingleThreaded); + let mut rt = Runtime::new(10_000, RuntimeFlavor::SingleThreaded); let inbox: Inbox = rt.new_inbox(); let mut next = rt diff --git a/src/actor.rs b/src/actor.rs index 890243c..f303ed2 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -1,4 +1,4 @@ -use crate::{ring_buffer::Receiver, runtime::Context, WATERLEVEL}; +use crate::{Runtime, WATERLEVEL, ring_buffer::Receiver}; pub trait Message: 'static + Sized + Clone + Send {} impl Message for T {} @@ -6,27 +6,22 @@ impl Message for T {} pub trait ActorInterface: 'static + Send { type Incoming: Message; type Response: Message; - fn handle(&mut self, ctx: &Context, msg: Self::Incoming); + fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming); } pub type ActorAddress = u64; -/// FIXME: If we never have the actor struct reference its own address, should -/// we even include it as a variable here? We could instead grab this information -/// from the runtime or router. pub struct Actor where A: ActorInterface, { - _addr: ActorAddress, inbox: Receiver, inner: A, } impl Actor { - pub(crate) fn new(addr: ActorAddress, inbox: Receiver, inner: A) -> Self { + pub(crate) fn new(inbox: Receiver, inner: A) -> Self { Self { - _addr: addr, inbox, inner, } @@ -35,14 +30,14 @@ impl Actor { /// Trait for type-erased actors pub(crate) trait AnyActor: Send { - fn tick(&mut self, ctx: &Context); + fn tick(&mut self, ctx: &Runtime); } impl AnyActor for Actor where A: ActorInterface, { - fn tick(&mut self, ctx: &Context) { + fn tick(&mut self, ctx: &Runtime) { let total_messages = self.inbox.len(); let messages_to_process = if total_messages < WATERLEVEL { total_messages diff --git a/src/lib.rs b/src/lib.rs index 849606a..a3d4244 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -9,7 +9,7 @@ pub mod runtime; // Re-export commonly used types pub use actor::{ActorAddress, ActorInterface, Message}; -pub use runtime::{Context, Inbox, Runtime, RuntimeFlavor}; +pub use runtime::{Inbox, Runtime, RuntimeFlavor}; #[cfg(feature = "getrandom")] pub(crate) fn get_random(buf: &mut [u8]) { @@ -17,7 +17,7 @@ pub(crate) fn get_random(buf: &mut [u8]) { } /// The strategy for message processing is such: -/// +/// /// ```ignore /// if total_messages < WATERLEVEL: /// process all diff --git a/src/ring_buffer.rs b/src/ring_buffer.rs index 5c78bf8..fea5113 100644 --- a/src/ring_buffer.rs +++ b/src/ring_buffer.rs @@ -48,17 +48,6 @@ pub(crate) struct Sender { queue: Arc>, } -/// FIXME: I don't like this. Why do we need to clone the Sender -/// Because there are no guarentees on the existence of the Receiver -/// we need to be very careful about passing around access to the buffer. -impl Clone for Sender { - fn clone(&self) -> Self { - Self { - queue: Arc::clone(&self.queue), - } - } -} - impl Sender { /// Attempt to push a value to the queue. Returns Err(value) if the queue is full. pub fn try_send(&self, value: T) -> Result<(), T> { diff --git a/src/runtime.rs b/src/runtime.rs index 7acff9d..d00e7b8 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,15 +1,15 @@ use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, OnceLock}; use std::thread::{self, JoinHandle}; use crossbeam_queue::ArrayQueue; use crate::{ + DEFAULT_INBOX_CAPACITY, Error, actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}, get_random, ring_buffer::{Receiver, Sender}, - router::{Envelope, Router, RouterMessage}, - Error, DEFAULT_INBOX_CAPACITY, + router::{Router, RouterMessage}, }; #[derive(Debug, Clone, Default)] @@ -36,36 +36,6 @@ impl Inbox { } } -/// A lightweight handle for sending messages to actors -/// This is what actors receive in their handle() method -#[derive(Clone)] -pub struct Context { - router_inbox: Sender, -} - -impl Context { - /// Send a message to an actor address - pub fn send_to(&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("Failed to send message: router inbox full")) - } -} - -/// Shared runtime state, wrapped in Arc for thread sharing -/// FIXME: We check the `running` variable too often. Should -/// push it somewhere where it is infreqently checked, and instead -/// have shutdown logic for everything else. -struct RuntimeInner { - running: AtomicBool, - router_inbox: Sender, - actor_queue: ArrayQueue>, -} - /// Thread handles for the multithreaded runtime struct RuntimeHandles { workers: Vec>, @@ -74,19 +44,21 @@ struct RuntimeHandles { /// The main runtime for executing actors pub struct Runtime { - inner: Arc, + running: AtomicBool, + router_inbox: Sender, + actor_queue: ArrayQueue>, flavor: RuntimeFlavor, /// Router is only accessed from a single thread (either main or dedicated router thread) - /// - /// FIXME: If single threaded, why do we have a mutex + /// FIXME: The only state Router needs access to is a hashmap. If we find something + /// lockfree, we can remove this mutex router: Mutex, /// Thread handles, created lazily when run() is called - handles: Mutex>, + handles: OnceLock, } impl Runtime { /// Create a new runtime with given actor queue capacity and flavor - /// + /// /// FIXME: I don't like this interface. Maybe a builder or config pattern. I shouldnt /// have to read code to understand what these variable names are. pub fn new(capacity: usize, flavor: RuntimeFlavor) -> Self { @@ -94,30 +66,19 @@ impl Runtime { let router = Router::new(DEFAULT_INBOX_CAPACITY); let router_inbox = router.new_sender(); Self { - inner: Arc::new(RuntimeInner { - running: AtomicBool::new(false), - router_inbox, - actor_queue: ArrayQueue::new(capacity), - }), + running: AtomicBool::new(false), + router_inbox, + actor_queue: ArrayQueue::new(capacity), flavor, router: Mutex::new(router), - handles: Mutex::new(None), - } - } - - /// Get a context handle for sending messages - /// - /// FIXME: Do we need all this indirection? - pub fn context(&self) -> Context { - Context { - router_inbox: self.inner.router_inbox.clone(), + handles: OnceLock::new(), } } /// Spawn an actor, returns its address pub fn spawn(&self, actor: A) -> Result { // FIXME: Figure out what to do with this. - // Its distracting to include here, but not used elsewhere for now. + // Its distracting to include here, but not used elsewhere for now. let addr = { let mut bytes = u64::to_le_bytes(0); get_random(&mut bytes); @@ -128,13 +89,11 @@ impl Runtime { // Register the sender with the router let _ = self - .inner .router_inbox .try_send(RouterMessage::AddAddr(addr, Box::new(sender))); - self.inner - .actor_queue - .push(Box::new(Actor::new(addr, inbox, actor))) + self.actor_queue + .push(Box::new(Actor::new(inbox, actor))) .map_err(|_| Error::from("Runtime error: Failed to spawn actor."))?; Ok(addr) @@ -142,7 +101,12 @@ impl Runtime { /// Send a message to an actor address pub fn send_to(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { - self.context().send_to(addr, msg) + self.router_inbox + .try_send(RouterMessage::SendToAddr { + addr, + msg: Box::new(msg), + }) + .map_err(|_| "Failed to send message to router.".into()) } /// Create an external inbox for receiving messages outside actors @@ -156,7 +120,6 @@ impl Runtime { let sender = receiver.new_sender(); // Register the sender with the router let _ = self - .inner .router_inbox .try_send(RouterMessage::AddAddr(addr, Box::new(sender))); Inbox { @@ -167,140 +130,74 @@ impl Runtime { /// Process one actor tick + router messages /// Works in both single/multi mode (useful for testing and fine-grained control) - /// + /// /// FIXME: `tick` does not make sense in multithreaded context. Add a check to ensure single threaded - pub fn tick(&self) { - let ctx = self.context(); - - if let Some(mut actor) = self.inner.actor_queue.pop() { - actor.tick(&ctx); - let _ = self.inner.actor_queue.push(actor); + pub fn tick(&mut self) { + if let Some(mut actor) = self.actor_queue.pop() { + actor.tick(&self); + let _ = self.actor_queue.push(actor); } // In single-threaded mode, also tick the router if matches!(self.flavor, RuntimeFlavor::SingleThreaded) - && let Ok(mut router) = self.router.lock() { - router.tick(); + self.router.lock().expect("single threaded mutex lock").tick(); } } - /// Run the runtime (blocking) - /// - SingleThreaded: runs in current thread until shutdown - /// - Multithreaded: spawns workers + router thread, blocks until shutdown - pub fn run(&self) { - self.inner.running.store(true, Ordering::Release); - - match &self.flavor { - RuntimeFlavor::SingleThreaded => { - self.run_single_threaded(); - } - RuntimeFlavor::Multithreaded { workers } => { - self.run_multi_threaded(*workers); - } - } - } - - /// FIXME: I don't like this loop. We can send shutdown signals from the process - /// that calls the actor runtime instead. It also does not make sense to have this - /// around for single threaded runtimes (people can instead loop over `runtime.tick()`). - fn run_single_threaded(&self) { - let ctx = self.context(); - - while self.inner.running.load(Ordering::Relaxed) { - // Pop actor, tick it, push it back - if let Some(mut actor) = self.inner.actor_queue.pop() { - actor.tick(&ctx); - let _ = self.inner.actor_queue.push(actor); - } - - // Tick the router - if let Ok(mut router) = self.router.lock() { - router.tick(); - } - - thread::yield_now(); - } - } - - fn run_multi_threaded(&self, num_workers: usize) { - // Take ownership of router for the dedicated router thread - // FIXME: this is weird and concerning. Introduces a class of logic errors - // wherein we try and call a dummy router. - let router = { - let mut guard = self.router.lock().unwrap(); - std::mem::replace(&mut *guard, Router::new(1)) // placeholder + /// Run the runtime + /// - Multithreaded: spawns workers + router thread, returns an `Arc` pointer to the + /// `Runtime` struct. + pub fn run(self) -> Arc { + // FIXME: gate access + let num_workers = match self.flavor { + RuntimeFlavor::Multithreaded { workers } => workers, + _ => panic!("'run' method requires a multithreaded runtime") }; + let rt = Arc::new(self); // Spawn dedicated router thread - // FIXME: what is this, these variables are named terribly - let router_running = Arc::clone(&self.inner); let router_handle = { - let running = router_running; + let ctx = rt.clone(); thread::spawn(move || { - router_loop(router, running); + router_loop(ctx); }) }; // Spawn worker threads - let worker_handles: Vec<_> = (0..num_workers) - .map(|_| { - let inner = Arc::clone(&self.inner); - thread::spawn(move || { - worker_loop(inner); - }) - }) - .collect(); - - // Store handles - *self.handles.lock().unwrap() = Some(RuntimeHandles { - workers: worker_handles, - router_thread: Some(router_handle), - }); - - // Block until shutdown - wait for all threads to complete - self.wait_for_shutdown(); - } - - fn wait_for_shutdown(&self) { - // Wait for the running flag to be set to false, then join threads - while self.inner.running.load(Ordering::Relaxed) { - thread::yield_now(); + let mut worker_handles: Vec> = vec![]; + for _ in 0..num_workers { + let ctx = rt.clone(); + let handle = thread::spawn(move || { + worker_loop(ctx); + }); + worker_handles.push(handle); } - // Join all threads - let handles = self.handles.lock().unwrap().take(); - if let Some(h) = handles { - for worker in h.workers { - let _ = worker.join(); - } - if let Some(rt) = h.router_thread { - let _ = rt.join(); - } - } + rt.handles.set(RuntimeHandles { workers: worker_handles, router_thread: Some(router_handle) }); + + rt } + /// Signal all workers to stop pub fn shutdown(&self) { - self.inner.running.store(false, Ordering::Release); + self.running.store(false, Ordering::Release); } /// Check if runtime is still active pub fn is_running(&self) -> bool { - self.inner.running.load(Ordering::Acquire) + self.running.load(Ordering::Acquire) } } /// Worker thread loop - processes actors from the shared queue -fn worker_loop(inner: Arc) { - let ctx = Context { - router_inbox: inner.router_inbox.clone(), - }; +fn worker_loop(ctx: Arc) { - while inner.running.load(Ordering::Relaxed) { - if let Some(mut actor) = inner.actor_queue.pop() { + while ctx.running.load(Ordering::Relaxed) { + if let Some(mut actor) = ctx.actor_queue.pop() { actor.tick(&ctx); - let _ = inner.actor_queue.push(actor); + let _ = ctx.actor_queue.push(actor); } else { thread::yield_now(); } @@ -308,9 +205,9 @@ fn worker_loop(inner: Arc) { } /// Router thread loop - processes router messages -fn router_loop(mut router: Router, inner: Arc) { - while inner.running.load(Ordering::Relaxed) { - router.tick(); +fn router_loop(ctx: Arc) { + while ctx.running.load(Ordering::Relaxed) { + ctx.router.lock().expect("failed to lock router mutex").tick(); thread::yield_now(); } } diff --git a/tests/runtime_tests.rs b/tests/runtime_tests.rs index 57e9023..4047506 100644 --- a/tests/runtime_tests.rs +++ b/tests/runtime_tests.rs @@ -1,10 +1,9 @@ -use std::sync::Arc; use std::thread; use std::time::Duration; use swactor::{ actor::{ActorAddress, ActorInterface}, - runtime::{Context, Inbox, Runtime, RuntimeFlavor}, + runtime::{Inbox, Runtime, RuntimeFlavor}, }; // ============================================================================ @@ -25,7 +24,7 @@ impl ActorInterface for PongActor { type Incoming = PingMessage; type Response = PongMessage; - fn handle(&mut self, ctx: &Context, msg: PingMessage) { + fn handle(&mut self, ctx: &Runtime, msg: PingMessage) { let _ = ctx.send_to(msg.reply_to, PongMessage); } } @@ -42,14 +41,14 @@ impl ActorInterface for ForwarderActor { type Incoming = ForwardMessage; type Response = (); - fn handle(&mut self, ctx: &Context, msg: ForwardMessage) { + 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(100, RuntimeFlavor::SingleThreaded); + let mut rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); let inbox: Inbox = rt.new_inbox(); let pong_addr = rt.spawn(PongActor).expect("spawn pong"); @@ -76,7 +75,7 @@ fn test_single_threaded_ping_pong() { #[test] fn test_single_threaded_message_chain() { - let rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); + let mut rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); let inbox: Inbox = rt.new_inbox(); // Create a chain: A -> B -> C -> inbox @@ -105,10 +104,7 @@ fn test_single_threaded_message_chain() { #[test] fn test_multithreaded_message_passing() { - let rt = Arc::new(Runtime::new( - 1000, - RuntimeFlavor::Multithreaded { workers: 4 }, - )); + let rt = Runtime::new(1000, RuntimeFlavor::Multithreaded { workers: 4 }); let inbox: Inbox = rt.new_inbox(); // Create a longer chain to exercise multi-threading @@ -123,21 +119,19 @@ fn test_multithreaded_message_passing() { rt.send_to(start_addr, ForwardMessage(999)).unwrap(); // Spawn thread to check for result and shutdown - let rt_clone = Arc::clone(&rt); + let ctx = rt.run(); let inbox_check = thread::spawn(move || { for _ in 0..100 { thread::sleep(Duration::from_millis(10)); if let Some(ForwardMessage(val)) = inbox.try_recv() { - rt_clone.shutdown(); + ctx.shutdown(); return Some(val); } } - rt_clone.shutdown(); + ctx.shutdown(); None }); - rt.run(); - let result = inbox_check.join().unwrap(); assert_eq!(result, Some(999)); } -- 2.45.2 From c62b20c7323868535bcc2c1c738da0acbbb0cb5b Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Sun, 25 Jan 2026 12:54:01 +0700 Subject: [PATCH 5/6] feat(WIP): refactor router execution In single threaded contexts, the Router now is treated as yet another actor on the queue. In multithreaded contexts, it gets its own dedicated thread. --- examples/hello.rs | 2 +- examples/ring.rs | 2 +- src/actor.rs | 4 +- src/lib.rs | 2 + src/router.rs | 63 ++++++++++++-------------------- src/runtime.rs | 83 +++++++++++++++++++++++++----------------- tests/runtime_tests.rs | 72 +++++++++++++++--------------------- 7 files changed, 110 insertions(+), 118 deletions(-) diff --git a/examples/hello.rs b/examples/hello.rs index ef306e5..ec73da4 100644 --- a/examples/hello.rs +++ b/examples/hello.rs @@ -35,7 +35,7 @@ impl ActorInterface for Greeter { } fn main() { - let mut rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); + let mut rt = Runtime::new(); let addr = rt .spawn(Greeter::default()) .expect("failed to spawn greeter"); diff --git a/examples/ring.rs b/examples/ring.rs index cc3af6e..d79e35c 100644 --- a/examples/ring.rs +++ b/examples/ring.rs @@ -38,7 +38,7 @@ impl ActorInterface for RingActor { } fn main() { - let mut rt = Runtime::new(10_000, RuntimeFlavor::SingleThreaded); + let mut rt = Runtime::new(); let inbox: Inbox = rt.new_inbox(); let mut next = rt diff --git a/src/actor.rs b/src/actor.rs index f303ed2..82aa2ce 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -1,7 +1,7 @@ use crate::{Runtime, WATERLEVEL, ring_buffer::Receiver}; -pub trait Message: 'static + Sized + Clone + Send {} -impl Message for T {} +pub trait Message: 'static + Sized + Clone + Send + Sync {} +impl Message for T {} pub trait ActorInterface: 'static + Send { type Incoming: Message; diff --git a/src/lib.rs b/src/lib.rs index a3d4244..665aed0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -16,6 +16,7 @@ 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 @@ -25,4 +26,5 @@ pub(crate) fn get_random(buf: &mut [u8]) { /// process total_messages >> 1 /// ``` const WATERLEVEL: usize = 10; +/// FIXME: remove hard coded defaults const DEFAULT_INBOX_CAPACITY: usize = 1_000; diff --git a/src/router.rs b/src/router.rs index bde52f0..c6816b8 100644 --- a/src/router.rs +++ b/src/router.rs @@ -1,86 +1,71 @@ -use std::collections::HashMap; +use std::{collections::HashMap, sync::Arc}; use crate::{ + ActorInterface, actor::{ActorAddress, Message}, - ring_buffer::{Receiver, Sender}, - WATERLEVEL, + ring_buffer::Sender, }; -pub(crate) type Envelope = Box; +/// FIXME: Go over with a fine-toothed comb and reassure yourself this typing +/// makes sense. With these types, what happens at the hardware level. +pub(crate) type Envelope = Arc; -pub(crate) trait SenderT: Send { +pub(crate) trait SenderT: Send + Sync { 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); + 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, Box), - + AddAddr(ActorAddress, Arc), + /// FIXME: this will be active when we allow actors to shut themselves /// down. For now, disable the warning. - #[allow(dead_code)] + #[allow(dead_code)] /// remove an actor from the address book RemoveAddr(ActorAddress), - + /// send to SendToAddr { addr: ActorAddress, msg: Envelope }, } pub(crate) struct Router { - directory: HashMap>, - inbox: Receiver, + directory: HashMap>, } impl Router { - pub fn new(cap: usize) -> Self { + pub fn new() -> 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 - }; +impl ActorInterface for Router { + type Incoming = RouterMessage; - 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."), - } - } - } + type Response = (); - pub fn new_sender(&self) -> Sender { - self.inbox.new_sender() - } - - fn handle(&mut self, msg: RouterMessage) { + fn handle(&mut self, _ctx: &crate::Runtime, msg: Self::Incoming) { match msg { RouterMessage::AddAddr(addr, sender) => { self.directory.insert(addr, sender); - } - RouterMessage::RemoveAddr(addr) => { - self.directory.remove(&addr); - } + }, + 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 index d00e7b8..a245130 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,5 +1,5 @@ use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Mutex, OnceLock}; +use std::sync::{Arc, OnceLock}; use std::thread::{self, JoinHandle}; use crossbeam_queue::ArrayQueue; @@ -48,33 +48,55 @@ pub struct Runtime { router_inbox: Sender, actor_queue: ArrayQueue>, flavor: RuntimeFlavor, - /// Router is only accessed from a single thread (either main or dedicated router thread) - /// FIXME: The only state Router needs access to is a hashmap. If we find something - /// lockfree, we can remove this mutex - router: Mutex, /// Thread handles, created lazily when run() is called handles: OnceLock, } impl Runtime { /// Create a new runtime with given actor queue capacity and flavor - /// - /// FIXME: I don't like this interface. Maybe a builder or config pattern. I shouldnt - /// have to read code to understand what these variable names are. - pub fn new(capacity: usize, flavor: RuntimeFlavor) -> Self { - // FIXME: Avoid hard coded defaults, or at least put them all in one place - let router = Router::new(DEFAULT_INBOX_CAPACITY); - let router_inbox = router.new_sender(); + pub fn new() -> Self { + let actor_queue = ArrayQueue::new(DEFAULT_INBOX_CAPACITY); + + let router = Router::new(); + let router_incoming = + Receiver::<::Incoming>::new(DEFAULT_INBOX_CAPACITY); + let router_inbox = router_incoming.new_sender(); + + actor_queue + .push(Box::new(Actor::new(router_incoming, router)) as Box) + .map_err(|_| "failed to add router to actor queue") + .expect("failed to spawn router at runtime initialization."); + Self { running: AtomicBool::new(false), router_inbox, - actor_queue: ArrayQueue::new(capacity), - flavor, - router: Mutex::new(router), + actor_queue, + flavor: RuntimeFlavor::SingleThreaded, handles: OnceLock::new(), } } + /// FIXME: Abstract out by making two separate `Runtime` structs, one for singlethreaded, another for multi + pub fn new_multithreaded(num_workers: usize) -> (Self, Actor) { + let router = Router::new(); + let router_incoming = + Receiver::<::Incoming>::new(DEFAULT_INBOX_CAPACITY); + let router_inbox = router_incoming.new_sender(); + + ( + Self { + running: AtomicBool::new(false), + router_inbox, + actor_queue: ArrayQueue::new(DEFAULT_INBOX_CAPACITY), + flavor: RuntimeFlavor::Multithreaded { + workers: num_workers, + }, + handles: OnceLock::new(), + }, + Actor::new(router_incoming, router), + ) + } + /// Spawn an actor, returns its address pub fn spawn(&self, actor: A) -> Result { // FIXME: Figure out what to do with this. @@ -90,7 +112,7 @@ impl Runtime { // Register the sender with the router let _ = self .router_inbox - .try_send(RouterMessage::AddAddr(addr, Box::new(sender))); + .try_send(RouterMessage::AddAddr(addr, Arc::new(sender))); self.actor_queue .push(Box::new(Actor::new(inbox, actor))) @@ -104,7 +126,7 @@ impl Runtime { self.router_inbox .try_send(RouterMessage::SendToAddr { addr, - msg: Box::new(msg), + msg: Arc::new(msg), }) .map_err(|_| "Failed to send message to router.".into()) } @@ -121,7 +143,7 @@ impl Runtime { // Register the sender with the router let _ = self .router_inbox - .try_send(RouterMessage::AddAddr(addr, Box::new(sender))); + .try_send(RouterMessage::AddAddr(addr, Arc::new(sender))); Inbox { addr, inner: receiver, @@ -137,22 +159,16 @@ impl Runtime { actor.tick(&self); let _ = self.actor_queue.push(actor); } - - // In single-threaded mode, also tick the router - if matches!(self.flavor, RuntimeFlavor::SingleThreaded) - { - self.router.lock().expect("single threaded mutex lock").tick(); - } } - /// Run the runtime + /// Run the runtime /// - Multithreaded: spawns workers + router thread, returns an `Arc` pointer to the /// `Runtime` struct. - pub fn run(self) -> Arc { + pub fn run(self, router: Actor) -> Arc { // FIXME: gate access let num_workers = match self.flavor { RuntimeFlavor::Multithreaded { workers } => workers, - _ => panic!("'run' method requires a multithreaded runtime") + _ => panic!("'run' method requires a multithreaded runtime"), }; let rt = Arc::new(self); @@ -160,7 +176,7 @@ impl Runtime { let router_handle = { let ctx = rt.clone(); thread::spawn(move || { - router_loop(ctx); + router_loop(ctx, router); }) }; @@ -174,12 +190,14 @@ impl Runtime { worker_handles.push(handle); } - rt.handles.set(RuntimeHandles { workers: worker_handles, router_thread: Some(router_handle) }); + rt.handles.set(RuntimeHandles { + workers: worker_handles, + router_thread: Some(router_handle), + }); rt } - /// Signal all workers to stop pub fn shutdown(&self) { self.running.store(false, Ordering::Release); @@ -193,7 +211,6 @@ impl Runtime { /// Worker thread loop - processes actors from the shared queue fn worker_loop(ctx: Arc) { - while ctx.running.load(Ordering::Relaxed) { if let Some(mut actor) = ctx.actor_queue.pop() { actor.tick(&ctx); @@ -205,9 +222,9 @@ fn worker_loop(ctx: Arc) { } /// Router thread loop - processes router messages -fn router_loop(ctx: Arc) { +fn router_loop(ctx: Arc, mut router: Actor) { while ctx.running.load(Ordering::Relaxed) { - ctx.router.lock().expect("failed to lock router mutex").tick(); + router.tick(&ctx); thread::yield_now(); } } diff --git a/tests/runtime_tests.rs b/tests/runtime_tests.rs index 4047506..02fb75b 100644 --- a/tests/runtime_tests.rs +++ b/tests/runtime_tests.rs @@ -48,7 +48,7 @@ impl ActorInterface for ForwarderActor { #[test] fn test_single_threaded_ping_pong() { - let mut rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); + let mut rt = Runtime::new(); let inbox: Inbox = rt.new_inbox(); let pong_addr = rt.spawn(PongActor).expect("spawn pong"); @@ -75,7 +75,7 @@ fn test_single_threaded_ping_pong() { #[test] fn test_single_threaded_message_chain() { - let mut rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); + let mut rt = Runtime::new(); let inbox: Inbox = rt.new_inbox(); // Create a chain: A -> B -> C -> inbox @@ -102,48 +102,36 @@ fn test_single_threaded_message_chain() { panic!("Message did not traverse the chain"); } -#[test] -fn test_multithreaded_message_passing() { - let rt = Runtime::new(1000, RuntimeFlavor::Multithreaded { workers: 4 }); - let inbox: Inbox = rt.new_inbox(); +// #[test] +// fn test_multithreaded_message_passing() { +// let rt = Runtime::new(1000, RuntimeFlavor::Multithreaded { workers: 4 }); +// let inbox: Inbox = rt.new_inbox(); - // Create a longer chain to exercise multi-threading - let mut target = *inbox.addr(); - for _ in 0..20 { - target = rt.spawn(ForwarderActor { target }).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; +// let start_addr = target; - // Send message - rt.send_to(start_addr, ForwardMessage(999)).unwrap(); +// // Send message +// rt.send_to(start_addr, ForwardMessage(999)).unwrap(); - // Spawn thread to check for result and shutdown - let ctx = rt.run(); - let inbox_check = thread::spawn(move || { - for _ in 0..100 { - thread::sleep(Duration::from_millis(10)); - if let Some(ForwardMessage(val)) = inbox.try_recv() { - ctx.shutdown(); - return Some(val); - } - } - ctx.shutdown(); - None - }); +// // Spawn thread to check for result and shutdown +// let ctx = rt.run(); +// let inbox_check = thread::spawn(move || { +// for _ in 0..100 { +// thread::sleep(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)); -} - -#[test] -fn test_is_running_flag() { - let rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); - - // Before run(), is_running should be false - assert!(!rt.is_running()); - - // After shutdown before run, still false - rt.shutdown(); - assert!(!rt.is_running()); -} +// let result = inbox_check.join().unwrap(); +// assert_eq!(result, Some(999)); +// } -- 2.45.2 From 98a2733173d8176187cafdd2e21cd563ae3f844f Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Sun, 25 Jan 2026 20:31:34 +0700 Subject: [PATCH 6/6] feat: multithreaded runtime Runtime is now configurable. Adds a tunable config for modifying the size of pre-allocations for actor messaging channels, and for selecting the number of threads the runtime will use. --- examples/hello.rs | 14 +- examples/ring.rs | 13 +- src/actor.rs | 58 ++++++- src/error.rs | 14 ++ src/lib.rs | 6 - src/ring_buffer.rs | 5 +- src/router.rs | 13 +- src/runtime.rs | 332 ++++++++++++++++++++++------------------- tests/runtime_tests.rs | 81 +++++----- 9 files changed, 311 insertions(+), 225 deletions(-) diff --git a/examples/hello.rs b/examples/hello.rs index ec73da4..d67c439 100644 --- a/examples/hello.rs +++ b/examples/hello.rs @@ -1,6 +1,6 @@ use swactor::{ actor::{ActorAddress, ActorInterface}, - runtime::{Runtime, RuntimeFlavor}, + runtime::{Runtime, RuntimeConfig}, }; #[derive(Debug, Default)] @@ -35,12 +35,17 @@ impl ActorInterface for Greeter { } fn main() { - let mut rt = Runtime::new(); + 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 { @@ -49,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 index d79e35c..d3c6031 100644 --- a/examples/ring.rs +++ b/examples/ring.rs @@ -1,6 +1,6 @@ use swactor::{ actor::{ActorAddress, ActorInterface}, - runtime::{Inbox, Runtime, RuntimeFlavor}, + runtime::{Inbox, Runtime, RuntimeConfig}, }; #[derive(Debug, Default, Clone)] @@ -38,13 +38,15 @@ impl ActorInterface for RingActor { } fn main() { - let mut rt = Runtime::new(); - let inbox: Inbox = rt.new_inbox(); + 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"); - for _ in 0..500 { + let num_passes = 500; + for _ in 0..num_passes { let new = rt.spawn(RingActor::new(next)).expect("failed to spawn"); next = new; } @@ -63,6 +65,7 @@ fn main() { } } } + assert_eq!(msg.count, num_passes + 1); // count should equal the number of passes plus the return to main process inbox - println!("{msg:?}") + println!("{msg:?}"); } diff --git a/src/actor.rs b/src/actor.rs index 82aa2ce..2b80d40 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -1,17 +1,67 @@ -use crate::{Runtime, WATERLEVEL, ring_buffer::Receiver}; +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); } -pub type ActorAddress = u64; +/// 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) + } +} -pub struct Actor +/// The actor process as represented in the Runtime, with the actor state stored with it's inbox. +pub(crate) struct Actor where A: ActorInterface, { @@ -38,6 +88,8 @@ 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 diff --git a/src/error.rs b/src/error.rs index 91a260a..12183e2 100644 --- a/src/error.rs +++ b/src/error.rs @@ -1,3 +1,17 @@ +/// 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(crate) fn convert_err(e: E) -> Error { diff --git a/src/lib.rs b/src/lib.rs index 665aed0..5ea2d6b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,10 +7,6 @@ mod ring_buffer; mod router; pub mod runtime; -// Re-export commonly used types -pub use actor::{ActorAddress, ActorInterface, Message}; -pub use runtime::{Inbox, Runtime, RuntimeFlavor}; - #[cfg(feature = "getrandom")] pub(crate) fn get_random(buf: &mut [u8]) { getrandom::getrandom(buf).unwrap() @@ -26,5 +22,3 @@ pub(crate) fn get_random(buf: &mut [u8]) { /// process total_messages >> 1 /// ``` const WATERLEVEL: usize = 10; -/// FIXME: remove hard coded defaults -const DEFAULT_INBOX_CAPACITY: usize = 1_000; 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 index c6816b8..82fb10a 100644 --- a/src/router.rs +++ b/src/router.rs @@ -1,13 +1,14 @@ use std::{collections::HashMap, sync::Arc}; use crate::{ - ActorInterface, - actor::{ActorAddress, Message}, - ring_buffer::Sender, + actor::{ActorAddress, ActorInterface, Message}, + ring_buffer::Sender, runtime::Runtime, }; /// FIXME: Go over with a fine-toothed comb and reassure yourself this typing -/// makes sense. With these types, what happens at the hardware level. +/// 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 { @@ -38,6 +39,7 @@ pub(crate) enum RouterMessage { 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>, } @@ -52,10 +54,9 @@ impl Router { impl ActorInterface for Router { type Incoming = RouterMessage; - type Response = (); - fn handle(&mut self, _ctx: &crate::Runtime, msg: Self::Incoming) { + fn handle(&mut self, _ctx: &Runtime, msg: Self::Incoming) { match msg { RouterMessage::AddAddr(addr, sender) => { self.directory.insert(addr, sender); diff --git a/src/runtime.rs b/src/runtime.rs index a245130..6a7a76a 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,26 +1,17 @@ +use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, OnceLock}; use std::thread::{self, JoinHandle}; use crossbeam_queue::ArrayQueue; use crate::{ - DEFAULT_INBOX_CAPACITY, Error, + Error, actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}, - get_random, ring_buffer::{Receiver, Sender}, router::{Router, RouterMessage}, }; -#[derive(Debug, Clone, Default)] -pub enum RuntimeFlavor { - #[default] - SingleThreaded, - Multithreaded { - workers: usize, - }, -} - +/// Generic message inbox for receiving messages outside of the runtime. pub struct Inbox { addr: ActorAddress, inner: Receiver, @@ -36,195 +27,226 @@ impl Inbox { } } -/// Thread handles for the multithreaded runtime -struct RuntimeHandles { - workers: Vec>, - router_thread: Option>, +/// 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, } -/// The main runtime for executing actors -pub struct Runtime { - running: AtomicBool, - router_inbox: Sender, - actor_queue: ArrayQueue>, - flavor: RuntimeFlavor, - /// Thread handles, created lazily when run() is called - handles: OnceLock, -} +/// 8kB for the `Box<..>` before counting the rest of the memory +const DEFAULT_MAX_ACTORS: usize = 1_000; -impl Runtime { - /// Create a new runtime with given actor queue capacity and flavor - pub fn new() -> Self { - let actor_queue = ArrayQueue::new(DEFAULT_INBOX_CAPACITY); +/// 160kB for the `Arc<..>` before counting the rest of the memory +const DEFAULT_ROUTER_MAX_MESSAGES: usize = 10_000; - let router = Router::new(); - let router_incoming = - Receiver::<::Incoming>::new(DEFAULT_INBOX_CAPACITY); - let router_inbox = router_incoming.new_sender(); - - actor_queue - .push(Box::new(Actor::new(router_incoming, router)) as Box) - .map_err(|_| "failed to add router to actor queue") - .expect("failed to spawn router at runtime initialization."); +/// 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 { - running: AtomicBool::new(false), - router_inbox, - actor_queue, - flavor: RuntimeFlavor::SingleThreaded, - handles: OnceLock::new(), + 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(); } } - /// FIXME: Abstract out by making two separate `Runtime` structs, one for singlethreaded, another for multi - pub fn new_multithreaded(num_workers: usize) -> (Self, Actor) { - let router = Router::new(); - let router_incoming = - Receiver::<::Incoming>::new(DEFAULT_INBOX_CAPACITY); - let router_inbox = router_incoming.new_sender(); + /// Simple helper, calls the inner `Runtime::shutdown()` method + pub fn shutdown(&self) { + self.runtime.shutdown(); + } +} - ( - Self { - running: AtomicBool::new(false), - router_inbox, - actor_queue: ArrayQueue::new(DEFAULT_INBOX_CAPACITY), - flavor: RuntimeFlavor::Multithreaded { - workers: num_workers, - }, - handles: OnceLock::new(), - }, - Actor::new(router_incoming, router), - ) +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 { - // FIXME: Figure out what to do with this. - // Its distracting to include here, but not used elsewhere for now. - 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); + // 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 - let _ = self - .router_inbox - .try_send(RouterMessage::AddAddr(addr, Arc::new(sender))); + 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."))?; + .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_inbox + self.router_interface .try_send(RouterMessage::SendToAddr { addr, msg: Arc::new(msg), }) - .map_err(|_| "Failed to send message to router.".into()) + .map_err(|_| Error::from("Failed to send message to router.")) } - /// Create an external inbox for receiving messages outside actors - 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); + /// 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 - let _ = self - .router_inbox - .try_send(RouterMessage::AddAddr(addr, Arc::new(sender))); - Inbox { + 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, - } + }) } - /// Process one actor tick + router messages - /// Works in both single/multi mode (useful for testing and fine-grained control) + /// Spawn worker threads and start processing, returning a set of handles and + /// a Runtime object to interface with. /// - /// FIXME: `tick` does not make sense in multithreaded context. Add a check to ensure single threaded - pub fn tick(&mut self) { + /// ### 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); } } - /// Run the runtime - /// - Multithreaded: spawns workers + router thread, returns an `Arc` pointer to the - /// `Runtime` struct. - pub fn run(self, router: Actor) -> Arc { - // FIXME: gate access - let num_workers = match self.flavor { - RuntimeFlavor::Multithreaded { workers } => workers, - _ => panic!("'run' method requires a multithreaded runtime"), - }; - let rt = Arc::new(self); - - // Spawn dedicated router thread - let router_handle = { - let ctx = rt.clone(); - thread::spawn(move || { - router_loop(ctx, router); - }) - }; - - // Spawn worker threads - let mut worker_handles: Vec> = vec![]; - for _ in 0..num_workers { - let ctx = rt.clone(); - let handle = thread::spawn(move || { - worker_loop(ctx); - }); - worker_handles.push(handle); - } - - rt.handles.set(RuntimeHandles { - workers: worker_handles, - router_thread: Some(router_handle), - }); - - rt - } - /// Signal all workers to stop pub fn shutdown(&self) { - self.running.store(false, Ordering::Release); - } - - /// Check if runtime is still active - pub fn is_running(&self) -> bool { - self.running.load(Ordering::Acquire) - } -} - -/// Worker thread loop - processes actors from the shared queue -fn worker_loop(ctx: Arc) { - while ctx.running.load(Ordering::Relaxed) { - if let Some(mut actor) = ctx.actor_queue.pop() { - actor.tick(&ctx); - let _ = ctx.actor_queue.push(actor); - } else { - thread::yield_now(); - } - } -} - -/// Router thread loop - processes router messages -fn router_loop(ctx: Arc, mut router: Actor) { - while ctx.running.load(Ordering::Relaxed) { - router.tick(&ctx); - thread::yield_now(); + self.is_running.store(false, Ordering::Release); } } diff --git a/tests/runtime_tests.rs b/tests/runtime_tests.rs index 02fb75b..56ca321 100644 --- a/tests/runtime_tests.rs +++ b/tests/runtime_tests.rs @@ -1,14 +1,4 @@ -use std::thread; -use std::time::Duration; - -use swactor::{ - actor::{ActorAddress, ActorInterface}, - runtime::{Inbox, Runtime, RuntimeFlavor}, -}; - -// ============================================================================ -// Test Helpers -// ============================================================================ +use swactor::{actor::{ActorAddress, ActorInterface}, runtime::{Inbox, Runtime, RuntimeConfig}}; #[derive(Clone)] struct PingMessage { @@ -48,8 +38,8 @@ impl ActorInterface for ForwarderActor { #[test] fn test_single_threaded_ping_pong() { - let mut rt = Runtime::new(); - let inbox: Inbox = rt.new_inbox(); + let rt = Runtime::new(RuntimeConfig::default()); + let inbox: Inbox = rt.new_inbox().unwrap(); let pong_addr = rt.spawn(PongActor).expect("spawn pong"); @@ -75,8 +65,8 @@ fn test_single_threaded_ping_pong() { #[test] fn test_single_threaded_message_chain() { - let mut rt = Runtime::new(); - let inbox: Inbox = rt.new_inbox(); + let rt = Runtime::new(RuntimeConfig::default()); + let inbox: Inbox = rt.new_inbox().unwrap(); // Create a chain: A -> B -> C -> inbox let c_addr = rt @@ -102,36 +92,39 @@ fn test_single_threaded_message_chain() { panic!("Message did not traverse the chain"); } -// #[test] -// fn test_multithreaded_message_passing() { -// let rt = Runtime::new(1000, RuntimeFlavor::Multithreaded { workers: 4 }); -// let inbox: Inbox = rt.new_inbox(); +#[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(); -// } + // 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; -// let start_addr = target; + // Send message + rt.send_to(start_addr, ForwardMessage(999)).unwrap(); -// // 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 + }); -// // Spawn thread to check for result and shutdown -// let ctx = rt.run(); -// let inbox_check = thread::spawn(move || { -// for _ in 0..100 { -// thread::sleep(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)); -// } + let result = inbox_check.join().unwrap(); + assert_eq!(result, Some(999)); +} -- 2.45.2