#![feature(impl_trait_in_assoc_type)] use std::{ any::Any, collections::HashMap, fmt::Display, future::{ready, Future, Ready}, marker::PhantomData, net::IpAddr, ops::DerefMut, pin::Pin, sync::Arc, }; use btlib::{bterr, crypto::Creds, error::StringError, BlockPath, Result}; use btserde::{field_helpers::smart_ptr, from_slice, to_vec, write_to}; use bttp::{DeserCallback, MsgCallback, Receiver, Replier, Transmitter}; use serde::{de::DeserializeOwned, Deserialize, Serialize}; use tokio::{ sync::{mpsc, oneshot, Mutex, RwLock}, task::JoinHandle, }; use uuid::Uuid; /// Declares a new [Runtime] which listens for messages at the given IP address and uses the given /// [Creds]. Runtimes are intended to be created once in a process's lifetime and continue running /// until the process exits. #[macro_export] macro_rules! declare_runtime { ($name:ident, $ip_addr:expr, $creds:expr) => { ::lazy_static::lazy_static! { static ref $name: &'static ::btrun::Runtime = { ::lazy_static::lazy_static! { static ref RUNTIME: ::btrun::Runtime = ::btrun::Runtime::_new($creds).unwrap(); static ref RECEIVER: ::bttp::Receiver = _new_receiver($ip_addr, $creds, &*RUNTIME); } // By dereferencing RECEIVER we ensure it is started. let _ = &*RECEIVER; &*RUNTIME }; } }; } /// This function is not intended to be called by downstream crates. #[doc(hidden)] pub fn _new_receiver(ip_addr: IpAddr, creds: Arc, runtime: &'static Runtime) -> Receiver where C: 'static + Send + Sync + Creds, { let callback = RuntimeCallback::new(runtime); Receiver::new(ip_addr, creds, callback).unwrap() } /// An actor runtime. /// /// Actors can be activated by the runtime and execute autonomously until they return. Running /// actors can be sent messages using the `send` method, which does not wait for a response from the /// recipient. If a reply is needed, then `call` can be used, which returns a future that will /// be ready when the reply has been received. pub struct Runtime { path: Arc, handles: RwLock>, peers: RwLock, Transmitter>>, } impl Runtime { /// This method is not intended to be called directly by downstream crates. Use the macro /// [declare_runtime] to create a [Runtime]. /// /// If you create a non-static [Runtime], your process will panic when it is dropped. #[doc(hidden)] pub fn _new(creds: Arc) -> Result { let path = Arc::new(creds.bind_path()?); Ok(Runtime { path, handles: RwLock::new(HashMap::new()), peers: RwLock::new(HashMap::new()), }) } pub fn path(&self) -> &Arc { &self.path } /// Returns the number of actors that are currently executing in this [Runtime]. pub async fn num_running(&self) -> usize { let guard = self.handles.read().await; guard.len() } /// Sends a message to the actor identified by the given [ActorName]. pub async fn send( &self, to: ActorName, from: ActorName, msg: T, ) -> Result<()> { if to.path == self.path { let guard = self.handles.read().await; if let Some(handle) = guard.get(&to.act_id) { handle.send(from, msg).await } else { Err(bterr!("invalid actor name")) } } else { let guard = self.peers.read().await; if let Some(peer) = guard.get(&to.path) { let buf = to_vec(&msg)?; let wire_msg = WireMsg { to, from, payload: &buf, }; peer.send(wire_msg).await } else { // TODO: Use the filesystem to discover the address of the recipient and connect to // it. todo!() } } } /// Sends a message to the actor identified by the given [ActorName] and returns a future which /// is ready when a reply has been received. pub async fn call( &self, to: ActorName, from: ActorName, msg: T, ) -> Result { if to.path == self.path { let guard = self.handles.read().await; if let Some(handle) = guard.get(&to.act_id) { handle.call_through(from, msg).await } else { Err(bterr!("invalid actor name")) } } else { let guard = self.peers.read().await; if let Some(peer) = guard.get(&to.path) { let buf = to_vec(&msg)?; let wire_msg = WireMsg { to, from, payload: &buf, }; peer.call(wire_msg, ReplyCallback::::new()).await? } else { // TODO: Use the filesystem to discover the address of the recipient and connect to // it. todo!() } } } /// Resolves the given [ServiceName] to an [ActorName] which is part of it. pub async fn resolve<'a>(&'a self, _service: &ServiceName) -> Result { todo!() } /// Activates a new actor using the given activator function and returns a handle to it. pub async fn activate(&'static self, activator: F) -> ActorName where Msg: 'static + CallMsg, Fut: 'static + Send + Future, F: FnOnce(&'static Runtime, mpsc::Receiver>, Uuid) -> Fut, { let mut guard = self.handles.write().await; let act_id = { let mut act_id = Uuid::new_v4(); while guard.contains_key(&act_id) { act_id = Uuid::new_v4(); } act_id }; let act_name = self.actor_name(act_id); let (tx, rx) = mpsc::channel::>(MAILBOX_LIMIT); // The deliverer closure is responsible for deserializing messages received over the wire // and delivering them to the actor's mailbox, and sending replies to call messages. let deliverer = { let buffer = Arc::new(Mutex::new(Vec::::new())); let tx = tx.clone(); let act_name = act_name.clone(); move |envelope: WireEnvelope| { let (wire_msg, replier) = envelope.into_parts(); let result = from_slice(wire_msg.payload); let buffer = buffer.clone(); let tx = tx.clone(); let act_name = act_name.clone(); let fut: FutureResult = Box::pin(async move { let msg = result?; if let Some(mut replier) = replier { let (envelope, rx) = Envelope::new_call(act_name, msg); tx.send(envelope).await.map_err(|_| { bterr!("failed to deliver message. Recipient may have halted.") })?; match rx.await { Ok(reply) => { let mut guard = buffer.lock().await; guard.clear(); write_to(&reply, guard.deref_mut())?; let wire_reply = WireReply::Ok(&guard); replier.reply(wire_reply).await } Err(err) => replier.reply_err(err.to_string(), None).await, } } else { tx.send(Envelope::new_send(act_name, msg)) .await .map_err(|_| { bterr!("failed to deliver message. Recipient may have halted.") }) } }); fut } }; let handle = tokio::task::spawn(activator(self, rx, act_id)); let actor_handle = ActorHandle::new(handle, tx, deliverer); guard.insert(act_id, actor_handle); act_name } /// Registers an actor as a service with the given [ServiceId]. pub async fn register( &self, _id: ServiceId, _activator: F, _deserializer: G, ) -> Result<()> where Msg: 'static + CallMsg, Fut: 'static + Send + Future, F: Fn(mpsc::Receiver>, Uuid) -> Fut, G: 'static + Send + Sync + Fn(&[u8]) -> Result, { todo!() } /// Returns the [ActorHandle] for the actor with the given name. /// /// If there is no such actor in this runtime then a [RuntimeError::BadActorName] error is /// returned. /// /// Note that the actor will be aborted when the given handle is dropped (unless it has already /// returned when the handle is dropped), and no further messages will be delivered to it by /// this runtime. pub async fn take(&self, name: &ActorName) -> Result { if name.path == self.path { let mut guard = self.handles.write().await; if let Some(handle) = guard.remove(&name.act_id) { Ok(handle) } else { Err(RuntimeError::BadActorName(name.clone()).into()) } } else { Err(RuntimeError::BadActorName(name.clone()).into()) } } /// Returns the name of the actor in this runtime with the given actor ID. pub fn actor_name(&self, act_id: Uuid) -> ActorName { ActorName::new(self.path.clone(), act_id) } } impl Drop for Runtime { fn drop(&mut self) { panic!("A Runtime was dropped. Panicking to avoid undefined behavior."); } } #[derive(Debug, Clone, PartialEq, Eq)] pub enum RuntimeError { BadActorName(ActorName), } impl Display for RuntimeError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::BadActorName(name) => write!(f, "bad actor name: {name}"), } } } impl std::error::Error for RuntimeError {} /// Represents the terminal state of an actor, where it stops processing messages and halts. pub struct End; impl End { /// Returns the identifier for this type which is expected in protocol definitions. pub fn ident() -> &'static str { stringify!(End) } } #[allow(dead_code)] /// Delivered to an actor implementation when it starts up. pub struct Activate { /// A reference to the `Runtime` which is running this actor. rt: &'static Runtime, /// The ID assigned to this actor. act_id: Uuid, } impl Activate { pub fn new(rt: &'static Runtime, act_id: Uuid) -> Self { Self { rt, act_id } } } /// Deserializes replies sent over the wire. pub struct ReplyCallback { _phantom: PhantomData, } impl ReplyCallback { pub fn new() -> Self { Self { _phantom: PhantomData, } } } impl Default for ReplyCallback { fn default() -> Self { Self::new() } } impl DeserCallback for ReplyCallback { type Arg<'de> = WireReply<'de> where T: 'de; type Return = Result; type CallFut<'de> = Ready where T: 'de, T::Reply: 'de; fn call<'de>(&'de mut self, arg: Self::Arg<'de>) -> Self::CallFut<'de> { let result = match arg { WireReply::Ok(slice) => from_slice(slice).map_err(|err| err.into()), WireReply::Err(msg) => Err(StringError::new(msg.to_string()).into()), }; ready(result) } } struct SendReplyCallback { replier: Option, } impl SendReplyCallback { fn new(replier: Replier) -> Self { Self { replier: Some(replier), } } } impl DeserCallback for SendReplyCallback { type Arg<'de> = WireReply<'de>; type Return = Result<()>; type CallFut<'de> = impl 'de + Future; fn call<'de>(&'de mut self, arg: Self::Arg<'de>) -> Self::CallFut<'de> { async move { if let Some(mut replier) = self.replier.take() { replier.reply(arg).await } else { Ok(()) } } } } /// This struct implements the server callback for network messages. #[derive(Clone)] struct RuntimeCallback { rt: &'static Runtime, } impl RuntimeCallback { fn new(rt: &'static Runtime) -> Self { Self { rt } } async fn deliver_local(&self, msg: WireMsg<'_>, replier: Option) -> Result<()> { let guard = self.rt.handles.read().await; if let Some(handle) = guard.get(&msg.to.act_id) { let envelope = if let Some(replier) = replier { WireEnvelope::Call { msg, replier } } else { WireEnvelope::Send { msg } }; (handle.deliverer)(envelope).await } else { Err(bterr!("invalid actor name: {}", msg.to)) } } async fn route_msg(&self, msg: WireMsg<'_>, replier: Option) -> Result<()> { let guard = self.rt.peers.read().await; if let Some(tx) = guard.get(msg.to.path()) { if let Some(replier) = replier { let callback = SendReplyCallback::new(replier); tx.call(msg, callback).await? } else { tx.send(msg).await } } else { Err(bterr!( "unable to deliver message to peer at '{}'", msg.to.path() )) } } } impl MsgCallback for RuntimeCallback { type Arg<'de> = WireMsg<'de>; type CallFut<'de> = impl 'de + Future>; fn call<'de>(&'de self, arg: bttp::MsgReceived>) -> Self::CallFut<'de> { async move { let (_, body, replier) = arg.into_parts(); if body.to.path() == self.rt.path() { self.deliver_local(body, replier).await } else { self.route_msg(body, replier).await } } } } /// A unique identifier for a particular service. #[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Serialize, Deserialize)] pub struct ServiceId(#[serde(with = "smart_ptr")] Arc); impl From for ServiceId { fn from(value: String) -> Self { Self(Arc::new(value)) } } impl<'a> From<&'a str> for ServiceId { fn from(value: &'a str) -> Self { Self(Arc::new(value.to_owned())) } } /// A unique identifier for a service. /// /// A service is a collection of actors in the same directory which provide some functionality. #[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Serialize, Deserialize)] pub struct ServiceName { /// The path to the directory containing the service. #[serde(with = "smart_ptr")] path: Arc, /// The id of the service. service_id: ServiceId, } /// A unique identifier for a specific actor activation. #[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Serialize, Deserialize)] pub struct ActorName { /// The path to the directory containing this actor. #[serde(with = "smart_ptr")] path: Arc, /// A unique identifier for an actor activation. Even as an actor transitions to different types /// as it handles messages, this value does not change. Thus this value can be used to trace an /// actor through a series of state transitions. act_id: Uuid, } impl ActorName { pub fn new(path: Arc, act_id: Uuid) -> Self { Self { path, act_id } } pub fn path(&self) -> &Arc { &self.path } pub fn act_id(&self) -> Uuid { self.act_id } } impl Display for ActorName { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(f, "{}@{}", self.act_id, self.path) } } /// Trait for messages which expect exactly one reply. pub trait CallMsg: Serialize + DeserializeOwned + Send + Sync { /// The reply type expected for this message. type Reply: Serialize + DeserializeOwned + Send + Sync; } /// Trait for messages which expect exactly zero replies. pub trait SendMsg: CallMsg {} /// A type used to express when a reply is not expected for a message type. #[derive(Serialize, Deserialize)] enum NoReply {} /// The maximum number of messages which can be kept in an actor's mailbox. const MAILBOX_LIMIT: usize = 32; /// The type of messages sent over the wire between runtimes. #[derive(Serialize, Deserialize)] pub struct WireMsg<'a> { to: ActorName, from: ActorName, payload: &'a [u8], } impl<'a> WireMsg<'a> { pub fn new(to: ActorName, from: ActorName, payload: &'a [u8]) -> Self { Self { to, from, payload } } } impl<'a> bttp::CallMsg<'a> for WireMsg<'a> { type Reply<'r> = WireReply<'r>; } impl<'a> bttp::SendMsg<'a> for WireMsg<'a> {} #[derive(Serialize, Deserialize)] pub enum WireReply<'a> { Ok(&'a [u8]), Err(&'a str), } /// A wrapper around [WireMsg] which indicates whether a call or send was executed. enum WireEnvelope<'de> { Send { msg: WireMsg<'de> }, Call { msg: WireMsg<'de>, replier: Replier }, } impl<'de> WireEnvelope<'de> { fn into_parts(self) -> (WireMsg<'de>, Option) { match self { Self::Send { msg } => (msg, None), Self::Call { msg, replier } => (msg, Some(replier)), } } } /// Wrapper around a message type `T` which indicates who the message is from and, if the message /// was dispatched with `call`, provides a channel to reply to it. pub struct Envelope { from: ActorName, reply: Option>, msg: T, } impl Envelope { pub fn new(msg: T, reply: Option>, from: ActorName) -> Self { Self { from, reply, msg } } /// Creates a new envelope containing the given message which does not expect a reply. fn new_send(from: ActorName, msg: T) -> Self { Self { from, msg, reply: None, } } /// Creates a new envelope containing the given message which expects exactly one reply. fn new_call(from: ActorName, msg: T) -> (Self, oneshot::Receiver) { let (tx, rx) = oneshot::channel::(); let envelope = Self { from, msg, reply: Some(tx), }; (envelope, rx) } /// Returns the name of the actor which sent this message. pub fn from(&self) -> &ActorName { &self.from } /// Returns a reference to the message in this envelope. pub fn msg(&self) -> &T { &self.msg } /// Sends a reply to this message. /// /// If this message is not expecting a reply, or if this message has already been replied to, /// then an error is returned. pub fn reply(&mut self, reply: T::Reply) -> Result<()> { if let Some(tx) = self.reply.take() { if tx.send(reply).is_ok() { Ok(()) } else { Err(bterr!("failed to send reply")) } } else { Err(bterr!("reply already sent")) } } /// Returns true if this message expects a reply and it has not already been replied to. pub fn needs_reply(&self) -> bool { self.reply.is_some() } pub fn split(self) -> (T, Option>, ActorName) { (self.msg, self.reply, self.from) } } type FutureResult = Pin>>>; pub struct ActorHandle { handle: Option>, sender: Box, deliverer: Box) -> FutureResult>, } impl ActorHandle { fn new(handle: JoinHandle<()>, sender: mpsc::Sender>, deliverer: F) -> Self where T: 'static + CallMsg, F: 'static + Send + Sync + Fn(WireEnvelope<'_>) -> FutureResult, { Self { handle: Some(handle), sender: Box::new(sender), deliverer: Box::new(deliverer), } } fn sender(&self) -> Result<&mpsc::Sender>> { self.sender .downcast_ref::>>() .ok_or_else(|| bterr!("unexpected message type")) } /// Sends a message to the actor represented by this handle. pub async fn send(&self, from: ActorName, msg: T) -> Result<()> { let sender = self.sender()?; sender .send(Envelope::new_send(from, msg)) .await .map_err(|_| bterr!("failed to enqueue message"))?; Ok(()) } pub async fn call_through( &self, from: ActorName, msg: T, ) -> Result { let sender = self.sender()?; let (envelope, rx) = Envelope::new_call(from, msg); sender .send(envelope) .await .map_err(|_| bterr!("failed to enqueue call"))?; let reply = rx.await?; Ok(reply) } pub async fn returned(&mut self) -> Result<()> { if let Some(handle) = self.handle.take() { handle.await?; } Ok(()) } pub fn abort(&mut self) { if let Some(handle) = self.handle.take() { handle.abort(); } } } impl Drop for ActorHandle { fn drop(&mut self) { self.abort(); } }