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 7000329..8ad6f69 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -41,8 +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" } @@ -51,20 +67,23 @@ 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" [dependencies] -bincode = "1.1.4" -dashmap = "1.0.3" -futures = "0.3.1" -num_cpus = "1.10.1" -log = "0.4" +smol = "0.1.4" # Async Executor +piper = { path = "lib/piper" } # Async pipes, channels, mutexes, and more +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 once_cell = "1.0.2" -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 } - +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/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/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/examples/philosophers.rs b/examples/philosophers.rs index 3ef2318..48c4036 100644 --- a/examples/philosophers.rs +++ b/examples/philosophers.rs @@ -21,8 +21,8 @@ //! badly timed messages. This is largely up to the user. use maxim::prelude::*; -use log::LevelFilter; -use log::{error, info}; +use tracing::{error, info}; + use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::env; @@ -438,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. diff --git a/lib/piper b/lib/piper new file mode 160000 index 0000000..73e7b41 --- /dev/null +++ b/lib/piper @@ -0,0 +1 @@ +Subproject commit 73e7b4121da3edf4fe99463c179c85de7dafe828 diff --git a/src/actors.rs b/src/actors.rs index 84ac5e6..55a4ce8 100644 --- a/src/actors.rs +++ b/src/actors.rs @@ -6,10 +6,8 @@ //! 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}; -use log::{debug, error, trace, warn}; #[cfg(feature = "actor-pool")] use rand::{ distributions::{Distribution, Uniform}, @@ -17,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; @@ -33,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. @@ -109,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 { @@ -191,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 @@ -333,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, @@ -343,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(()) } } @@ -400,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 @@ -440,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 @@ -483,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, .. @@ -491,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(()) } @@ -547,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 @@ -588,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`] @@ -639,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 @@ -678,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 } } @@ -757,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")] @@ -824,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 } } @@ -971,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 { @@ -980,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, @@ -988,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. @@ -1000,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 } } @@ -1026,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, @@ -1040,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 @@ -1058,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 } } @@ -1116,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 { @@ -1193,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 { @@ -1202,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 @@ -1252,85 +1272,84 @@ 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 { use super::*; use crate::tests::*; - use log::*; use std::thread; use std::time::Instant; @@ -1437,6 +1456,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. @@ -1467,7 +1487,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 { @@ -1484,7 +1504,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/cluster.rs b/src/cluster.rs index 6d84112..71c9c74 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::*; diff --git a/src/executor.rs b/src/executor.rs index bfef62e..d5ababc 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -1,18 +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 log::{debug, info, trace, warn}; +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 @@ -22,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(); - 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(), } } @@ -417,7 +402,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; diff --git a/src/executor/thread_pool.rs b/src/executor/thread_pool.rs index 0a98deb..41c0bd3 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; diff --git a/src/lib.rs b/src/lib.rs index f6d65c1..69c1dea 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}; @@ -198,9 +202,13 @@ use std::fmt::{Display, Formatter}; pub use futures; use prelude::*; +// Skip Enabled Concurrent Channels +pub(crate) mod secc; + 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; @@ -244,8 +252,7 @@ mod tests { use std::thread; use std::time::Duration; - use log::LevelFilter; - use secc::{SeccReceiver, SeccSender}; + use crate::secc::{SeccReceiver, SeccSender}; use serde::{Deserialize, Serialize}; use super::*; @@ -285,10 +292,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/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/secc.rs b/src/secc.rs new file mode 100644 index 0000000..7f9a85f --- /dev/null +++ b/src/secc.rs @@ -0,0 +1,347 @@ +//! 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 (raw_sender, raw_receiver) = async_channel::unbounded(); + + ( + SeccSender::new(raw_sender), + SeccReceiver::new(raw_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. +#[tracing::instrument] +pub fn secc_bounded(capacity: usize) -> (SeccSender, SeccReceiver) { + let (raw_sender, raw_receiver) = async_channel::bounded(capacity); + + ( + SeccSender::new(raw_sender), + SeccReceiver::new(raw_receiver), + ) +} + +/// 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(async_channel::Sender); + +impl SeccSender { + // Create a [`SeccSender`] from a `async_channel` Sender. + fn new(sender: async_channel::Sender) -> Self { + SeccSender(sender) + } + + /// See [`async_channel::Sender::send`]. + pub async fn send(&self, msg: T) -> Result<(), async_channel::SendError> { + self.0.send(msg).await + } + + /// 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 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 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 + 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 `async_channel` Receiver. + fn new(receiver: async_channel::Receiver) -> Self { + SeccReceiver { + receiver, + peeked_message: None, + skipped: VecDeque::new(), + is_resetting: false, + reset_until: 0, + } + } + + /// 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 { + 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 + trace!("Grabbing the next element in the channel"); + let msg = self.receiver.recv().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 + Ok(msg) + } + } + + /// 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() { + self.reset_until -= 1; + trace!(self.reset_until, "Decremented reset_until"); + + if self.reset_until == 0 { + trace!("reset_unitl == 0: going out of reset mode"); + self.is_resetting = false; + } + + trace!("Returning skipped message"); + Ok(msg) + + } else { + 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().await + } + } + + /// Skip the next message in the channel + #[tracing::instrument(skip(self))] + 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 + 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 \ + queue."); + } + + // Otherwise, get the next message from the channel and skip it + } else { + trace!("Selecting the next message from the channel"); + self.receiver.recv().await? + }; + + // Add it to the skipped message queue + trace!("Skipping selected message"); + 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 + #[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(); + 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)] +mod test { + use super::*; + + #[derive(Debug)] + enum Mode { + Bounded(usize), + Unbounded, + } + + fn init_logging() { + tracing_subscriber::fmt::try_init().ok(); + } + + #[tracing::instrument] + 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).await.unwrap(); + + // Receive the message + assert_eq!(receiver.recv().await.unwrap(), 0); + + // Send another message + sender.send(1).await.unwrap(); + + // Peek at the message + assert_eq!(receiver.peek().await.unwrap(), 1); + + // Send another message + sender.send(2).await.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).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); + + // 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).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(); + 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 secc_bounded_basic() { + init_logging(); + basic(Mode::Bounded(100)); + } + + #[test] + fn secc_unbounded_basic() { + init_logging(); + basic(Mode::Unbounded); + } +} diff --git a/src/system.rs b/src/system.rs index b02307f..305caf2 100644 --- a/src/system.rs +++ b/src/system.rs @@ -9,26 +9,28 @@ //! //! 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 log::{debug, error, info, trace, warn}; use once_cell::sync::OnceCell; -use secc::{SeccReceiver, SeccSender}; +use piper::ChangeNotifier; use serde::{Deserialize, Serialize}; -use std::collections::{BinaryHeap, HashSet}; +use smol::{Task, Timer}; +use uuid::Uuid; + +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 uuid::Uuid; +use std::time::Duration; + +#[cfg(feature = "actor-pool")] +use crate::actors::ActorPoolBuilder; +use crate::actors::{Actor, ActorBuilder, ActorChannelKind, ActorStream}; +use crate::prelude::*; +use crate::secc::{SeccReceiver, SeccSender}; +use crate::system::system_actor::SystemActor; mod system_actor; @@ -75,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 @@ -93,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 @@ -108,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 @@ -116,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, @@ -134,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 } @@ -156,29 +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 { - thread_pool_size: (num_cpus::get() * 4) as u16, + default_channel_kind: ActorChannelKind::Bounded(32), + #[cfg(not(feature = "auto-num-threads"))] + thread_pool_size: 4, // Default to the assumed default number of CPUs ( 4 ) + #[cfg(feature = "auto-num-threads")] + 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, } @@ -217,53 +182,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 @@ -277,8 +207,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. @@ -296,10 +224,28 @@ 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()); + 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; @@ -308,8 +254,6 @@ impl ActorSystem { data: Arc::new(ActorSystemData { uuid, config, - threads, - executor, started: AtomicBool::new(false), shutdown_triggered, actors_by_aid: Arc::new(DashMap::default()), @@ -317,7 +261,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())), }), }; @@ -330,75 +273,23 @@ 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 .compare_and_swap(false, true, Ordering::Relaxed) { 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() .name("System") .with(SystemActor, SystemActor::processor) + .await .unwrap(); } } - /// 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 { - trace!("Sending delayed message"); - msg.destination - .send(msg.message.clone()) - .unwrap_or_else(|error| { - 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 @@ -412,117 +303,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(); - 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); - } - } - } + // 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. @@ -556,90 +448,89 @@ 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 { - 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) + pub fn trigger_and_await_shutdown(&self, timeout: impl Into>) { + unimplemented!("FIXME: Re-implement graceful shutdown"); + // self.trigger_shutdown(); + // self.await_shutdown(timeout) } // 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, @@ -657,8 +548,8 @@ 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 + // self.data.executor.register_actor(stream); + aid.send(Message::new(SystemMsg::Start)).await.unwrap(); // Actor was just made Ok(aid) } @@ -686,7 +577,7 @@ impl ActorSystem { ActorBuilder { system: self.clone(), name: None, - channel_size: None, + channel_kind: self.data.config.default_channel_kind, } } @@ -721,34 +612,12 @@ impl ActorSystem { ActorBuilder { system: self.clone(), name: None, - channel_size: None, + channel_kind: ActorChannelKind::Bounded(32), }, count, ) } - /// 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. - 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. @@ -761,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; @@ -783,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 + ); + }); } } } @@ -844,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) }); } @@ -860,22 +732,12 @@ 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(); } } @@ -902,13 +764,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); }); @@ -925,6 +789,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)); @@ -941,16 +806,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. @@ -1123,7 +988,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 diff --git a/src/system/system_actor.rs b/src/system/system_actor.rs index fe9e406..6aae826 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; @@ -24,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