diff --git a/benches/scaling.rs b/benches/scaling.rs index 33d7b23..0febea5 100644 --- a/benches/scaling.rs +++ b/benches/scaling.rs @@ -8,9 +8,8 @@ use crate::harness::{black_box, Bench, BenchSuite}; use std::thread; use swactor::{ - Ctx, actor::ActorInterface, - runtime::{Runtime, RuntimeConfig}, + runtime::{Ctx, Runtime, RuntimeConfig}, }; // ============================================================================ diff --git a/benches/throughput.rs b/benches/throughput.rs index eaed827..900a4d0 100644 --- a/benches/throughput.rs +++ b/benches/throughput.rs @@ -8,9 +8,8 @@ use crate::harness::{black_box, Bench, BenchSuite}; use swactor::{ - Ctx, actor::{ActorAddress, ActorInterface}, - runtime::{Runtime, RuntimeConfig}, + runtime::{Ctx, Runtime, RuntimeConfig}, }; // ============================================================================ diff --git a/examples/hello.rs b/examples/hello.rs index 18378d9..7f9b483 100644 --- a/examples/hello.rs +++ b/examples/hello.rs @@ -1,7 +1,6 @@ use swactor::{ - Ctx, actor::{ActorAddress, ActorInterface}, - runtime::{Runtime, RuntimeConfig}, + runtime::{Ctx, Runtime, RuntimeConfig}, }; #[derive(Debug, Default)] diff --git a/examples/ring.rs b/examples/ring.rs index 703a3be..dd4e87b 100644 --- a/examples/ring.rs +++ b/examples/ring.rs @@ -1,7 +1,6 @@ use swactor::{ - Ctx, actor::{ActorAddress, ActorInterface}, - runtime::{Inbox, Runtime, RuntimeConfig}, + runtime::{Ctx, Inbox, Runtime, RuntimeConfig}, }; #[derive(Debug, Default, Clone)] diff --git a/src/actor.rs b/src/actor.rs index e14cf82..a5ae92a 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -1,7 +1,6 @@ use std::any::Any; -use crate::{get_random, worker::mailbox::Mailbox}; -use crate::context::{ContextInner, Ctx}; +use crate::{get_random, runtime::{ContextInner, Ctx}, worker::Mailbox}; /// The primary trait defining data that can be passed to and from actor processes pub trait Message: 'static + Sized + Clone + Send + Sync {} diff --git a/src/address_map.rs b/src/address_map.rs index adabe3a..48ac949 100644 --- a/src/address_map.rs +++ b/src/address_map.rs @@ -1,5 +1,6 @@ use std::collections::HashMap; use std::sync::RwLock; +use std::sync::atomic::{AtomicUsize, Ordering}; use crate::actor::ActorAddress; @@ -50,6 +51,26 @@ impl AddressMap { } } +/// 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::*; @@ -87,4 +108,13 @@ mod tests { map.insert(addr1, WorkerId(0)); assert_eq!(map.len(), 1); } + + #[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/channel/mod.rs b/src/channel.rs similarity index 77% rename from src/channel/mod.rs rename to src/channel.rs index fbccdb5..c487ec4 100644 --- a/src/channel/mod.rs +++ b/src/channel.rs @@ -1,13 +1,11 @@ use std::sync::Arc; -use std::sync::atomic::{AtomicUsize, Ordering}; use crossbeam_queue::{ArrayQueue, SegQueue}; pub struct HybridChannel { ring: ArrayQueue, overflow: SegQueue, - overflow_len: AtomicUsize, } impl HybridChannel { @@ -15,7 +13,6 @@ impl HybridChannel { Self { ring: ArrayQueue::new(capacity), overflow: SegQueue::new(), - overflow_len: AtomicUsize::new(0), } } @@ -24,7 +21,6 @@ impl HybridChannel { Ok(()) => Ok(()), Err(v) => { self.overflow.push(v); - self.overflow_len.fetch_add(1, Ordering::Relaxed); Ok(()) } } @@ -37,16 +33,12 @@ impl HybridChannel { match self.overflow.pop() { Some(value) => { - self.overflow_len.fetch_sub(1, Ordering::Relaxed); Some(value) } None => None, } } - pub fn len(&self) -> usize { - self.ring.len() + self.overflow_len.load(Ordering::Relaxed) - } } pub(crate) struct Receiver { @@ -59,11 +51,6 @@ impl Receiver { Self { queue } } - - pub fn len(&self) -> usize { - self.queue.len() - } - pub fn try_recv(&self) -> Option { return self.queue.pop(); } diff --git a/src/context.rs b/src/context.rs deleted file mode 100644 index f5e5bb6..0000000 --- a/src/context.rs +++ /dev/null @@ -1,47 +0,0 @@ -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>; - fn mailbox_waterlevel(&self) -> usize; -} - -/// 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 waterlevel = self.inner.mailbox_waterlevel(); - let actor = Actor::new(addr, Mailbox::new(waterlevel), 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 index d55fb8a..c3a716d 100644 --- a/src/envelope.rs +++ b/src/envelope.rs @@ -2,31 +2,6 @@ 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 { diff --git a/src/lib.rs b/src/lib.rs index 08f7b07..fc57124 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,12 +5,8 @@ pub(crate) mod channel; pub(crate) mod error; pub use error::Error; -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; diff --git a/src/placement.rs b/src/placement.rs deleted file mode 100644 index d88e63d..0000000 --- a/src/placement.rs +++ /dev/null @@ -1,37 +0,0 @@ -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/runtime.rs b/src/runtime.rs index 8c2e7ab..622c0db 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -6,14 +6,11 @@ use std::sync::{Arc, RwLock}; use std::thread::{self, JoinHandle}; use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}; -use crate::address_map::{AddressMap, WorkerId}; +use crate::address_map::{AddressMap, Placement, WorkerId}; use crate::channel::{Receiver, Sender}; // Re-export config types so existing code using `runtime::RuntimeConfig` still works pub use crate::config::{BackoffPolicy, RuntimeConfig}; -use crate::context::ContextInner; -use crate::envelope::Envelope; -use crate::placement::Placement; -use crate::worker::mailbox::Mailbox; +use crate::worker::Mailbox; use crate::worker::{TickContext, Worker}; use crate::Error; @@ -32,6 +29,34 @@ impl SenderT for Sender { } } + +/// 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 + } +} + + // ─── InboxRegistry ─────────────────────────────────────────────────────────── /// Registry of external inboxes — replaces the Router's role for non-actor receivers. @@ -83,6 +108,50 @@ impl Inbox { } } + +/// 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>; + fn mailbox_waterlevel(&self) -> usize; +} + +/// 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 waterlevel = self.inner.mailbox_waterlevel(); + let actor = Actor::new(addr, Mailbox::new(waterlevel), actor); + let boxed: Box = Box::new(actor); + self.inner.spawn_any(addr, boxed)?; + Ok(addr) + } +} + + // ─── Runtime ───────────────────────────────────────────────────────────────── /// The `Runtime` struct is the primary gateway for interacting with the framework. diff --git a/src/worker/mod.rs b/src/worker.rs similarity index 66% rename from src/worker/mod.rs rename to src/worker.rs index f8752a8..c6d0e3c 100644 --- a/src/worker/mod.rs +++ b/src/worker.rs @@ -1,19 +1,13 @@ -pub mod mailbox; -pub(crate) mod pool; - use std::any::Any; use std::cell::RefCell; +use std::collections::{HashMap, VecDeque}; -use crate::actor::{ActorAddress, AnyActor}; -use crate::address_map::{AddressMap, WorkerId}; +use crate::actor::{ActorAddress, AnyActor, Message}; +use crate::address_map::{AddressMap, Placement, WorkerId}; use crate::channel::{Receiver, Sender}; use crate::config::RuntimeConfig; -use crate::context::ContextInner; -use crate::envelope::Envelope; -use crate::placement::Placement; -use crate::runtime::InboxRegistry; +use crate::runtime::{ContextInner, Envelope, InboxRegistry}; use crate::Error; -use pool::ActorPool; /// Shared state passed to tick_once — single thin pointer avoids register spill. pub(crate) struct TickContext<'a> { @@ -146,3 +140,93 @@ impl ContextInner for WorkerContext<'_> { self.config.mailbox_waterlevel } } + +/// 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() + } +} + + + +pub struct Mailbox { + queue: VecDeque, + waterlevel: usize, +} + +impl Mailbox { + pub fn new(waterlevel: usize) -> Self { + Self { + queue: VecDeque::new(), + waterlevel, + } + } + + pub fn push(&mut self, msg: M) { + self.queue.push_back(msg); + } + + pub fn pop(&mut self) -> Option { + self.queue.pop_front() + } + + pub fn len(&self) -> usize { + self.queue.len() + } + + pub fn is_empty(&self) -> bool { + self.queue.is_empty() + } + + /// How many messages to process this tick: + /// - `len < waterlevel` → process all (`len`) + /// - `len >= waterlevel` → process half (`len >> 1`) + pub fn drain_count(&self) -> usize { + let len = self.queue.len(); + if len < self.waterlevel { + len + } else { + len >> 1 + } + } +} diff --git a/src/worker/mailbox.rs b/src/worker/mailbox.rs deleted file mode 100644 index a331e22..0000000 --- a/src/worker/mailbox.rs +++ /dev/null @@ -1,45 +0,0 @@ -use std::collections::VecDeque; - -use crate::actor::Message; - -pub struct Mailbox { - queue: VecDeque, - waterlevel: usize, -} - -impl Mailbox { - pub fn new(waterlevel: usize) -> Self { - Self { - queue: VecDeque::new(), - waterlevel, - } - } - - pub fn push(&mut self, msg: M) { - self.queue.push_back(msg); - } - - pub fn pop(&mut self) -> Option { - self.queue.pop_front() - } - - pub fn len(&self) -> usize { - self.queue.len() - } - - pub fn is_empty(&self) -> bool { - self.queue.is_empty() - } - - /// How many messages to process this tick: - /// - `len < waterlevel` → process all (`len`) - /// - `len >= waterlevel` → process half (`len >> 1`) - pub fn drain_count(&self) -> usize { - let len = self.queue.len(); - if len < self.waterlevel { - len - } else { - len >> 1 - } - } -} diff --git a/src/worker/pool.rs b/src/worker/pool.rs deleted file mode 100644 index 95c7321..0000000 --- a/src/worker/pool.rs +++ /dev/null @@ -1,51 +0,0 @@ -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/mailbox_tests.rs b/tests/mailbox_tests.rs index 2deebba..66a671e 100644 --- a/tests/mailbox_tests.rs +++ b/tests/mailbox_tests.rs @@ -1,4 +1,4 @@ -use swactor::worker::mailbox::Mailbox; +use swactor::worker::Mailbox; // ── Basic operations ── diff --git a/tests/runtime_tests.rs b/tests/runtime_tests.rs index d57a514..ed21044 100644 --- a/tests/runtime_tests.rs +++ b/tests/runtime_tests.rs @@ -1,4 +1,4 @@ -use swactor::{Ctx, actor::{ActorAddress, ActorInterface}, runtime::{Inbox, Runtime, RuntimeConfig}}; +use swactor::{actor::{ActorAddress, ActorInterface}, runtime::{Ctx, Inbox, Runtime, RuntimeConfig}}; #[derive(Clone)] struct PingMessage { diff --git a/tests/stress/mod.rs b/tests/stress/mod.rs index 87cfdd7..2839ac9 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::{Ctx, actor::ActorInterface}; +use swactor::{actor::ActorInterface, runtime::Ctx}; /// An actor that just absorbs messages pub struct BlackHole;