diff --git a/Cargo.lock b/Cargo.lock index 561715c..a25b4f1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8,25 +8,6 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" -[[package]] -name = "crossbeam-deque" -version = "0.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9dd111b7b7f7d55b72c0a6ae361660ee5853c9af73f70c3c2ef6858b950e2e51" -dependencies = [ - "crossbeam-epoch", - "crossbeam-utils", -] - -[[package]] -name = "crossbeam-epoch" -version = "0.9.18" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" -dependencies = [ - "crossbeam-utils", -] - [[package]] name = "crossbeam-queue" version = "0.3.12" @@ -69,7 +50,6 @@ checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" name = "swactor" version = "0.1.0" dependencies = [ - "crossbeam-deque", "crossbeam-queue", "crossbeam-utils", "getrandom", diff --git a/Cargo.toml b/Cargo.toml index 6471309..78f8ddb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,7 +15,6 @@ stress = [] # Enable stress tests [dependencies] getrandom = { version = "0.2", optional = true } -crossbeam-deque = "0.8" crossbeam-queue = "0.3.12" crossbeam-utils = "0.8.21" smallvec = "1.13" diff --git a/README.md b/README.md index 181a624..bcce6fe 100644 --- a/README.md +++ b/README.md @@ -5,6 +5,7 @@ ```rust use swactor::{ + Ctx, actor::{ActorAddress, ActorInterface}, runtime::{Runtime, RuntimeConfig}, }; @@ -22,10 +23,10 @@ impl ActorInterface for Greeter { type Incoming = GreetMessage; type Response = GreetResponse; - fn handle(&mut self, ctx: &Runtime, msg: GreetMessage) { + fn handle(&mut self, ctx: &Ctx, msg: GreetMessage) { let res = GreetResponse(format!("Hello, {}!", msg.who)); self.num_greeted += 1; - if let Err(_) = ctx.send_to(msg.return_addr, res) { + if let Err(_) = ctx.send(msg.return_addr, res) { self.num_greeted -= 1; } } diff --git a/benches/scaling.rs b/benches/scaling.rs index e54bf64..85440cd 100644 --- a/benches/scaling.rs +++ b/benches/scaling.rs @@ -8,6 +8,7 @@ use crate::harness::{black_box, Bench, BenchSuite}; use std::thread; use swactor::{ + Ctx, actor::ActorInterface, runtime::{Runtime, RuntimeConfig}, }; @@ -34,7 +35,7 @@ impl ActorInterface for CounterActor { type Incoming = Increment; type Response = (); - fn handle(&mut self, _ctx: &Runtime, _msg: Increment) { + fn handle(&mut self, _ctx: &Ctx, _msg: Increment) { self.count += 1; } } @@ -47,7 +48,7 @@ impl ActorInterface for SharedCounter { type Incoming = Increment; type Response = (); - fn handle(&mut self, _ctx: &Runtime, _msg: Increment) { + fn handle(&mut self, _ctx: &Ctx, _msg: Increment) { self.count .fetch_add(1, std::sync::atomic::Ordering::Relaxed); } @@ -71,7 +72,7 @@ impl ActorInterface for PayloadActor { type Incoming = Payload; type Response = (); - fn handle(&mut self, _ctx: &Runtime, msg: Payload) { + fn handle(&mut self, _ctx: &Ctx, msg: Payload) { self.bytes_received += msg.0.len(); black_box(&msg.0); } @@ -97,8 +98,7 @@ pub fn bench_actor_count_scaling(suite: &mut BenchSuite) { || { let config = RuntimeConfig { max_actors: (actor_count as usize) + 100, - router_max_messages: (total_messages as usize) * 3, - actor_max_messages: (messages_per_actor as usize) * 2, + actor_max_messages: (total_messages as usize) * 3, num_threads: 1, }; let runtime = Runtime::new(config); @@ -110,7 +110,7 @@ pub fn bench_actor_count_scaling(suite: &mut BenchSuite) { actors.push(addr); } - // Process registrations + // Process spawns for _ in 0..(actor_count * 2) { runtime.tick(); } @@ -156,8 +156,7 @@ pub fn bench_thread_count_scaling(suite: &mut BenchSuite) { let counter = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); let config = RuntimeConfig { max_actors: (actor_count as usize) + 100, - router_max_messages: (total_messages as usize) * 3, - actor_max_messages: (messages_per_actor as usize) * 2, + actor_max_messages: (total_messages as usize) * 3, num_threads: thread_count, }; let runtime = Runtime::new(config); @@ -233,14 +232,13 @@ pub fn bench_payload_size_scaling(suite: &mut BenchSuite) { || { let config = RuntimeConfig { max_actors: 10, - router_max_messages: (message_count as usize) * 2, actor_max_messages: (message_count as usize) * 2, num_threads: 1, }; let runtime = Runtime::new(config); let sink = runtime.spawn(PayloadActor::new()).unwrap(); - // Process registration + // Process spawn for _ in 0..10 { runtime.tick(); } diff --git a/benches/throughput.rs b/benches/throughput.rs index 58c67b3..001b0a1 100644 --- a/benches/throughput.rs +++ b/benches/throughput.rs @@ -8,6 +8,7 @@ use crate::harness::{black_box, Bench, BenchSuite}; use swactor::{ + Ctx, actor::{ActorAddress, ActorInterface}, runtime::{Runtime, RuntimeConfig}, }; @@ -34,7 +35,7 @@ impl ActorInterface for SinkActor { type Incoming = Ping; type Response = (); - fn handle(&mut self, _ctx: &Runtime, _msg: Ping) { + fn handle(&mut self, _ctx: &Ctx, _msg: Ping) { self.count += 1; } } @@ -58,9 +59,9 @@ impl ActorInterface for ForwardActor { type Incoming = Ping; type Response = Ping; - fn handle(&mut self, ctx: &Runtime, msg: Ping) { + fn handle(&mut self, ctx: &Ctx, msg: Ping) { if let Some(next) = self.next { - let _ = ctx.send_to(next, msg); + let _ = ctx.send(next, msg); } } } @@ -69,7 +70,7 @@ impl ActorInterface for ForwardActor { // Benchmarks // ============================================================================ -/// Benchmark: Messages sent through the router to a single sink actor +/// Benchmark: Messages sent through the runtime to a single sink actor pub fn bench_message_throughput(suite: &mut BenchSuite) { for msg_count in [1_000u64, 10_000, 100_000] { let name = format!("message_throughput_{}", msg_count); @@ -83,7 +84,6 @@ pub fn bench_message_throughput(suite: &mut BenchSuite) { // Setup: create runtime and sink actor let config = RuntimeConfig { max_actors: 100, - router_max_messages: (msg_count as usize) * 2, actor_max_messages: (msg_count as usize) * 2, num_threads: 1, }; @@ -97,8 +97,6 @@ pub fn bench_message_throughput(suite: &mut BenchSuite) { let _ = runtime.send_to::(sink, Ping); } // Process until done - // Tick enough times to process all messages - // (router tick + actor tick) * messages / WATERLEVEL for _ in 0..(count * 3) { runtime.tick(); } @@ -123,7 +121,6 @@ pub fn bench_spawn_rate(suite: &mut BenchSuite) { || { let config = RuntimeConfig { max_actors: 1000, - router_max_messages: 10_000, actor_max_messages: 100, num_threads: 1, }; @@ -133,7 +130,7 @@ pub fn bench_spawn_rate(suite: &mut BenchSuite) { for _ in 0..actor_count { let _ = runtime.spawn(SinkActor::new()); } - // Process router messages to register all actors + // Process spawns for _ in 0..(actor_count * 2) { runtime.tick(); } @@ -159,10 +156,9 @@ pub fn bench_fanout(suite: &mut BenchSuite) { || { let config = RuntimeConfig { max_actors: (fan_count as usize) + 10, - router_max_messages: (fan_count as usize) + actor_max_messages: (fan_count as usize) * (messages_per_receiver as usize) * 2, - actor_max_messages: (messages_per_receiver as usize) * 2, num_threads: 1, }; let runtime = Runtime::new(config); @@ -174,7 +170,7 @@ pub fn bench_fanout(suite: &mut BenchSuite) { sinks.push(addr); } - // Process router registrations + // Process spawns for _ in 0..(fan_count * 2) { runtime.tick(); } @@ -217,8 +213,7 @@ pub fn bench_fanin(suite: &mut BenchSuite) { let total_messages = (sender_count * messages_per_sender) as usize; let config = RuntimeConfig { max_actors: (sender_count as usize) + 10, - router_max_messages: total_messages * 3, - actor_max_messages: total_messages * 2, + actor_max_messages: total_messages * 3, num_threads: 1, }; let runtime = Runtime::new(config); @@ -233,7 +228,7 @@ pub fn bench_fanin(suite: &mut BenchSuite) { senders.push(addr); } - // Process router registrations + // Process spawns for _ in 0..((sender_count + 1) * 2) { runtime.tick(); } @@ -275,8 +270,7 @@ pub fn bench_ring(suite: &mut BenchSuite) { || { let config = RuntimeConfig { max_actors: (ring_size as usize) + 10, - router_max_messages: 10_000, - actor_max_messages: 1_000, + actor_max_messages: 10_000, num_threads: 1, }; let runtime = Runtime::new(config); @@ -292,7 +286,7 @@ pub fn bench_ring(suite: &mut BenchSuite) { // so instead we'll use an inbox to receive the final message // For now, we'll just measure message passing through a chain - // Process registrations + // Process spawns for _ in 0..(ring_size * 2) { runtime.tick(); } @@ -301,7 +295,7 @@ pub fn bench_ring(suite: &mut BenchSuite) { }, |(runtime, actors, laps)| { // Send to first actor (even though they don't forward, we're - // measuring the router + inbox overhead) + // measuring the transfer queue + inbox overhead) for _ in 0..laps { for actor in &actors { let _ = runtime.send_to::(*actor, Ping); diff --git a/examples/hello.rs b/examples/hello.rs index d67c439..18378d9 100644 --- a/examples/hello.rs +++ b/examples/hello.rs @@ -1,4 +1,5 @@ use swactor::{ + Ctx, actor::{ActorAddress, ActorInterface}, runtime::{Runtime, RuntimeConfig}, }; @@ -24,10 +25,10 @@ impl ActorInterface for Greeter { type Incoming = GreetMessage; type Response = GreetResponse; - fn handle(&mut self, ctx: &Runtime, msg: GreetMessage) { + fn handle(&mut self, ctx: &Ctx, msg: GreetMessage) { let res = GreetResponse(format!("Hello, {}!", msg.who)); self.num_greeted += 1; - if let Err(_) = ctx.send_to(msg.return_addr, res) { + if let Err(_) = ctx.send(msg.return_addr, res) { // no error handling self.num_greeted -= 1; } diff --git a/examples/ring.rs b/examples/ring.rs index d3c6031..703a3be 100644 --- a/examples/ring.rs +++ b/examples/ring.rs @@ -1,4 +1,5 @@ use swactor::{ + Ctx, actor::{ActorAddress, ActorInterface}, runtime::{Inbox, Runtime, RuntimeConfig}, }; @@ -30,8 +31,8 @@ impl RingActor { impl ActorInterface for RingActor { type Incoming = RingMessage; type Response = (); - fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming) { - if let Err(_) = ctx.send_to(self.next, msg.next()) { + fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming) { + if let Err(_) = ctx.send(self.next, msg.next()) { // do nothing } } diff --git a/src/actor.rs b/src/actor.rs index c3938ea..e14cf82 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -1,4 +1,7 @@ -use crate::{WATERLEVEL, channel::Receiver, get_random, runtime::Runtime}; +use std::any::Any; + +use crate::{get_random, worker::mailbox::Mailbox}; +use crate::context::{ContextInner, Ctx}; /// The primary trait defining data that can be passed to and from actor processes pub trait Message: 'static + Sized + Clone + Send + Sync {} @@ -14,7 +17,7 @@ impl Message for T {} /// /// # Example /// ``` -/// use swactor::{actor::{ActorAddress, ActorInterface}, runtime::Runtime}; +/// use swactor::{Ctx, actor::{ActorAddress, ActorInterface}}; /// /// struct Greeter { /// num_greeted: usize, @@ -32,10 +35,10 @@ impl Message for T {} /// impl ActorInterface for Greeter { /// type Incoming = GreetMessage; /// type Response = GreetResponse; -/// -/// fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming) { +/// +/// fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming) { /// let response = GreetResponse(format!("Hello, {}!", msg.who).to_string()); -/// if let Ok(_) = ctx.send_to(msg.return_addr, response) { +/// if let Ok(_) = ctx.send(msg.return_addr, response) { /// self.num_greeted += 1; /// } /// } @@ -44,7 +47,7 @@ impl Message for T {} pub trait ActorInterface: 'static + Send { type Incoming: Message; type Response: Message; - fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming); + fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming); } /// A unique address for this actor. 32 bytes is overkill for a small application, @@ -60,48 +63,58 @@ impl ActorAddress { } } -/// The actor process as represented in the Runtime, with the actor state stored with it's inbox. +/// The actor process as represented in the Runtime, with the actor state stored with its mailbox. pub(crate) struct Actor where A: ActorInterface, { - inbox: Receiver, + addr: ActorAddress, + mailbox: Mailbox, inner: A, } impl Actor { - pub(crate) fn new(inbox: Receiver, inner: A) -> Self { - Self { inbox, inner } + pub(crate) fn new(addr: ActorAddress, mailbox: Mailbox, inner: A) -> Self { + Self { + addr, + mailbox, + inner, + } } } /// Trait for type-erased actors pub(crate) trait AnyActor: Send { - fn tick(&mut self, ctx: &Runtime); + /// Tick the actor, processing pending messages. Returns `true` if any work was done. + fn tick(&mut self, inner: &dyn ContextInner) -> bool; + /// Deliver a type-erased message into this actor's mailbox. + /// Returns `true` if the downcast succeeded. + fn deliver(&mut self, msg: Box) -> bool; } impl AnyActor for Actor where A: ActorInterface, { - fn tick(&mut self, ctx: &Runtime) { - // TODO: WATERLEVEL is hard coded, and so is this message handling scheme. We should - // make it so both are more flexible, with sane defaults. - // FIXME: lots of indirection just to get length on a hot path - let total_messages = self.inbox.len(); - let messages_to_process = if total_messages < WATERLEVEL { - total_messages - } else { - total_messages >> 1 - }; - - for _ in 0..messages_to_process { - match self.inbox.try_recv() { - Some(msg) => self.inner.handle(ctx, msg), - None => unreachable!( - "We checked number of unprocessed messages in the queue ahead of processing" - ), + fn tick(&mut self, inner: &dyn ContextInner) -> bool { + let n = self.mailbox.drain_count(); + if n > 0 { + let ctx = Ctx::new(inner, self.addr); + for _ in 0..n { + if let Some(msg) = self.mailbox.pop() { + self.inner.handle(&ctx, msg); + } } } + n > 0 + } + + fn deliver(&mut self, msg: Box) -> bool { + if let Ok(typed) = msg.downcast::() { + self.mailbox.push(*typed); + true + } else { + false + } } } diff --git a/src/address_map.rs b/src/address_map.rs new file mode 100644 index 0000000..adabe3a --- /dev/null +++ b/src/address_map.rs @@ -0,0 +1,90 @@ +use std::collections::HashMap; +use std::sync::RwLock; + +use crate::actor::ActorAddress; + +/// Identifies a worker thread. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub(crate) struct WorkerId(pub(crate) usize); + +impl WorkerId { + pub fn as_usize(self) -> usize { + self.0 + } +} + +/// Maps actor addresses to the worker that owns them. +/// +/// `RwLock` — zero contention for parallel reads, write-rare (only on spawn). +pub(crate) struct AddressMap { + inner: RwLock>, +} + +impl AddressMap { + pub fn new() -> Self { + Self { + inner: RwLock::new(HashMap::new()), + } + } + + pub fn with_capacity(cap: usize) -> Self { + Self { + inner: RwLock::new(HashMap::with_capacity(cap)), + } + } + + pub fn insert(&self, addr: ActorAddress, worker: WorkerId) { + self.inner.write().unwrap().insert(addr, worker); + } + + pub fn remove(&self, addr: &ActorAddress) { + self.inner.write().unwrap().remove(addr); + } + + pub fn lookup(&self, addr: &ActorAddress) -> Option { + self.inner.read().unwrap().get(addr).copied() + } + + pub fn len(&self) -> usize { + self.inner.read().unwrap().len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn insert_and_lookup() { + let map = AddressMap::new(); + let addr = ActorAddress::default(); + let wid = WorkerId(3); + map.insert(addr, wid); + assert_eq!(map.lookup(&addr), Some(wid)); + } + + #[test] + fn lookup_missing_returns_none() { + let map = AddressMap::new(); + let addr = ActorAddress::default(); + assert_eq!(map.lookup(&addr), None); + } + + #[test] + fn remove_works() { + let map = AddressMap::new(); + let addr = ActorAddress::default(); + map.insert(addr, WorkerId(0)); + map.remove(&addr); + assert_eq!(map.lookup(&addr), None); + } + + #[test] + fn len_tracks_entries() { + let map = AddressMap::with_capacity(10); + assert_eq!(map.len(), 0); + let addr1 = ActorAddress::default(); + map.insert(addr1, WorkerId(0)); + assert_eq!(map.len(), 1); + } +} diff --git a/src/config.rs b/src/config.rs new file mode 100644 index 0000000..f6d2a8b --- /dev/null +++ b/src/config.rs @@ -0,0 +1,24 @@ +/// The tunable settings for the runtime. +pub struct RuntimeConfig { + pub max_actors: usize, + pub actor_max_messages: usize, + pub num_threads: usize, +} + +/// 8kB for the `Box<..>` before counting the rest of the memory +const DEFAULT_MAX_ACTORS: usize = 1_000; + +/// 16kB PER ACTOR to alloc space for storing messages. +/// With default setting of [DEFAULT_MAX_ACTORS] this is: +/// 1_000 * 16kB = 16MB +const DEFAULT_ACTOR_MAX_MESSAGES: usize = 1_000; + +impl Default for RuntimeConfig { + fn default() -> Self { + Self { + max_actors: DEFAULT_MAX_ACTORS, + actor_max_messages: DEFAULT_ACTOR_MAX_MESSAGES, + num_threads: 1, + } + } +} diff --git a/src/context.rs b/src/context.rs new file mode 100644 index 0000000..6bd2387 --- /dev/null +++ b/src/context.rs @@ -0,0 +1,45 @@ +use std::any::Any; + +use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}; +use crate::worker::mailbox::Mailbox; +use crate::Error; + +/// Object-safe inner trait for sending type-erased messages. +pub(crate) trait ContextInner { + fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error>; + fn spawn_any(&self, addr: ActorAddress, actor: Box) -> Result<(), Error>; +} + +/// Actor syscall interface — passed to `ActorInterface::handle()`. +/// +/// Wraps a `&dyn ContextInner` to solve the object-safety problem while +/// providing a typed public API. +pub struct Ctx<'a> { + inner: &'a dyn ContextInner, + self_addr: ActorAddress, +} + +impl<'a> Ctx<'a> { + pub(crate) fn new(inner: &'a dyn ContextInner, self_addr: ActorAddress) -> Self { + Self { inner, self_addr } + } + + /// Returns the address of the actor currently being ticked. + pub fn self_addr(&self) -> ActorAddress { + self.self_addr + } + + /// Send a typed message to an actor address. + pub fn send(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { + self.inner.send_any(addr, Box::new(msg)) + } + + /// Spawn a new actor, returning its address. + pub fn spawn(&self, actor: A) -> Result { + let addr = ActorAddress::new_random(); + let actor = Actor::new(addr, Mailbox::new(), actor); + let boxed: Box = Box::new(actor); + self.inner.spawn_any(addr, boxed)?; + Ok(addr) + } +} diff --git a/src/envelope.rs b/src/envelope.rs new file mode 100644 index 0000000..d55fb8a --- /dev/null +++ b/src/envelope.rs @@ -0,0 +1,49 @@ +use std::any::Any; + +use crate::actor::ActorAddress; + +/// A type-erased message envelope for cross-worker delivery. +/// +/// Uses `Box` (no atomic refcount) and move semantics (no clone). +pub(crate) struct Envelope { + dest: ActorAddress, + payload: Box, +} + +impl Envelope { + pub fn new(dest: ActorAddress, payload: Box) -> Self { + Self { dest, payload } + } + + pub fn dest(&self) -> ActorAddress { + self.dest + } + + pub fn downcast(self) -> Option { + self.payload.downcast::().ok().map(|b| *b) + } + + pub fn into_payload(self) -> Box { + self.payload + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn envelope_roundtrip() { + let addr = ActorAddress::default(); + let env = Envelope::new(addr, Box::new(42u64)); + assert_eq!(env.dest(), addr); + assert_eq!(env.downcast::(), Some(42u64)); + } + + #[test] + fn envelope_wrong_type_returns_none() { + let addr = ActorAddress::default(); + let env = Envelope::new(addr, Box::new(42u64)); + assert!(env.downcast::().is_none()); + } +} diff --git a/src/lib.rs b/src/lib.rs index ae7d08a..08f7b07 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,11 +1,18 @@ pub mod actor; pub mod worker; -mod channel; +pub(crate) mod channel; pub(crate) mod error; pub use error::Error; -mod router; +pub mod context; +pub use context::Ctx; + +pub(crate) mod envelope; +pub(crate) mod address_map; +pub(crate) mod placement; +pub mod config; + pub mod runtime; #[cfg(feature = "getrandom")] @@ -26,14 +33,3 @@ pub(crate) fn get_random(buf: &mut [u8]) { *byte = bytes[i % core::mem::size_of::()]; } } - -/// FIXME: remove hard coded defaults -/// The strategy for message processing is such: -/// -/// ```ignore -/// if total_messages < WATERLEVEL: -/// process all -/// else -/// process total_messages >> 1 -/// ``` -const WATERLEVEL: usize = 10; diff --git a/src/placement.rs b/src/placement.rs new file mode 100644 index 0000000..d88e63d --- /dev/null +++ b/src/placement.rs @@ -0,0 +1,37 @@ +use std::sync::atomic::{AtomicUsize, Ordering}; + +use crate::address_map::WorkerId; + +/// Round-robin actor placement strategy. +pub(crate) struct Placement { + next: AtomicUsize, + num_workers: usize, +} + +impl Placement { + pub fn new(num_workers: usize) -> Self { + Self { + next: AtomicUsize::new(0), + num_workers, + } + } + + pub fn next_worker(&self) -> WorkerId { + let id = self.next.fetch_add(1, Ordering::Relaxed) % self.num_workers; + WorkerId(id) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn round_robin() { + let p = Placement::new(3); + assert_eq!(p.next_worker(), WorkerId(0)); + assert_eq!(p.next_worker(), WorkerId(1)); + assert_eq!(p.next_worker(), WorkerId(2)); + assert_eq!(p.next_worker(), WorkerId(0)); + } +} diff --git a/src/router.rs b/src/router.rs deleted file mode 100644 index d015fe1..0000000 --- a/src/router.rs +++ /dev/null @@ -1,83 +0,0 @@ -use std::{collections::HashMap, sync::Arc}; - -use crate::{ - actor::{ActorAddress, ActorInterface, Message}, - channel::Sender, - runtime::Runtime, -}; - -/// FIXME: Go over with a fine-toothed comb and reassure yourself this typing -/// makes sense, that we are not doing loads of indirection on a hot path. -/// -/// A type erased `Message` to be routed between actor processes. -pub(crate) type Envelope = Arc; - -pub(crate) trait SenderT: Send + Sync { - fn try_send(&self, envelope: Envelope); -} - -impl SenderT for Sender { - fn try_send(&self, envelope: Envelope) { - if let Some(msg) = envelope.downcast_ref::() { - // FIXME: we are directly cloning the contents of the Arc pointer here - // Do we want to? Should we provide another way? - // - // The standard concept of an actor has message and state - // isolation, so we should leave this as is. However, we should - // make it clear and obvious this pathway is the heavy, contained - // pathway, and include a shared memory pathway for logic that - // may need it. - let _ = Sender::try_send(self, msg.clone()); - } - } -} - -/// Internal messages for the Router's own inbox -#[derive(Clone)] -pub(crate) enum RouterMessage { - /// register addrs with sender - AddAddr(ActorAddress, Arc), - - /// FIXME: this will be active when we allow actors to shut themselves - /// down. For now, disable the warning. - #[allow(dead_code)] - /// remove an actor from the address book - RemoveAddr(ActorAddress), - - /// send to - SendToAddr { addr: ActorAddress, msg: Envelope }, -} - -/// The `Router` is responsible for taking in and delivering all messages in the runtime. -pub(crate) struct Router { - directory: HashMap>, -} - -impl Router { - pub fn new() -> Self { - Self { - directory: HashMap::new(), - } - } -} - -impl ActorInterface for Router { - type Incoming = RouterMessage; - type Response = (); - - fn handle(&mut self, _ctx: &Runtime, msg: Self::Incoming) { - match msg { - RouterMessage::AddAddr(addr, sender) => { - self.directory.insert(addr, sender); - } - RouterMessage::RemoveAddr(addr) => { - self.directory.remove(&addr); - } - RouterMessage::SendToAddr { addr, msg } => { - if let Some(sender) = self.directory.get(&addr) { - sender.try_send(msg); - } - } - } - } -} diff --git a/src/runtime.rs b/src/runtime.rs index 1a5df3b..e602db2 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,17 +1,71 @@ +use std::any::Any; +use std::cell::RefCell; +use std::collections::HashMap; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, RwLock}; use std::thread::{self, JoinHandle}; -use std::time::Duration; -use crossbeam_deque::{Injector, Steal, Stealer, Worker}; +use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}; +use crate::address_map::{AddressMap, WorkerId}; +use crate::channel::{Receiver, Sender}; +// Re-export RuntimeConfig so existing code using `runtime::RuntimeConfig` still works +pub use crate::config::RuntimeConfig; +use crate::context::ContextInner; +use crate::envelope::Envelope; +use crate::placement::Placement; +use crate::worker::mailbox::Mailbox; +use crate::worker::Worker; +use crate::Error; -use crate::channel::HybridChannel; -use crate::{ - actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}, - channel::{Receiver, Sender}, - router::{Router, RouterMessage}, - Error, -}; +// ─── SenderT trait (moved from router.rs) ──────────────────────────────────── + +/// Type-erased sender for external inboxes. +pub(crate) trait SenderT: Send + Sync { + fn try_send_any(&self, msg: Box); +} + +impl SenderT for Sender { + fn try_send_any(&self, msg: Box) { + if let Ok(typed) = msg.downcast::() { + let _ = Sender::try_send(self, *typed); + } + } +} + +// ─── InboxRegistry ─────────────────────────────────────────────────────────── + +/// Registry of external inboxes — replaces the Router's role for non-actor receivers. +pub(crate) struct InboxRegistry { + senders: RwLock>>, +} + +impl InboxRegistry { + pub fn new() -> Self { + Self { + senders: RwLock::new(HashMap::new()), + } + } + + pub fn register(&self, addr: ActorAddress, sender: Arc) { + self.senders.write().unwrap().insert(addr, sender); + } + + pub fn try_deliver( + &self, + addr: ActorAddress, + msg: Box, + ) -> Result<(), Error> { + let senders = self.senders.read().unwrap(); + if let Some(sender) = senders.get(&addr) { + sender.try_send_any(msg); + Ok(()) + } else { + Err(Error::from("Address not found")) + } + } +} + +// ─── Inbox ─────────────────────────────────────────────────────────────────── /// Generic message inbox for receiving messages outside of the runtime. pub struct Inbox { @@ -29,59 +83,210 @@ impl Inbox { } } -/// The tunable settings for the runtime. -pub struct RuntimeConfig { - pub max_actors: usize, - pub router_max_messages: usize, - pub actor_max_messages: usize, - pub num_threads: usize, -} - -/// 8kB for the `Box<..>` before counting the rest of the memory -const DEFAULT_MAX_ACTORS: usize = 1_000; - -/// 160kB for the `Arc<..>` before counting the rest of the memory -const DEFAULT_ROUTER_MAX_MESSAGES: usize = 10_000; - -/// 16kB PER ACTOR to alloc space for storing the `Arc<..>` pointers -/// With default setting of [DEFAULT_MAX_ACTORS] this is: -/// 1_000 * 16kB = 16MB -const DEFAULT_ACTOR_MAX_MESSAGES: usize = 1_000; - -impl Default for RuntimeConfig { - fn default() -> Self { - Self { - max_actors: DEFAULT_MAX_ACTORS, - router_max_messages: DEFAULT_ROUTER_MAX_MESSAGES, - actor_max_messages: DEFAULT_ACTOR_MAX_MESSAGES, - num_threads: 1, - } - } -} - -/// Per-worker state for work-stealing scheduler -struct WorkerState { - local: Worker>, // FIFO for fairness - id: usize, -} - -/// Shared scheduler state for work-stealing -struct SchedulerState { - injector: Injector>, // For spawn() - stealers: Vec>>, // For work-stealing - is_running: AtomicBool, -} +// ─── Runtime ───────────────────────────────────────────────────────────────── /// The `Runtime` struct is the primary gateway for interacting with the framework. pub struct Runtime { config: RuntimeConfig, - actor_queue: HybridChannel>, // Used for single-threaded mode - scheduler: Option>, // Used for multi-threaded mode - router_interface: Sender, - router: Option>, // `None` if single-threaded - - // for multithreaded contexts + address_map: Arc, + inbox_registry: Arc, + transfer_txs: Vec>, + spawn_txs: Vec)>>, + placement: Placement, is_running: AtomicBool, + /// Single-threaded mode: worker stored inline + single_worker: Option>, + /// Multi-threaded mode: workers waiting to be assigned to threads by run() + pending_workers: Option>, +} + +// Safety: RefCell is only accessed from the thread that owns the Runtime +// in single-threaded mode. In multi-threaded mode, single_worker is None and +// pending_workers is consumed by run() before Arc sharing. +unsafe impl Sync for Runtime {} + +impl Runtime { + /// Builds a new `Runtime` struct, but does not yet run anything. If multithreaded, call + /// `run()`, if single threaded, needs to be driven by calls to the `tick()` method. + pub fn new(config: RuntimeConfig) -> Self { + let num_workers = if config.num_threads < 2 { + 1 + } else { + config.num_threads + }; + + let address_map = Arc::new(AddressMap::with_capacity(config.max_actors)); + let inbox_registry = Arc::new(InboxRegistry::new()); + let placement = Placement::new(num_workers); + + let mut transfer_txs = Vec::with_capacity(num_workers); + let mut spawn_txs = Vec::with_capacity(num_workers); + let mut workers = Vec::with_capacity(num_workers); + + for i in 0..num_workers { + let transfer_rx = Receiver::::new(config.actor_max_messages); + let transfer_tx = transfer_rx.new_sender(); + transfer_txs.push(transfer_tx); + + let spawn_rx = + Receiver::<(ActorAddress, Box)>::new(config.max_actors); + let spawn_tx = spawn_rx.new_sender(); + spawn_txs.push(spawn_tx); + + workers.push(Worker::new(WorkerId(i), transfer_rx, spawn_rx)); + } + + if config.num_threads < 2 { + // Single-threaded: store one worker inline + let worker = workers.remove(0); + Self { + config, + address_map, + inbox_registry, + transfer_txs, + spawn_txs, + placement, + is_running: AtomicBool::new(false), + single_worker: Some(RefCell::new(worker)), + pending_workers: None, + } + } else { + // Multi-threaded: stash workers for run() + Self { + config, + address_map, + inbox_registry, + transfer_txs, + spawn_txs, + placement, + is_running: AtomicBool::new(false), + single_worker: None, + pending_workers: Some(workers), + } + } + } + + /// Spawn an actor, returns its address + pub fn spawn(&self, actor: A) -> Result { + let addr = ActorAddress::new_random(); + let worker_id = self.placement.next_worker(); + self.address_map.insert(addr, worker_id); + let actor = Actor::new(addr, Mailbox::new(), actor); + let boxed: Box = Box::new(actor); + self.spawn_txs[worker_id.as_usize()] + .try_send((addr, boxed)) + .map_err(|_| Error::from("Runtime error: spawn queue full"))?; + Ok(addr) + } + + /// Send a message to an actor address + pub fn send_to(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { + let msg_box: Box = Box::new(msg); + match self.address_map.lookup(&addr) { + Some(wid) => self.transfer_txs[wid.as_usize()] + .try_send(Envelope::new(addr, msg_box)) + .map_err(|_| Error::from("Transfer queue full")), + None => self.inbox_registry.try_deliver(addr, msg_box), + } + } + + /// Create an external inbox for receiving messages in the outer process containing the runtime + pub fn new_inbox(&self) -> Result, Error> { + let addr = ActorAddress::new_random(); + let receiver = Receiver::::new(self.config.actor_max_messages); + let sender = receiver.new_sender(); + self.inbox_registry.register(addr, Arc::new(sender)); + Ok(Inbox { + addr, + inner: receiver, + }) + } + + /// Drive one tick of the single-threaded worker. + pub fn tick(&self) { + if let Some(ref worker) = self.single_worker { + worker.borrow_mut().tick_once( + &self.address_map, + &self.transfer_txs, + &self.spawn_txs, + &self.placement, + &self.inbox_registry, + ); + } + } + + /// Spawn worker threads and start processing, returning a set of handles and + /// a Runtime object to interface with. + /// + /// ### WARN: + /// ##### Returns an error if the configuration is set as single threaded + /// `config.num_threads == 1` + pub fn run(mut self) -> Result { + if self.config.num_threads < 2 { + return Err(Error::from( + "Runtime error: cannot call `Runtime::run()` from a single-threaded context.", + )); + } + + self.is_running.store(true, Ordering::Release); + + let workers = self + .pending_workers + .take() + .expect("Workers must be present for multi-threaded runtime"); + + let rt = Arc::new(self); + let mut handles: Vec> = Vec::with_capacity(workers.len()); + + for mut worker in workers { + let rt_clone = rt.clone(); + let handle = thread::spawn(move || { + worker_loop(&mut worker, &rt_clone); + }); + handles.push(handle); + } + + Ok(RuntimeHandle { + runtime: rt, + threads: handles, + }) + } + + /// Signal all workers to stop + pub fn shutdown(&self) { + self.is_running.store(false, Ordering::Release); + } +} + +/// Worker thread loop for multi-threaded runtime. +/// Uses spin → yield → park backoff to reduce CPU usage when idle. +fn worker_loop(worker: &mut Worker, rt: &Runtime) { + let mut idle_count: u32 = 0; + + while rt.is_running.load(Ordering::Acquire) { + let did_work = worker.tick_once( + &rt.address_map, + &rt.transfer_txs, + &rt.spawn_txs, + &rt.placement, + &rt.inbox_registry, + ); + + if did_work { + idle_count = 0; + } else { + idle_count = idle_count.saturating_add(1); + if idle_count < 64 { + core::hint::spin_loop(); + } else if idle_count < 256 { + thread::yield_now(); + } else { + // Park: sleep briefly, cap at 1ms + let micros = std::cmp::min((idle_count - 256) as u64 * 50, 1000); + thread::sleep(std::time::Duration::from_micros(micros)); + } + } + } } /// Handle for dealing with a runtime that has started via the `Runtime::run()` method. @@ -103,258 +308,22 @@ impl RuntimeHandle { } } -impl Runtime { - /// Builds a new `Runtime` struct, but does not yet run anything. If multithreaded, call - /// `run()`, if single threaded, needs to be driven by calls to the `tick()` method. - pub fn new(config: RuntimeConfig) -> Self { - let actor_queue = HybridChannel::new(config.max_actors); - - // router is a unique actor in that the runtime needs access to it's `Sender` handle - let router_inner = Router::new(); - let router_inbox: Receiver = - 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, - scheduler: None, // Initialized in run() for multi-threaded mode - router_interface: router_sender, - is_running: AtomicBool::new(false), - router: router_option, +impl ContextInner for Runtime { + fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error> { + match self.address_map.lookup(&addr) { + Some(wid) => { + let _ = self.transfer_txs[wid.as_usize()].try_send(Envelope::new(addr, msg)); + Ok(()) + } + None => self.inbox_registry.try_deliver(addr, msg), } } - /// Spawn an actor, returns its address - pub fn spawn(&self, actor: A) -> Result { - // assign a stochastic - let addr = ActorAddress::new_random(); - let inbox = Receiver::::new(self.config.actor_max_messages); - let sender = inbox.new_sender(); - - // Register the sender with the router - self.router_interface - .try_send(RouterMessage::AddAddr(addr, Arc::new(sender))) - .map_err(|_| { - Error::from("Runtime error: failed to add actor to router. Router inbox full") - })?; - - let boxed_actor: Box = Box::new(Actor::new(inbox, actor)); - - // Multi-threaded: use injector, single-threaded: use shared queue - if let Some(ref scheduler) = self.scheduler { - scheduler.injector.push(boxed_actor); - } else { - self.actor_queue - .push(boxed_actor) - .map_err(|_| Error::from("Runtime error: Failed to spawn actor. Queue full."))?; - } - - Ok(addr) - } - - /// Send a message to an actor address - pub fn send_to(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { - self.router_interface - .try_send(RouterMessage::SendToAddr { - addr, - msg: Arc::new(msg), - }) - .map_err(|_| Error::from("Failed to send message to router.")) - } - - /// Create an external inbox for receiving messages in the outer process containing the runtime - pub fn new_inbox(&self) -> Result, Error> { - let addr = ActorAddress::new_random(); - - let receiver = Receiver::::new(self.config.actor_max_messages); - let sender = receiver.new_sender(); - - // Register the sender with the router - self.router_interface - .try_send(RouterMessage::AddAddr(addr, Arc::new(sender))) - .map_err(|_| { - Error::from( - "Runtime error: failed to add a new inbox channel. Router inbox is full.", - ) - })?; - - Ok(Inbox { - addr, - inner: receiver, - }) - } - - /// Spawn worker threads and start processing, returning a set of handles and - /// a Runtime object to interface with. - /// - /// ### WARN: - /// ##### This function panics if the configuration is set as single threaded - /// `config.num_threads == 1` - pub fn run(mut self) -> Result { - if self.config.num_threads < 2 { - return Err(Error::from( - "Runtime error: cannot call `Runtime::run()` from a single-threaded context.", - )); - } - - self.is_running.store(true, Ordering::Release); - - // Take router out before wrapping in Arc - it will be owned by router thread - let mut router = self - .router - .take() - .expect("Router must be present for multi-threaded runtime"); - - let num_workers = self.config.num_threads - 1; - - // Create workers and collect stealers for work-stealing - let mut workers = Vec::with_capacity(num_workers); - let mut stealers = Vec::with_capacity(num_workers); - for id in 0..num_workers { - let worker = WorkerState { - local: Worker::new_fifo(), // FIFO for fairness - id, - }; - stealers.push(worker.local.stealer()); - workers.push(worker); - } - - let scheduler = Arc::new(SchedulerState { - injector: Injector::new(), - stealers, - is_running: AtomicBool::new(true), - }); - - // Transfer any actors spawned before run() to the injector - while let Some(actor) = self.actor_queue.pop() { - scheduler.injector.push(actor); - } - - self.scheduler = Some(scheduler.clone()); - let rt = Arc::new(self); - let mut handles: Vec> = vec![]; - - // Router thread owns the router directly - no synchronization needed - // Process multiple ticks per cycle to maximize throughput - const ROUTER_BATCH_SIZE: usize = 64; - let router_handle = { - let ctx = rt.clone(); - let sched = scheduler.clone(); - thread::spawn(move || { - while sched.is_running.load(Ordering::Acquire) { - for _ in 0..ROUTER_BATCH_SIZE { - router.tick(&ctx); - } - } - }) - }; - handles.push(router_handle); - - // Spawn worker threads, each owns its WorkerState - for worker in workers { - let ctx = rt.clone(); - let sched = scheduler.clone(); - let handle = thread::spawn(move || worker_loop(ctx, sched, worker)); - handles.push(handle); - } - - Ok(RuntimeHandle { - runtime: rt, - threads: handles, - }) - } - - /// Pop the actor off the top of the queue and process it's messages, returning it to the back of - /// the queue upon completion. - pub fn tick(&self) { - if let Some(mut actor) = self.actor_queue.pop() { - actor.tick(&self); - let _ = self.actor_queue.push(actor); - } - } - - /// Signal all workers to stop - pub fn shutdown(&self) { - self.is_running.store(false, Ordering::Release); - // Also stop the scheduler if multi-threaded - if let Some(ref scheduler) = self.scheduler { - scheduler.is_running.store(false, Ordering::Release); - } - } -} - -/// Work-stealing worker loop for multi-threaded runtime -fn worker_loop(ctx: Arc, scheduler: Arc, worker: WorkerState) { - const TICKS_PER_ACTOR: usize = 4; - const INJECTOR_CHECK_INTERVAL: usize = 64; // Check injector every N iterations - let mut spin_count: usize = 0; - let mut iteration: usize = 0; - - while scheduler.is_running.load(Ordering::Acquire) { - iteration = iteration.wrapping_add(1); - - // Priority 1: Local queue - let mut actor = worker.local.pop(); - - // Priority 2: Periodically check injector for new actors - // This ensures newly spawned actors get picked up even when workers have local work - if actor.is_none() || (iteration % INJECTOR_CHECK_INTERVAL == 0) { - if let Steal::Success(a) = scheduler.injector.steal_batch_and_pop(&worker.local) { - if actor.is_some() { - // We already had an actor, push the stolen one to local - worker.local.push(a); - } else { - actor = Some(a); - } - } - } - - // Priority 3: Steal from peer workers when idle - if actor.is_none() { - for (i, stealer) in scheduler.stealers.iter().enumerate() { - if i == worker.id { - continue; - } - if let Steal::Success(a) = stealer.steal_batch_and_pop(&worker.local) { - actor = Some(a); - break; - } - } - } - - // Process or backoff - match actor { - Some(mut actor) => { - spin_count = 0; - for _ in 0..TICKS_PER_ACTOR { - actor.tick(&ctx); - } - worker.local.push(actor); - } - None => { - // Exponential backoff - spin_count = spin_count.saturating_add(1); - if spin_count < 10 { - std::hint::spin_loop(); - } else if spin_count < 100 { - thread::yield_now(); - } else { - thread::sleep(Duration::from_micros(10)); - } - } - } + fn spawn_any(&self, addr: ActorAddress, actor: Box) -> Result<(), Error> { + let worker_id = self.placement.next_worker(); + self.address_map.insert(addr, worker_id); + self.spawn_txs[worker_id.as_usize()] + .try_send((addr, actor)) + .map_err(|_| Error::from("Spawn queue full")) } } diff --git a/src/worker/mod.rs b/src/worker/mod.rs index 808ccf2..bc56b5b 100644 --- a/src/worker/mod.rs +++ b/src/worker/mod.rs @@ -1 +1,138 @@ pub mod mailbox; +pub(crate) mod pool; + +use std::any::Any; +use std::cell::RefCell; + +use crate::actor::{ActorAddress, AnyActor}; +use crate::address_map::{AddressMap, WorkerId}; +use crate::channel::{Receiver, Sender}; +use crate::context::ContextInner; +use crate::envelope::Envelope; +use crate::placement::Placement; +use crate::runtime::InboxRegistry; +use crate::Error; +use pool::ActorPool; + +/// A worker owns a set of actors and runs them in a loop. +pub(crate) struct Worker { + id: WorkerId, + pool: ActorPool, + transfer_rx: Receiver, + spawn_rx: Receiver<(ActorAddress, Box)>, +} + +impl Worker { + pub fn new( + id: WorkerId, + transfer_rx: Receiver, + spawn_rx: Receiver<(ActorAddress, Box)>, + ) -> Self { + Self { + id, + pool: ActorPool::new(), + transfer_rx, + spawn_rx, + } + } + + /// Run one iteration of the worker loop. Returns `true` if any work was done. + pub fn tick_once( + &mut self, + address_map: &AddressMap, + transfer_txs: &[Sender], + spawn_txs: &[Sender<(ActorAddress, Box)>], + placement: &Placement, + inbox_registry: &InboxRegistry, + ) -> bool { + let mut did_work = false; + + // 1. Drain spawn queue → add actors to pool + while let Some((addr, actor)) = self.spawn_rx.try_recv() { + self.pool.insert(addr, actor); + did_work = true; + } + + // 2. Drain transfer queue → deliver envelopes to actors + while let Some(envelope) = self.transfer_rx.try_recv() { + let dest = envelope.dest(); + let payload = envelope.into_payload(); + self.pool.deliver(&dest, payload); + did_work = true; + } + + // 3. Tick all actors with WorkerContext + let pending_local: RefCell)>> = + RefCell::new(Vec::new()); + + { + let worker_ctx = WorkerContext { + worker_id: self.id, + address_map, + transfer_txs, + spawn_txs, + placement, + inbox_registry, + pending_local: &pending_local, + }; + if self.pool.tick_all(&worker_ctx) { + did_work = true; + } + } + + // 4. Drain pending_local buffer → deliver to local actors + let pending = pending_local.into_inner(); + if !pending.is_empty() { + did_work = true; + } + for (addr, msg) in pending { + self.pool.deliver(&addr, msg); + } + + did_work + } +} + +/// The `ContextInner` impl for worker threads. +/// +/// Same-worker sends are buffered in `pending_local` (delivered after current tick round). +/// Cross-worker sends go through the transfer queue. +struct WorkerContext<'a> { + worker_id: WorkerId, + address_map: &'a AddressMap, + transfer_txs: &'a [Sender], + spawn_txs: &'a [Sender<(ActorAddress, Box)>], + placement: &'a Placement, + inbox_registry: &'a InboxRegistry, + pending_local: &'a RefCell)>>, +} + +impl ContextInner for WorkerContext<'_> { + fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error> { + match self.address_map.lookup(&addr) { + Some(wid) if wid == self.worker_id => { + // Same worker: buffer for local delivery (after current tick round) + self.pending_local.borrow_mut().push((addr, msg)); + Ok(()) + } + Some(wid) => { + // Cross worker: envelope through transfer queue + let envelope = Envelope::new(addr, msg); + let _ = self.transfer_txs[wid.as_usize()].try_send(envelope); + Ok(()) + } + None => { + // Try inbox registry (external inboxes) + self.inbox_registry.try_deliver(addr, msg) + } + } + } + + fn spawn_any(&self, addr: ActorAddress, actor: Box) -> Result<(), Error> { + let worker_id = self.placement.next_worker(); + self.address_map.insert(addr, worker_id); + self.spawn_txs[worker_id.as_usize()] + .try_send((addr, actor)) + .map_err(|_| Error::from("Spawn queue full")) + } +} diff --git a/src/worker/pool.rs b/src/worker/pool.rs new file mode 100644 index 0000000..95c7321 --- /dev/null +++ b/src/worker/pool.rs @@ -0,0 +1,51 @@ +use std::any::Any; +use std::collections::HashMap; + +use crate::actor::{ActorAddress, AnyActor}; +use crate::context::ContextInner; + +/// Per-worker actor storage. +pub(crate) struct ActorPool { + actors: HashMap>, +} + +impl ActorPool { + pub fn new() -> Self { + Self { + actors: HashMap::new(), + } + } + + pub fn insert(&mut self, addr: ActorAddress, actor: Box) { + self.actors.insert(addr, actor); + } + + pub fn remove(&mut self, addr: &ActorAddress) -> Option> { + self.actors.remove(addr) + } + + /// Deliver a type-erased message to the actor at `addr`. + /// Returns `true` if the actor was found and the message type matched. + pub fn deliver(&mut self, addr: &ActorAddress, msg: Box) -> bool { + if let Some(actor) = self.actors.get_mut(addr) { + actor.deliver(msg) + } else { + false + } + } + + /// Tick all actors in the pool. Returns `true` if any actor processed messages. + pub fn tick_all(&mut self, inner: &dyn ContextInner) -> bool { + let mut did_work = false; + for actor in self.actors.values_mut() { + if actor.tick(inner) { + did_work = true; + } + } + did_work + } + + pub fn len(&self) -> usize { + self.actors.len() + } +} diff --git a/tests/runtime_tests.rs b/tests/runtime_tests.rs index 56ca321..d57a514 100644 --- a/tests/runtime_tests.rs +++ b/tests/runtime_tests.rs @@ -1,4 +1,4 @@ -use swactor::{actor::{ActorAddress, ActorInterface}, runtime::{Inbox, Runtime, RuntimeConfig}}; +use swactor::{Ctx, actor::{ActorAddress, ActorInterface}, runtime::{Inbox, Runtime, RuntimeConfig}}; #[derive(Clone)] struct PingMessage { @@ -14,8 +14,8 @@ impl ActorInterface for PongActor { type Incoming = PingMessage; type Response = PongMessage; - fn handle(&mut self, ctx: &Runtime, msg: PingMessage) { - let _ = ctx.send_to(msg.reply_to, PongMessage); + fn handle(&mut self, ctx: &Ctx, msg: PingMessage) { + let _ = ctx.send(msg.reply_to, PongMessage); } } @@ -31,8 +31,8 @@ impl ActorInterface for ForwarderActor { type Incoming = ForwardMessage; type Response = (); - fn handle(&mut self, ctx: &Runtime, msg: ForwardMessage) { - let _ = ctx.send_to(self.target, msg); + fn handle(&mut self, ctx: &Ctx, msg: ForwardMessage) { + let _ = ctx.send(self.target, msg); } } diff --git a/tests/stress/concurrency.rs b/tests/stress/concurrency.rs index 2511745..acb3cf4 100644 --- a/tests/stress/concurrency.rs +++ b/tests/stress/concurrency.rs @@ -25,7 +25,6 @@ fn shutdown_under_load() { let result = std::panic::catch_unwind(|| { let config = RuntimeConfig { max_actors: 100, - router_max_messages: 10_000, actor_max_messages: 1000, num_threads: 4, }; @@ -100,7 +99,6 @@ fn send_to_newborn() { for _ in 0..100 { let config = RuntimeConfig { max_actors: 1000, - router_max_messages: 10_000, actor_max_messages: 100, num_threads: 4, }; @@ -149,7 +147,6 @@ fn rapid_spawn_churn() { let config = RuntimeConfig { max_actors: 100, - router_max_messages: 10_000, actor_max_messages: 100, num_threads: 4, }; @@ -210,8 +207,7 @@ fn inbox_contention() { let config = RuntimeConfig { max_actors: 10, - router_max_messages: 100_000, - actor_max_messages: 10_000, + actor_max_messages: 100_000, num_threads: 4, }; let runtime = Runtime::new(config); @@ -284,7 +280,6 @@ fn shutdown_timing_fuzz() { let result = std::panic::catch_unwind(|| { let config = RuntimeConfig { max_actors: 50, - router_max_messages: 1000, actor_max_messages: 100, num_threads: 4, }; diff --git a/tests/stress/mod.rs b/tests/stress/mod.rs index be625c4..87cfdd7 100644 --- a/tests/stress/mod.rs +++ b/tests/stress/mod.rs @@ -165,7 +165,7 @@ impl Stress { } // Test actors used across stress tests -use swactor::{actor::ActorInterface, runtime::Runtime}; +use swactor::{Ctx, actor::ActorInterface}; /// An actor that just absorbs messages pub struct BlackHole; @@ -176,7 +176,7 @@ pub struct Msg; impl ActorInterface for BlackHole { type Incoming = Msg; type Response = (); - fn handle(&mut self, _ctx: &Runtime, _msg: Msg) {} + fn handle(&mut self, _ctx: &Ctx, _msg: Msg) {} } /// An actor that counts messages received @@ -193,7 +193,7 @@ impl Counter { impl ActorInterface for Counter { type Incoming = Msg; type Response = (); - fn handle(&mut self, _ctx: &Runtime, _msg: Msg) { + fn handle(&mut self, _ctx: &Ctx, _msg: Msg) { self.count += 1; } } diff --git a/tests/stress/saturation.rs b/tests/stress/saturation.rs index a8fc97e..9345cd2 100644 --- a/tests/stress/saturation.rs +++ b/tests/stress/saturation.rs @@ -6,23 +6,25 @@ use super::{BlackHole, Counter, Msg, Stress, StressResult}; use std::time::Duration; use swactor::runtime::{Runtime, RuntimeConfig}; -/// Blast the router inbox +/// Blast the transfer queue (replaces router_inbox_overflow) #[test] #[cfg(feature = "stress")] -fn router_inbox_overflow() { - println!("\n>>> STRESS: Router Inbox Overflow"); +fn transfer_queue_overflow() { + println!("\n>>> STRESS: Transfer Queue Overflow"); let config = RuntimeConfig { max_actors: 10, - router_max_messages: 100, // Tiny buffer - actor_max_messages: 1000, + actor_max_messages: 100, // Tiny buffer num_threads: 1, }; let runtime = Runtime::new(config); let sink = runtime.spawn(BlackHole).unwrap(); + // Process spawn + runtime.tick(); + // Blast messages without processing - let mut result = StressResult::new("router_inbox_overflow"); + let mut result = StressResult::new("transfer_queue_overflow"); let start = std::time::Instant::now(); for _ in 0..10_000 { @@ -35,15 +37,15 @@ fn router_inbox_overflow() { } result.duration = start.elapsed(); - result.note(format!("Router buffer: 100, Messages sent: 10,000")); + result.note(format!("Transfer buffer: 100, Messages sent: 10,000")); - // With hybrid, no failures expected + // With hybrid channel, no failures expected assert_eq!(result.failures, 0, "Hybrid channel should not reject"); result.print(); - println!(">>> PASS: Hybrid channel prevented router overflow\n"); + println!(">>> PASS: Hybrid channel prevented transfer queue overflow\n"); } -/// Blast a single actor's inbox +/// Blast a single actor's mailbox via transfer queue #[test] #[cfg(feature = "stress")] fn actor_inbox_overflow() { @@ -51,26 +53,25 @@ fn actor_inbox_overflow() { let config = RuntimeConfig { max_actors: 10, - router_max_messages: 100_000, // Large router buffer - actor_max_messages: 100, // Tiny actor inbox + actor_max_messages: 100_000, // Large transfer buffer num_threads: 1, }; let runtime = Runtime::new(config); let sink = runtime.spawn(Counter::new()).unwrap(); - // Process router registration + // Process spawn runtime.tick(); - // Now blast messages - router will accept them but actor inbox will fill + // Now blast messages let mut sent = 0u64; - let mut router_failed = 0u64; + let mut failed = 0u64; for _ in 0..10_000 { if runtime.send_to::(sink, Msg).is_ok() { sent += 1; } else { - router_failed += 1; + failed += 1; } - // Tick occasionally to let router deliver + // Tick occasionally to let worker deliver if sent % 100 == 0 { runtime.tick(); } @@ -81,10 +82,10 @@ fn actor_inbox_overflow() { runtime.tick(); } - println!(" Router accepted: {}", sent); - println!(" Router rejected: {}", router_failed); + println!(" Sent: {}", sent); + println!(" Failed: {}", failed); - assert_eq!(router_failed, 0, "Router rejected message under load"); + assert_eq!(failed, 0, "Transfer queue rejected message under load"); println!(">>> PASS: No message loss with hybrid channel\n"); } @@ -96,7 +97,6 @@ fn actor_queue_overflow() { let config = RuntimeConfig { max_actors: 100, // Small actor queue - router_max_messages: 10_000, actor_max_messages: 100, num_threads: 1, }; @@ -118,12 +118,12 @@ fn actor_queue_overflow() { result.note(format!("Queue capacity: 100, Spawn attempts: 500")); result.print(); - // Note: Router also takes a slot, so we expect ~99 actors max + // Note: With hybrid channel (overflow to SegQueue), we expect no failures assert_eq!( result.failures, 0, "Spawned more actors than queue capacity" ); - println!(">>> PASS: Actor queue correctly rejects when full\n"); + println!(">>> PASS: Actor queue correctly handles overflow\n"); } /// FIXME: IS this actually testing what it should be? @@ -136,8 +136,7 @@ fn sustained_overload() { let config = RuntimeConfig { max_actors: 100, - router_max_messages: 1000, - actor_max_messages: 100, + actor_max_messages: 1000, num_threads: 1, }; let runtime = Runtime::new(config); @@ -150,7 +149,7 @@ fn sustained_overload() { } } - // Process registrations + // Process spawn registrations for _ in 0..200 { runtime.tick(); }