diff --git a/DESIGN.md b/DESIGN.md index 1a50043..7870a2e 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -74,4 +74,198 @@ impl HybridChannel { } } } -``` \ No newline at end of file +``` + +### Kimi Suggestions + +IMPROVEMENTS FITTING DESIGN GOALS +Here are improvements aligned with the stated goals: "maximum usability and speed while keeping line count low" and "no footguns." +Priority 1: Critical Bug Fixes & MVP Completion +1. Fix Runtime Constructor (~5 lines) + - Implement Runtime::new() + - Implement Builder::build() + - Fix examples to compile +2. Handle Full Inboxes Gracefully (~15 lines) + - Return Result<(), Error> from send_to on full inbox + - Provide backpressure signal instead of silent drop + - Add try_send() vs send() distinction +3. Implement Multithreaded Runtime (~30-40 lines) + - Complete threading infrastructure (already partially designed) + - Router runs in separate thread + - Actor processing pool with work-stealing (simple round-robin first) +Priority 2: Usability Enhancements (Low Line Count) +4. Ergonomic Macros (~20-30 lines procedural macro crate) + #[derive(Actor)] + struct MyActor { ... } + - Auto-impl ActorInterface for simple cases + - Reduces boilerplate significantly +5. Request/Response Helper (~15 lines) + let resp: Response = rt.request(addr, msg).await?; + - Common pattern many users need + - Maintains simplicity +6. Inbox Capacity Configuration (~5 lines) + - Per-actor capacity instead of global constant + - Allow spawn_with_capacity() +Priority 3: Performance Optimizations +7. Sharded Router (~30-40 lines) + - Multiple HashMaps based on address hash + - Reduces contention on messaging hot path + - Maintains O(1) lookups +8. Actor Work Stealing (~40-50 lines) + - Multiple actor queues instead of single global queue + - Threads steal work when idle + - Improves cache locality +9. Hybrid Channel (from DESIGN.md) (~25 lines) + - Implements the overflow mechanism described + - Ring buffer + Mutex for emergencies + - Prevent message loss under burst loads +10. Actor State Colocation (~15 lines) + - Group related actors by affinity + - Optional "actor system" or "node" concept + - Better cache locality +Priority 4: Observability (Minimal Overhead) +11. Lightweight Metrics (~15-20 lines) + - Message counts per actor (atomic counters) + - Overflow/drop tracking + - Optional, compile-time feature flag +12. Message Tracing (~10-15 lines opt-in) + - Optional trace ID in envelope + - Zero-cost when disabled (feature flag) +Priority 5: Reliability Patterns +13. Bounded Channels with Overflow (~20 lines) + - Implement HybridChannel from design doc + - Graceful degradation under load +14. Watchdog Timer Pattern (~15 lines example) + - Show pattern: actor checking heartbeats + - Keep library simple, document patterns +--- +SPECIFIC CODE IMPROVEMENTS +Fix Silent Failures (Priority: CRITICAL) +Current (src/runtime.rs:85-93): +pub fn send_to(&self, addr: ActorAddress, msg: M) -> Result<(), ()> { + let envelope: Envelope = Box::new(msg); + self.router_inbox + .try_send(RouterMessage::SendToAddr { addr, msg: envelope }) + .map_err(|_| ()) +} +Improved: +pub fn try_send(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { + let envelope: Envelope = Box::new(msg); + self.router_inbox + .try_send(RouterMessage::SendToAddr { addr, msg: envelope }) + .map_err(|_| Error::from("Router inbox full")) +} +// Add send that blocks/resizes +pub fn send(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { ... } +Implement HybridChannel (Priority: HIGH) +From design doc, add to ring_buffer.rs: +pub struct HybridChannel { + ring: ArrayQueue, + overflow: Mutex>, + overflow_count: AtomicUsize, +} +impl HybridChannel { + fn push(&self, value: T) -> Result<(), T> { + if self.ring.push(value).is_err() { + self.overflow.lock().push_back(value); + self.overflow_count.fetch_add(1, Relaxed); + // Optionally resize ring if overflow_count > threshold + } + Ok(()) + } +} +Fix Runtime Construction (Priority: CRITICAL) +Current: Runtime::new() doesn't exist but examples use it. +Add to runtime.rs: +impl Runtime { + pub fn new(capacity: usize, flavor: Option) -> Self { + let router = Router::new(capacity); + let router_inbox = router.new_sender(); + + Self { + flavor: flavor.unwrap_or_default(), + router, + router_inbox, + actor_queue: ArrayQueue::new(capacity), + thread_pool: Vec::new(), + } + } +} +Add Sharded Router (Priority: MEDIUM) +Current: Single HashMap for all addresses +Improved: N HashMaps based on address modulo +pub(crate) struct Router { + shards: Vec>>, + shard_mask: usize, // shards.len() - 1 (power of 2) + inbox: Receiver, +} +impl Router { + fn get_shard(&self, addr: ActorAddress) -> &HashMap<...> { + &self.shards[(addr as usize) & self.shard_mask] + } +} +--- +RECOMMENDED ROADMAP +Phase 1: Bug Fixes & MVP (1-2 days) +1. Implement Runtime::new() +2. Implement Builder::build() +3. Fix compilation errors +4. Add error handling for full inboxes +5. Document API +Phase 2: Single-Threaded Polish (1 week) +1. Ergonomic macros +2. Request/response helpers +3. Inbox capacity configuration +4. Example improvements +5. Basic tests +Phase 3: Multi-Threaded (2 weeks) +1. Implement threaded runtime +2. Worker thread pool +3. Router in separate thread +4. Work-stealing queues +5. Performance benchmarks +Phase 4: Production Hardening (2 weeks) +1. Sharded router +2. Hybrid channels +3. Metrics (opt-in) +4. Message tracing (opt-in) +5. Stress testing +Phase 5: Documentation & Patterns (1 week) +1. Actor patterns guide +2. Performance tuning guide +3. WASM integration examples +4. Best practices documentation +--- +ALTERNATIVE ARCHITECTURES TO CONSIDER +For Even Smaller Line Count +If the goal is absolutely minimal code, consider: +- Single-threaded only: Remove multi-threading complexity entirely +- No router: Direct mpsc channels between actors (more Erlang-like) +- Simpler scheduling: Round-robin over actors array +Tradeoff: Less flexible, but potentially <200 lines total. +For Better Performance +If performance outweighs minimalism: +- Lock-free HashMap: Use dashmap or equivalent for router +- SegQueue: Better for work-stealing than ArrayQueue +- Pre-allocated: Fixed-size actor pool with object pool pattern +- Lock-free message passing: Use crossbeam or tokio channels throughout +Tradeoff: More dependencies, larger binary size. +For Better Ergonomics +If usability is primary goal: +- Async/Await native: Integrate with tokio or async-std +- Actor supervision: Basic supervisor trees (small implementation) +- Message DSL: Macro for pattern-matching message handlers +Tradeoff: Increases complexity substantially, may conflict with "small" goal. +--- +CONCLUSION +swactor has a solid, minimalist architecture that delivers on its core promise: a small, WASM-compatible actor library. The design is clean, modular, and avoids unnecessary complexity. +Current Grade: C+ (Incomplete MVP) +- Architecture: B+ +- Ease of Use: D (examples don't compile, silent failures) +- Performance: B (good primitives but scalability concerns) +Potential Grade with improvements: A- +- Fixing critical bugs would make it immediately usable +- Sharded router + work-stealing would address scalability +- Ergonomic macros would dramatically improve UX +- Hybrid channels would solve burst-load scenarios +Recommendation: Focus on completing Phase 1 (bug fixes) and Phase 2 (usability). The architecture is sound—it's just incomplete. Avoid premature optimization; measure performance first, then implement sharding/work-stealing if benchmarks show contention. \ No newline at end of file diff --git a/examples/hello.rs b/examples/hello.rs index db46a0e..ef306e5 100644 --- a/examples/hello.rs +++ b/examples/hello.rs @@ -1,6 +1,6 @@ use swactor::{ actor::{ActorAddress, ActorInterface}, - runtime::{Context, Runtime, RuntimeFlavor}, + runtime::{Runtime, RuntimeFlavor}, }; #[derive(Debug, Default)] @@ -24,7 +24,7 @@ impl ActorInterface for Greeter { type Incoming = GreetMessage; type Response = GreetResponse; - fn handle(&mut self, ctx: &Context, msg: GreetMessage) { + fn handle(&mut self, ctx: &Runtime, msg: GreetMessage) { let res = GreetResponse(format!("Hello, {}!", msg.who)); self.num_greeted += 1; if let Err(_) = ctx.send_to(msg.return_addr, res) { @@ -35,7 +35,7 @@ impl ActorInterface for Greeter { } fn main() { - let rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); + let mut rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); let addr = rt .spawn(Greeter::default()) .expect("failed to spawn greeter"); diff --git a/examples/ring.rs b/examples/ring.rs index 313b779..cc3af6e 100644 --- a/examples/ring.rs +++ b/examples/ring.rs @@ -1,6 +1,6 @@ use swactor::{ actor::{ActorAddress, ActorInterface}, - runtime::{Context, Inbox, Runtime, RuntimeFlavor}, + runtime::{Inbox, Runtime, RuntimeFlavor}, }; #[derive(Debug, Default, Clone)] @@ -30,7 +30,7 @@ impl RingActor { impl ActorInterface for RingActor { type Incoming = RingMessage; type Response = (); - fn handle(&mut self, ctx: &Context, msg: Self::Incoming) { + fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming) { if let Err(_) = ctx.send_to(self.next, msg.next()) { // do nothing } @@ -38,7 +38,7 @@ impl ActorInterface for RingActor { } fn main() { - let rt = Runtime::new(10_000, RuntimeFlavor::SingleThreaded); + let mut rt = Runtime::new(10_000, RuntimeFlavor::SingleThreaded); let inbox: Inbox = rt.new_inbox(); let mut next = rt diff --git a/src/actor.rs b/src/actor.rs index 890243c..f303ed2 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -1,4 +1,4 @@ -use crate::{ring_buffer::Receiver, runtime::Context, WATERLEVEL}; +use crate::{Runtime, WATERLEVEL, ring_buffer::Receiver}; pub trait Message: 'static + Sized + Clone + Send {} impl Message for T {} @@ -6,27 +6,22 @@ impl Message for T {} pub trait ActorInterface: 'static + Send { type Incoming: Message; type Response: Message; - fn handle(&mut self, ctx: &Context, msg: Self::Incoming); + fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming); } pub type ActorAddress = u64; -/// FIXME: If we never have the actor struct reference its own address, should -/// we even include it as a variable here? We could instead grab this information -/// from the runtime or router. pub struct Actor where A: ActorInterface, { - _addr: ActorAddress, inbox: Receiver, inner: A, } impl Actor { - pub(crate) fn new(addr: ActorAddress, inbox: Receiver, inner: A) -> Self { + pub(crate) fn new(inbox: Receiver, inner: A) -> Self { Self { - _addr: addr, inbox, inner, } @@ -35,14 +30,14 @@ impl Actor { /// Trait for type-erased actors pub(crate) trait AnyActor: Send { - fn tick(&mut self, ctx: &Context); + fn tick(&mut self, ctx: &Runtime); } impl AnyActor for Actor where A: ActorInterface, { - fn tick(&mut self, ctx: &Context) { + fn tick(&mut self, ctx: &Runtime) { let total_messages = self.inbox.len(); let messages_to_process = if total_messages < WATERLEVEL { total_messages diff --git a/src/lib.rs b/src/lib.rs index 849606a..a3d4244 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -9,7 +9,7 @@ pub mod runtime; // Re-export commonly used types pub use actor::{ActorAddress, ActorInterface, Message}; -pub use runtime::{Context, Inbox, Runtime, RuntimeFlavor}; +pub use runtime::{Inbox, Runtime, RuntimeFlavor}; #[cfg(feature = "getrandom")] pub(crate) fn get_random(buf: &mut [u8]) { @@ -17,7 +17,7 @@ pub(crate) fn get_random(buf: &mut [u8]) { } /// The strategy for message processing is such: -/// +/// /// ```ignore /// if total_messages < WATERLEVEL: /// process all diff --git a/src/ring_buffer.rs b/src/ring_buffer.rs index 5c78bf8..fea5113 100644 --- a/src/ring_buffer.rs +++ b/src/ring_buffer.rs @@ -48,17 +48,6 @@ pub(crate) struct Sender { queue: Arc>, } -/// FIXME: I don't like this. Why do we need to clone the Sender -/// Because there are no guarentees on the existence of the Receiver -/// we need to be very careful about passing around access to the buffer. -impl Clone for Sender { - fn clone(&self) -> Self { - Self { - queue: Arc::clone(&self.queue), - } - } -} - impl Sender { /// Attempt to push a value to the queue. Returns Err(value) if the queue is full. pub fn try_send(&self, value: T) -> Result<(), T> { diff --git a/src/runtime.rs b/src/runtime.rs index 7acff9d..d00e7b8 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,15 +1,15 @@ use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, OnceLock}; use std::thread::{self, JoinHandle}; use crossbeam_queue::ArrayQueue; use crate::{ + DEFAULT_INBOX_CAPACITY, Error, actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}, get_random, ring_buffer::{Receiver, Sender}, - router::{Envelope, Router, RouterMessage}, - Error, DEFAULT_INBOX_CAPACITY, + router::{Router, RouterMessage}, }; #[derive(Debug, Clone, Default)] @@ -36,36 +36,6 @@ impl Inbox { } } -/// A lightweight handle for sending messages to actors -/// This is what actors receive in their handle() method -#[derive(Clone)] -pub struct Context { - router_inbox: Sender, -} - -impl Context { - /// Send a message to an actor address - pub fn send_to(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { - let envelope: Envelope = Box::new(msg); - self.router_inbox - .try_send(RouterMessage::SendToAddr { - addr, - msg: envelope, - }) - .map_err(|_| Error::from("Failed to send message: router inbox full")) - } -} - -/// Shared runtime state, wrapped in Arc for thread sharing -/// FIXME: We check the `running` variable too often. Should -/// push it somewhere where it is infreqently checked, and instead -/// have shutdown logic for everything else. -struct RuntimeInner { - running: AtomicBool, - router_inbox: Sender, - actor_queue: ArrayQueue>, -} - /// Thread handles for the multithreaded runtime struct RuntimeHandles { workers: Vec>, @@ -74,19 +44,21 @@ struct RuntimeHandles { /// The main runtime for executing actors pub struct Runtime { - inner: Arc, + running: AtomicBool, + router_inbox: Sender, + actor_queue: ArrayQueue>, flavor: RuntimeFlavor, /// Router is only accessed from a single thread (either main or dedicated router thread) - /// - /// FIXME: If single threaded, why do we have a mutex + /// FIXME: The only state Router needs access to is a hashmap. If we find something + /// lockfree, we can remove this mutex router: Mutex, /// Thread handles, created lazily when run() is called - handles: Mutex>, + handles: OnceLock, } impl Runtime { /// Create a new runtime with given actor queue capacity and flavor - /// + /// /// FIXME: I don't like this interface. Maybe a builder or config pattern. I shouldnt /// have to read code to understand what these variable names are. pub fn new(capacity: usize, flavor: RuntimeFlavor) -> Self { @@ -94,30 +66,19 @@ impl Runtime { let router = Router::new(DEFAULT_INBOX_CAPACITY); let router_inbox = router.new_sender(); Self { - inner: Arc::new(RuntimeInner { - running: AtomicBool::new(false), - router_inbox, - actor_queue: ArrayQueue::new(capacity), - }), + running: AtomicBool::new(false), + router_inbox, + actor_queue: ArrayQueue::new(capacity), flavor, router: Mutex::new(router), - handles: Mutex::new(None), - } - } - - /// Get a context handle for sending messages - /// - /// FIXME: Do we need all this indirection? - pub fn context(&self) -> Context { - Context { - router_inbox: self.inner.router_inbox.clone(), + handles: OnceLock::new(), } } /// Spawn an actor, returns its address pub fn spawn(&self, actor: A) -> Result { // FIXME: Figure out what to do with this. - // Its distracting to include here, but not used elsewhere for now. + // Its distracting to include here, but not used elsewhere for now. let addr = { let mut bytes = u64::to_le_bytes(0); get_random(&mut bytes); @@ -128,13 +89,11 @@ impl Runtime { // Register the sender with the router let _ = self - .inner .router_inbox .try_send(RouterMessage::AddAddr(addr, Box::new(sender))); - self.inner - .actor_queue - .push(Box::new(Actor::new(addr, inbox, actor))) + self.actor_queue + .push(Box::new(Actor::new(inbox, actor))) .map_err(|_| Error::from("Runtime error: Failed to spawn actor."))?; Ok(addr) @@ -142,7 +101,12 @@ impl Runtime { /// Send a message to an actor address pub fn send_to(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { - self.context().send_to(addr, msg) + self.router_inbox + .try_send(RouterMessage::SendToAddr { + addr, + msg: Box::new(msg), + }) + .map_err(|_| "Failed to send message to router.".into()) } /// Create an external inbox for receiving messages outside actors @@ -156,7 +120,6 @@ impl Runtime { let sender = receiver.new_sender(); // Register the sender with the router let _ = self - .inner .router_inbox .try_send(RouterMessage::AddAddr(addr, Box::new(sender))); Inbox { @@ -167,140 +130,74 @@ impl Runtime { /// Process one actor tick + router messages /// Works in both single/multi mode (useful for testing and fine-grained control) - /// + /// /// FIXME: `tick` does not make sense in multithreaded context. Add a check to ensure single threaded - pub fn tick(&self) { - let ctx = self.context(); - - if let Some(mut actor) = self.inner.actor_queue.pop() { - actor.tick(&ctx); - let _ = self.inner.actor_queue.push(actor); + pub fn tick(&mut self) { + if let Some(mut actor) = self.actor_queue.pop() { + actor.tick(&self); + let _ = self.actor_queue.push(actor); } // In single-threaded mode, also tick the router if matches!(self.flavor, RuntimeFlavor::SingleThreaded) - && let Ok(mut router) = self.router.lock() { - router.tick(); + self.router.lock().expect("single threaded mutex lock").tick(); } } - /// Run the runtime (blocking) - /// - SingleThreaded: runs in current thread until shutdown - /// - Multithreaded: spawns workers + router thread, blocks until shutdown - pub fn run(&self) { - self.inner.running.store(true, Ordering::Release); - - match &self.flavor { - RuntimeFlavor::SingleThreaded => { - self.run_single_threaded(); - } - RuntimeFlavor::Multithreaded { workers } => { - self.run_multi_threaded(*workers); - } - } - } - - /// FIXME: I don't like this loop. We can send shutdown signals from the process - /// that calls the actor runtime instead. It also does not make sense to have this - /// around for single threaded runtimes (people can instead loop over `runtime.tick()`). - fn run_single_threaded(&self) { - let ctx = self.context(); - - while self.inner.running.load(Ordering::Relaxed) { - // Pop actor, tick it, push it back - if let Some(mut actor) = self.inner.actor_queue.pop() { - actor.tick(&ctx); - let _ = self.inner.actor_queue.push(actor); - } - - // Tick the router - if let Ok(mut router) = self.router.lock() { - router.tick(); - } - - thread::yield_now(); - } - } - - fn run_multi_threaded(&self, num_workers: usize) { - // Take ownership of router for the dedicated router thread - // FIXME: this is weird and concerning. Introduces a class of logic errors - // wherein we try and call a dummy router. - let router = { - let mut guard = self.router.lock().unwrap(); - std::mem::replace(&mut *guard, Router::new(1)) // placeholder + /// Run the runtime + /// - Multithreaded: spawns workers + router thread, returns an `Arc` pointer to the + /// `Runtime` struct. + pub fn run(self) -> Arc { + // FIXME: gate access + let num_workers = match self.flavor { + RuntimeFlavor::Multithreaded { workers } => workers, + _ => panic!("'run' method requires a multithreaded runtime") }; + let rt = Arc::new(self); // Spawn dedicated router thread - // FIXME: what is this, these variables are named terribly - let router_running = Arc::clone(&self.inner); let router_handle = { - let running = router_running; + let ctx = rt.clone(); thread::spawn(move || { - router_loop(router, running); + router_loop(ctx); }) }; // Spawn worker threads - let worker_handles: Vec<_> = (0..num_workers) - .map(|_| { - let inner = Arc::clone(&self.inner); - thread::spawn(move || { - worker_loop(inner); - }) - }) - .collect(); - - // Store handles - *self.handles.lock().unwrap() = Some(RuntimeHandles { - workers: worker_handles, - router_thread: Some(router_handle), - }); - - // Block until shutdown - wait for all threads to complete - self.wait_for_shutdown(); - } - - fn wait_for_shutdown(&self) { - // Wait for the running flag to be set to false, then join threads - while self.inner.running.load(Ordering::Relaxed) { - thread::yield_now(); + let mut worker_handles: Vec> = vec![]; + for _ in 0..num_workers { + let ctx = rt.clone(); + let handle = thread::spawn(move || { + worker_loop(ctx); + }); + worker_handles.push(handle); } - // Join all threads - let handles = self.handles.lock().unwrap().take(); - if let Some(h) = handles { - for worker in h.workers { - let _ = worker.join(); - } - if let Some(rt) = h.router_thread { - let _ = rt.join(); - } - } + rt.handles.set(RuntimeHandles { workers: worker_handles, router_thread: Some(router_handle) }); + + rt } + /// Signal all workers to stop pub fn shutdown(&self) { - self.inner.running.store(false, Ordering::Release); + self.running.store(false, Ordering::Release); } /// Check if runtime is still active pub fn is_running(&self) -> bool { - self.inner.running.load(Ordering::Acquire) + self.running.load(Ordering::Acquire) } } /// Worker thread loop - processes actors from the shared queue -fn worker_loop(inner: Arc) { - let ctx = Context { - router_inbox: inner.router_inbox.clone(), - }; +fn worker_loop(ctx: Arc) { - while inner.running.load(Ordering::Relaxed) { - if let Some(mut actor) = inner.actor_queue.pop() { + while ctx.running.load(Ordering::Relaxed) { + if let Some(mut actor) = ctx.actor_queue.pop() { actor.tick(&ctx); - let _ = inner.actor_queue.push(actor); + let _ = ctx.actor_queue.push(actor); } else { thread::yield_now(); } @@ -308,9 +205,9 @@ fn worker_loop(inner: Arc) { } /// Router thread loop - processes router messages -fn router_loop(mut router: Router, inner: Arc) { - while inner.running.load(Ordering::Relaxed) { - router.tick(); +fn router_loop(ctx: Arc) { + while ctx.running.load(Ordering::Relaxed) { + ctx.router.lock().expect("failed to lock router mutex").tick(); thread::yield_now(); } } diff --git a/tests/runtime_tests.rs b/tests/runtime_tests.rs index 57e9023..4047506 100644 --- a/tests/runtime_tests.rs +++ b/tests/runtime_tests.rs @@ -1,10 +1,9 @@ -use std::sync::Arc; use std::thread; use std::time::Duration; use swactor::{ actor::{ActorAddress, ActorInterface}, - runtime::{Context, Inbox, Runtime, RuntimeFlavor}, + runtime::{Inbox, Runtime, RuntimeFlavor}, }; // ============================================================================ @@ -25,7 +24,7 @@ impl ActorInterface for PongActor { type Incoming = PingMessage; type Response = PongMessage; - fn handle(&mut self, ctx: &Context, msg: PingMessage) { + fn handle(&mut self, ctx: &Runtime, msg: PingMessage) { let _ = ctx.send_to(msg.reply_to, PongMessage); } } @@ -42,14 +41,14 @@ impl ActorInterface for ForwarderActor { type Incoming = ForwardMessage; type Response = (); - fn handle(&mut self, ctx: &Context, msg: ForwardMessage) { + fn handle(&mut self, ctx: &Runtime, msg: ForwardMessage) { let _ = ctx.send_to(self.target, msg); } } #[test] fn test_single_threaded_ping_pong() { - let rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); + let mut rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); let inbox: Inbox = rt.new_inbox(); let pong_addr = rt.spawn(PongActor).expect("spawn pong"); @@ -76,7 +75,7 @@ fn test_single_threaded_ping_pong() { #[test] fn test_single_threaded_message_chain() { - let rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); + let mut rt = Runtime::new(100, RuntimeFlavor::SingleThreaded); let inbox: Inbox = rt.new_inbox(); // Create a chain: A -> B -> C -> inbox @@ -105,10 +104,7 @@ fn test_single_threaded_message_chain() { #[test] fn test_multithreaded_message_passing() { - let rt = Arc::new(Runtime::new( - 1000, - RuntimeFlavor::Multithreaded { workers: 4 }, - )); + let rt = Runtime::new(1000, RuntimeFlavor::Multithreaded { workers: 4 }); let inbox: Inbox = rt.new_inbox(); // Create a longer chain to exercise multi-threading @@ -123,21 +119,19 @@ fn test_multithreaded_message_passing() { rt.send_to(start_addr, ForwardMessage(999)).unwrap(); // Spawn thread to check for result and shutdown - let rt_clone = Arc::clone(&rt); + let ctx = rt.run(); let inbox_check = thread::spawn(move || { for _ in 0..100 { thread::sleep(Duration::from_millis(10)); if let Some(ForwardMessage(val)) = inbox.try_recv() { - rt_clone.shutdown(); + ctx.shutdown(); return Some(val); } } - rt_clone.shutdown(); + ctx.shutdown(); None }); - rt.run(); - let result = inbox_check.join().unwrap(); assert_eq!(result, Some(999)); }