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

feat: multithreaded runtime

Browse source 
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.
This commit is contained in:
Zachery Aaron Shores-Chmielewski 2026-01-25 20:31:34 +07:00
parent c62b20c732
commit 98a2733173
9 changed files with 311 additions and 225 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
examples/hello.rs
View file

@ -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::<GreetResponse>();
// create an `Inbox` that allows us to receive messages from the runtime
let inbox = rt.new_inbox::<GreetResponse>().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);

13
examples/ring.rs
View file

@ -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<RingMessage> = rt.new_inbox();
let config = RuntimeConfig::default();
let rt = Runtime::new(config);
let inbox: Inbox<RingMessage> = 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:?}");
}

58
src/actor.rs
View file

@ -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<T: 'static + Sized + Clone + Send + Sync> 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<A>
/// The actor process as represented in the Runtime, with the actor state stored with it's inbox.
pub(crate) struct Actor<A>
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

14
src/error.rs
View file

@ -1,3 +1,17 @@
/// Simple, ergonomic, local `Error` type.
/// # Usage
/// ```
/// use swactor::Error;
///
/// fn foo_if_even(num: u64) -> Result<String, Error> {
/// if num % 2 == 0 {
/// return Ok("foo".into());
/// }
/// else {
/// return Err(Error::from("baz"));
/// }
/// }
/// ```
#[derive(Debug)]
pub struct Error(Box<dyn std::error::Error + Send + Sync + 'static>);
pub(crate) fn convert_err<E: std::fmt::Debug>(e: E) -> Error {

6
src/lib.rs
View file

@ -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;

5
src/ring_buffer.rs
View file

@ -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.
///

13
src/router.rs
View file

@ -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<dyn std::any::Any + Send + Sync>;
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<ActorAddress, Arc<dyn SenderT>>,
}
@ -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);

320
src/runtime.rs
View file

