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