From 98a2733173d8176187cafdd2e21cd563ae3f844f Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Sun, 25 Jan 2026 20:31:34 +0700 Subject: [PATCH] feat: multithreaded runtime Runtime is now configurable. Adds a tunable config for modifying the size of pre-allocations for actor messaging channels, and for selecting the number of threads the runtime will use. --- examples/hello.rs | 14 +- examples/ring.rs | 13 +- src/actor.rs | 58 ++++++- src/error.rs | 14 ++ src/lib.rs | 6 - src/ring_buffer.rs | 5 +- src/router.rs | 13 +- src/runtime.rs | 332 ++++++++++++++++++++++------------------- tests/runtime_tests.rs | 81 +++++----- 9 files changed, 311 insertions(+), 225 deletions(-) diff --git a/examples/hello.rs b/examples/hello.rs index ec73da4..d67c439 100644 --- a/examples/hello.rs +++ b/examples/hello.rs @@ -1,6 +1,6 @@ use swactor::{ actor::{ActorAddress, ActorInterface}, - runtime::{Runtime, RuntimeFlavor}, + runtime::{Runtime, RuntimeConfig}, }; #[derive(Debug, Default)] @@ -35,12 +35,17 @@ impl ActorInterface for Greeter { } fn main() { - let mut rt = Runtime::new(); + let rt = Runtime::new(RuntimeConfig::default()); + + // spawn a `Greeter` in the runtime, returning an address to contact it with let addr = rt .spawn(Greeter::default()) .expect("failed to spawn greeter"); - let inbox = rt.new_inbox::(); + // create an `Inbox` that allows us to receive messages from the runtime + let inbox = rt.new_inbox::().unwrap(); + + // send a message to the `Greeter` we spawned rt.send_to( addr, GreetMessage { @@ -49,10 +54,11 @@ fn main() { }, ) .unwrap(); + + // default runtime is single threaded, and requires the parent process to drive for _ in 0..3 { rt.tick(); } - let resp = inbox.try_recv().expect("greeter should have said hello"); println!("{}", resp.0); diff --git a/examples/ring.rs b/examples/ring.rs index d79e35c..d3c6031 100644 --- a/examples/ring.rs +++ b/examples/ring.rs @@ -1,6 +1,6 @@ use swactor::{ actor::{ActorAddress, ActorInterface}, - runtime::{Inbox, Runtime, RuntimeFlavor}, + runtime::{Inbox, Runtime, RuntimeConfig}, }; #[derive(Debug, Default, Clone)] @@ -38,13 +38,15 @@ impl ActorInterface for RingActor { } fn main() { - let mut rt = Runtime::new(); - let inbox: Inbox = rt.new_inbox(); + let config = RuntimeConfig::default(); + let rt = Runtime::new(config); + let inbox: Inbox = rt.new_inbox().unwrap(); let mut next = rt .spawn(RingActor::new(*inbox.addr())) .expect("failed to spawn"); - for _ in 0..500 { + let num_passes = 500; + for _ in 0..num_passes { let new = rt.spawn(RingActor::new(next)).expect("failed to spawn"); next = new; } @@ -63,6 +65,7 @@ fn main() { } } } + assert_eq!(msg.count, num_passes + 1); // count should equal the number of passes plus the return to main process inbox - println!("{msg:?}") + println!("{msg:?}"); } diff --git a/src/actor.rs b/src/actor.rs index 82aa2ce..2b80d40 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -1,17 +1,67 @@ -use crate::{Runtime, WATERLEVEL, ring_buffer::Receiver}; +use crate::{runtime::Runtime, WATERLEVEL, get_random, ring_buffer::Receiver}; +/// The primary trait defining data that can be passed to and from actor processes pub trait Message: 'static + Sized + Clone + Send + Sync {} impl Message for T {} +/// The trait that needs to be implemented in order to run a process as an `Actor` +/// +/// The `Incoming` type represents `Messages` that can be delivered to the `Actor`. +/// +/// The `Response` type represents possible `Messages` the actor may attempt to reply with. +/// +/// The `fn handle(..)` is where you implement the logic for handling `Incoming` messages +/// +/// # Example +/// ``` +/// use swactor::{actor::{ActorAddress, ActorInterface}, runtime::Runtime}; +/// +/// struct Greeter { +/// num_greeted: usize, +/// } +/// +/// #[derive(Clone)] // required to auto implement `Message` +/// struct GreetMessage { +/// who: String, +/// return_addr: ActorAddress, +/// } +/// +/// #[derive(Clone)] +/// struct GreetResponse(String); +/// +/// impl ActorInterface for Greeter { +/// type Incoming = GreetMessage; +/// type Response = GreetResponse; +/// +/// fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming) { +/// let response = GreetResponse(format!("Hello, {}!", msg.who).to_string()); +/// if let Ok(_) = ctx.send_to(msg.return_addr, response) { +/// self.num_greeted += 1; +/// } +/// } +/// } +/// ``` pub trait ActorInterface: 'static + Send { type Incoming: Message; type Response: Message; fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming); } -pub type ActorAddress = u64; +/// A unique address for this actor. 32 bytes is overkill for a small application, +/// but most systems are powerful, and this allows us to create a global map of +/// actor processes in the future, without worrying about collision. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash)] +pub struct ActorAddress(pub [u8; 32]); +impl ActorAddress { + pub fn new_random() -> Self { + let mut bytes = [0u8; 32]; + get_random(&mut bytes); + Self(bytes) + } +} -pub struct Actor +/// The actor process as represented in the Runtime, with the actor state stored with it's inbox. +pub(crate) struct Actor where A: ActorInterface, { @@ -38,6 +88,8 @@ where A: ActorInterface, { fn tick(&mut self, ctx: &Runtime) { + // TODO: WATERLEVEL is hard coded, and so is this message handling scheme. We should + // make it so both are more flexible, with sane defaults. let total_messages = self.inbox.len(); let messages_to_process = if total_messages < WATERLEVEL { total_messages diff --git a/src/error.rs b/src/error.rs index 91a260a..12183e2 100644 --- a/src/error.rs +++ b/src/error.rs @@ -1,3 +1,17 @@ +/// Simple, ergonomic, local `Error` type. +/// # Usage +/// ``` +/// use swactor::Error; +/// +/// fn foo_if_even(num: u64) -> Result { +/// if num % 2 == 0 { +/// return Ok("foo".into()); +/// } +/// else { +/// return Err(Error::from("baz")); +/// } +/// } +/// ``` #[derive(Debug)] pub struct Error(Box); pub(crate) fn convert_err(e: E) -> Error { diff --git a/src/lib.rs b/src/lib.rs index 665aed0..5ea2d6b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,10 +7,6 @@ mod ring_buffer; mod router; pub mod runtime; -// Re-export commonly used types -pub use actor::{ActorAddress, ActorInterface, Message}; -pub use runtime::{Inbox, Runtime, RuntimeFlavor}; - #[cfg(feature = "getrandom")] pub(crate) fn get_random(buf: &mut [u8]) { getrandom::getrandom(buf).unwrap() @@ -26,5 +22,3 @@ pub(crate) fn get_random(buf: &mut [u8]) { /// process total_messages >> 1 /// ``` const WATERLEVEL: usize = 10; -/// FIXME: remove hard coded defaults -const DEFAULT_INBOX_CAPACITY: usize = 1_000; diff --git a/src/ring_buffer.rs b/src/ring_buffer.rs index fea5113..b016b3e 100644 --- a/src/ring_buffer.rs +++ b/src/ring_buffer.rs @@ -1,7 +1,8 @@ -pub use crossbeam_queue::ArrayQueue; +//! Shallow wrapper around the `crossbeam_queue::ArrayQueue` implementation of a mpmc ring buffer. use std::sync::Arc; +pub use crossbeam_queue::ArrayQueue; -/// The receiving end of a `crossbeam_queue::ArrayQueue`, a lock-free mpsc queue. +/// The receiving end of a `crossbeam_queue::ArrayQueue`, a lock-free mpmc queue. /// The queue is constructed by the `Receiver::new()` method. /// Responsible for creating the `Sender` ends of itself. /// diff --git a/src/router.rs b/src/router.rs index c6816b8..82fb10a 100644 --- a/src/router.rs +++ b/src/router.rs @@ -1,13 +1,14 @@ use std::{collections::HashMap, sync::Arc}; use crate::{ - ActorInterface, - actor::{ActorAddress, Message}, - ring_buffer::Sender, + actor::{ActorAddress, ActorInterface, Message}, + ring_buffer::Sender, runtime::Runtime, }; /// FIXME: Go over with a fine-toothed comb and reassure yourself this typing -/// makes sense. With these types, what happens at the hardware level. +/// makes sense, that we are not doing loads of indirection on a hot path. +/// +/// A type erased `Message` to be routed between actor processes. pub(crate) type Envelope = Arc; pub(crate) trait SenderT: Send + Sync { @@ -38,6 +39,7 @@ pub(crate) enum RouterMessage { SendToAddr { addr: ActorAddress, msg: Envelope }, } +/// The `Router` is responsible for taking in and delivering all messages in the runtime. pub(crate) struct Router { directory: HashMap>, } @@ -52,10 +54,9 @@ impl Router { impl ActorInterface for Router { type Incoming = RouterMessage; - type Response = (); - fn handle(&mut self, _ctx: &crate::Runtime, msg: Self::Incoming) { + fn handle(&mut self, _ctx: &Runtime, msg: Self::Incoming) { match msg { RouterMessage::AddAddr(addr, sender) => { self.directory.insert(addr, sender); diff --git a/src/runtime.rs b/src/runtime.rs index a245130..6a7a76a 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,26 +1,17 @@ +use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, OnceLock}; use std::thread::{self, JoinHandle}; use crossbeam_queue::ArrayQueue; use crate::{ - DEFAULT_INBOX_CAPACITY, Error, + Error, actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}, - get_random, ring_buffer::{Receiver, Sender}, router::{Router, RouterMessage}, }; -#[derive(Debug, Clone, Default)] -pub enum RuntimeFlavor { - #[default] - SingleThreaded, - Multithreaded { - workers: usize, - }, -} - +/// Generic message inbox for receiving messages outside of the runtime. pub struct Inbox { addr: ActorAddress, inner: Receiver, @@ -36,195 +27,226 @@ impl Inbox { } } -/// Thread handles for the multithreaded runtime -struct RuntimeHandles { - workers: Vec>, - router_thread: Option>, +/// The tunable settings for the runtime. +pub struct RuntimeConfig { + pub max_actors: usize, + pub router_max_messages: usize, + pub actor_max_messages: usize, + pub num_threads: usize, } -/// The main runtime for executing actors -pub struct Runtime { - running: AtomicBool, - router_inbox: Sender, - actor_queue: ArrayQueue>, - flavor: RuntimeFlavor, - /// Thread handles, created lazily when run() is called - handles: OnceLock, -} +/// 8kB for the `Box<..>` before counting the rest of the memory +const DEFAULT_MAX_ACTORS: usize = 1_000; -impl Runtime { - /// Create a new runtime with given actor queue capacity and flavor - pub fn new() -> Self { - let actor_queue = ArrayQueue::new(DEFAULT_INBOX_CAPACITY); +/// 160kB for the `Arc<..>` before counting the rest of the memory +const DEFAULT_ROUTER_MAX_MESSAGES: usize = 10_000; - let router = Router::new(); - let router_incoming = - Receiver::<::Incoming>::new(DEFAULT_INBOX_CAPACITY); - let router_inbox = router_incoming.new_sender(); - - actor_queue - .push(Box::new(Actor::new(router_incoming, router)) as Box) - .map_err(|_| "failed to add router to actor queue") - .expect("failed to spawn router at runtime initialization."); +/// 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 { - running: AtomicBool::new(false), - router_inbox, - actor_queue, - flavor: RuntimeFlavor::SingleThreaded, - handles: OnceLock::new(), + max_actors: DEFAULT_MAX_ACTORS, + router_max_messages: DEFAULT_ROUTER_MAX_MESSAGES, + actor_max_messages: DEFAULT_ACTOR_MAX_MESSAGES, + num_threads: 1, + } + } +} + +/// The `Runtime` struct is the primary gateway for interacting with the framework. +pub struct Runtime { + config: RuntimeConfig, + actor_queue: ArrayQueue>, + router_interface: Sender, + router: Option>, // `None` if single-threaded + + // for multithreaded contexts + is_running: AtomicBool, +} + +/// Handle for dealing with a runtime that has started via the `Runtime::run()` method. +pub struct RuntimeHandle { + pub runtime: Arc, + threads: Vec>, +} + +impl RuntimeHandle { + pub fn join(self) { + for handle in self.threads { + let _ = handle.join(); } } - /// FIXME: Abstract out by making two separate `Runtime` structs, one for singlethreaded, another for multi - pub fn new_multithreaded(num_workers: usize) -> (Self, Actor) { - let router = Router::new(); - let router_incoming = - Receiver::<::Incoming>::new(DEFAULT_INBOX_CAPACITY); - let router_inbox = router_incoming.new_sender(); + /// Simple helper, calls the inner `Runtime::shutdown()` method + pub fn shutdown(&self) { + self.runtime.shutdown(); + } +} - ( - Self { - running: AtomicBool::new(false), - router_inbox, - actor_queue: ArrayQueue::new(DEFAULT_INBOX_CAPACITY), - flavor: RuntimeFlavor::Multithreaded { - workers: num_workers, - }, - handles: OnceLock::new(), - }, - Actor::new(router_incoming, router), - ) +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 = ArrayQueue::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, + router_interface: router_sender, + is_running: AtomicBool::new(false), + router: router_option, + } } /// Spawn an actor, returns its address pub fn spawn(&self, actor: A) -> Result { - // FIXME: Figure out what to do with this. - // Its distracting to include here, but not used elsewhere for now. - let addr = { - let mut bytes = u64::to_le_bytes(0); - get_random(&mut bytes); - u64::from_le_bytes(bytes) - }; - let inbox = Receiver::::new(DEFAULT_INBOX_CAPACITY); + // 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 - let _ = self - .router_inbox - .try_send(RouterMessage::AddAddr(addr, Arc::new(sender))); + self.router_interface + .try_send(RouterMessage::AddAddr(addr, Arc::new(sender))) + .map_err(|_| { + Error::from("Runtime error: failed to add actor to router. Router inbox full") + })?; self.actor_queue .push(Box::new(Actor::new(inbox, actor))) - .map_err(|_| Error::from("Runtime error: Failed to spawn actor."))?; + .map_err(|_| Error::from("Runtime error: Failed to spawn actor. Queue full."))?; Ok(addr) } /// Send a message to an actor address pub fn send_to(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { - self.router_inbox + self.router_interface .try_send(RouterMessage::SendToAddr { addr, msg: Arc::new(msg), }) - .map_err(|_| "Failed to send message to router.".into()) + .map_err(|_| Error::from("Failed to send message to router.")) } - /// Create an external inbox for receiving messages outside actors - pub fn new_inbox(&self) -> Inbox { - let addr = { - let mut bytes = u64::to_le_bytes(0); - get_random(&mut bytes); - u64::from_le_bytes(bytes) - }; - let receiver = Receiver::::new(DEFAULT_INBOX_CAPACITY); + /// 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 - let _ = self - .router_inbox - .try_send(RouterMessage::AddAddr(addr, Arc::new(sender))); - Inbox { + self.router_interface + .try_send(RouterMessage::AddAddr(addr, Arc::new(sender))) + .map_err(|_| { + Error::from( + "Runtime error: failed to add a new inbox channel. Router inbox is full.", + ) + })?; + + Ok(Inbox { addr, inner: receiver, - } + }) } - /// Process one actor tick + router messages - /// Works in both single/multi mode (useful for testing and fine-grained control) + /// Spawn worker threads and start processing, returning a set of handles and + /// a Runtime object to interface with. /// - /// FIXME: `tick` does not make sense in multithreaded context. Add a check to ensure single threaded - pub fn tick(&mut self) { + /// ### WARN: + /// ##### This function panics if the configuration is set as single threaded + /// `config.num_threads == 1` + pub fn run(mut self) -> Result { + if self.config.num_threads < 2 { + return Err(Error::from( + "Runtime error: cannot call `Runtime::run()` from a single-threaded context.", + )); + } + + self.is_running.store(true, Ordering::Release); + + // Take router out before wrapping in Arc - it will be owned by router thread + let mut router = self + .router + .take() + .expect("Router must be present for multi-threaded runtime"); + + let rt = Arc::new(self); + let mut handles: Vec> = vec![]; + + // Router thread owns the router directly - no synchronization needed + let router_handle = { + let ctx = rt.clone(); + thread::spawn(move || { + while ctx.is_running.load(Ordering::Acquire) { + router.tick(&ctx); + thread::yield_now(); + } + }) + }; + handles.push(router_handle); + + // Spawn worker threads + let num_workers = rt.config.num_threads - 1; + for _ in 0..num_workers { + let ctx = rt.clone(); + let handle = thread::spawn(move || { + while ctx.is_running.load(Ordering::Acquire) { + if let Some(mut actor) = ctx.actor_queue.pop() { + actor.tick(&ctx); + if let Err(_) = ctx.actor_queue.push(actor) { + panic!( + "Runtime panic: attempted to return an actor to the queue, but queue was full." + ) + } + } else { + thread::yield_now(); + } + } + }); + handles.push(handle); + } + + Ok(RuntimeHandle { + runtime: rt, + threads: handles, + }) + } + + /// Pop the actor off the top of the queue and process it's messages, returning it to the back of + /// the queue upon completion. + pub fn tick(&self) { if let Some(mut actor) = self.actor_queue.pop() { actor.tick(&self); let _ = self.actor_queue.push(actor); } } - /// Run the runtime - /// - Multithreaded: spawns workers + router thread, returns an `Arc` pointer to the - /// `Runtime` struct. - pub fn run(self, router: Actor) -> 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 - let router_handle = { - let ctx = rt.clone(); - thread::spawn(move || { - router_loop(ctx, router); - }) - }; - - // Spawn worker threads - 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); - } - - rt.handles.set(RuntimeHandles { - workers: worker_handles, - router_thread: Some(router_handle), - }); - - rt - } - /// Signal all workers to stop pub fn shutdown(&self) { - self.running.store(false, Ordering::Release); - } - - /// Check if runtime is still active - pub fn is_running(&self) -> bool { - self.running.load(Ordering::Acquire) - } -} - -/// Worker thread loop - processes actors from the shared queue -fn worker_loop(ctx: Arc) { - while ctx.running.load(Ordering::Relaxed) { - if let Some(mut actor) = ctx.actor_queue.pop() { - actor.tick(&ctx); - let _ = ctx.actor_queue.push(actor); - } else { - thread::yield_now(); - } - } -} - -/// Router thread loop - processes router messages -fn router_loop(ctx: Arc, mut router: Actor) { - while ctx.running.load(Ordering::Relaxed) { - router.tick(&ctx); - thread::yield_now(); + self.is_running.store(false, Ordering::Release); } } diff --git a/tests/runtime_tests.rs b/tests/runtime_tests.rs index 02fb75b..56ca321 100644 --- a/tests/runtime_tests.rs +++ b/tests/runtime_tests.rs @@ -1,14 +1,4 @@ -use std::thread; -use std::time::Duration; - -use swactor::{ - actor::{ActorAddress, ActorInterface}, - runtime::{Inbox, Runtime, RuntimeFlavor}, -}; - -// ============================================================================ -// Test Helpers -// ============================================================================ +use swactor::{actor::{ActorAddress, ActorInterface}, runtime::{Inbox, Runtime, RuntimeConfig}}; #[derive(Clone)] struct PingMessage { @@ -48,8 +38,8 @@ impl ActorInterface for ForwarderActor { #[test] fn test_single_threaded_ping_pong() { - let mut rt = Runtime::new(); - let inbox: Inbox = rt.new_inbox(); + let rt = Runtime::new(RuntimeConfig::default()); + let inbox: Inbox = rt.new_inbox().unwrap(); let pong_addr = rt.spawn(PongActor).expect("spawn pong"); @@ -75,8 +65,8 @@ fn test_single_threaded_ping_pong() { #[test] fn test_single_threaded_message_chain() { - let mut rt = Runtime::new(); - let inbox: Inbox = rt.new_inbox(); + let rt = Runtime::new(RuntimeConfig::default()); + let inbox: Inbox = rt.new_inbox().unwrap(); // Create a chain: A -> B -> C -> inbox let c_addr = rt @@ -102,36 +92,39 @@ fn test_single_threaded_message_chain() { panic!("Message did not traverse the chain"); } -// #[test] -// fn test_multithreaded_message_passing() { -// let rt = Runtime::new(1000, RuntimeFlavor::Multithreaded { workers: 4 }); -// let inbox: Inbox = rt.new_inbox(); +#[test] +fn test_multithreaded_message_passing() { + let config = RuntimeConfig { + num_threads: 4, + ..Default::default() + }; + let rt = Runtime::new(config); + let inbox: Inbox = rt.new_inbox().unwrap(); -// // Create a longer chain to exercise multi-threading -// let mut target = *inbox.addr(); -// for _ in 0..20 { -// target = rt.spawn(ForwarderActor { target }).unwrap(); -// } + // Create a longer chain to exercise multi-threading + let mut target = *inbox.addr(); + for _ in 0..20 { + target = rt.spawn(ForwarderActor { target }).unwrap(); + } + let start_addr = target; -// let start_addr = target; + // Send message + rt.send_to(start_addr, ForwardMessage(999)).unwrap(); -// // Send message -// rt.send_to(start_addr, ForwardMessage(999)).unwrap(); + // Spawn thread to check for result and shutdown + let ctx = rt.run().unwrap(); + let inbox_check = std::thread::spawn(move || { + for _ in 0..100 { + std::thread::sleep(std::time::Duration::from_millis(10)); + if let Some(ForwardMessage(val)) = inbox.try_recv() { + ctx.shutdown(); + return Some(val); + } + } + ctx.shutdown(); + None + }); -// // Spawn thread to check for result and shutdown -// let ctx = rt.run(); -// let inbox_check = thread::spawn(move || { -// for _ in 0..100 { -// thread::sleep(Duration::from_millis(10)); -// if let Some(ForwardMessage(val)) = inbox.try_recv() { -// ctx.shutdown(); -// return Some(val); -// } -// } -// ctx.shutdown(); -// None -// }); - -// let result = inbox_check.join().unwrap(); -// assert_eq!(result, Some(999)); -// } + let result = inbox_check.join().unwrap(); + assert_eq!(result, Some(999)); +}