@ -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<M: Message> {
addr: ActorAddress,
inner: Receiver<M>,
@ -36,195 +27,226 @@ impl<M: Message> Inbox<M> {
}
}
/// Thread handles for the multithreaded runtime
struct RuntimeHandles {
workers: Vec<JoinHandle<()>>,
router_thread: Option<JoinHandle<()>>,
/// 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
/// 8kB for the `Box<..>` before counting the rest of the memory
const DEFAULT_MAX_ACTORS: usize = 1_000;
/// 160kB for the `Arc<..>` before counting the rest of the memory
const DEFAULT_ROUTER_MAX_MESSAGES: usize = 10_000;
/// 16kB PER ACTOR to alloc space for storing the `Arc<..>` pointers
/// With default setting of [DEFAULT_MAX_ACTORS] this is:
/// 1_000 * 16kB = 16MB
const DEFAULT_ACTOR_MAX_MESSAGES: usize = 1_000;
impl Default for RuntimeConfig {
fn default() -> Self {
Self {
max_actors: DEFAULT_MAX_ACTORS,
router_max_messages: DEFAULT_ROUTER_MAX_MESSAGES,
actor_max_messages: DEFAULT_ACTOR_MAX_MESSAGES,
num_threads: 1,
}
}
}
/// The `Runtime` struct is the primary gateway for interacting with the framework.
pub struct Runtime {
running: AtomicBool,
router_inbox: Sender<RouterMessage>,
config: RuntimeConfig,
actor_queue: ArrayQueue<Box<dyn AnyActor>>,
flavor: RuntimeFlavor,
/// Thread handles, created lazily when run() is called
handles: OnceLock<RuntimeHandles>,
router_interface: Sender<RouterMessage>,
router: Option<Actor<Router>>, // `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<Runtime>,
threads: Vec<JoinHandle<()>>,
}
impl RuntimeHandle {
pub fn join(self) {
for handle in self.threads {
let _ = handle.join();
}
}
/// Simple helper, calls the inner `Runtime::shutdown()` method
pub fn shutdown(&self) {
self.runtime.shutdown();
}
}
impl Runtime {
/// Create a new runtime with given actor queue capacity and flavor
pub fn new() -> Self {
let actor_queue = ArrayQueue::new(DEFAULT_INBOX_CAPACITY);
/// 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);
let router = Router::new();
let router_incoming =
Receiver::<<Router as ActorInterface>::Incoming>::new(DEFAULT_INBOX_CAPACITY);
let router_inbox = router_incoming.new_sender();
// router is a unique actor in that the runtime needs access to it's `Sender` handle
let router_inner = Router::new();
let router_inbox: Receiver<RouterMessage> =
Receiver::<<Router as ActorInterface>::Incoming>::new(config.router_max_messages);
let router_sender = router_inbox.new_sender();
let router = Actor::new(router_inbox, router_inner);
// Single-threaded: router goes in queue. Multi-threaded: stays in Option
let router_option = if config.num_threads < 2 {
actor_queue
.push(Box::new(Actor::new(router_incoming, router)) as Box<dyn AnyActor>)
.push(Box::new(router) as Box<dyn AnyActor>)
.map_err(|_| "failed to add router to actor queue")
.expect("failed to spawn router at runtime initialization.");
None
} else {
Some(router)
};
Self {
running: AtomicBool::new(false),
router_inbox,
config,
actor_queue,
flavor: RuntimeFlavor::SingleThreaded,
handles: OnceLock::new(),
router_interface: router_sender,
is_running: AtomicBool::new(false),
router: router_option,
}
}
/// FIXME: Abstract out by making two separate `Runtime` structs, one for singlethreaded, another for multi
pub fn new_multithreaded(num_workers: usize) -> (Self, Actor<Router>) {
let router = Router::new();
let router_incoming =
Receiver::<<Router as ActorInterface>::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<A: ActorInterface>(&self, actor: A) -> Result<ActorAddress, Error> {
// 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::<A::Incoming>::new(DEFAULT_INBOX_CAPACITY);
// assign a stochastic
let addr = ActorAddress::new_random();
let inbox = Receiver::<A::Incoming>::new(self.config.actor_max_messages);
let sender = inbox.new_sender();
// Register the sender with the router
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<M: Message>(&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<M: Message>(&self) -> Inbox<M> {
let addr = {
let mut bytes = u64::to_le_bytes(0);
get_random(&mut bytes);
u64::from_le_bytes(bytes)
};
let receiver = Receiver::<M>::new(DEFAULT_INBOX_CAPACITY);
/// Create an external inbox for receiving messages in the outer process containing the runtime
pub fn new_inbox<M: Message>(&self) -> Result<Inbox<M>, Error> {
let addr = ActorAddress::new_random();
let receiver = Receiver::<M>::new(self.config.actor_max_messages);
let sender = receiver.new_sender();
// Register the sender with the router
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<RuntimeHandle, Error> {
if self.config.num_threads < 2 {
return Err(Error::from(
"Runtime error: cannot call `Runtime::run()` from a single-threaded context.",
));
}
self.is_running.store(true, Ordering::Release);
// Take router out before wrapping in Arc - it will be owned by router thread
let mut router = self
.router
.take()
.expect("Router must be present for multi-threaded runtime");
let rt = Arc::new(self);
let mut handles: Vec<JoinHandle<()>> = 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<Router>) -> Arc<Self> {
// 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<JoinHandle<()>> = 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<Runtime>) {
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<Runtime>, mut router: Actor<Router>) {
while ctx.running.load(Ordering::Relaxed) {
router.tick(&ctx);
thread::yield_now();
self.is_running.store(false, Ordering::Release);
}
}

81
tests/runtime_tests.rs
View file

@ -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<PongMessage> = rt.new_inbox();
let rt = Runtime::new(RuntimeConfig::default());
let inbox: Inbox<PongMessage> = 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<ForwardMessage> = rt.new_inbox();
let rt = Runtime::new(RuntimeConfig::default());
let inbox: Inbox<ForwardMessage> = 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<ForwardMessage> = 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<ForwardMessage> = 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));
}