From ac6ac5b5349beebcdaa46064376f61095e809e55 Mon Sep 17 00:00:00 2001 From: Zicklag Date: Wed, 29 Apr 2020 02:38:35 +0000 Subject: [PATCH 1/9] Make num_cpus Dependency Optional Allows you to specify the `auto-num-threads` feature to get the automatically detected threads based on the CPU count. --- Cargo.toml | 6 ++++-- src/system.rs | 6 +++++- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 7000329..5aaeca1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -42,6 +42,9 @@ exclude = [ [features] default = ["actor-pool"] actor-pool = ["rand", "rand_xoshiro"] +# Automatically determine the default number of threads from the number of CPUs. Adds the +# num_cpus dependency. +auto-num-threads = ["num_cpus"] [badges] # We won't be using Travis in a bit @@ -59,7 +62,7 @@ serde_json = "^1.0.40" bincode = "1.1.4" dashmap = "1.0.3" futures = "0.3.1" -num_cpus = "1.10.1" +num_cpus = { version = "1.10.1", optional = true } log = "0.4" once_cell = "1.0.2" secc = "0.0.10" @@ -67,4 +70,3 @@ serde = { version = "1.0.97", features = ["derive", "rc"] } uuid = { version = "0.8.1", features = ["serde", "v4"]} rand = { version = "0.7.3", optional = true } rand_xoshiro = { version = "0.4.0", optional = true } - diff --git a/src/system.rs b/src/system.rs index b02307f..55a9cef 100644 --- a/src/system.rs +++ b/src/system.rs @@ -174,7 +174,11 @@ impl Default for ActorSystemConfig { /// Create the config with the default values. fn default() -> ActorSystemConfig { ActorSystemConfig { - thread_pool_size: (num_cpus::get() * 4) as u16, + #[cfg(not(feature = "auto-num-threads"))] + thread_pool_size: 16, // Default to 4 times the assumed default number of CPUs ( 4 ) + #[cfg(feature = "auto-num-threads")] + thread_pool_size: (num_cpus::get() * 4) as u16, // Run 4 times the number of detected CPUs + warn_threshold: Duration::from_millis(1), time_slice: Duration::from_millis(1), thread_wait_time: Duration::from_millis(100), From 4f71bafd641b20473b32f94330e53d469890fbe2 Mon Sep 17 00:00:00 2001 From: Zicklag Date: Wed, 29 Apr 2020 02:43:32 +0000 Subject: [PATCH 2/9] Run Rustfmt --- examples/philosophers.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/philosophers.rs b/examples/philosophers.rs index 3ef2318..035cc49 100644 --- a/examples/philosophers.rs +++ b/examples/philosophers.rs @@ -20,9 +20,9 @@ //! panics ensue. Some FSM implementations might be quite a bit more lose, preferring to ignore //! badly timed messages. This is largely up to the user. -use maxim::prelude::*; use log::LevelFilter; use log::{error, info}; +use maxim::prelude::*; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::env; From 01c6a6c6c436637bd09d363fe915a5bfd077ef48 Mon Sep 17 00:00:00 2001 From: Zicklag Date: Sun, 3 May 2020 02:51:12 +0000 Subject: [PATCH 3/9] Remove Log Imports And Use log::Level Code style preference to remove all `use log::error` or other log macro imports and just use `log::error!` at the log site. --- examples/philosophers.rs | 31 +++++++++++++---------- lib/piper | 1 + src/actors.rs | 39 ++++++++++++++--------------- src/cluster.rs | 12 ++++----- src/executor.rs | 32 ++++++++++++------------ src/executor/thread_pool.rs | 13 +++++----- src/system.rs | 49 +++++++++++++++++++------------------ src/system/system_actor.rs | 7 +++--- 8 files changed, 93 insertions(+), 91 deletions(-) create mode 160000 lib/piper diff --git a/examples/philosophers.rs b/examples/philosophers.rs index 035cc49..a067883 100644 --- a/examples/philosophers.rs +++ b/examples/philosophers.rs @@ -21,7 +21,6 @@ //! badly timed messages. This is largely up to the user. use log::LevelFilter; -use log::{error, info}; use maxim::prelude::*; use serde::{Deserialize, Serialize}; use std::collections::HashMap; @@ -90,17 +89,20 @@ impl Fork { // Resetting the skip allows fork requests to be processed. Ok(Status::reset(self)) } else { - error!( + log::error!( "[{}] fork_put_down() from non-owner: {} real owner is: {}", - context.aid, sender, owner + context.aid, + sender, + owner ); Ok(Status::done(self)) } } None => { - error!( + log::error!( "[{}] fork_put_down() from non-owner: {} real owner is: None:", - context.aid, sender + context.aid, + sender ); Ok(Status::done(self)) } @@ -119,12 +121,12 @@ impl Fork { // has been marked as being dirty. Ok(Status::reset(self)) } else { - error!("[{}] Got UsingFork from non-owner: {}", context.aid, sender); + log::error!("[{}] Got UsingFork from non-owner: {}", context.aid, sender); Ok(Status::done(self)) } } _ => { - error!("[{}] Got UsingFork from non-owner: {}", context.aid, sender); + log::error!("[{}] Got UsingFork from non-owner: {}", context.aid, sender); Ok(Status::done(self)) } } @@ -317,10 +319,10 @@ impl Philosopher { self.request_missing_forks(context)?; } PhilosopherState::Hungry => { - error!("[{}] Got BecomeHungry while eating!", context.aid); + log::error!("[{}] Got BecomeHungry while eating!", context.aid); } PhilosopherState::Eating => { - error!("[{}] Got BecomeHungry while eating!", context.aid); + log::error!("[{}] Got BecomeHungry while eating!", context.aid); } }; } @@ -371,9 +373,12 @@ impl Philosopher { fork_aid.send_new(ForkCommand::ForkPutDown(context.aid.clone()))?; } } else { - error!( + log::error!( "[{}] Unknown fork asked for: {}:\n left ==> {}\n right ==> {}", - context.aid, fork_aid, self.left_fork_aid, self.right_fork_aid + context.aid, + fork_aid, + self.left_fork_aid, + self.right_fork_aid ); } @@ -513,9 +518,9 @@ pub fn main() { // output the results of the simulation and end the program by shutting // down the actor system. if !state.iter().any(|(_, metrics)| metrics.is_none()) { - info!("Final Metrics:"); + log::info!("Final Metrics:"); for (aid, metrics) in state.iter() { - info!("{}: {:?}", aid, metrics); + log::info!("{}: {:?}", aid, metrics); } context.system.trigger_shutdown(); } diff --git a/lib/piper b/lib/piper new file mode 160000 index 0000000..c7f72a9 --- /dev/null +++ b/lib/piper @@ -0,0 +1 @@ +Subproject commit c7f72a9209a4ab04794322bbf7b0d64aadc07a90 diff --git a/src/actors.rs b/src/actors.rs index 84ac5e6..19332be 100644 --- a/src/actors.rs +++ b/src/actors.rs @@ -9,7 +9,6 @@ use crate::message::ActorMessage; use crate::prelude::*; use futures::{FutureExt, Stream}; -use log::{debug, error, trace, warn}; #[cfg(feature = "actor-pool")] use rand::{ distributions::{Distribution, Uniform}, @@ -664,7 +663,7 @@ impl Aid { pub(crate) fn stop(&self) -> Result<(), AidError> { match &self.data.sender { ActorSender::Local { stopped, .. } => { - trace!("Stopping local Actor"); + log::trace!("Stopping local Actor"); stopped.fetch_or(true, Ordering::AcqRel); Ok(()) } @@ -987,7 +986,7 @@ impl ActorBuilder { F: Processor + 'static, { let (actor, stream) = Actor::new(self.system.clone(), &self, state, processor); - debug!("Actor created: {}", actor.context.aid.uuid()); + log::debug!("Actor created: {}", actor.context.aid.uuid()); self.system.register_actor(actor, stream) } @@ -1152,12 +1151,12 @@ impl Actor { Ok(future) => match AssertUnwindSafe(future).catch_unwind().await { Ok(x) => x, Err(panic) => { - warn!("Actor panicked! Catching as error"); + log::warn!("Actor panicked! Catching as error"); Err(Panic::from(panic).into()) } }, Err(err) => { - warn!("Actor panicked! Catching as error"); + log::warn!("Actor panicked! Catching as error"); Err(Panic::from(err).into()) } } @@ -1198,21 +1197,21 @@ impl ActorStream { match result { Ok(Status::Done) => { - trace!( + log::trace!( "Actor {} finished processing a message", self.context.aid.uuid() ); self.receiver.pop().unwrap() } Ok(Status::Skip) => { - trace!( + log::trace!( "Actor {} skipped processing a message", self.context.aid.uuid() ); self.receiver.skip().unwrap() } Ok(Status::Reset) => { - trace!( + log::trace!( "Actor {} finished processing a message and reset the cursor", self.context.aid.uuid() ); @@ -1220,7 +1219,7 @@ impl ActorStream { self.receiver.reset_skip().unwrap(); } Ok(Status::Stop) => { - debug!("Actor \"{}\" stopping", self.context.aid.name_or_uuid()); + log::debug!("Actor \"{}\" stopping", self.context.aid.name_or_uuid()); self.receiver.pop().unwrap(); self.context .system @@ -1229,9 +1228,10 @@ impl ActorStream { } Err(e) => { self.receiver.pop().unwrap(); - error!( + log::error!( "[{}] returned an error when processing: {}", - self.context.aid, &e + self.context.aid, + &e ); self.context .system @@ -1260,7 +1260,7 @@ impl Stream for ActorStream { mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>, ) -> Poll> { - trace!("Actor {} is being polled", self.context.aid.name_or_uuid()); + log::trace!("Actor {} is being polled", self.context.aid.name_or_uuid()); // If we have a pending future, that's what we poll. if let Some(pending) = self.pending.as_mut() { // Poll, ensure we respect stopping condition. @@ -1270,7 +1270,7 @@ impl Stream for ActorStream { .map(|r| Some(self.overwrite_on_stop(r))); if let Poll::Pending = &poll { - trace!("Actor {} is pending", self.context.aid.uuid()); + log::trace!("Actor {} is pending", self.context.aid.uuid()); } else { drop(self.pending.take()); } @@ -1288,7 +1288,7 @@ impl Stream for ActorStream { // We're stopping after this future, mark as such if let Some(m) = msg.content_as::() { if let SystemMsg::Stop = *m { - trace!("Actor {} received stop message", self.context.aid.uuid()); + log::trace!("Actor {} received stop message", self.context.aid.uuid()); self.stopping = true; } } @@ -1300,7 +1300,7 @@ impl Stream for ActorStream { match future.as_mut().poll(cx) { Poll::Ready(r) => Poll::Ready(Some(self.overwrite_on_stop(r))), Poll::Pending => { - trace!("Actor {} is pending", self.context.aid.uuid()); + log::trace!("Actor {} is pending", self.context.aid.uuid()); self.pending = Some(future); Poll::Pending } @@ -1314,7 +1314,7 @@ impl Stream for ActorStream { // While this is exhaustive, we're avoiding a catchall to in anticipation of // future Secc errors we would *want* to handle. SeccErrors::Empty | SeccErrors::Full(_) => { - trace!( + log::trace!( "Actor `{}` has no more messages, return to sleep", self.context.aid.name_or_uuid() ); @@ -1330,8 +1330,7 @@ impl Stream for ActorStream { mod tests { use super::*; use crate::tests::*; - use log::*; - use std::thread; + use std::thread; use std::time::Instant; /// This is identical to the documentation but here so that its formatted by rust and we can @@ -1351,8 +1350,8 @@ mod tests { .unwrap(); match aid.send(Message::new(11)) { - Ok(_) => info!("OK Then!"), - Err(e) => info!("Ooops {:?}", e), + Ok(_) => log::info!("OK Then!"), + Err(e) => log::info!("Ooops {:?}", e), } system.await_shutdown(None); diff --git a/src/cluster.rs b/src/cluster.rs index 6d84112..9e33121 100644 --- a/src/cluster.rs +++ b/src/cluster.rs @@ -9,7 +9,6 @@ //! robust and well tested like the rest of Maxim. use crate::prelude::*; -use log::{error, info}; use secc::*; use std::collections::HashMap; use std::io::prelude::*; @@ -102,7 +101,7 @@ impl TcpClusterMgr { system.init_current(); let sys_uuid = system.uuid(); let listener = TcpListener::bind(address).unwrap(); - info!("{}: Listening for connections on {}.", sys_uuid, address); + log::info!("{}: Listening for connections on {}.", sys_uuid, address); // Notify the cluster manager that the listener is ready. let (mutex, condvar) = &*pair; @@ -116,14 +115,15 @@ impl TcpClusterMgr { while manager.data.running.load(Ordering::Relaxed) { match listener.accept() { Ok((stream, socket_address)) => { - info!( + log::info!( "{}: Accepting connection from: {}.", - sys_uuid, socket_address + sys_uuid, + socket_address ); manager.start_tcp_threads(stream, socket_address); } Err(e) => { - error!("couldn't get client: {:?}", e); + log::error!("couldn't get client: {:?}", e); } } } @@ -160,7 +160,7 @@ impl TcpClusterMgr { rx_handle, }; - info!( + log::info!( "{:?}: Connected to {:?}@{:?}", self.data.system.uuid(), system_uuid, diff --git a/src/executor.rs b/src/executor.rs index bfef62e..c2130c6 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -6,7 +6,6 @@ use crate::prelude::*; use dashmap::DashMap; use futures::task::ArcWake; use futures::Stream; -use log::{debug, info, trace, warn}; use std::collections::{BTreeMap, VecDeque}; use std::pin::Pin; use std::sync::{Arc, Condvar, Mutex, RwLock}; @@ -55,7 +54,7 @@ impl MaximExecutor { self.reactors.insert(i, reactor.clone()); self.actors_per_reactor.insert(i, 0); let sys = system.clone(); - info!("Spawning Reactors"); + log::info!("Spawning Reactors"); self.thread_pool .spawn(format!("Reactor-{}", reactor.name), move || { sys.init_current(); @@ -82,12 +81,12 @@ impl MaximExecutor { /// This wakes an ActorStream in the Executor which will cause its future to be polled. The Aid, /// through the ActorSystem, will call this on Message Send. pub(crate) fn wake(&self, id: Aid) { - trace!("Waking Actor `{}`", id.name_or_uuid()); + log::trace!("Waking Actor `{}`", id.name_or_uuid()); // Pull the Task let task = match self.sleeping.remove(&id) { Some((_, task)) => task, None => { - debug!( + log::debug!( "Actor `{}` not in Executor - already woken or stopped", id.name_or_uuid() ); @@ -119,7 +118,7 @@ impl MaximExecutor { /// When a Reactor is done with an task, it will be sent here, and the Executor will decrement /// the Actor count for that Reactor. fn return_task(&self, task: Task, reactor: &MaximReactor) { - trace!( + log::trace!( "Actor {} returned from Reactor {}", task.id.name_or_uuid(), reactor.name @@ -134,7 +133,7 @@ impl MaximExecutor { /// triggered. pub(crate) fn await_shutdown(&self, timeout: impl Into>) -> ShutdownResult { let start = Instant::now(); - info!("Notifying Reactor threads, so they can end gracefully"); + log::info!("Notifying Reactor threads, so they can end gracefully"); for r in self.reactors.iter() { match r.thread_condvar.read() { Ok(g) => g.1.notify_one(), @@ -142,7 +141,7 @@ impl MaximExecutor { } } let timeout = timeout.into().map(|t| t - (Instant::now() - start)); - info!("Awaiting the threadpool's shutdown"); + log::info!("Awaiting the threadpool's shutdown"); self.thread_pool.await_shutdown(timeout) } } @@ -192,7 +191,7 @@ impl MaximReactor { /// Creates a new Reactor fn new(executor: MaximExecutor, system: &ActorSystem, id: u16) -> MaximReactor { let name = format!("{:08x?}-{}", system.data.uuid.as_fields().0, id); - debug!("Creating Reactor {}", name); + log::debug!("Creating Reactor {}", name); MaximReactor { id, @@ -234,7 +233,7 @@ impl MaximReactor { .lock() .expect("Poisoned shutdown_triggered condvar") { - debug!("Reactor-{} acknowledging shutdown", self.name); + log::debug!("Reactor-{} acknowledging shutdown", self.name); return false; } } @@ -279,13 +278,13 @@ impl MaximReactor { // Still pending, return to wait_queue. Drop the wakeup, because the futures // will re-add it later through their wakers. Poll::Pending => { - trace!("Reactor-{} waiting on pending Actor", self.name); + log::trace!("Reactor-{} waiting on pending Actor", self.name); self.wait(task); break; } } if Instant::now().duration_since(start) >= self.warn_threshold { - warn!( + log::warn!( "Actor {} took longer than configured warning threshold", aid.name_or_uuid() ); @@ -304,14 +303,14 @@ impl MaximReactor { fn get_work(&self) -> LoopResult<(Wakeup, Task)> { if let Some(w) = self.get_woken() { if let Some(task) = self.remove_waiting(&w.id) { - trace!( + log::trace!( "Reactor-{} received Wakeup for Actor `{}`", self.name, task.id.name_or_uuid() ); LoopResult::Ok((w, task)) } else { - trace!("Reactor-{} dropping spurious WakeUp", self.name); + log::trace!("Reactor-{} dropping spurious WakeUp", self.name); LoopResult::Continue } } else { @@ -320,12 +319,12 @@ impl MaximReactor { .read() .expect("Poisoned Reactor condvar"); - trace!("Reactor-{} waiting on condvar", self.name); + log::trace!("Reactor-{} waiting on condvar", self.name); let g = mutex.lock().expect("Poisoned Reactor condvar"); let _ = condvar .wait_timeout(g, self.thread_wait_time) .expect("Poisoned Reactor condvar"); - trace!("Reactor-{} resuming", self.name); + log::trace!("Reactor-{} resuming", self.name); LoopResult::Continue } } @@ -417,7 +416,6 @@ mod tests { use crate::executor::ShutdownResult; use crate::prelude::*; use crate::tests::*; - use log::*; use std::future::Future; use std::pin::Pin; use std::task::Poll; @@ -446,7 +444,7 @@ mod tests { 0 => Poll::Ready(Ok(Status::done(()))), count => { *count -= 1; - debug!("Pending, {} times left", count); + log::debug!("Pending, {} times left", count); let waker = cx.waker().clone(); let sleep_for = self.sleep_for; thread::spawn(move || { diff --git a/src/executor/thread_pool.rs b/src/executor/thread_pool.rs index 0a98deb..4152835 100644 --- a/src/executor/thread_pool.rs +++ b/src/executor/thread_pool.rs @@ -1,5 +1,4 @@ use crate::executor::ShutdownResult; -use log::{debug, error, trace}; use std::sync::{Arc, Condvar, Mutex}; use std::thread; use std::time::Duration; @@ -33,7 +32,7 @@ impl MaximThreadPool { .spawn(move || { let lease = ThreadLease::new(deed); lease.deed.drain.increment(); - debug!("Thread {} has started", lease.deed.name); + log::debug!("Thread {} has started", lease.deed.name); lease.deed.set_running(); f(); lease.deed.set_stopped(); @@ -94,9 +93,9 @@ impl Drop for ThreadLease { // If the Lease dropped while Running, it Panicked. if let ThreadState::Running = *g { *g = ThreadState::Panicked; - error!("Thread {} panicked!", self.deed.name) + log::error!("Thread {} panicked!", self.deed.name) } else { - debug!("Thread {} has stopped", self.deed.name) + log::debug!("Thread {} has stopped", self.deed.name) } // Either way, it's dead, let's decrement the thread counter. self.deed.drain.decrement(); @@ -124,7 +123,7 @@ impl DrainAwait { pub fn increment(&self) { let mut g = self.mutex.lock().expect("DrainAwait poisoned"); let new = *g + 1; - trace!("Incrementing DrainAwait to {}", new); + log::trace!("Incrementing DrainAwait to {}", new); *g += 1; } @@ -132,9 +131,9 @@ impl DrainAwait { pub fn decrement(&self) { let mut guard = self.mutex.lock().expect("DrainAwait poisoned"); *guard -= 1; - trace!("Decrementing DrainAwait to {}", *guard); + log::trace!("Decrementing DrainAwait to {}", *guard); if *guard == 0 { - debug!("DrainAwait is depleted, notifying blocked threads"); + log::debug!("DrainAwait is depleted, notifying blocked threads"); self.condvar.notify_all(); } } diff --git a/src/system.rs b/src/system.rs index 55a9cef..e83ae93 100644 --- a/src/system.rs +++ b/src/system.rs @@ -16,7 +16,6 @@ use crate::executor::MaximExecutor; use crate::prelude::*; use crate::system::system_actor::SystemActor; use dashmap::DashMap; -use log::{debug, error, info, trace, warn}; use once_cell::sync::OnceCell; use secc::{SeccReceiver, SeccSender}; use serde::{Deserialize, Serialize}; @@ -340,7 +339,7 @@ impl ActorSystem { .started .compare_and_swap(false, true, Ordering::Relaxed) { - info!("ActorSystem {} has spawned", self.data.uuid); + log::info!("ActorSystem {} has spawned", self.data.uuid); self.data.executor.init(self); // We have the thread pool in a mutex to avoid a chicken & egg situation with the actor @@ -383,13 +382,14 @@ impl ActorSystem { Some(msg) => { let now = Instant::now(); if now >= msg.instant { - trace!("Sending delayed message"); + log::trace!("Sending delayed message"); msg.destination .send(msg.message.clone()) .unwrap_or_else(|error| { - warn!( + log::warn!( "Cannot send scheduled message to {}: Error {:?}", - msg.destination, error + msg.destination, + error ); }); data.pop(); @@ -428,7 +428,7 @@ impl ActorSystem { system_actor_aid: self.system_actor_aid(), }; sender.send(hello).unwrap(); - debug!("Sending hello from {}", self.data.uuid); + log::debug!("Sending hello from {}", self.data.uuid); // FIXME (Issue #75) Make error handling in ActorSystem::connect more robust. let system_actor_aid = @@ -508,7 +508,7 @@ impl ActorSystem { } => { if let Some(aid) = self.find_aid(&system_uuid, &actor_uuid) { aid.send(message.clone()).unwrap_or_else(|error| { - warn!("Could not send wire message to {}. Error: {}", aid, error); + log::warn!("Could not send wire message to {}. Error: {}", aid, error); }) } } @@ -523,7 +523,7 @@ impl ActorSystem { .expect("Error not handled yet"); } WireMessage::Hello { system_actor_aid } => { - debug!("{:?} Got Hello from {}", self.data.uuid, system_actor_aid); + log::debug!("{:?} Got Hello from {}", self.data.uuid, system_actor_aid); } } } @@ -569,7 +569,7 @@ impl ActorSystem { /// will wait on after [`ActorSystem::trigger_shutdown`] is called, blocking until all Reactors /// have stopped. pub fn await_shutdown(&self, timeout: impl Into>) -> ShutdownResult { - info!("System awaiting shutdown"); + log::info!("System awaiting shutdown"); let start = Instant::now(); let timeout = timeout.into(); @@ -744,7 +744,7 @@ impl ActorSystem { } else { // The actor was removed from the map so ignore the problem and just log // a warning. - warn!( + log::warn!( "Attempted to schedule actor with aid {:?} on system with node_id {:?} but the actor does not exist.", aid, @@ -788,9 +788,10 @@ impl ActorSystem { error: error.clone(), }; m_aid.send(Message::new(value)).unwrap_or_else(|error| { - error!( + log::error!( "Could not send 'Stopped' to monitoring actor {}: Error: {:?}", - m_aid, error + m_aid, + error ); }); } @@ -850,11 +851,11 @@ impl ActorSystem { // FIXME (Issue #72) Add try_send ability. pub fn send_to_system_actors(&self, message: Message) { let remotes = &*self.data.remotes; - trace!("Sending message to Remote System Actors"); + log::trace!("Sending message to Remote System Actors"); for remote in remotes.iter() { let aid = &remote.value().system_actor_aid; aid.send(message.clone()).unwrap_or_else(|error| { - error!("Could not send to system actor {}. Error: {}", aid, error) + log::error!("Could not send to system actor {}. Error: {}", aid, error) }); } } @@ -1045,17 +1046,17 @@ mod tests { fn test_send_after() { init_test_log(); - info!("Preparing test"); + log::info!("Preparing test"); let system = ActorSystem::create(ActorSystemConfig::default().thread_pool_size(2)); let aid = system.spawn().name("A").with((), simple_handler).unwrap(); await_received(&aid, 1, 1000).unwrap(); - info!("Test prepared, sending delayed message"); + log::info!("Test prepared, sending delayed message"); system.send_after(Message::new(11), aid.clone(), Duration::from_millis(10)); - info!("Sleeping for initial check"); + log::info!("Sleeping for initial check"); sleep(5); assert_eq!(1, aid.received().unwrap()); - info!("Sleeping till we're 100% sure we should have the message"); + log::info!("Sleeping till we're 100% sure we should have the message"); sleep(10); assert_eq!(2, aid.received().unwrap()); @@ -1216,11 +1217,11 @@ mod tests { .spawn() .with((), |_: (), _: Context, msg: Message| { if let Some(_) = msg.content_as::() { - debug!("Not panicking this time"); + log::debug!("Not panicking this time"); return future::ok(Status::done(())); } - debug!("About to panic"); + log::debug!("About to panic"); panic!("I panicked") }) .unwrap(); @@ -1335,13 +1336,13 @@ mod tests { .spawn() .with((), move |_: (), context: Context, message: Message| { if let Some(_) = message.content_as::() { - debug!("Received reply, shutting down"); + log::debug!("Received reply, shutting down"); context.system.trigger_shutdown(); future::ok(Status::stop(())) } else if let Some(msg) = message.content_as::() { match &*msg { SystemMsg::Start => { - debug!("Starting request actor"); + log::debug!("Starting request actor"); let target_aid: Aid = bincode::deserialize(&serialized).unwrap(); target_aid .send_new(Request { @@ -1396,7 +1397,7 @@ mod tests { if let Some(msg) = message.content_as::() { match &*msg { SystemActorMessage::FindByNameResult { aid: found, .. } => { - debug!("FindByNameResult received"); + log::debug!("FindByNameResult received"); if let Some(target) = found { t.assert( target.uuid() == aid1.uuid(), @@ -1412,7 +1413,7 @@ mod tests { _ => t.panic("Unexpected message received!"), } } else if let Some(msg) = message.content_as::() { - debug!("Actor started, attempting to send FindByName request"); + log::debug!("Actor started, attempting to send FindByName request"); if let SystemMsg::Start = &*msg { context.system.send_to_system_actors(Message::new( SystemActorMessage::FindByName { diff --git a/src/system/system_actor.rs b/src/system/system_actor.rs index fe9e406..f9c39eb 100644 --- a/src/system/system_actor.rs +++ b/src/system/system_actor.rs @@ -1,5 +1,4 @@ use crate::prelude::*; -use log::{debug, error}; use serde::{Deserialize, Serialize}; use uuid::Uuid; @@ -14,7 +13,7 @@ impl SystemActor { if let Some(msg) = message.content_as::() { // Someone requested that this system actor find an actor by name. if let SystemActorMessage::FindByName { reply_to, name } = &*msg { - debug!("Attempting to locate Actor by name: {}", name); + log::debug!("Attempting to locate Actor by name: {}", name); let found = context.system.find_aid_by_name(&name); let reply = Message::new(SystemActorMessage::FindByNameResult { system_uuid: context.system.uuid(), @@ -25,7 +24,7 @@ impl SystemActor { // there is a problem sending the reply. In this case, the error is logged but the // actor moves on. reply_to.send(reply).unwrap_or_else(|error| { - error!( + log::error!( "Could not send reply to FindByName to actor {}. Error: {:?}", reply_to, error ) @@ -37,7 +36,7 @@ impl SystemActor { Ok(Status::done(self)) // Log an error if we get an unexpected message kind, but continue processing as normal. } else { - error!("Unhandled message received."); + log::error!("Unhandled message received."); Ok(Status::done(self)) } } From 2b16defa6b92cac6c6846161d42597fbfdd8fbae Mon Sep 17 00:00:00 2001 From: Zicklag Date: Sun, 3 May 2020 03:02:32 +0000 Subject: [PATCH 4/9] WIP Smol Executor Refactor --- .gitmodules | 3 + Cargo.toml | 22 +- deny.toml | 3 + src/actors.rs | 11 +- src/executor.rs | 68 +++--- src/lib.rs | 5 +- src/prelude.rs | 2 +- src/system.rs | 562 ++++++++++++++++++++---------------------------- 8 files changed, 286 insertions(+), 390 deletions(-) create mode 100644 .gitmodules diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 0000000..3dbf8bd --- /dev/null +++ b/.gitmodules @@ -0,0 +1,3 @@ +[submodule "lib/piper"] + path = lib/piper + url = git@github.com:katharostech/piper.git diff --git a/Cargo.toml b/Cargo.toml index 5aaeca1..dc3b920 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -59,14 +59,16 @@ rand = "^0.7" serde_json = "^1.0.40" [dependencies] -bincode = "1.1.4" -dashmap = "1.0.3" -futures = "0.3.1" -num_cpus = { version = "1.10.1", optional = true } -log = "0.4" -once_cell = "1.0.2" +smol = "0.1.4" # Async Executor +piper = { path = "lib/piper" } # Async pipes, channels, mutexes, and more +futures = "0.3.1" # Async Utils +serde = { version = "1.0.97", features = ["derive", "rc"] } # Serialization support +bincode = "1.1.4" # RPC Serialization +dashmap = "1.0.3" # Concurrent hashmap +log = "0.4" # Logging Facade +uuid = { version = "0.8.1", features = ["serde", "v4"]} # UUID Generation secc = "0.0.10" -serde = { version = "1.0.97", features = ["derive", "rc"] } -uuid = { version = "0.8.1", features = ["serde", "v4"]} -rand = { version = "0.7.3", optional = true } -rand_xoshiro = { version = "0.4.0", optional = true } +once_cell = "1.0.2" +rand = { version = "0.7.3", optional = true } # Random support +rand_xoshiro = { version = "0.4.0", optional = true } # Fast random number generator +num_cpus = { version = "1.10.1", optional = true } # Detect number of CPUs diff --git a/deny.toml b/deny.toml index 1a0ef73..fc55045 100644 --- a/deny.toml +++ b/deny.toml @@ -73,6 +73,9 @@ allow = [ "MIT", "Apache-2.0", "BSD-2-Clause", + # TODO: This the license for `wepoll-binding` and `wepoll-sys` and I'm + # pretty sure we don't mind MPL, but it would be good to double-check. + "MPL-2.0", ] # List of explictly disallowed licenses # See https://spdx.org/licenses/ for list of possible licenses diff --git a/src/actors.rs b/src/actors.rs index 19332be..a36d3f6 100644 --- a/src/actors.rs +++ b/src/actors.rs @@ -344,9 +344,9 @@ impl Aid { } else { match sender.send_await_timeout(message, system.config().send_timeout) { Ok(_) => { - if sender.receivable() == 1 { - system.schedule(self.clone()); - }; + // if sender.receivable() == 1 { + // system.schedule(self.clone()); + // }; Ok(()) } Err(_) => Err(AidError::SendTimedOut(self.clone())), @@ -1436,6 +1436,7 @@ mod tests { /// instead of panic. #[test] fn test_aid_serialization() { + unimplemented!("FIXME: Re-implement cluster support."); let system = ActorSystem::create(ActorSystemConfig::default().thread_pool_size(2)); let aid1 = system.spawn().with((), simple_handler).unwrap(); system.init_current(); // Required by Aid serialization. @@ -1466,7 +1467,7 @@ mod tests { let system2 = ActorSystem::create(ActorSystemConfig::default().thread_pool_size(2)); system2.init_current(); // Connect the systems so the remote channel can be used. - ActorSystem::connect_with_channels(&system, &system2); + // ActorSystem::connect_with_channels(&system, &system2); let deserialized: Aid = bincode::deserialize(&aid1_serialized).unwrap(); match deserialized.data.sender { @@ -1483,7 +1484,7 @@ mod tests { // Disconnecting the remote then attempting to deserialize the Aid should result in a // deserialization error. - system2.disconnect(aid1.system_uuid()).unwrap(); + // system2.disconnect(aid1.system_uuid()).unwrap(); let aid1_deserialized = bincode::deserialize::(&aid1_serialized); assert!(aid1_deserialized.is_err()); }); diff --git a/src/executor.rs b/src/executor.rs index c2130c6..b05d3b8 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -1,17 +1,25 @@ //! The Executor is responsible for the high-level scheduling of Actors. -use crate::actors::ActorStream; -use crate::executor::thread_pool::MaximThreadPool; -use crate::prelude::*; +// use dashmap::DashMap; use dashmap::DashMap; use futures::task::ArcWake; use futures::Stream; +use piper::ChangeNotifier; + use std::collections::{BTreeMap, VecDeque}; use std::pin::Pin; -use std::sync::{Arc, Condvar, Mutex, RwLock}; +use std::sync::{ + atomic::{AtomicBool, Ordering::SeqCst}, + Arc, Condvar, Mutex, RwLock, +}; use std::task::{Context, Poll, Waker}; +use std::thread; use std::time::{Duration, Instant}; +use crate::actors::ActorStream; +// use crate::executor::thread_pool::MaximThreadPool; +use crate::prelude::*; + mod thread_pool; /// The Executor is responsible for the high-level scheduling of Actors. When an Actor is @@ -21,51 +29,29 @@ mod thread_pool; #[derive(Clone)] pub(crate) struct MaximExecutor { /// The system's "is shutting down" flag. - shutdown_triggered: Arc<(Mutex, Condvar)>, - /// Barrier to await shutdown on. - thread_pool: Arc, + shutdown_triggered: ChangeNotifier>, + // /// Barrier to await shutdown on. + // thread_pool: Arc, /// Actors that have no messages available. sleeping: Arc>, - /// All Reactors owned by this Executor. - reactors: Arc>, - /// Counting actors per reactor for even distribution of Actors. - actors_per_reactor: Arc>, + // /// All Reactors owned by this Executor. + // reactors: Arc>, + // /// Counting actors per reactor for even distribution of Actors. + // actors_per_reactor: Arc>, } impl MaximExecutor { - /// Creates a new Executor with the given actor system configuration. This will govern the - /// configuration of the executor. - pub(crate) fn new(shutdown_triggered: Arc<(Mutex, Condvar)>) -> Self { + /// Creates a new Executor with the given actor system configuration. This starts the thread pool. + pub(crate) fn new( + config: &ActorSystemConfig, + shutdown_triggered: Arc<(Mutex, Condvar)>, + ) -> Self { Self { shutdown_triggered, - thread_pool: Default::default(), + // thread_pool: Default::default(), sleeping: Default::default(), - reactors: Default::default(), - actors_per_reactor: Default::default(), - } - } - - /// Initializes the executor and starts the MaximReactor instances based on the count of the - /// number of threads configured in the actor system. This must be called before any work can - /// be performed with the actor system. - pub(crate) fn init(&self, system: &ActorSystem) { - for i in 0..system.data.config.thread_pool_size { - let reactor = MaximReactor::new(self.clone(), system, i); - self.reactors.insert(i, reactor.clone()); - self.actors_per_reactor.insert(i, 0); - let sys = system.clone(); - log::info!("Spawning Reactors"); - self.thread_pool - .spawn(format!("Reactor-{}", reactor.name), move || { - sys.init_current(); - futures::executor::enter().expect("Executor nested in other executor"); - loop { - // `MaximReactor::thread` returns true if it's set to be ran again. - if !reactor.thread() { - break; - } - } - }); + // reactors: Default::default(), + // actors_per_reactor: Default::default(), } } diff --git a/src/lib.rs b/src/lib.rs index f6d65c1..23cf224 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -199,8 +199,9 @@ pub use futures; use prelude::*; pub mod actors; -pub mod cluster; -mod executor; +// FIXME: Implement new async cluster support +// pub mod cluster; +// mod executor; pub mod message; pub mod system; diff --git a/src/prelude.rs b/src/prelude.rs index cbbc724..0c740fd 100644 --- a/src/prelude.rs +++ b/src/prelude.rs @@ -8,7 +8,7 @@ pub use crate::actors::RandomAidPool; pub use crate::actors::Status; #[cfg(feature = "actor-pool")] pub use crate::actors::SyncAidPool; -pub use crate::executor::ShutdownResult; +// pub use crate::executor::ShutdownResult; pub use crate::message::Message; pub use crate::system::ActorSystem; pub use crate::system::ActorSystemConfig; diff --git a/src/system.rs b/src/system.rs index e83ae93..0fe3038 100644 --- a/src/system.rs +++ b/src/system.rs @@ -9,16 +9,14 @@ //! //! The user should refer to test cases and examples as "how-to" guides for using Maxim. -#[cfg(feature = "actor-pool")] -use crate::actors::ActorPoolBuilder; -use crate::actors::{Actor, ActorBuilder, ActorStream}; -use crate::executor::MaximExecutor; -use crate::prelude::*; -use crate::system::system_actor::SystemActor; use dashmap::DashMap; use once_cell::sync::OnceCell; use secc::{SeccReceiver, SeccSender}; use serde::{Deserialize, Serialize}; +use uuid::Uuid; +use smol::{Task, Timer}; +use piper::ChangeNotifier; + use std::collections::{BinaryHeap, HashSet}; use std::error::Error; use std::fmt; @@ -27,7 +25,12 @@ use std::sync::{Arc, Condvar, Mutex}; use std::thread; use std::thread::JoinHandle; use std::time::{Duration, Instant}; -use uuid::Uuid; + +#[cfg(feature = "actor-pool")] +use crate::actors::ActorPoolBuilder; +use crate::actors::{Actor, ActorBuilder, ActorStream}; +use crate::prelude::*; +use crate::system::system_actor::SystemActor; mod system_actor; @@ -220,53 +223,18 @@ pub struct RemoteInfo { _handle: JoinHandle<()>, } -/// Stores a message that will be sent to an actor with a delay. -struct DelayedMessage { - /// A unique identifier for a message. - uuid: Uuid, - /// The Aid that the message will be sent to. - destination: Aid, - /// The minimum instant that the message should be sent. - instant: Instant, - /// The message to sent. - message: Message, -} - -impl std::cmp::PartialEq for DelayedMessage { - fn eq(&self, other: &Self) -> bool { - self.uuid == other.uuid - } -} - -impl std::cmp::Eq for DelayedMessage {} - -impl std::cmp::PartialOrd for DelayedMessage { - fn partial_cmp(&self, other: &DelayedMessage) -> Option { - Some(other.instant.cmp(&self.instant)) // Uses an inverted sort. - } -} - -impl std::cmp::Ord for DelayedMessage { - fn cmp(&self, other: &Self) -> std::cmp::Ordering { - self.partial_cmp(other) - .expect("DelayedMessage::partial_cmp() returned None; can't happen") - } -} - /// Contains the inner data used by the actor system. pub(crate) struct ActorSystemData { /// Unique version 4 UUID for this actor system. pub(crate) uuid: Uuid, /// The config for the actor system which was passed to it when created. pub(crate) config: ActorSystemConfig, - /// Holds handles to the pool of threads processing the work channel. - threads: Mutex>>, - /// The Executor responsible for managing the runtime of the Actors - executor: MaximExecutor, + // /// The Executor responsible for managing the runtime of the Actors + // executor: MaximExecutor, /// Whether the ActorSystem has started or not. started: AtomicBool, /// A flag and condvar that can be used to send a signal when the system begins to shutdown. - shutdown_triggered: Arc<(Mutex, Condvar)>, + shutdown_triggered: ChangeNotifier>, /// Holds the [`Actor`] objects keyed by the [`Aid`]. actors_by_aid: Arc>>, /// Holds a map of the actor ids by the UUID in the actor id. UUIDs of actor ids are assigned @@ -280,8 +248,6 @@ pub(crate) struct ActorSystemData { monitoring_by_monitored: Arc>>, /// Holds a map of information objects about links to remote actor systems. remotes: Arc>, - /// Holds the messages that have been enqueued for delayed send. - delayed_messages: Arc<(Mutex>, Condvar)>, } /// An actor system that contains and manages the actors spawned inside it. @@ -299,10 +265,27 @@ impl ActorSystem { /// on in order to satisfy the requirements of the software they are creating. pub fn create(config: ActorSystemConfig) -> ActorSystem { let uuid = Uuid::new_v4(); - let threads = Mutex::new(Vec::with_capacity(config.thread_pool_size as usize)); - let shutdown_triggered = Arc::new((Mutex::new(false), Condvar::new())); - let executor = MaximExecutor::new(shutdown_triggered.clone()); + log::trace!("Starting executor thread pool"); + + // Flag to indicate the system should shutdown + let shutdown_triggered = ChangeNotifier::new(Arc::new(AtomicBool::new(false))); + + // Create an executor thread pool. + let mut threads = Vec::with_capacity(config.thread_pool_size as usize); + let shutdown_triggered_ = shutdown_triggered.clone(); + for _ in 0..config.thread_pool_size { + let shutdown_triggered__ = shutdown_triggered_.clone(); + // Spawn an executor thread that waits for the shutdown signal. + threads.push(thread::spawn(move || smol::run(async move { + loop { + shutdown_triggered__.listen().await; + if shutdown_triggered__.load(Ordering::SeqCst) { + break; + } + } + }))); + } let start_on_launch = config.start_on_launch; @@ -311,8 +294,6 @@ impl ActorSystem { data: Arc::new(ActorSystemData { uuid, config, - threads, - executor, started: AtomicBool::new(false), shutdown_triggered, actors_by_aid: Arc::new(DashMap::default()), @@ -320,7 +301,6 @@ impl ActorSystem { aids_by_name: Arc::new(DashMap::default()), monitoring_by_monitored: Arc::new(DashMap::default()), remotes: Arc::new(DashMap::default()), - delayed_messages: Arc::new((Mutex::new(BinaryHeap::new()), Condvar::new())), }), }; @@ -340,21 +320,6 @@ impl ActorSystem { .compare_and_swap(false, true, Ordering::Relaxed) { log::info!("ActorSystem {} has spawned", self.data.uuid); - self.data.executor.init(self); - - // We have the thread pool in a mutex to avoid a chicken & egg situation with the actor - // system not being created yet but needed by the thread. We put this code in a block to - // get around rust borrow constraints without unnecessarily copying things. - { - let mut guard = self.data.threads.lock().unwrap(); - - // Start the thread that reads from the `delayed_messages` queue. - // FIXME Put in ability to confirm how many of these to start. - for _ in 0..1 { - let thread = self.start_send_after_thread(); - guard.push(thread); - } - } // Launch the SystemActor and give it the name "System" self.spawn() @@ -364,45 +329,6 @@ impl ActorSystem { } } - /// Starts a thread that monitors the delayed_messages and sends the messages when their - /// delays have elapsed. - // FIXME Add a graceful shutdown to this thread and notifications. - fn start_send_after_thread(&self) -> JoinHandle<()> { - let system = self.clone(); - let delayed_messages = self.data.delayed_messages.clone(); - thread::spawn(move || { - while !*system.data.shutdown_triggered.0.lock().unwrap() { - let (ref mutex, ref condvar) = &*delayed_messages; - let mut data = mutex.lock().unwrap(); - match data.peek() { - None => { - // wait to be notified something is added. - let _ = condvar.wait(data).unwrap(); - } - Some(msg) => { - let now = Instant::now(); - if now >= msg.instant { - log::trace!("Sending delayed message"); - msg.destination - .send(msg.message.clone()) - .unwrap_or_else(|error| { - log::warn!( - "Cannot send scheduled message to {}: Error {:?}", - msg.destination, - error - ); - }); - data.pop(); - } else { - let duration = msg.instant.duration_since(now); - let _result = condvar.wait_timeout(data, duration).unwrap(); - } - } - } - } - }) - } - /// Returns a reference to the config for this actor system. pub fn config(&self) -> &ActorSystemConfig { &self.data.config @@ -416,117 +342,118 @@ impl ActorSystem { .map(|info| info.sender.clone()) } - /// Adds a connection to a remote actor system. When the connection is established the - /// actor system will announce itself to the remote system with a [`WireMessage::Hello`]. - pub fn connect( - &self, - sender: &SeccSender, - receiver: &SeccReceiver, - ) -> Uuid { - // Announce ourselves to the other system and get their info. - let hello = WireMessage::Hello { - system_actor_aid: self.system_actor_aid(), - }; - sender.send(hello).unwrap(); - log::debug!("Sending hello from {}", self.data.uuid); - - // FIXME (Issue #75) Make error handling in ActorSystem::connect more robust. - let system_actor_aid = - match receiver.receive_await_timeout(self.data.config.thread_wait_time) { - Ok(message) => match message { - WireMessage::Hello { system_actor_aid } => system_actor_aid, - _ => panic!("Expected first message to be a Hello"), - }, - Err(e) => panic!("Expected to read a Hello message {:?}", e), - }; - - // Starts a thread to read incoming wire messages and process them. - let system = self.clone(); - let receiver_clone = receiver.clone(); - let thread_timeout = self.data.config.thread_wait_time; - let sys_uuid = system_actor_aid.system_uuid(); - let handle = thread::spawn(move || { - system.init_current(); - // FIXME (Issue #76) Add graceful shutdown for threads handling remotes including - // informing remote that the system is exiting. - while !*system.data.shutdown_triggered.0.lock().unwrap() { - match receiver_clone.receive_await_timeout(thread_timeout) { - Err(_) => (), // not an error, just loop and try again. - Ok(wire_msg) => system.process_wire_message(&sys_uuid, &wire_msg), - } - } - }); - - // Save the info and thread to the remotes map. - let info = RemoteInfo { - system_uuid: system_actor_aid.system_uuid(), - sender: sender.clone(), - receiver: receiver.clone(), - _handle: handle, - system_actor_aid, - }; - - let uuid = info.system_uuid; - self.data.remotes.insert(uuid.clone(), info); - uuid - } - - /// Disconnects this actor system from the remote actor system with the given UUID. - // FIXME Connectivity management needs a lot of work and testing. - pub fn disconnect(&self, system_uuid: Uuid) -> Result<(), AidError> { - self.data.remotes.remove(&system_uuid); - Ok(()) - } - - /// Connects two actor systems using two channels directly. This can be used as a utility - /// in testing or to link two actor systems directly within the same process. - pub fn connect_with_channels(system1: &ActorSystem, system2: &ActorSystem) { - let (tx1, rx1) = secc::create::(32, system1.data.config.thread_wait_time); - let (tx2, rx2) = secc::create::(32, system2.data.config.thread_wait_time); - - // We will do this in 2 threads because one connect would block waiting on a message - // from the other actor system, causing a deadlock. - let system1_clone = system1.clone(); - let system2_clone = system2.clone(); - let h1 = thread::spawn(move || system1_clone.connect(&tx1, &rx2)); - let h2 = thread::spawn(move || system2_clone.connect(&tx2, &rx1)); - - // Wait for the completion of the connection. - h1.join().unwrap(); - h2.join().unwrap(); - } - - /// A helper function to process a wire message from another actor system. The passed uuid - /// is the uuid of the remote that sent the message. - // FIXME (Issue #74) Make error handling in ActorSystem::process_wire_message more robust. - fn process_wire_message(&self, _uuid: &Uuid, wire_message: &WireMessage) { - match wire_message { - WireMessage::ActorMessage { - actor_uuid, - system_uuid, - message, - } => { - if let Some(aid) = self.find_aid(&system_uuid, &actor_uuid) { - aid.send(message.clone()).unwrap_or_else(|error| { - log::warn!("Could not send wire message to {}. Error: {}", aid, error); - }) - } - } - WireMessage::DelayedActorMessage { - duration, - actor_uuid, - system_uuid, - message, - } => { - self.find_aid(&system_uuid, &actor_uuid) - .map(|aid| self.send_after(message.clone(), aid, *duration)) - .expect("Error not handled yet"); - } - WireMessage::Hello { system_actor_aid } => { - log::debug!("{:?} Got Hello from {}", self.data.uuid, system_actor_aid); - } - } - } + // FIXME: Reimplement cluster support + // /// Adds a connection to a remote actor system. When the connection is established the + // /// actor system will announce itself to the remote system with a [`WireMessage::Hello`]. + // pub fn connect( + // &self, + // sender: &SeccSender, + // receiver: &SeccReceiver, + // ) -> Uuid { + // // Announce ourselves to the other system and get their info. + // let hello = WireMessage::Hello { + // system_actor_aid: self.system_actor_aid(), + // }; + // sender.send(hello).unwrap(); + // debug!("Sending hello from {}", self.data.uuid); + + // // FIXME (Issue #75) Make error handling in ActorSystem::connect more robust. + // let system_actor_aid = + // match receiver.receive_await_timeout(self.data.config.thread_wait_time) { + // Ok(message) => match message { + // WireMessage::Hello { system_actor_aid } => system_actor_aid, + // _ => panic!("Expected first message to be a Hello"), + // }, + // Err(e) => panic!("Expected to read a Hello message {:?}", e), + // }; + + // // Starts a thread to read incoming wire messages and process them. + // let system = self.clone(); + // let receiver_clone = receiver.clone(); + // let thread_timeout = self.data.config.thread_wait_time; + // let sys_uuid = system_actor_aid.system_uuid(); + // let handle = thread::spawn(move || { + // system.init_current(); + // // FIXME (Issue #76) Add graceful shutdown for threads handling remotes including + // // informing remote that the system is exiting. + // while !*system.data.shutdown_triggered.0.lock().unwrap() { + // match receiver_clone.receive_await_timeout(thread_timeout) { + // Err(_) => (), // not an error, just loop and try again. + // Ok(wire_msg) => system.process_wire_message(&sys_uuid, &wire_msg), + // } + // } + // }); + + // // Save the info and thread to the remotes map. + // let info = RemoteInfo { + // system_uuid: system_actor_aid.system_uuid(), + // sender: sender.clone(), + // receiver: receiver.clone(), + // _handle: handle, + // system_actor_aid, + // }; + + // let uuid = info.system_uuid; + // self.data.remotes.insert(uuid.clone(), info); + // uuid + // } + + // /// Disconnects this actor system from the remote actor system with the given UUID. + // // FIXME Connectivity management needs a lot of work and testing. + // pub fn disconnect(&self, system_uuid: Uuid) -> Result<(), AidError> { + // self.data.remotes.remove(&system_uuid); + // Ok(()) + // } + + // /// Connects two actor systems using two channels directly. This can be used as a utility + // /// in testing or to link two actor systems directly within the same process. + // pub fn connect_with_channels(system1: &ActorSystem, system2: &ActorSystem) { + // let (tx1, rx1) = secc::create::(32, system1.data.config.thread_wait_time); + // let (tx2, rx2) = secc::create::(32, system2.data.config.thread_wait_time); + + // // We will do this in 2 threads because one connect would block waiting on a message + // // from the other actor system, causing a deadlock. + // let system1_clone = system1.clone(); + // let system2_clone = system2.clone(); + // let h1 = thread::spawn(move || system1_clone.connect(&tx1, &rx2)); + // let h2 = thread::spawn(move || system2_clone.connect(&tx2, &rx1)); + + // // Wait for the completion of the connection. + // h1.join().unwrap(); + // h2.join().unwrap(); + // } + + // /// A helper function to process a wire message from another actor system. The passed uuid + // /// is the uuid of the remote that sent the message. + // // FIXME (Issue #74) Make error handling in ActorSystem::process_wire_message more robust. + // fn process_wire_message(&self, _uuid: &Uuid, wire_message: &WireMessage) { + // match wire_message { + // WireMessage::ActorMessage { + // actor_uuid, + // system_uuid, + // message, + // } => { + // if let Some(aid) = self.find_aid(&system_uuid, &actor_uuid) { + // aid.send(message.clone()).unwrap_or_else(|error| { + // warn!("Could not send wire message to {}. Error: {}", aid, error); + // }) + // } + // } + // WireMessage::DelayedActorMessage { + // duration, + // actor_uuid, + // system_uuid, + // message, + // } => { + // self.find_aid(&system_uuid, &actor_uuid) + // .map(|aid| self.send_after(message.clone(), aid, *duration)) + // .expect("Error not handled yet"); + // } + // WireMessage::Hello { system_actor_aid } => { + // debug!("{:?} Got Hello from {}", self.data.uuid, system_actor_aid); + // } + // } + // } /// Initializes this actor system to use for the current thread which is necessary if the /// user wishes to serialize and deserialize [`Aid`]s. @@ -560,86 +487,88 @@ impl ActorSystem { /// Triggers a shutdown but doesn't wait for the Reactors to stop. pub fn trigger_shutdown(&self) { - let (ref mutex, ref condvar) = &*self.data.shutdown_triggered; - *mutex.lock().unwrap() = true; - condvar.notify_all(); + self.data.shutdown_triggered.store(true, Ordering::SeqCst); } /// Awaits the Executor shutting down all Reactors. This is backed by a barrier that Reactors /// will wait on after [`ActorSystem::trigger_shutdown`] is called, blocking until all Reactors /// have stopped. - pub fn await_shutdown(&self, timeout: impl Into>) -> ShutdownResult { - log::info!("System awaiting shutdown"); - - let start = Instant::now(); - let timeout = timeout.into(); - - let result = match timeout { - Some(dur) => self.await_shutdown_trigger_with_timeout(dur), - None => self.await_shutdown_trigger_without_timeout(), - }; - - if let Some(r) = result { - return r; - } - - let timeout = { - match timeout { - Some(timeout) => { - let elapsed = Instant::now().duration_since(start); - if let Some(t) = timeout.checked_sub(elapsed) { - Some(t) - } else { - return ShutdownResult::TimedOut; - } - } - None => None, - } - }; - - // Wait for the executor to finish shutting down - self.data.executor.await_shutdown(timeout) + pub fn await_shutdown(&self, timeout: impl Into>) { + unimplemented!("FIXME: Reimplement graceful shutdown"); + // info!("System awaiting shutdown"); + + // let start = Instant::now(); + // let timeout = timeout.into(); + + // let result = match timeout { + // Some(dur) => self.await_shutdown_trigger_with_timeout(dur), + // None => self.await_shutdown_trigger_without_timeout(), + // }; + + // if let Some(r) = result { + // return r; + // } + + // let timeout = { + // match timeout { + // Some(timeout) => { + // let elapsed = Instant::now().duration_since(start); + // if let Some(t) = timeout.checked_sub(elapsed) { + // Some(t) + // } else { + // return ShutdownResult::TimedOut; + // } + // } + // None => None, + // } + // }; + + // // Wait for the executor to finish shutting down + // self.data.executor.await_shutdown(timeout) } - fn await_shutdown_trigger_with_timeout(&self, mut dur: Duration) -> Option { - let (ref mutex, ref condvar) = &*self.data.shutdown_triggered; - let mut guard = mutex.lock().unwrap(); - while !*guard { - let started = Instant::now(); - let (new_guard, timeout) = match condvar.wait_timeout(guard, dur) { - Ok(ret) => ret, - Err(_) => return Some(ShutdownResult::Panicked), - }; - - if timeout.timed_out() { - return Some(ShutdownResult::TimedOut); - } - - guard = new_guard; - dur -= started.elapsed(); - } - None + fn await_shutdown_trigger_with_timeout(&self, mut dur: Duration) { + unimplemented!("FIXME: Re-implement graceful shutdown") + // let (ref mutex, ref condvar) = &*self.data.shutdown_triggered; + // let mut guard = mutex.lock().unwrap(); + // while !*guard { + // let started = Instant::now(); + // let (new_guard, timeout) = match condvar.wait_timeout(guard, dur) { + // Ok(ret) => ret, + // Err(_) => return Some(ShutdownResult::Panicked), + // }; + + // if timeout.timed_out() { + // return Some(ShutdownResult::TimedOut); + // } + + // guard = new_guard; + // dur -= started.elapsed(); + // } + // None } - fn await_shutdown_trigger_without_timeout(&self) -> Option { - let (ref mutex, ref condvar) = &*self.data.shutdown_triggered; - let mut guard = mutex.lock().unwrap(); - while !*guard { - guard = match condvar.wait(guard) { - Ok(ret) => ret, - Err(_) => return Some(ShutdownResult::Panicked), - }; - } - None + fn await_shutdown_trigger_without_timeout(&self) { + unimplemented!("FIXME: Re-implement graceful shutdown"); + // let (ref mutex, ref condvar) = &*self.data.shutdown_triggered; + // let mut guard = mutex.lock().unwrap(); + // while !*guard { + // guard = match condvar.wait(guard) { + // Ok(ret) => ret, + // Err(_) => return Some(ShutdownResult::Panicked), + // }; + // } + // None } /// Triggers a shutdown of the system and returns only when all Reactors have shutdown. pub fn trigger_and_await_shutdown( &self, timeout: impl Into>, - ) -> ShutdownResult { - self.trigger_shutdown(); - self.await_shutdown(timeout) + ) { + unimplemented!("FIXME: Re-implement graceful shutdown"); + // self.trigger_shutdown(); + // self.await_shutdown(timeout) } // An internal helper to register an actor in the actor system. @@ -661,7 +590,7 @@ impl ActorSystem { } actors_by_aid.insert(aid.clone(), actor); aids_by_uuid.insert(aid.uuid(), aid.clone()); - self.data.executor.register_actor(stream); + // self.data.executor.register_actor(stream); aid.send(Message::new(SystemMsg::Start)).unwrap(); // Actor was just made Ok(aid) } @@ -731,28 +660,6 @@ impl ActorSystem { ) } - /// Schedules the `aid` for work. Note that this is the only time that we have to use the - /// lookup table. This function gets called when an actor goes from 0 receivable messages to - /// 1 receivable message. If the actor has more receivable messages then this will not be - /// needed to be called because the dispatcher threads will handle the process of resending - /// the actor to the work channel. - // TODO Put tests verifying the resend on multiple messages. - pub(crate) fn schedule(&self, aid: Aid) { - let actors_by_aid = &self.data.actors_by_aid; - if actors_by_aid.contains_key(&aid) { - self.data.executor.wake(aid); - } else { - // The actor was removed from the map so ignore the problem and just log - // a warning. - log::warn!( - "Attempted to schedule actor with aid {:?} on system with node_id {:?} but - the actor does not exist.", - aid, - self.data.uuid.to_string(), - ); - } - } - /// Stops an actor by shutting down its channels and removing it from the actors list and /// telling the [`Aid`] to not allow messages to be sent to the actor since the receiving /// side of the actor is gone. @@ -865,22 +772,11 @@ impl ActorSystem { /// not be sent on exactly the delay passed. However, the message will never be sent before /// the given delay. pub(crate) fn send_after(&self, message: Message, destination: Aid, delay: Duration) { - let instant = Instant::now().checked_add(delay).unwrap(); - let entry = DelayedMessage { - uuid: Uuid::new_v4(), - destination, - instant, - message, - }; - let (ref mutex, ref condvar) = &*self.data.delayed_messages; - let mut data = mutex.lock().unwrap(); - data.push(entry); - condvar.notify_all(); - } + Task::spawn(async move { + Timer::after(delay).await; - #[cfg(test)] - pub(crate) fn executor(&self) -> &MaximExecutor { - &self.data.executor + destination.send(message); + }).detach(); } } @@ -907,13 +803,15 @@ mod tests { fn start_and_connect_two_systems() -> (ActorSystem, ActorSystem) { let system1 = ActorSystem::create(ActorSystemConfig::default().thread_pool_size(2)); let system2 = ActorSystem::create(ActorSystemConfig::default().thread_pool_size(2)); - ActorSystem::connect_with_channels(&system1, &system2); + unimplemented!("FIXME: Re-implement cluster support"); + // ActorSystem::connect_with_channels(&system1, &system2); (system1, system2) } /// Helper to wait for 2 actor systems to shutdown or panic if they don't do so within /// 2000 milliseconds. fn await_two_system_shutdown(system1: ActorSystem, system2: ActorSystem) { + unimplemented!("FIXME: Re-implement graceful shutdown"); let h1 = thread::spawn(move || { system1.await_shutdown(None); }); @@ -930,6 +828,7 @@ mod tests { /// timeout work properly. #[test] fn test_shutdown_await_timeout() { + unimplemented!("FIXME: Re-implement graceful shutdown"); use std::time::Duration; let system = ActorSystem::create(ActorSystemConfig::default().thread_pool_size(2)); @@ -946,16 +845,16 @@ mod tests { .unwrap(); // Expecting to timeout - assert_eq!( - system.await_shutdown(Duration::from_millis(10)), - ShutdownResult::TimedOut - ); + // assert_eq!( + // system.await_shutdown(Duration::from_millis(10)), + // ShutdownResult::TimedOut + // ); // Expecting to NOT timeout - assert_eq!( - system.await_shutdown(Duration::from_millis(200)), - ShutdownResult::Ok - ); + // assert_eq!( + // system.await_shutdown(Duration::from_millis(200)), + // ShutdownResult::Ok + // ); // Validate that if the system is already shutdown the method doesn't hang. // FIXME Design a means that this cannot ever hang the test. @@ -1128,7 +1027,8 @@ mod tests { fn test_connect_with_channels() { let system1 = ActorSystem::create(ActorSystemConfig::default().thread_pool_size(2)); let system2 = ActorSystem::create(ActorSystemConfig::default().thread_pool_size(2)); - ActorSystem::connect_with_channels(&system1, &system2); + unimplemented!("FIXME: Re-implement cluster support"); + // ActorSystem::connect_with_channels(&system1, &system2); { system1 .data From ec3aac14e862d525ab7daa0c998cab173acc2f75 Mon Sep 17 00:00:00 2001 From: Zicklag Date: Mon, 4 May 2020 03:41:27 +0000 Subject: [PATCH 5/9] WIP SECC Channel Based on Flume --- Cargo.toml | 1 + lib/piper | 2 +- src/lib.rs | 4 + src/secc.rs | 310 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 316 insertions(+), 1 deletion(-) create mode 100644 src/secc.rs diff --git a/Cargo.toml b/Cargo.toml index dc3b920..f152419 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -61,6 +61,7 @@ serde_json = "^1.0.40" [dependencies] smol = "0.1.4" # Async Executor piper = { path = "lib/piper" } # Async pipes, channels, mutexes, and more +flume = { version = "0.7.1", default-features = false, features = ["async"] } futures = "0.3.1" # Async Utils serde = { version = "1.0.97", features = ["derive", "rc"] } # Serialization support bincode = "1.1.4" # RPC Serialization diff --git a/lib/piper b/lib/piper index c7f72a9..73e7b41 160000 --- a/lib/piper +++ b/lib/piper @@ -1 +1 @@ -Subproject commit c7f72a9209a4ab04794322bbf7b0d64aadc07a90 +Subproject commit 73e7b4121da3edf4fe99463c179c85de7dafe828 diff --git a/src/lib.rs b/src/lib.rs index 23cf224..103a237 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -198,6 +198,9 @@ use std::fmt::{Display, Formatter}; pub use futures; use prelude::*; +// Skip Enabled Concurrent Channels +pub(crate) mod secc; + pub mod actors; // FIXME: Implement new async cluster support // pub mod cluster; @@ -246,6 +249,7 @@ mod tests { use std::time::Duration; use log::LevelFilter; + extern crate secc; use secc::{SeccReceiver, SeccSender}; use serde::{Deserialize, Serialize}; diff --git a/src/secc.rs b/src/secc.rs new file mode 100644 index 0000000..2b0e0f0 --- /dev/null +++ b/src/secc.rs @@ -0,0 +1,310 @@ +//! Async SECC ( Skip Enabled Concurrent Channel ) implementation based on [`flume`]. +//! +//! This is the channel implementation used by actors to send and receive messages. + +use std::collections::VecDeque; + +/// Create an unbounded SECC channel +/// +/// > **note:** The type `T` should be efficiently clonable as calls to [`SeccReceiver::peek`] +/// > must clone the value. Using an [`Arc`] is one way to do this. +pub fn secc_unbounded() -> (SeccSender, SeccReceiver) { + let (flume_sender, flume_receiver) = flume::unbounded(); + + ( + SeccSender::new(flume_sender), + SeccReceiver::new(flume_receiver), + ) +} + +/// Create a bounded SECC channel +/// +/// > **note:** The type `T` should be efficiently clonable as calls to [`SeccReceiver::peek`] +/// > must clone the value. Using an [`Arc`] is one way to do this. +pub fn secc_bounded(capacity: usize) -> (SeccSender, SeccReceiver) { + let (flume_sender, flume_receiver) = flume::bounded(capacity); + + ( + SeccSender::new(flume_sender), + SeccReceiver::new(flume_receiver), + ) +} + +/// A SECC sender, which is actually just a newtype over a `flume::Sender`. +/// +/// Implemented as a newtype just in case we have to add more to it later, so that we can modify +/// its internals without breaking its usage. +#[derive(Clone)] +pub struct SeccSender(flume::Sender); + +impl SeccSender { + // Create a [`SeccSender`] from a `flume` Sender. + fn new(sender: flume::Sender) -> Self { + SeccSender(sender) + } + + /// See [`flume::Sender::send`]. + pub fn send(&self, msg: T) -> Result<(), flume::SendError> { + self.0.send(msg) + } + + /// See [`flume::Sender::try_send`]. + pub fn try_send(&self, msg: T) -> Result<(), flume::TrySendError> { + self.0.try_send(msg) + } +} + +/// A receiver for a SECC channel. It is a wrapper around a flume reciever along with a skipped +/// messages queue that is used to store any messages that are skipped with the skip function. +pub struct SeccReceiver { + /// The underlying flume channel receiver + receiver: flume::Receiver, + /// A message that has been received and peeked with `peek()` + peeked_message: Option, + /// The queue of messages that have been skipped by the receiver + skipped: VecDeque, + /// The index in the skipped deque at which to stop resetting + reset_until: usize, + /// Whether or not we are currently in the process of resetting a skip + is_resetting: bool, +} + +impl SeccReceiver { + // Create a [`SeccReceiver`] from a `flume` Receiver. + fn new(receiver: flume::Receiver) -> Self { + SeccReceiver { + receiver, + peeked_message: None, + skipped: VecDeque::new(), + is_resetting: false, + reset_until: 0, + } + } + + /// Peek at the next message in the channel + pub async fn peek(&mut self) -> Result { + // If we already have a peeked message, return it + if let Some(msg) = &self.peeked_message { + Ok(msg.clone()) + + // If we are resetting, peek the message from the skipped queue + } else if self.is_resetting { + // Get the next message in the queue + if let Some(msg) = self.skipped.get(0) { + Ok(msg.clone()) + + } else { + unreachable!("If we are resetting there shoud always be a message in the \ + skipped queue."); + } + + // If we don't already have a peeked message and we aren't resetting + } else { + // Get the next message in the channel + let msg = self.receiver.recv_async().await?; + + // Clone it and put it in our peeked message slot + self.peeked_message = Some(msg.clone()); + + // Return the message + Ok(msg) + } + } + + /// Receive the next message in the channel + pub async fn recv(&mut self) -> Result { + // If we are currently resetting + if self.is_resetting { + // Pop the next message off of the skipped queue + if let Some(msg) = self.skipped.pop_front() { + // Decrement the reset until cursor to make sure it stays pointing at the same message + self.reset_until -= 1; + + // If this was the last message we were supposed to reset until + if self.reset_until == 0 { + // Go out of resetting mode + self.is_resetting = false; + } + + Ok(msg) + // If there is no message, go out of resetting mode and return the next message in the channel + } else { + self.is_resetting = false; + self.receiver.recv_async().await + } + + // If we have a peeked message, return that one + } else if let Some(msg) = self.peeked_message.take() { + Ok(msg) + + // Get the message from the channel + } else { + self.receiver.recv_async().await + } + } + + /// Skip the next message in the channel + pub async fn skip(&mut self) -> Result<(), flume::RecvError> { + // Get the message to skip + let msg = + // If we have a peeked message skip that one + if let Some(msg) = self.peeked_message.take() { + msg + + // If we are resetting, skip the one off of the top of the skipped queue + } else if self.is_resetting { + if let Some(msg) = self.skipped.pop_front() { + msg + } else { + unreachable!("If we are resetting there should be a message in the skipped \ + queue."); + } + + // Otherwise, get the next message from the channel and skip it + } else { + self.receiver.recv_async().await? + }; + + // Add it to the skipped message queue + self.skipped.push_back(msg); + + Ok(()) + } + + /// Causes `recv` to return previously skipped messages untill ther are none, where it starts + /// collecting the messages from the channel again + pub fn reset_skip(&mut self) { + // Go into resetting mode + self.is_resetting = true; + // Reset until the end of the skipped message queue + self.reset_until = self.skipped.len() - 1; + } +} + +#[cfg(test)] +mod test { + use super::*; + + enum Mode { + Bounded(usize), + Unbounded, + } + + fn get_channel(mode: Mode) -> (SeccSender, SeccReceiver) { + match mode { + Mode::Bounded(capacity) => secc_bounded(capacity), + Mode::Unbounded => secc_unbounded(), + } + } + + fn basic(mode: Mode) { + smol::run(async move { + // Create a secc channel + let (sender, mut receiver) = secc_bounded(100); + + // Send a message + sender.send(0).unwrap(); + + // Receive the message + assert_eq!(receiver.recv().await.unwrap(), 0); + + // Send another message + sender.send(1).unwrap(); + + // Peek at the message + assert_eq!(receiver.peek().await.unwrap(), 1); + + // Send another message + sender.send(2).unwrap(); + + // Peek at the message again ( it shouldn't change ) + assert_eq!(receiver.peek().await.unwrap(), 1); + + // Receive the next message ( it should be the peeked one ) + assert_eq!(receiver.recv().await.unwrap(), 1); + + // Receive the next message ( it should be the next one in line ) + assert_eq!(receiver.recv().await.unwrap(), 2); + + // Send 4 new messages + sender.send(3).unwrap(); + sender.send(4).unwrap(); + sender.send(5).unwrap(); + sender.send(6).unwrap(); + + // Peek at the next message + assert_eq!(receiver.peek().await.unwrap(), 3); + + // Skip the message ( should skip 3 ) + receiver.skip().await.unwrap(); + + // Skip the next message as well ( should skip 4 ) + receiver.skip().await.unwrap(); + + // Peek the next message ( should be 5 ) + assert_eq!(receiver.peek().await.unwrap(), 5); + + // Receive the next message ( should be 5 ) + assert_eq!(receiver.recv().await.unwrap(), 5); + + // Reset the skip + receiver.reset_skip(); + + // Receive the next message ( should be previously skipped 3 ) + assert_eq!(receiver.recv().await.unwrap(), 3); + + // Receive the next message ( should be previously skipped 4 ) + assert_eq!(receiver.recv().await.unwrap(), 4); + + // Receive the next message ( should be 6, the next one after previously skipped ones ) + assert_eq!(receiver.recv().await.unwrap(), 6); + + // Send 5 new messages + sender.send(7).unwrap(); + sender.send(8).unwrap(); + sender.send(9).unwrap(); + sender.send(10).unwrap(); + sender.send(11).unwrap(); + + // Skip the next two messages ( skips 7 and 8 ) + receiver.skip().await.unwrap(); + receiver.skip().await.unwrap(); + + // Receive the next message ( should be 9 ) + assert_eq!(receiver.recv().await.unwrap(), 9); + + // Reset skip + receiver.reset_skip(); + + // Peek the next message ( should be the previously skipped message 7 ) + assert_eq!(receiver.peek().await.unwrap(), 7); + + // Skip this message ( 7 ) + receiver.skip().await.unwrap(); + + // Receive the next message ( should be previously skipped message 8 ) + assert_eq!(receiver.recv().await.unwrap(), 8); + + // Receive the next message ( should be 10 as we've exhausted our skip queue ) + assert_eq!(receiver.recv().await.unwrap(), 10); + + // Reset skip + receiver.reset_skip(); + + // Receive the next message ( should be 7 which was skipped a second time earlier ) + assert_eq!(receiver.recv().await.unwrap(), 7); + + // Receive the next message ( should be 11 ) + assert_eq!(receiver.recv().await.unwrap(), 11); + }); + } + #[test] + fn bounded_basic() { + basic(Mode::Bounded(100)); + } + + #[test] + fn unbounded_basic() { + basic(Mode::Unbounded); + } +} From e3d228e4361885fe5a0d59db17607e2d0ef7445c Mon Sep 17 00:00:00 2001 From: Zicklag Date: Wed, 6 May 2020 01:03:26 +0000 Subject: [PATCH 6/9] Switch to the `tracing` Crate for Logging - Also includes some Rustfmt changes --- Cargo.toml | 19 ++++++++++-- examples/philosophers.rs | 48 +++++++++++----------------- src/actors.rs | 37 +++++++++++----------- src/cluster.rs | 11 +++---- src/executor.rs | 28 ++++++++--------- src/executor/thread_pool.rs | 12 +++---- src/lib.rs | 12 ++++--- src/system.rs | 62 ++++++++++++++++++------------------- src/system/system_actor.rs | 6 ++-- 9 files changed, 118 insertions(+), 117 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index f152419..ea870d7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -41,11 +41,24 @@ exclude = [ [features] default = ["actor-pool"] + +# Actor System Features + +# Enable the use of AidPool's actor-pool = ["rand", "rand_xoshiro"] + # Automatically determine the default number of threads from the number of CPUs. Adds the # num_cpus dependency. auto-num-threads = ["num_cpus"] +# Logging features + +# Disables trace logging for release builds ( may improve performance for release builds ) +no-release-trace-logging = ["tracing/release_max_level_trace"] + +# Enables logging messages created by the `log` crate as well as the `tracing` crate +enable-log-crate = ["tracing/log"] + [badges] # We won't be using Travis in a bit #travis-ci = { repository = "katharostech/maxim" } @@ -54,7 +67,7 @@ is-it-maintained-open-issues = { repository = "katharostech/maxim" } maintenance = { status = "actively-developed" } [dev-dependencies] -env_logger = "^0.7.1" +tracing-subscriber = "0.2.5" rand = "^0.7" serde_json = "^1.0.40" @@ -66,7 +79,9 @@ futures = "0.3.1" # Async Utils serde = { version = "1.0.97", features = ["derive", "rc"] } # Serialization support bincode = "1.1.4" # RPC Serialization dashmap = "1.0.3" # Concurrent hashmap -log = "0.4" # Logging Facade +# TODO: Make sure we enable minimal required features for tracing and tracing futures +tracing = "0.1.13" # Tracing ( like logging with spans and extra event data ) +tracing-futures = { version = "0.2.4", default-features = false, features = ["futures-03"] } uuid = { version = "0.8.1", features = ["serde", "v4"]} # UUID Generation secc = "0.0.10" once_cell = "1.0.2" diff --git a/examples/philosophers.rs b/examples/philosophers.rs index a067883..48c4036 100644 --- a/examples/philosophers.rs +++ b/examples/philosophers.rs @@ -20,8 +20,9 @@ //! panics ensue. Some FSM implementations might be quite a bit more lose, preferring to ignore //! badly timed messages. This is largely up to the user. -use log::LevelFilter; use maxim::prelude::*; +use tracing::{error, info}; + use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::env; @@ -89,20 +90,17 @@ impl Fork { // Resetting the skip allows fork requests to be processed. Ok(Status::reset(self)) } else { - log::error!( + error!( "[{}] fork_put_down() from non-owner: {} real owner is: {}", - context.aid, - sender, - owner + context.aid, sender, owner ); Ok(Status::done(self)) } } None => { - log::error!( + error!( "[{}] fork_put_down() from non-owner: {} real owner is: None:", - context.aid, - sender + context.aid, sender ); Ok(Status::done(self)) } @@ -121,12 +119,12 @@ impl Fork { // has been marked as being dirty. Ok(Status::reset(self)) } else { - log::error!("[{}] Got UsingFork from non-owner: {}", context.aid, sender); + error!("[{}] Got UsingFork from non-owner: {}", context.aid, sender); Ok(Status::done(self)) } } _ => { - log::error!("[{}] Got UsingFork from non-owner: {}", context.aid, sender); + error!("[{}] Got UsingFork from non-owner: {}", context.aid, sender); Ok(Status::done(self)) } } @@ -319,10 +317,10 @@ impl Philosopher { self.request_missing_forks(context)?; } PhilosopherState::Hungry => { - log::error!("[{}] Got BecomeHungry while eating!", context.aid); + error!("[{}] Got BecomeHungry while eating!", context.aid); } PhilosopherState::Eating => { - log::error!("[{}] Got BecomeHungry while eating!", context.aid); + error!("[{}] Got BecomeHungry while eating!", context.aid); } }; } @@ -373,12 +371,9 @@ impl Philosopher { fork_aid.send_new(ForkCommand::ForkPutDown(context.aid.clone()))?; } } else { - log::error!( + error!( "[{}] Unknown fork asked for: {}:\n left ==> {}\n right ==> {}", - context.aid, - fork_aid, - self.left_fork_aid, - self.right_fork_aid + context.aid, fork_aid, self.left_fork_aid, self.right_fork_aid ); } @@ -443,17 +438,10 @@ struct EndSimulation {} /// actors. pub fn main() { let args: Vec = env::args().collect(); - let level = if args.contains(&"-v".to_string()) { - LevelFilter::Debug - } else { - LevelFilter::Info - }; - - env_logger::builder() - .filter_level(level) - .is_test(true) - .try_init() - .unwrap(); + + tracing_subscriber::fmt() + .with_max_level(tracing::Level::TRACE) + .init(); // FIXME Let the user pass in the number of philosophers at the table, time slice // and runtime as command line parameters. @@ -518,9 +506,9 @@ pub fn main() { // output the results of the simulation and end the program by shutting // down the actor system. if !state.iter().any(|(_, metrics)| metrics.is_none()) { - log::info!("Final Metrics:"); + info!("Final Metrics:"); for (aid, metrics) in state.iter() { - log::info!("{}: {:?}", aid, metrics); + info!("{}: {:?}", aid, metrics); } context.system.trigger_shutdown(); } diff --git a/src/actors.rs b/src/actors.rs index a36d3f6..f88c6ab 100644 --- a/src/actors.rs +++ b/src/actors.rs @@ -663,7 +663,7 @@ impl Aid { pub(crate) fn stop(&self) -> Result<(), AidError> { match &self.data.sender { ActorSender::Local { stopped, .. } => { - log::trace!("Stopping local Actor"); + trace!("Stopping local Actor"); stopped.fetch_or(true, Ordering::AcqRel); Ok(()) } @@ -986,7 +986,7 @@ impl ActorBuilder { F: Processor + 'static, { let (actor, stream) = Actor::new(self.system.clone(), &self, state, processor); - log::debug!("Actor created: {}", actor.context.aid.uuid()); + debug!("Actor created: {}", actor.context.aid.uuid()); self.system.register_actor(actor, stream) } @@ -1151,12 +1151,12 @@ impl Actor { Ok(future) => match AssertUnwindSafe(future).catch_unwind().await { Ok(x) => x, Err(panic) => { - log::warn!("Actor panicked! Catching as error"); + warn!("Actor panicked! Catching as error"); Err(Panic::from(panic).into()) } }, Err(err) => { - log::warn!("Actor panicked! Catching as error"); + warn!("Actor panicked! Catching as error"); Err(Panic::from(err).into()) } } @@ -1197,21 +1197,21 @@ impl ActorStream { match result { Ok(Status::Done) => { - log::trace!( + trace!( "Actor {} finished processing a message", self.context.aid.uuid() ); self.receiver.pop().unwrap() } Ok(Status::Skip) => { - log::trace!( + trace!( "Actor {} skipped processing a message", self.context.aid.uuid() ); self.receiver.skip().unwrap() } Ok(Status::Reset) => { - log::trace!( + trace!( "Actor {} finished processing a message and reset the cursor", self.context.aid.uuid() ); @@ -1219,7 +1219,7 @@ impl ActorStream { self.receiver.reset_skip().unwrap(); } Ok(Status::Stop) => { - log::debug!("Actor \"{}\" stopping", self.context.aid.name_or_uuid()); + debug!("Actor \"{}\" stopping", self.context.aid.name_or_uuid()); self.receiver.pop().unwrap(); self.context .system @@ -1228,10 +1228,9 @@ impl ActorStream { } Err(e) => { self.receiver.pop().unwrap(); - log::error!( + error!( "[{}] returned an error when processing: {}", - self.context.aid, - &e + self.context.aid, &e ); self.context .system @@ -1260,7 +1259,7 @@ impl Stream for ActorStream { mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>, ) -> Poll> { - log::trace!("Actor {} is being polled", self.context.aid.name_or_uuid()); + trace!("Actor {} is being polled", self.context.aid.name_or_uuid()); // If we have a pending future, that's what we poll. if let Some(pending) = self.pending.as_mut() { // Poll, ensure we respect stopping condition. @@ -1270,7 +1269,7 @@ impl Stream for ActorStream { .map(|r| Some(self.overwrite_on_stop(r))); if let Poll::Pending = &poll { - log::trace!("Actor {} is pending", self.context.aid.uuid()); + trace!("Actor {} is pending", self.context.aid.uuid()); } else { drop(self.pending.take()); } @@ -1288,7 +1287,7 @@ impl Stream for ActorStream { // We're stopping after this future, mark as such if let Some(m) = msg.content_as::() { if let SystemMsg::Stop = *m { - log::trace!("Actor {} received stop message", self.context.aid.uuid()); + trace!("Actor {} received stop message", self.context.aid.uuid()); self.stopping = true; } } @@ -1300,7 +1299,7 @@ impl Stream for ActorStream { match future.as_mut().poll(cx) { Poll::Ready(r) => Poll::Ready(Some(self.overwrite_on_stop(r))), Poll::Pending => { - log::trace!("Actor {} is pending", self.context.aid.uuid()); + trace!("Actor {} is pending", self.context.aid.uuid()); self.pending = Some(future); Poll::Pending } @@ -1314,7 +1313,7 @@ impl Stream for ActorStream { // While this is exhaustive, we're avoiding a catchall to in anticipation of // future Secc errors we would *want* to handle. SeccErrors::Empty | SeccErrors::Full(_) => { - log::trace!( + trace!( "Actor `{}` has no more messages, return to sleep", self.context.aid.name_or_uuid() ); @@ -1330,7 +1329,7 @@ impl Stream for ActorStream { mod tests { use super::*; use crate::tests::*; - use std::thread; + use std::thread; use std::time::Instant; /// This is identical to the documentation but here so that its formatted by rust and we can @@ -1350,8 +1349,8 @@ mod tests { .unwrap(); match aid.send(Message::new(11)) { - Ok(_) => log::info!("OK Then!"), - Err(e) => log::info!("Ooops {:?}", e), + Ok(_) => info!("OK Then!"), + Err(e) => info!("Ooops {:?}", e), } system.await_shutdown(None); diff --git a/src/cluster.rs b/src/cluster.rs index 9e33121..71c9c74 100644 --- a/src/cluster.rs +++ b/src/cluster.rs @@ -101,7 +101,7 @@ impl TcpClusterMgr { system.init_current(); let sys_uuid = system.uuid(); let listener = TcpListener::bind(address).unwrap(); - log::info!("{}: Listening for connections on {}.", sys_uuid, address); + info!("{}: Listening for connections on {}.", sys_uuid, address); // Notify the cluster manager that the listener is ready. let (mutex, condvar) = &*pair; @@ -115,15 +115,14 @@ impl TcpClusterMgr { while manager.data.running.load(Ordering::Relaxed) { match listener.accept() { Ok((stream, socket_address)) => { - log::info!( + info!( "{}: Accepting connection from: {}.", - sys_uuid, - socket_address + sys_uuid, socket_address ); manager.start_tcp_threads(stream, socket_address); } Err(e) => { - log::error!("couldn't get client: {:?}", e); + error!("couldn't get client: {:?}", e); } } } @@ -160,7 +159,7 @@ impl TcpClusterMgr { rx_handle, }; - log::info!( + info!( "{:?}: Connected to {:?}@{:?}", self.data.system.uuid(), system_uuid, diff --git a/src/executor.rs b/src/executor.rs index b05d3b8..d5ababc 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -67,12 +67,12 @@ impl MaximExecutor { /// This wakes an ActorStream in the Executor which will cause its future to be polled. The Aid, /// through the ActorSystem, will call this on Message Send. pub(crate) fn wake(&self, id: Aid) { - log::trace!("Waking Actor `{}`", id.name_or_uuid()); + trace!("Waking Actor `{}`", id.name_or_uuid()); // Pull the Task let task = match self.sleeping.remove(&id) { Some((_, task)) => task, None => { - log::debug!( + debug!( "Actor `{}` not in Executor - already woken or stopped", id.name_or_uuid() ); @@ -104,7 +104,7 @@ impl MaximExecutor { /// When a Reactor is done with an task, it will be sent here, and the Executor will decrement /// the Actor count for that Reactor. fn return_task(&self, task: Task, reactor: &MaximReactor) { - log::trace!( + trace!( "Actor {} returned from Reactor {}", task.id.name_or_uuid(), reactor.name @@ -119,7 +119,7 @@ impl MaximExecutor { /// triggered. pub(crate) fn await_shutdown(&self, timeout: impl Into>) -> ShutdownResult { let start = Instant::now(); - log::info!("Notifying Reactor threads, so they can end gracefully"); + info!("Notifying Reactor threads, so they can end gracefully"); for r in self.reactors.iter() { match r.thread_condvar.read() { Ok(g) => g.1.notify_one(), @@ -127,7 +127,7 @@ impl MaximExecutor { } } let timeout = timeout.into().map(|t| t - (Instant::now() - start)); - log::info!("Awaiting the threadpool's shutdown"); + info!("Awaiting the threadpool's shutdown"); self.thread_pool.await_shutdown(timeout) } } @@ -177,7 +177,7 @@ impl MaximReactor { /// Creates a new Reactor fn new(executor: MaximExecutor, system: &ActorSystem, id: u16) -> MaximReactor { let name = format!("{:08x?}-{}", system.data.uuid.as_fields().0, id); - log::debug!("Creating Reactor {}", name); + debug!("Creating Reactor {}", name); MaximReactor { id, @@ -219,7 +219,7 @@ impl MaximReactor { .lock() .expect("Poisoned shutdown_triggered condvar") { - log::debug!("Reactor-{} acknowledging shutdown", self.name); + debug!("Reactor-{} acknowledging shutdown", self.name); return false; } } @@ -264,13 +264,13 @@ impl MaximReactor { // Still pending, return to wait_queue. Drop the wakeup, because the futures // will re-add it later through their wakers. Poll::Pending => { - log::trace!("Reactor-{} waiting on pending Actor", self.name); + trace!("Reactor-{} waiting on pending Actor", self.name); self.wait(task); break; } } if Instant::now().duration_since(start) >= self.warn_threshold { - log::warn!( + warn!( "Actor {} took longer than configured warning threshold", aid.name_or_uuid() ); @@ -289,14 +289,14 @@ impl MaximReactor { fn get_work(&self) -> LoopResult<(Wakeup, Task)> { if let Some(w) = self.get_woken() { if let Some(task) = self.remove_waiting(&w.id) { - log::trace!( + trace!( "Reactor-{} received Wakeup for Actor `{}`", self.name, task.id.name_or_uuid() ); LoopResult::Ok((w, task)) } else { - log::trace!("Reactor-{} dropping spurious WakeUp", self.name); + trace!("Reactor-{} dropping spurious WakeUp", self.name); LoopResult::Continue } } else { @@ -305,12 +305,12 @@ impl MaximReactor { .read() .expect("Poisoned Reactor condvar"); - log::trace!("Reactor-{} waiting on condvar", self.name); + trace!("Reactor-{} waiting on condvar", self.name); let g = mutex.lock().expect("Poisoned Reactor condvar"); let _ = condvar .wait_timeout(g, self.thread_wait_time) .expect("Poisoned Reactor condvar"); - log::trace!("Reactor-{} resuming", self.name); + trace!("Reactor-{} resuming", self.name); LoopResult::Continue } } @@ -430,7 +430,7 @@ mod tests { 0 => Poll::Ready(Ok(Status::done(()))), count => { *count -= 1; - log::debug!("Pending, {} times left", count); + debug!("Pending, {} times left", count); let waker = cx.waker().clone(); let sleep_for = self.sleep_for; thread::spawn(move || { diff --git a/src/executor/thread_pool.rs b/src/executor/thread_pool.rs index 4152835..41c0bd3 100644 --- a/src/executor/thread_pool.rs +++ b/src/executor/thread_pool.rs @@ -32,7 +32,7 @@ impl MaximThreadPool { .spawn(move || { let lease = ThreadLease::new(deed); lease.deed.drain.increment(); - log::debug!("Thread {} has started", lease.deed.name); + debug!("Thread {} has started", lease.deed.name); lease.deed.set_running(); f(); lease.deed.set_stopped(); @@ -93,9 +93,9 @@ impl Drop for ThreadLease { // If the Lease dropped while Running, it Panicked. if let ThreadState::Running = *g { *g = ThreadState::Panicked; - log::error!("Thread {} panicked!", self.deed.name) + error!("Thread {} panicked!", self.deed.name) } else { - log::debug!("Thread {} has stopped", self.deed.name) + debug!("Thread {} has stopped", self.deed.name) } // Either way, it's dead, let's decrement the thread counter. self.deed.drain.decrement(); @@ -123,7 +123,7 @@ impl DrainAwait { pub fn increment(&self) { let mut g = self.mutex.lock().expect("DrainAwait poisoned"); let new = *g + 1; - log::trace!("Incrementing DrainAwait to {}", new); + trace!("Incrementing DrainAwait to {}", new); *g += 1; } @@ -131,9 +131,9 @@ impl DrainAwait { pub fn decrement(&self) { let mut guard = self.mutex.lock().expect("DrainAwait poisoned"); *guard -= 1; - log::trace!("Decrementing DrainAwait to {}", *guard); + trace!("Decrementing DrainAwait to {}", *guard); if *guard == 0 { - log::debug!("DrainAwait is depleted, notifying blocked threads"); + debug!("DrainAwait is depleted, notifying blocked threads"); self.condvar.notify_all(); } } diff --git a/src/lib.rs b/src/lib.rs index 103a237..8f4b4e0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -190,6 +190,10 @@ //! See the License for the specific language governing permissions and //! limitations under the License. +// Include the tracing macros globally +#[macro_use] +extern crate tracing; + use std::any::Any; use std::error::Error; use std::fmt::{Display, Formatter}; @@ -248,7 +252,6 @@ mod tests { use std::thread; use std::time::Duration; - use log::LevelFilter; extern crate secc; use secc::{SeccReceiver, SeccSender}; use serde::{Deserialize, Serialize}; @@ -290,10 +293,9 @@ mod tests { } pub fn init_test_log() { - let _ = env_logger::builder() - .filter_level(LevelFilter::Warn) - .is_test(true) - .try_init(); + tracing_subscriber::fmt() + .with_max_level(tracing::Level::TRACE) + .init(); } pub fn sleep(millis: u64) { diff --git a/src/system.rs b/src/system.rs index 0fe3038..03f19a1 100644 --- a/src/system.rs +++ b/src/system.rs @@ -11,11 +11,11 @@ use dashmap::DashMap; use once_cell::sync::OnceCell; +use piper::ChangeNotifier; use secc::{SeccReceiver, SeccSender}; use serde::{Deserialize, Serialize}; -use uuid::Uuid; use smol::{Task, Timer}; -use piper::ChangeNotifier; +use uuid::Uuid; use std::collections::{BinaryHeap, HashSet}; use std::error::Error; @@ -266,8 +266,7 @@ impl ActorSystem { pub fn create(config: ActorSystemConfig) -> ActorSystem { let uuid = Uuid::new_v4(); - log::trace!("Starting executor thread pool"); - + trace!("Starting executor thread pool"); // Flag to indicate the system should shutdown let shutdown_triggered = ChangeNotifier::new(Arc::new(AtomicBool::new(false))); @@ -277,14 +276,16 @@ impl ActorSystem { for _ in 0..config.thread_pool_size { let shutdown_triggered__ = shutdown_triggered_.clone(); // Spawn an executor thread that waits for the shutdown signal. - threads.push(thread::spawn(move || smol::run(async move { - loop { - shutdown_triggered__.listen().await; - if shutdown_triggered__.load(Ordering::SeqCst) { - break; + threads.push(thread::spawn(move || { + smol::run(async move { + loop { + shutdown_triggered__.listen().await; + if shutdown_triggered__.load(Ordering::SeqCst) { + break; + } } - } - }))); + }) + })); } let start_on_launch = config.start_on_launch; @@ -319,7 +320,7 @@ impl ActorSystem { .started .compare_and_swap(false, true, Ordering::Relaxed) { - log::info!("ActorSystem {} has spawned", self.data.uuid); + info!("ActorSystem {} has spawned", self.data.uuid); // Launch the SystemActor and give it the name "System" self.spawn() @@ -562,10 +563,7 @@ impl ActorSystem { } /// Triggers a shutdown of the system and returns only when all Reactors have shutdown. - pub fn trigger_and_await_shutdown( - &self, - timeout: impl Into>, - ) { + pub fn trigger_and_await_shutdown(&self, timeout: impl Into>) { unimplemented!("FIXME: Re-implement graceful shutdown"); // self.trigger_shutdown(); // self.await_shutdown(timeout) @@ -695,10 +693,9 @@ impl ActorSystem { error: error.clone(), }; m_aid.send(Message::new(value)).unwrap_or_else(|error| { - log::error!( + error!( "Could not send 'Stopped' to monitoring actor {}: Error: {:?}", - m_aid, - error + m_aid, error ); }); } @@ -758,11 +755,11 @@ impl ActorSystem { // FIXME (Issue #72) Add try_send ability. pub fn send_to_system_actors(&self, message: Message) { let remotes = &*self.data.remotes; - log::trace!("Sending message to Remote System Actors"); + trace!("Sending message to Remote System Actors"); for remote in remotes.iter() { let aid = &remote.value().system_actor_aid; aid.send(message.clone()).unwrap_or_else(|error| { - log::error!("Could not send to system actor {}. Error: {}", aid, error) + error!("Could not send to system actor {}. Error: {}", aid, error) }); } } @@ -776,7 +773,8 @@ impl ActorSystem { Timer::after(delay).await; destination.send(message); - }).detach(); + }) + .detach(); } } @@ -945,17 +943,17 @@ mod tests { fn test_send_after() { init_test_log(); - log::info!("Preparing test"); + info!("Preparing test"); let system = ActorSystem::create(ActorSystemConfig::default().thread_pool_size(2)); let aid = system.spawn().name("A").with((), simple_handler).unwrap(); await_received(&aid, 1, 1000).unwrap(); - log::info!("Test prepared, sending delayed message"); + info!("Test prepared, sending delayed message"); system.send_after(Message::new(11), aid.clone(), Duration::from_millis(10)); - log::info!("Sleeping for initial check"); + info!("Sleeping for initial check"); sleep(5); assert_eq!(1, aid.received().unwrap()); - log::info!("Sleeping till we're 100% sure we should have the message"); + info!("Sleeping till we're 100% sure we should have the message"); sleep(10); assert_eq!(2, aid.received().unwrap()); @@ -1117,11 +1115,11 @@ mod tests { .spawn() .with((), |_: (), _: Context, msg: Message| { if let Some(_) = msg.content_as::() { - log::debug!("Not panicking this time"); + debug!("Not panicking this time"); return future::ok(Status::done(())); } - log::debug!("About to panic"); + debug!("About to panic"); panic!("I panicked") }) .unwrap(); @@ -1236,13 +1234,13 @@ mod tests { .spawn() .with((), move |_: (), context: Context, message: Message| { if let Some(_) = message.content_as::() { - log::debug!("Received reply, shutting down"); + debug!("Received reply, shutting down"); context.system.trigger_shutdown(); future::ok(Status::stop(())) } else if let Some(msg) = message.content_as::() { match &*msg { SystemMsg::Start => { - log::debug!("Starting request actor"); + debug!("Starting request actor"); let target_aid: Aid = bincode::deserialize(&serialized).unwrap(); target_aid .send_new(Request { @@ -1297,7 +1295,7 @@ mod tests { if let Some(msg) = message.content_as::() { match &*msg { SystemActorMessage::FindByNameResult { aid: found, .. } => { - log::debug!("FindByNameResult received"); + debug!("FindByNameResult received"); if let Some(target) = found { t.assert( target.uuid() == aid1.uuid(), @@ -1313,7 +1311,7 @@ mod tests { _ => t.panic("Unexpected message received!"), } } else if let Some(msg) = message.content_as::() { - log::debug!("Actor started, attempting to send FindByName request"); + debug!("Actor started, attempting to send FindByName request"); if let SystemMsg::Start = &*msg { context.system.send_to_system_actors(Message::new( SystemActorMessage::FindByName { diff --git a/src/system/system_actor.rs b/src/system/system_actor.rs index f9c39eb..f922c32 100644 --- a/src/system/system_actor.rs +++ b/src/system/system_actor.rs @@ -13,7 +13,7 @@ impl SystemActor { if let Some(msg) = message.content_as::() { // Someone requested that this system actor find an actor by name. if let SystemActorMessage::FindByName { reply_to, name } = &*msg { - log::debug!("Attempting to locate Actor by name: {}", name); + debug!("Attempting to locate Actor by name: {}", name); let found = context.system.find_aid_by_name(&name); let reply = Message::new(SystemActorMessage::FindByNameResult { system_uuid: context.system.uuid(), @@ -24,7 +24,7 @@ impl SystemActor { // there is a problem sending the reply. In this case, the error is logged but the // actor moves on. reply_to.send(reply).unwrap_or_else(|error| { - log::error!( + error!( "Could not send reply to FindByName to actor {}. Error: {:?}", reply_to, error ) @@ -36,7 +36,7 @@ impl SystemActor { Ok(Status::done(self)) // Log an error if we get an unexpected message kind, but continue processing as normal. } else { - log::error!("Unhandled message received."); + error!("Unhandled message received."); Ok(Status::done(self)) } } From 2d1f4600f4ea4914fc311d523bdc0558e2a38134 Mon Sep 17 00:00:00 2001 From: Zicklag Date: Wed, 6 May 2020 02:07:17 +0000 Subject: [PATCH 7/9] Finish SECC Channel Implementation And Tests --- src/secc.rs | 56 +++++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 46 insertions(+), 10 deletions(-) diff --git a/src/secc.rs b/src/secc.rs index 2b0e0f0..ae6bc5c 100644 --- a/src/secc.rs +++ b/src/secc.rs @@ -8,6 +8,7 @@ use std::collections::VecDeque; /// /// > **note:** The type `T` should be efficiently clonable as calls to [`SeccReceiver::peek`] /// > must clone the value. Using an [`Arc`] is one way to do this. +#[tracing::instrument] pub fn secc_unbounded() -> (SeccSender, SeccReceiver) { let (flume_sender, flume_receiver) = flume::unbounded(); @@ -21,6 +22,7 @@ pub fn secc_unbounded() -> (SeccSender, SeccReceiver) { /// /// > **note:** The type `T` should be efficiently clonable as calls to [`SeccReceiver::peek`] /// > must clone the value. Using an [`Arc`] is one way to do this. +#[tracing::instrument] pub fn secc_bounded(capacity: usize) -> (SeccSender, SeccReceiver) { let (flume_sender, flume_receiver) = flume::bounded(capacity); @@ -82,15 +84,24 @@ impl SeccReceiver { } /// Peek at the next message in the channel + #[tracing::instrument(skip(self))] pub async fn peek(&mut self) -> Result { // If we already have a peeked message, return it if let Some(msg) = &self.peeked_message { + trace!("Returning the value we have previously peeked at"); Ok(msg.clone()) // If we are resetting, peek the message from the skipped queue } else if self.is_resetting { + trace!("We are in the middle of resetting"); + // Get the next message in the queue if let Some(msg) = self.skipped.get(0) { + trace!("Grabbing the message off the top of the skipped queue"); + + self.reset_until -= 1; + trace!(self.reset_until, "Decremented self.reset_until"); + Ok(msg.clone()) } else { @@ -101,9 +112,11 @@ impl SeccReceiver { // If we don't already have a peeked message and we aren't resetting } else { // Get the next message in the channel + trace!("Grabbing the next element in the channel"); let msg = self.receiver.recv_async().await?; // Clone it and put it in our peeked message slot + trace!("Sticking message in our peeked slot"); self.peeked_message = Some(msg.clone()); // Return the message @@ -112,48 +125,59 @@ impl SeccReceiver { } /// Receive the next message in the channel + #[tracing::instrument(skip(self))] pub async fn recv(&mut self) -> Result { // If we are currently resetting if self.is_resetting { + trace!("We are in the middle of resetting"); + // Pop the next message off of the skipped queue + trace!("Grabbing next element off of the skipped queue"); if let Some(msg) = self.skipped.pop_front() { - // Decrement the reset until cursor to make sure it stays pointing at the same message self.reset_until -= 1; - - // If this was the last message we were supposed to reset until + trace!(self.reset_until, "Decremented reset_until"); + if self.reset_until == 0 { - // Go out of resetting mode + trace!("reset_unitl == 0: going out of reset mode"); self.is_resetting = false; } + trace!("Returning skipped message"); Ok(msg) - // If there is no message, go out of resetting mode and return the next message in the channel + } else { - self.is_resetting = false; - self.receiver.recv_async().await + unreachable!("There should be an element in the skipped queue as we are in \ + the middle of resetting still") } // If we have a peeked message, return that one } else if let Some(msg) = self.peeked_message.take() { + trace!("Returning the message out of the peeked slot"); Ok(msg) // Get the message from the channel } else { + trace!("Getting next message from channel"); self.receiver.recv_async().await } } /// Skip the next message in the channel + #[tracing::instrument(skip(self))] pub async fn skip(&mut self) -> Result<(), flume::RecvError> { // Get the message to skip let msg = // If we have a peeked message skip that one if let Some(msg) = self.peeked_message.take() { + trace!("Selecting the message that is in the peeked slot"); msg // If we are resetting, skip the one off of the top of the skipped queue } else if self.is_resetting { + trace!("We are in the middle of resetting"); + if let Some(msg) = self.skipped.pop_front() { + trace!("Selecting the next message in the skipped queue"); msg } else { unreachable!("If we are resetting there should be a message in the skipped \ @@ -162,10 +186,12 @@ impl SeccReceiver { // Otherwise, get the next message from the channel and skip it } else { + trace!("Selecting the next message from the channel"); self.receiver.recv_async().await? }; // Add it to the skipped message queue + trace!("Skipping selected message"); self.skipped.push_back(msg); Ok(()) @@ -173,11 +199,13 @@ impl SeccReceiver { /// Causes `recv` to return previously skipped messages untill ther are none, where it starts /// collecting the messages from the channel again + #[tracing::instrument(skip(self))] pub fn reset_skip(&mut self) { // Go into resetting mode self.is_resetting = true; // Reset until the end of the skipped message queue - self.reset_until = self.skipped.len() - 1; + self.reset_until = self.skipped.len(); + trace!(self.reset_until, "Resetting skips. Going into reset mode.") } } @@ -185,11 +213,16 @@ impl SeccReceiver { mod test { use super::*; + #[derive(Debug)] enum Mode { Bounded(usize), Unbounded, } + fn init_logging() { + tracing_subscriber::fmt::try_init().ok(); + } + fn get_channel(mode: Mode) -> (SeccSender, SeccReceiver) { match mode { Mode::Bounded(capacity) => secc_bounded(capacity), @@ -197,6 +230,7 @@ mod test { } } + #[tracing::instrument] fn basic(mode: Mode) { smol::run(async move { // Create a secc channel @@ -299,12 +333,14 @@ mod test { }); } #[test] - fn bounded_basic() { + fn secc_bounded_basic() { + init_logging(); basic(Mode::Bounded(100)); } #[test] - fn unbounded_basic() { + fn secc_unbounded_basic() { + init_logging(); basic(Mode::Unbounded); } } From 87e8312e7caf68d2b1570c9e9793aa2e7e2d6b4a Mon Sep 17 00:00:00 2001 From: Zicklag Date: Wed, 6 May 2020 02:28:03 +0000 Subject: [PATCH 8/9] Remove Addressed TODO --- Cargo.toml | 1 - 1 file changed, 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index ea870d7..5e49473 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -79,7 +79,6 @@ futures = "0.3.1" # Async Utils serde = { version = "1.0.97", features = ["derive", "rc"] } # Serialization support bincode = "1.1.4" # RPC Serialization dashmap = "1.0.3" # Concurrent hashmap -# TODO: Make sure we enable minimal required features for tracing and tracing futures tracing = "0.1.13" # Tracing ( like logging with spans and extra event data ) tracing-futures = { version = "0.2.4", default-features = false, features = ["futures-03"] } uuid = { version = "0.8.1", features = ["serde", "v4"]} # UUID Generation From 98477ca8cd44fdeb2e15493f77fd40744287a2e5 Mon Sep 17 00:00:00 2001 From: Zicklag Date: Tue, 16 Jun 2020 00:29:25 +0000 Subject: [PATCH 9/9] Start Migration to Async Channels - Doesn't compile yet - Comments out lots of code that will need to be re-implemented --- Cargo.toml | 6 +- examples/montecarlo.rs | 2 + src/actors.rs | 473 +++++++++++++++++++------------------ src/lib.rs | 3 +- src/secc.rs | 93 ++++---- src/system.rs | 101 +++----- src/system/system_actor.rs | 2 +- 7 files changed, 333 insertions(+), 347 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 5e49473..8ad6f69 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -74,16 +74,16 @@ serde_json = "^1.0.40" [dependencies] smol = "0.1.4" # Async Executor piper = { path = "lib/piper" } # Async pipes, channels, mutexes, and more -flume = { version = "0.7.1", default-features = false, features = ["async"] } -futures = "0.3.1" # Async Utils +futures = { version = "0.3.1", default-features = false } # Async Utils serde = { version = "1.0.97", features = ["derive", "rc"] } # Serialization support bincode = "1.1.4" # RPC Serialization dashmap = "1.0.3" # Concurrent hashmap tracing = "0.1.13" # Tracing ( like logging with spans and extra event data ) tracing-futures = { version = "0.2.4", default-features = false, features = ["futures-03"] } +async-trait = "0.1.30" # Macro for creating async traits uuid = { version = "0.8.1", features = ["serde", "v4"]} # UUID Generation -secc = "0.0.10" once_cell = "1.0.2" rand = { version = "0.7.3", optional = true } # Random support rand_xoshiro = { version = "0.4.0", optional = true } # Fast random number generator num_cpus = { version = "1.10.1", optional = true } # Detect number of CPUs +async-channel = "1.1.0" diff --git a/examples/montecarlo.rs b/examples/montecarlo.rs index a98dbe3..a8ffb38 100644 --- a/examples/montecarlo.rs +++ b/examples/montecarlo.rs @@ -61,6 +61,7 @@ impl Game { // to the `GameManager`. results_aid .send_new(GameMsg::new(ctx.aid.clone(), results_vec)) + .await .unwrap(); // Because the `GameManager` is monitoring this actor, sending the `Stop` status // will inform the manager that this game is now completed. @@ -153,6 +154,7 @@ impl GameManager { .spawn() .name(&name) .with(game_conditions, Game::play) + .await .unwrap(); ctx.system.monitor(&ctx.aid, &aid); aid.send_new(ctx.aid.clone()).unwrap(); diff --git a/src/actors.rs b/src/actors.rs index f88c6ab..55a4ce8 100644 --- a/src/actors.rs +++ b/src/actors.rs @@ -6,8 +6,7 @@ //! are created by calling `system::spawn().with()` with any kind of function or closure that //! implements the `Processor` trait. -use crate::message::ActorMessage; -use crate::prelude::*; +use async_trait::async_trait; use futures::{FutureExt, Stream}; #[cfg(feature = "actor-pool")] use rand::{ @@ -16,10 +15,12 @@ use rand::{ }; #[cfg(feature = "actor-pool")] use rand_xoshiro::Xoshiro256Plus; -use secc::*; use serde::de::Deserializer; use serde::ser::Serializer; use serde::{Deserialize, Serialize}; +use smol::Timer; +use uuid::Uuid; + use std::cell::UnsafeCell; use std::fmt::Debug; use std::future::Future; @@ -32,7 +33,10 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::task::Poll; use std::time::Duration; -use uuid::Uuid; + +use crate::message::ActorMessage; +use crate::prelude::*; +use crate::secc::{secc_bounded, secc_unbounded, SeccReceiver, SeccSender}; /// Status of the message and potentially the actor as a resulting from processing a message /// with the actor. @@ -108,19 +112,9 @@ pub enum AidError { /// only work on local Aid instances. AidNotLocal, - /// Used when unable to send to an actor's message channel within the scheduled timeout - /// configured in the actor system. This could result from the actor's channel being too - /// small to accommodate the message flow, the lack of thread count to process messages fast - /// enough to keep up with the flow or something wrong with the actor itself that it is - /// taking too long to clear the messages. - SendTimedOut(Aid), - - /// Used when unable to schedule the actor for work in the work channel. This could be a - /// result of having a work channel that is too small to accommodate the number of actors - /// being concurrently scheduled, not enough threads to process actors in the channel fast - /// enough or simply an actor that misbehaves, causing dispatcher threads to take a lot of - /// time or not finish at all. - UnableToSchedule, + /// Channel error + /// TODO: Clarify error + ChannelError, } impl std::fmt::Display for AidError { @@ -190,6 +184,74 @@ struct AidSerializedForm { name: Option, } +/// The kind of channel that an actor will use for its messages. You can use either bounded +/// channels with a configurable capacity or unbounded channels that will grow to fit however many +/// messages are sent to it. +/// +/// > **Note:** While unbounded channels are provided and allow you to have backlog/queue-like +/// > actor message queues, it is best practice for actors to respond very quickly to the messages +/// > are sent and to prevent backing messages up. If an actor cannot process the messages it is +/// > sent fast enough, it should probably be broken into multiple smaller actors, or it should be +/// > spawned in an [`AidPool`] in order to split up the load. +// TODO: Move to actor.rs +#[derive(Copy, Clone, Debug, Serialize, Deserialize)] +pub enum ActorChannelKind { + /// A bounded channel with the specified capacity + Bounded(usize), + /// An unbounded channel + Unbounded, +} + +/// Represents a pool of actor ids in which you don't care *which* actor recieves a +/// message. +/// +/// When a message is sent to a pool, only one actor in the pool will receive the message. Different +/// [`AidPool`] implementations may have different ways of determining which actor to send a message +/// to. The implmentation may send a message to a random actor or it may go in order, for example. +/// +/// [`Aid`]'s also implement [`AidPool`] so an [`Aid`] can be used wherever a generic [`AidPool`] is +/// expected. +#[async_trait] +pub trait AidPool { + /// See [`Aid::send`] + async fn send(&mut self, message: Message) -> Result<(), AidError>; + + /// See [`Aid::send_arc`] + async fn send_arc(&mut self, value: Arc) -> Result<(), AidError> + where + T: 'static + ActorMessage; + + /// See [`Aid::send_new`] + async fn send_new(&mut self, value: T) -> Result<(), AidError> + where + T: 'static + ActorMessage; + + /// See [`Aid::send_after`] + async fn send_after(&mut self, message: Message, duration: Duration) -> Result<(), AidError>; + + /// See [`Aid::send_arc_after`] + async fn send_arc_after( + &mut self, + value: Arc, + duration: Duration, + ) -> Result<(), AidError> + where + T: 'static + ActorMessage; + + /// See [`Aid::send_new_after`] + async fn send_new_after(&mut self, value: T, duration: Duration) -> Result<(), AidError> + where + T: 'static + ActorMessage; +} + +/// A helper trait that is simply an AidPool that can be passed between actors, i.e. it is +/// `Sync + Send + Clone + 'static`. This is useful when you want to make a function generic over +/// [`AidPool`] but you need to be able to give the poool to actor. +pub trait SyncAidPool: AidPool + Sync + Send + Clone + 'static {} + +// Auto implement SyncAidPool for complying [`AidPool`]s +impl SyncAidPool for T {} + /// Encapsulates an Actor ID and is used to send messages to the actor. /// /// This is a unique reference to the actor within the entire cluster and can be used to send @@ -332,7 +394,7 @@ impl Aid { /// /// system.await_shutdown(None); /// ``` - pub fn send(&self, message: Message) -> Result<(), AidError> { + pub async fn send(&self, message: Message) -> Result<(), AidError> { match &self.data.sender { ActorSender::Local { stopped, @@ -342,25 +404,22 @@ impl Aid { if stopped.load(Ordering::Relaxed) { Err(AidError::ActorAlreadyStopped) } else { - match sender.send_await_timeout(message, system.config().send_timeout) { - Ok(_) => { - // if sender.receivable() == 1 { - // system.schedule(self.clone()); - // }; - Ok(()) - } - Err(_) => Err(AidError::SendTimedOut(self.clone())), + match sender.send(message).await { + Ok(_) => Ok(()), + // TODO: Clarify this error + Err(e) => Err(AidError::ChannelError), } } } ActorSender::Remote { sender } => { sender - .send_await(WireMessage::ActorMessage { + .send(WireMessage::ActorMessage { actor_uuid: self.data.uuid, system_uuid: self.data.system_uuid, message, }) - .unwrap(); + .await + .map_err(|_| AidError::ChannelError)?; Ok(()) } } @@ -399,11 +458,11 @@ impl Aid { /// /// system.await_shutdown(None); /// ``` - pub fn send_arc(&self, value: Arc) -> Result<(), AidError> + pub async fn send_arc(&self, value: Arc) -> Result<(), AidError> where T: 'static + ActorMessage, { - self.send(Message::from_arc(value)) + self.send(Message::from_arc(value)).await } /// Shortcut for calling `send(Message::new(value))` This method will internally wrap @@ -439,11 +498,11 @@ impl Aid { /// /// system.await_shutdown(None); /// ``` - pub fn send_new(&self, value: T) -> Result<(), AidError> + pub async fn send_new(&self, value: T) -> Result<(), AidError> where T: 'static + ActorMessage, { - self.send(Message::new(value)) + self.send(Message::new(value)).await } /// Schedules the given message to be sent after a minimum of the specified duration. Note @@ -482,7 +541,11 @@ impl Aid { /// /// system.await_shutdown(None); /// ``` - pub fn send_after(&self, message: Message, duration: Duration) -> Result<(), AidError> { + pub async fn send_after(&self, message: Message, delay: Duration) -> Result<(), AidError> { + // Pause for the given delay + Timer::after(delay).await; + + // Send the message match &self.data.sender { ActorSender::Local { stopped, system, .. @@ -490,22 +553,21 @@ impl Aid { if stopped.load(Ordering::Relaxed) { Err(AidError::ActorAlreadyStopped) } else { - system.send_after(message, self.clone(), duration); + system.send_after(message, self.clone(), delay); Ok(()) } } ActorSender::Remote { sender } => { - if let Err(err) = sender.send_await(WireMessage::DelayedActorMessage { - duration, - actor_uuid: self.data.uuid, - system_uuid: self.data.system_uuid, - message, - }) { - // Right now, this is the full extent of errors, but if that should change, it - // should create a compiler error. - return match err { - SeccErrors::Full(_) | SeccErrors::Empty => Ok(()), - }; + if let Err(err) = sender + .send(WireMessage::ActorMessage { + actor_uuid: self.data.uuid, + system_uuid: self.data.system_uuid, + message, + }) + .await + { + // TODO: Clarify error + return Err(AidError::ChannelError); } Ok(()) } @@ -546,11 +608,11 @@ impl Aid { /// /// system.await_shutdown(None); /// ``` - pub fn send_arc_after(&self, value: Arc, duration: Duration) -> Result<(), AidError> + pub async fn send_arc_after(&self, value: Arc, duration: Duration) -> Result<(), AidError> where T: 'static + ActorMessage, { - self.send_after(Message::from_arc(value), duration) + self.send_after(Message::from_arc(value), duration).await } /// Shortcut for calling `send_after(Message::new(value))` This method will internally wrap @@ -587,11 +649,11 @@ impl Aid { /// /// system.await_shutdown(None); /// ``` - pub fn send_new_after(&self, value: T, duration: Duration) -> Result<(), AidError> + pub async fn send_new_after(&self, value: T, duration: Duration) -> Result<(), AidError> where T: 'static + ActorMessage, { - self.send_after(Message::new(value), duration) + self.send_after(Message::new(value), duration).await } /// The unique UUID for this actor within the entire cluster. The UUID for an [`Aid`] @@ -638,23 +700,7 @@ impl Aid { } } - /// Determines how many messages the actor with the [`Aid`] has been sent. This method works only - /// for local [`Aid`]s, remote [`Aid`]s will return an error if this is called. - pub fn sent(&self) -> Result { - match &self.data.sender { - ActorSender::Local { sender, .. } => Ok(sender.sent()), - _ => Err(AidError::AidNotLocal), - } - } - - /// Determines how many messages the actor with the [`Aid`] has received. This method works only - /// for local [`Aid`]s, remote [`Aid`]s will return an error if this is called. - pub fn received(&self) -> Result { - match &self.data.sender { - ActorSender::Local { sender, .. } => Ok(sender.received()), - _ => Err(AidError::AidNotLocal), - } - } + // TODO: Add sent() and received() functions again? /// Marks the actor referenced by the [`Aid`] as stopped and puts mechanisms in place to /// cause no more messages to be sent to the actor. Note that once stopped, an [`Aid`] can @@ -677,53 +723,54 @@ impl Aid { } } +#[async_trait] impl AidPool for Aid { /// See [`Aid::send`] #[inline] - fn send(&mut self, message: Message) -> Result<(), AidError> { - Aid::send(self, message) + async fn send(&mut self, message: Message) -> Result<(), AidError> { + Aid::send(self, message).await } /// See [`Aid::send_arc`] #[inline] - fn send_arc(&mut self, value: Arc) -> Result<(), AidError> + async fn send_arc(&mut self, value: Arc) -> Result<(), AidError> where T: 'static + ActorMessage, { - Aid::send_arc(self, value) + Aid::send_arc(self, value).await } /// See [`Aid::send_new`] #[inline] - fn send_new(&mut self, value: T) -> Result<(), AidError> + async fn send_new(&mut self, value: T) -> Result<(), AidError> where T: 'static + ActorMessage, { - Aid::send_new(self, value) + Aid::send_new(self, value).await } /// See [`Aid::send_after`] #[inline] - fn send_after(&mut self, message: Message, duration: Duration) -> Result<(), AidError> { - Aid::send_after(self, message, duration) + async fn send_after(&mut self, message: Message, duration: Duration) -> Result<(), AidError> { + Aid::send_after(self, message, duration).await } /// See [`Aid::send_arc_after`] #[inline] - fn send_arc_after(&mut self, value: Arc, duration: Duration) -> Result<(), AidError> + async fn send_arc_after(&mut self, value: Arc, duration: Duration) -> Result<(), AidError> where T: 'static + ActorMessage, { - Aid::send_arc_after(self, value, duration) + Aid::send_arc_after(self, value, duration).await } /// See [`Aid::send_new_after`] #[inline] - fn send_new_after(&mut self, value: T, duration: Duration) -> Result<(), AidError> + async fn send_new_after(&mut self, value: T, duration: Duration) -> Result<(), AidError> where T: 'static + ActorMessage, { - Aid::send_new_after(self, value, duration) + Aid::send_new_after(self, value, duration).await } } @@ -756,46 +803,6 @@ impl Hash for Aid { } } -/// Represents a pool of actor ids in which you don't care *which* actor recieves a -/// message. -/// -/// When a message is sent to a pool, only one actor in the pool will receive the message. Different -/// [`AidPool`] implementations may have different ways of determining which actor to send a message -/// to. The implmentation may send a message to a random actor or it may go in order, for example. -/// -/// [`Aid`]'s also implement [`AidPool`] so an [`Aid`] can be used wherever a generic [`AidPool`] is -/// expected. -pub trait AidPool { - /// See [`Aid::send`] - fn send(&mut self, message: Message) -> Result<(), AidError>; - /// See [`Aid::send_arc`] - fn send_arc(&mut self, value: Arc) -> Result<(), AidError> - where - T: 'static + ActorMessage; - /// See [`Aid::send_new`] - fn send_new(&mut self, value: T) -> Result<(), AidError> - where - T: 'static + ActorMessage; - /// See [`Aid::send_after`] - fn send_after(&mut self, message: Message, duration: Duration) -> Result<(), AidError>; - /// See [`Aid::send_arc_after`] - fn send_arc_after(&mut self, value: Arc, duration: Duration) -> Result<(), AidError> - where - T: 'static + ActorMessage; - /// See [`Aid::send_new_after`] - fn send_new_after(&mut self, value: T, duration: Duration) -> Result<(), AidError> - where - T: 'static + ActorMessage; -} - -/// A helper trait that is simply an AidPool that can be passed between actors, i.e. it is -/// `Sync + Send + Clone + 'static`. This is useful when you want to make a function generic over -/// [`AidPool`] but you need to be able to give the poool to actor. -pub trait SyncAidPool: AidPool + Sync + Send + Clone + 'static {} - -// Auto implement SyncAidPool for complying [`AidPool`]s -impl SyncAidPool for T {} - /// An [`AidPool`] that sends messages to a random [`Aid`] in the pool. #[derive(Debug)] #[cfg(feature = "actor-pool")] @@ -823,53 +830,70 @@ impl RandomAidPool { } #[cfg(feature = "actor-pool")] +#[async_trait] impl AidPool for RandomAidPool { /// See [`Aid::send`] #[inline] - fn send(&mut self, message: Message) -> Result<(), AidError> { - self.aids[self.uniform.sample(&mut self.rng)].send(message) + async fn send(&mut self, message: Message) -> Result<(), AidError> { + self.aids[self.uniform.sample(&mut self.rng)] + .send(message) + .await } /// See [`Aid::send_arc`] #[inline] - fn send_arc(&mut self, value: Arc) -> Result<(), AidError> + async fn send_arc(&mut self, value: Arc) -> Result<(), AidError> where T: 'static + ActorMessage, { - self.aids[self.uniform.sample(&mut self.rng)].send_arc(value) + self.aids[self.uniform.sample(&mut self.rng)] + .send_arc(value) + .await } /// See [`Aid::send_new`] #[inline] - fn send_new(&mut self, value: T) -> Result<(), AidError> + async fn send_new(&mut self, value: T) -> Result<(), AidError> where T: 'static + ActorMessage, { - self.aids[self.uniform.sample(&mut self.rng)].send_new(value) + self.aids[self.uniform.sample(&mut self.rng)] + .send_new(value) + .await } /// See [`Aid::send_after`] #[inline] - fn send_after(&mut self, message: Message, duration: Duration) -> Result<(), AidError> { - self.aids[self.uniform.sample(&mut self.rng)].send_after(message, duration) + async fn send_after(&mut self, message: Message, duration: Duration) -> Result<(), AidError> { + self.aids[_self.uniform.sample(&mut self.rng)] + .send_after(message, duration) + .await } /// See [`Aid::send_arc_after`] #[inline] - fn send_arc_after(&mut self, value: Arc, duration: Duration) -> Result<(), AidError> + async fn send_arc_after<'a, T>( + &'a mut self, + value: Arc, + duration: Duration, + ) -> Result<(), AidError> where T: 'static + ActorMessage, { - self.aids[self.uniform.sample(&mut self.rng)].send_arc_after(value, duration) + self.aids[_self.uniform.sample(&mut self.rng)] + .send_arc_after(value, duration) + .await } /// See [`Aid::send_new_after`] #[inline] - fn send_new_after(&mut self, value: T, duration: Duration) -> Result<(), AidError> + async fn send_new_after(&mut self, value: T, duration: Duration) -> Result<(), AidError> where T: 'static + ActorMessage, { - self.aids[self.uniform.sample(&mut self.rng)].send_new_after(value, duration) + self.aids[self.uniform.sample(&mut self.rng)] + .send_new_after(value, duration) + .await } } @@ -970,7 +994,7 @@ pub struct ActorBuilder { pub name: Option, /// The size of the message channel for the actor which defaults to `None`; meaning the /// default for the actor system will be used for the message channel. - pub channel_size: Option, + pub channel_kind: ActorChannelKind, } impl ActorBuilder { @@ -979,7 +1003,7 @@ impl ActorBuilder { /// `ActorSystem::spawn` for more information and examples. /// // FIXME Consider implementing `using` to spawn a stateless actor. - pub fn with(self, state: S, processor: F) -> Result + pub async fn with(self, state: S, processor: F) -> Result where S: Send + Sync + 'static, R: Future> + Send + 'static, @@ -987,7 +1011,7 @@ impl ActorBuilder { { let (actor, stream) = Actor::new(self.system.clone(), &self, state, processor); debug!("Actor created: {}", actor.context.aid.uuid()); - self.system.register_actor(actor, stream) + self.system.register_actor(actor, stream).await } /// Set the name of the actor to the given string. @@ -999,9 +1023,8 @@ impl ActorBuilder { /// Set the size of the channel to the given value instead of the default for the actor system /// that the actor is spawned on. Note that passing a value less than 1 will cause a panic and /// there would be little reason to do so anyway. - pub fn channel_size(mut self, size: u16) -> Self { - assert!(size > 0); - self.channel_size = Some(size); + pub fn channel_kind(mut self, kind: ActorChannelKind) -> Self { + self.channel_kind = kind; self } } @@ -1025,7 +1048,7 @@ impl ActorPoolBuilder { } /// See [`ActorBuilder::with`] - pub fn with(self, state: S, processor: F) -> Result + pub async fn with(self, state: S, processor: F) -> Result where S: Clone + Send + Sync + 'static, R: Future> + Send + 'static, @@ -1039,7 +1062,7 @@ impl ActorPoolBuilder { // Add index as name suffix b.name = b.name.map(|name| format!("{}_{}", name, i)); // Add aid to list - aids.push(b.with(state.clone(), processor.clone())?) + aids.push(b.with(state.clone(), processor.clone()).await?) } // Return an `AidPool` from the list of `Aid`s @@ -1057,8 +1080,8 @@ impl ActorPoolBuilder { } /// See [`ActorBuilder::channel_size`] - pub fn channel_size(mut self, size: u16) -> Self { - self.builder = self.builder.channel_size(size); + pub fn channel_kind(mut self, kind: ActorChannelKind) -> Self { + self.builder = self.builder.channel_kind(kind); self } } @@ -1115,12 +1138,10 @@ impl Actor { R: Future> + Send + 'static, F: Processor + 'static, { - let (sender, receiver) = secc::create::( - builder - .channel_size - .unwrap_or(system.config().message_channel_size), - Duration::from_millis(10), - ); + let (sender, receiver) = match builder.channel_kind { + ActorChannelKind::Bounded(capacity) => secc_bounded(capacity), + ActorChannelKind::Unbounded => secc_unbounded(), + }; // The sender will be put inside the actor id. let aid = Aid { @@ -1192,7 +1213,7 @@ impl ActorStream { /// This takes the result and executes the subsequent steps in respect to the result. Namely, /// handling the Actor's message channel and informing the ActorSystem of errors. Returns /// whether the Actor is stopping or not. - pub(crate) fn handle_result(&self, result: Result) -> bool { + pub(crate) async fn handle_result(&mut self, result: Result) -> bool { let mut stopping = false; match result { @@ -1201,33 +1222,33 @@ impl ActorStream { "Actor {} finished processing a message", self.context.aid.uuid() ); - self.receiver.pop().unwrap() + self.receiver.pop().await.unwrap() } Ok(Status::Skip) => { trace!( "Actor {} skipped processing a message", self.context.aid.uuid() ); - self.receiver.skip().unwrap() + self.receiver.skip().await.unwrap() } Ok(Status::Reset) => { trace!( "Actor {} finished processing a message and reset the cursor", self.context.aid.uuid() ); - self.receiver.pop().unwrap(); - self.receiver.reset_skip().unwrap(); + self.receiver.pop().await.unwrap(); + self.receiver.reset_skip(); } Ok(Status::Stop) => { debug!("Actor \"{}\" stopping", self.context.aid.name_or_uuid()); - self.receiver.pop().unwrap(); + self.receiver.pop().await.unwrap(); self.context .system .internal_stop_actor(&self.context.aid, None); stopping = true; } Err(e) => { - self.receiver.pop().unwrap(); + self.receiver.pop().await.unwrap(); error!( "[{}] returned an error when processing: {}", self.context.aid, &e @@ -1251,79 +1272,79 @@ impl ActorStream { } } -/// The meat of the Actor's handling -impl Stream for ActorStream { - type Item = Result; - - fn poll_next( - mut self: Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - ) -> Poll> { - trace!("Actor {} is being polled", self.context.aid.name_or_uuid()); - // If we have a pending future, that's what we poll. - if let Some(pending) = self.pending.as_mut() { - // Poll, ensure we respect stopping condition. - let poll = pending - .as_mut() - .poll(cx) - .map(|r| Some(self.overwrite_on_stop(r))); - - if let Poll::Pending = &poll { - trace!("Actor {} is pending", self.context.aid.uuid()); - } else { - drop(self.pending.take()); - } - - poll - } else { - // Are we stopped? If so, we should not have been polled, panic. This is only acceptable - // because it means a bug in the Executor or Reactor. - if self.stopping { - panic!("Stopped ActorStream was polled after stopping. Please open a bug report.") - } - // Else, we go for another. - match self.receiver.peek() { - Ok(msg) => { - // We're stopping after this future, mark as such - if let Some(m) = msg.content_as::() { - if let SystemMsg::Stop = *m { - trace!("Actor {} received stop message", self.context.aid.uuid()); - self.stopping = true; - } - } - - // Get the next future - let ctx = self.context.clone(); - let mut future = (&mut self.handler)(ctx, msg); - // Just. give it a ~~wave~~ poll!! - match future.as_mut().poll(cx) { - Poll::Ready(r) => Poll::Ready(Some(self.overwrite_on_stop(r))), - Poll::Pending => { - trace!("Actor {} is pending", self.context.aid.uuid()); - self.pending = Some(future); - Poll::Pending - } - } - } - Err(err) => match err { - // Ready(None) is standard for "Stream is depleted". The stream is effectively - // monadic around the message queue, so if the channel is depleted, the stream - // is as well. `Full` is non-contextual. - // - // While this is exhaustive, we're avoiding a catchall to in anticipation of - // future Secc errors we would *want* to handle. - SeccErrors::Empty | SeccErrors::Full(_) => { - trace!( - "Actor `{}` has no more messages, return to sleep", - self.context.aid.name_or_uuid() - ); - Poll::Ready(None) - } - }, - } - } - } -} +// /// The meat of the Actor's handling +// impl Stream for ActorStream { +// type Item = Result; + +// fn poll_next( +// mut self: Pin<&mut Self>, +// cx: &mut std::task::Context<'_>, +// ) -> Poll> { +// trace!("Actor {} is being polled", self.context.aid.name_or_uuid()); +// // If we have a pending future, that's what we poll. +// if let Some(pending) = self.pending.as_mut() { +// // Poll, ensure we respect stopping condition. +// let poll = pending +// .as_mut() +// .poll(cx) +// .map(|r| Some(self.overwrite_on_stop(r))); + +// if let Poll::Pending = &poll { +// trace!("Actor {} is pending", self.context.aid.uuid()); +// } else { +// drop(self.pending.take()); +// } + +// poll +// } else { +// // Are we stopped? If so, we should not have been polled, panic. This is only acceptable +// // because it means a bug in the Executor or Reactor. +// if self.stopping { +// panic!("Stopped ActorStream was polled after stopping. Please open a bug report.") +// } +// // Else, we go for another. +// match self.receiver.peek().await { +// Ok(msg) => { +// // We're stopping after this future, mark as such +// if let Some(m) = msg.content_as::() { +// if let SystemMsg::Stop = *m { +// trace!("Actor {} received stop message", self.context.aid.uuid()); +// self.stopping = true; +// } +// } + +// // Get the next future +// let ctx = self.context.clone(); +// let mut future = (&mut self.handler)(ctx, msg); +// // Just. give it a ~~wave~~ poll!! +// match future.as_mut().poll(cx) { +// Poll::Ready(r) => Poll::Ready(Some(self.overwrite_on_stop(r))), +// Poll::Pending => { +// trace!("Actor {} is pending", self.context.aid.uuid()); +// self.pending = Some(future); +// Poll::Pending +// } +// } +// } +// Err(err) => match err { +// // Ready(None) is standard for "Stream is depleted". The stream is effectively +// // monadic around the message queue, so if the channel is depleted, the stream +// // is as well. `Full` is non-contextual. +// // +// // While this is exhaustive, we're avoiding a catchall to in anticipation of +// // future Secc errors we would *want* to handle. +// SeccErrors::Empty | SeccErrors::Full(_) => { +// trace!( +// "Actor `{}` has no more messages, return to sleep", +// self.context.aid.name_or_uuid() +// ); +// Poll::Ready(None) +// } +// }, +// } +// } +// } +// } #[cfg(test)] mod tests { diff --git a/src/lib.rs b/src/lib.rs index 8f4b4e0..69c1dea 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -252,8 +252,7 @@ mod tests { use std::thread; use std::time::Duration; - extern crate secc; - use secc::{SeccReceiver, SeccSender}; + use crate::secc::{SeccReceiver, SeccSender}; use serde::{Deserialize, Serialize}; use super::*; diff --git a/src/secc.rs b/src/secc.rs index ae6bc5c..7f9a85f 100644 --- a/src/secc.rs +++ b/src/secc.rs @@ -1,20 +1,22 @@ -//! Async SECC ( Skip Enabled Concurrent Channel ) implementation based on [`flume`]. +//! Async SECC ( Skip Enabled Concurrent Channel ) implementation based on [`async-channel`]. //! //! This is the channel implementation used by actors to send and receive messages. use std::collections::VecDeque; +use tracing::trace; + /// Create an unbounded SECC channel /// /// > **note:** The type `T` should be efficiently clonable as calls to [`SeccReceiver::peek`] /// > must clone the value. Using an [`Arc`] is one way to do this. #[tracing::instrument] pub fn secc_unbounded() -> (SeccSender, SeccReceiver) { - let (flume_sender, flume_receiver) = flume::unbounded(); + let (raw_sender, raw_receiver) = async_channel::unbounded(); ( - SeccSender::new(flume_sender), - SeccReceiver::new(flume_receiver), + SeccSender::new(raw_sender), + SeccReceiver::new(raw_receiver), ) } @@ -24,43 +26,43 @@ pub fn secc_unbounded() -> (SeccSender, SeccReceiver) { /// > must clone the value. Using an [`Arc`] is one way to do this. #[tracing::instrument] pub fn secc_bounded(capacity: usize) -> (SeccSender, SeccReceiver) { - let (flume_sender, flume_receiver) = flume::bounded(capacity); + let (raw_sender, raw_receiver) = async_channel::bounded(capacity); ( - SeccSender::new(flume_sender), - SeccReceiver::new(flume_receiver), + SeccSender::new(raw_sender), + SeccReceiver::new(raw_receiver), ) } -/// A SECC sender, which is actually just a newtype over a `flume::Sender`. +/// A SECC sender, which is actually just a newtype over a `async_channel::Sender`. /// /// Implemented as a newtype just in case we have to add more to it later, so that we can modify /// its internals without breaking its usage. #[derive(Clone)] -pub struct SeccSender(flume::Sender); +pub struct SeccSender(async_channel::Sender); impl SeccSender { - // Create a [`SeccSender`] from a `flume` Sender. - fn new(sender: flume::Sender) -> Self { + // Create a [`SeccSender`] from a `async_channel` Sender. + fn new(sender: async_channel::Sender) -> Self { SeccSender(sender) } - /// See [`flume::Sender::send`]. - pub fn send(&self, msg: T) -> Result<(), flume::SendError> { - self.0.send(msg) + /// See [`async_channel::Sender::send`]. + pub async fn send(&self, msg: T) -> Result<(), async_channel::SendError> { + self.0.send(msg).await } - /// See [`flume::Sender::try_send`]. - pub fn try_send(&self, msg: T) -> Result<(), flume::TrySendError> { + /// See [`async_channel::Sender::try_send`]. + pub fn try_send(&self, msg: T) -> Result<(), async_channel::TrySendError> { self.0.try_send(msg) } } -/// A receiver for a SECC channel. It is a wrapper around a flume reciever along with a skipped +/// A receiver for a SECC channel. It is a wrapper around a async_channel reciever along with a skipped /// messages queue that is used to store any messages that are skipped with the skip function. pub struct SeccReceiver { - /// The underlying flume channel receiver - receiver: flume::Receiver, + /// The underlying async_channel channel receiver + receiver: async_channel::Receiver, /// A message that has been received and peeked with `peek()` peeked_message: Option, /// The queue of messages that have been skipped by the receiver @@ -72,8 +74,8 @@ pub struct SeccReceiver { } impl SeccReceiver { - // Create a [`SeccReceiver`] from a `flume` Receiver. - fn new(receiver: flume::Receiver) -> Self { + // Create a [`SeccReceiver`] from a `async_channel` Receiver. + fn new(receiver: async_channel::Receiver) -> Self { SeccReceiver { receiver, peeked_message: None, @@ -85,7 +87,7 @@ impl SeccReceiver { /// Peek at the next message in the channel #[tracing::instrument(skip(self))] - pub async fn peek(&mut self) -> Result { + pub async fn peek(&mut self) -> Result { // If we already have a peeked message, return it if let Some(msg) = &self.peeked_message { trace!("Returning the value we have previously peeked at"); @@ -113,7 +115,7 @@ impl SeccReceiver { } else { // Get the next message in the channel trace!("Grabbing the next element in the channel"); - let msg = self.receiver.recv_async().await?; + let msg = self.receiver.recv().await?; // Clone it and put it in our peeked message slot trace!("Sticking message in our peeked slot"); @@ -126,7 +128,7 @@ impl SeccReceiver { /// Receive the next message in the channel #[tracing::instrument(skip(self))] - pub async fn recv(&mut self) -> Result { + pub async fn recv(&mut self) -> Result { // If we are currently resetting if self.is_resetting { trace!("We are in the middle of resetting"); @@ -158,13 +160,13 @@ impl SeccReceiver { // Get the message from the channel } else { trace!("Getting next message from channel"); - self.receiver.recv_async().await + self.receiver.recv().await } } /// Skip the next message in the channel #[tracing::instrument(skip(self))] - pub async fn skip(&mut self) -> Result<(), flume::RecvError> { + pub async fn skip(&mut self) -> Result<(), async_channel::RecvError> { // Get the message to skip let msg = // If we have a peeked message skip that one @@ -187,7 +189,7 @@ impl SeccReceiver { // Otherwise, get the next message from the channel and skip it } else { trace!("Selecting the next message from the channel"); - self.receiver.recv_async().await? + self.receiver.recv().await? }; // Add it to the skipped message queue @@ -207,6 +209,12 @@ impl SeccReceiver { self.reset_until = self.skipped.len(); trace!(self.reset_until, "Resetting skips. Going into reset mode.") } + + /// Gets the next receivable message and discards it, returning an error if the channel is empty + #[tracing::instrument(skip(self))] + pub async fn pop(&mut self) -> Result<(), async_channel::RecvError> { + self.recv().await.map(|_| ()) + } } #[cfg(test)] @@ -223,13 +231,6 @@ mod test { tracing_subscriber::fmt::try_init().ok(); } - fn get_channel(mode: Mode) -> (SeccSender, SeccReceiver) { - match mode { - Mode::Bounded(capacity) => secc_bounded(capacity), - Mode::Unbounded => secc_unbounded(), - } - } - #[tracing::instrument] fn basic(mode: Mode) { smol::run(async move { @@ -237,19 +238,19 @@ mod test { let (sender, mut receiver) = secc_bounded(100); // Send a message - sender.send(0).unwrap(); + sender.send(0).await.unwrap(); // Receive the message assert_eq!(receiver.recv().await.unwrap(), 0); // Send another message - sender.send(1).unwrap(); + sender.send(1).await.unwrap(); // Peek at the message assert_eq!(receiver.peek().await.unwrap(), 1); // Send another message - sender.send(2).unwrap(); + sender.send(2).await.unwrap(); // Peek at the message again ( it shouldn't change ) assert_eq!(receiver.peek().await.unwrap(), 1); @@ -261,10 +262,10 @@ mod test { assert_eq!(receiver.recv().await.unwrap(), 2); // Send 4 new messages - sender.send(3).unwrap(); - sender.send(4).unwrap(); - sender.send(5).unwrap(); - sender.send(6).unwrap(); + sender.send(3).await.unwrap(); + sender.send(4).await.unwrap(); + sender.send(5).await.unwrap(); + sender.send(6).await.unwrap(); // Peek at the next message assert_eq!(receiver.peek().await.unwrap(), 3); @@ -294,11 +295,11 @@ mod test { assert_eq!(receiver.recv().await.unwrap(), 6); // Send 5 new messages - sender.send(7).unwrap(); - sender.send(8).unwrap(); - sender.send(9).unwrap(); - sender.send(10).unwrap(); - sender.send(11).unwrap(); + sender.send(7).await.unwrap(); + sender.send(8).await.unwrap(); + sender.send(9).await.unwrap(); + sender.send(10).await.unwrap(); + sender.send(11).await.unwrap(); // Skip the next two messages ( skips 7 and 8 ) receiver.skip().await.unwrap(); diff --git a/src/system.rs b/src/system.rs index 03f19a1..305caf2 100644 --- a/src/system.rs +++ b/src/system.rs @@ -12,24 +12,24 @@ use dashmap::DashMap; use once_cell::sync::OnceCell; use piper::ChangeNotifier; -use secc::{SeccReceiver, SeccSender}; use serde::{Deserialize, Serialize}; use smol::{Task, Timer}; use uuid::Uuid; -use std::collections::{BinaryHeap, HashSet}; +use std::collections::HashSet; use std::error::Error; use std::fmt; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Condvar, Mutex}; +use std::sync::Arc; use std::thread; use std::thread::JoinHandle; -use std::time::{Duration, Instant}; +use std::time::Duration; #[cfg(feature = "actor-pool")] use crate::actors::ActorPoolBuilder; -use crate::actors::{Actor, ActorBuilder, ActorStream}; +use crate::actors::{Actor, ActorBuilder, ActorChannelKind, ActorStream}; use crate::prelude::*; +use crate::secc::{SeccReceiver, SeccSender}; use crate::system::system_actor::SystemActor; mod system_actor; @@ -77,17 +77,6 @@ pub enum WireMessage { /// The message to be sent. message: Message, }, - /// A container for sending a message with a specified duration delay. - DelayedActorMessage { - /// The duration to use to delay the message. - duration: Duration, - /// The UUID of the [`Aid`] that the message is being sent to. - actor_uuid: Uuid, - /// The UUID of the system that the destination [`Aid`] is local to. - system_uuid: Uuid, - /// The message to be sent. - message: Message, - }, } /// Configuration structure for the Maxim actor system. Note that this configuration implements @@ -95,13 +84,9 @@ pub enum WireMessage { /// means. #[derive(Clone, Debug, Serialize, Deserialize)] pub struct ActorSystemConfig { - /// The default size for the channel that is created for each actor. This can be overridden on - /// a per-actor basis during spawning as well. Making the default channel size bigger allows - /// for more bandwidth in sending messages to actors but also takes more memory. Also the - /// user should consider if their actor needs a large channel then it might need to be - /// refactored or the threads size should be increased because messages aren't being processed - /// fast enough. The default value for this is 32. - pub message_channel_size: u16, + /// The default channel kind that is created for each actor. This can be overridden on + /// a per-actor basis during spawning as well. See `[ActorChannelKind`]. + pub default_channel_kind: ActorChannelKind, /// Max duration to wait between attempts to send to an actor's message channel. This is used /// to poll a busy channel that is at its capacity limit. The larger this value is, the longer /// `send` will wait for capacity in the channel but the user should be aware that if the @@ -110,7 +95,7 @@ pub struct ActorSystemConfig { pub send_timeout: Duration, /// The size of the thread pool which governs how many worker threads there are in the system. /// The number of threads should be carefully considered to have sufficient parallelism but not - /// over-schedule the CPU on the target hardware. The default value is 4 * the number of logical + /// over-schedule the CPU on the target hardware. The default value is the number of logical /// CPUs. pub thread_pool_size: u16, /// The threshold at which the dispatcher thread will warn the user that the message took too @@ -118,17 +103,6 @@ pub struct ActorSystemConfig { /// how their message processing works and refactor big tasks into a number of smaller tasks. /// The default value is 1 millisecond. pub warn_threshold: Duration, - /// This controls how long a processor will spend working on messages for an actor before - /// yielding to work on other actors in the system. The dispatcher will continue to pluck - /// messages off the actor's channel and process them until this time slice is exceeded. Note - /// that actors themselves can exceed this in processing a single message and if so, only one - /// message will be processed before yielding. The default value is 1 millisecond. - pub time_slice: Duration, - /// While Reactors will constantly attempt to get more work, they may run out. At that point, - /// they will idle for this duration, or until they get a wakeup notification. Said - /// notifications can be missed, so it's best to not set this too high. The default value is 10 - /// milliseconds. This implementation is backed by a [`Condvar`]. - pub thread_wait_time: Duration, /// Determines whether the actor system will immediately start when it is created. The default /// value is true. pub start_on_launch: bool, @@ -136,8 +110,8 @@ pub struct ActorSystemConfig { impl ActorSystemConfig { /// Return a new config with the changed `message_channel_size`. - pub fn message_channel_size(mut self, value: u16) -> Self { - self.message_channel_size = value; + pub fn default_channel_kind(mut self, value: ActorChannelKind) -> Self { + self.default_channel_kind = value; self } @@ -158,33 +132,18 @@ impl ActorSystemConfig { self.warn_threshold = value; self } - - /// Return a new config with the changed `time_slice`. - pub fn time_slice(mut self, value: Duration) -> Self { - self.time_slice = value; - self - } - - /// Return a new config with the changed `thread_wait_time`. - pub fn thread_wait_time(mut self, value: Duration) -> Self { - self.thread_wait_time = value; - self - } } impl Default for ActorSystemConfig { /// Create the config with the default values. fn default() -> ActorSystemConfig { ActorSystemConfig { + default_channel_kind: ActorChannelKind::Bounded(32), #[cfg(not(feature = "auto-num-threads"))] - thread_pool_size: 16, // Default to 4 times the assumed default number of CPUs ( 4 ) + thread_pool_size: 4, // Default to the assumed default number of CPUs ( 4 ) #[cfg(feature = "auto-num-threads")] - thread_pool_size: (num_cpus::get() * 4) as u16, // Run 4 times the number of detected CPUs - + thread_pool_size: num_cpus::get() as u16, warn_threshold: Duration::from_millis(1), - time_slice: Duration::from_millis(1), - thread_wait_time: Duration::from_millis(100), - message_channel_size: 32, send_timeout: Duration::from_millis(1), start_on_launch: true, } @@ -314,7 +273,7 @@ impl ActorSystem { } /// Starts an unstarted ActorSystem. The function will do nothing if the ActorSystem has already been started. - pub fn start(&self) { + pub async fn start(&self) { if !self .data .started @@ -326,6 +285,7 @@ impl ActorSystem { self.spawn() .name("System") .with(SystemActor, SystemActor::processor) + .await .unwrap(); } } @@ -570,7 +530,7 @@ impl ActorSystem { } // An internal helper to register an actor in the actor system. - pub(crate) fn register_actor( + pub(crate) async fn register_actor( &self, actor: Arc, stream: ActorStream, @@ -589,7 +549,7 @@ impl ActorSystem { actors_by_aid.insert(aid.clone(), actor); aids_by_uuid.insert(aid.uuid(), aid.clone()); // self.data.executor.register_actor(stream); - aid.send(Message::new(SystemMsg::Start)).unwrap(); // Actor was just made + aid.send(Message::new(SystemMsg::Start)).await.unwrap(); // Actor was just made Ok(aid) } @@ -617,7 +577,7 @@ impl ActorSystem { ActorBuilder { system: self.clone(), name: None, - channel_size: None, + channel_kind: self.data.config.default_channel_kind, } } @@ -652,7 +612,7 @@ impl ActorSystem { ActorBuilder { system: self.clone(), name: None, - channel_size: None, + channel_kind: ActorChannelKind::Bounded(32), }, count, ) @@ -670,7 +630,7 @@ impl ActorSystem { /// Internal implementation of stop_actor, so we have the ability to send an error along with /// the notification of stop. - pub(crate) fn internal_stop_actor(&self, aid: &Aid, error: impl Into>) { + pub(crate) async fn internal_stop_actor(&self, aid: &Aid, error: impl Into>) { { let actors_by_aid = &self.data.actors_by_aid; let aids_by_uuid = &self.data.aids_by_uuid; @@ -692,12 +652,15 @@ impl ActorSystem { aid: aid.clone(), error: error.clone(), }; - m_aid.send(Message::new(value)).unwrap_or_else(|error| { - error!( - "Could not send 'Stopped' to monitoring actor {}: Error: {:?}", - m_aid, error - ); - }); + m_aid + .send(Message::new(value)) + .await + .unwrap_or_else(|error| { + error!( + "Could not send 'Stopped' to monitoring actor {}: Error: {:?}", + m_aid, error + ); + }); } } } @@ -753,12 +716,12 @@ impl ActorSystem { /// Asynchronously send a message to the system actors on all connected actor systems. // FIXME (Issue #72) Add try_send ability. - pub fn send_to_system_actors(&self, message: Message) { + pub async fn send_to_system_actors(&self, message: Message) { let remotes = &*self.data.remotes; trace!("Sending message to Remote System Actors"); for remote in remotes.iter() { let aid = &remote.value().system_actor_aid; - aid.send(message.clone()).unwrap_or_else(|error| { + aid.send(message.clone()).await.unwrap_or_else(|error| { error!("Could not send to system actor {}. Error: {}", aid, error) }); } diff --git a/src/system/system_actor.rs b/src/system/system_actor.rs index f922c32..6aae826 100644 --- a/src/system/system_actor.rs +++ b/src/system/system_actor.rs @@ -23,7 +23,7 @@ impl SystemActor { // Note that you can't just unwrap or you could panic the dispatcher thread if // there is a problem sending the reply. In this case, the error is logged but the // actor moves on. - reply_to.send(reply).unwrap_or_else(|error| { + reply_to.send(reply).await.unwrap_or_else(|error| { error!( "Could not send reply to FindByName to actor {}. Error: {:?}", reply_to, error