Repository: slawlor/ractor
Pull request #10 (2023-01-07 16:50:44)

Adding support for a global singleton registry for named actors
This follows the Erlang notion of [name registration via `register_name/3`](https://www.erlang.org/doc/man/global.html#register_name-2).

In our case, it's an in-memory naming, but allows global addressing of actors.

Additionally this PR addresses limitations in the port listening routines which wasn't using priorities due to the random way `tokio::select!` operates by default. Now we add the `biased;` indication to identify that we're handling the fairness of the ports, in that there is an inherent priority in them (Signal > Stop > Supervision > Message). 


Patch:
diff --git a/ractor/Cargo.toml b/ractor/Cargo.toml
index b3d8922..ea3be38 100644
--- a/ractor/Cargo.toml
+++ b/ractor/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "ractor"
-version = "0.1.1"
+version = "0.1.2"
 authors = ["Sean Lawlor <seanlawlor@fb.com>"]
 description = "A actor framework for Rust"
 license = "MIT"
@@ -15,6 +15,8 @@ categories = ["actor"]
 [dependencies]
 async-trait = "0.1"
 dashmap = "5"
+futures = "0.3"
+once_cell = "1"
 tokio = { version = "1.23", features = ["rt", "time", "sync", "macros"]}
 
 [dev-dependencies]
diff --git a/ractor/examples/supervisor.rs b/ractor/examples/supervisor.rs
new file mode 100644
index 0000000..c478b56
--- /dev/null
+++ b/ractor/examples/supervisor.rs
@@ -0,0 +1,261 @@
+// Copyright (c) Sean Lawlor
+//
+// This source code is licensed under both the MIT license found in the
+// LICENSE-MIT file in the root directory of this source tree.
+
+//! An example supervisory tree of actors
+//!
+//! Execute with
+//!
+//! ```text
+//! cargo run --example supervisor
+//! ```
+
+use ractor::{rpc, Actor, ActorCell, ActorHandler, RpcReplyPort, SupervisionEvent};
+
+use tokio::time::Duration;
+
+// ============================== Main ============================== //
+
+#[tokio::main]
+async fn main() {
+    let (root, handle) = Actor::spawn(Some("root"), RootActor)
+        .await
+        .expect("Failed to start root actor");
+    let mid_level = rpc::call::<RootActor, _, _>(
+        &root,
+        RootActorMessage::GetMidLevel,
+        Some(Duration::from_millis(100)),
+    )
+    .await
+    .expect("Failed to send message to root actor")
+    .expect("Failed to get mid level actor");
+
+    let leaf = rpc::call::<MidLevelActor, _, _>(
+        &mid_level,
+        MidLevelActorMessage::GetLeaf,
+        Some(Duration::from_millis(100)),
+    )
+    .await
+    .expect("Failed to send message to mid-level actor")
+    .expect("Failed to get leaf actor");
+
+    // send some no-op's to the leaf
+    for _ in 1..10 {
+        rpc::cast::<LeafActor>(&leaf, LeafActorMessage::NoOp)
+            .expect("Failed to send message to leaf actor");
+        tokio::time::sleep(Duration::from_millis(500)).await;
+    }
+
+    // send the "Boom" message and follow the supervision flow
+    rpc::cast::<LeafActor>(&leaf, LeafActorMessage::Boom)
+        .expect("Failed to send message to leaf actor");
+
+    handle.await.expect("Failed waiting for root actor to die");
+}
+
+// ============================== Leaf actor ============================== //
+
+struct LeafActor;
+
+struct LeafActorState {}
+
+enum LeafActorMessage {
+    Boom,
+    NoOp,
+}
+
+#[async_trait::async_trait]
+impl ActorHandler for LeafActor {
+    type Msg = LeafActorMessage;
+    type State = LeafActorState;
+    async fn pre_start(&self, _myself: ActorCell) -> Self::State {
+        LeafActorState {}
+    }
+
+    // async fn post_start(&self, myself: ActorCell, state: &Self::State) -> Option<Self::State> {
+    //     None
+    // }
+
+    // async fn post_stop(&self, myself: ActorCell, state: Self::State) -> Self::State {
+    //     state
+    // }
+
+    async fn handle(
+        &self,
+        _myself: ActorCell,
+        message: Self::Msg,
+        _state: &Self::State,
+    ) -> Option<Self::State> {
+        match message {
+            Self::Msg::Boom => {
+                panic!("LeafActor: Oh crap!");
+            }
+            Self::Msg::NoOp => {
+                println!("LeafActor: No-op!");
+            }
+        }
+        None
+    }
+
+    // async fn handle_supervisor_evt(
+    //     &self,
+    //     myself: ActorCell,
+    //     message: SupervisionEvent,
+    //     state: &Self::State,
+    // ) -> Option<Self::State> {
+    //     None
+    // }
+}
+
+// ============================== Mid-level supervisor ============================== //
+
+struct MidLevelActor;
+
+struct MidLevelActorState {
+    leaf_actor: ActorCell,
+}
+
+enum MidLevelActorMessage {
+    GetLeaf(RpcReplyPort<ActorCell>),
+}
+
+#[async_trait::async_trait]
+impl ActorHandler for MidLevelActor {
+    type Msg = MidLevelActorMessage;
+    type State = MidLevelActorState;
+
+    async fn pre_start(&self, myself: ActorCell) -> Self::State {
+        let (leaf_actor, _) = Actor::spawn_linked(Some("leaf"), LeafActor, myself)
+            .await
+            .expect("Failed to start leaf actor");
+        MidLevelActorState { leaf_actor }
+    }
+
+    // async fn post_start(&self, myself: ActorCell, state: &Self::State) -> Option<Self::State> {
+    //     None
+    // }
+
+    // async fn post_stop(&self, myself: ActorCell, state: Self::State) -> Self::State {
+    //     state
+    // }
+
+    async fn handle(
+        &self,
+        _myself: ActorCell,
+        message: Self::Msg,
+        state: &Self::State,
+    ) -> Option<Self::State> {
+        match message {
+            MidLevelActorMessage::GetLeaf(reply) => {
+                if !reply.is_closed() {
+                    reply
+                        .send(state.leaf_actor.clone())
+                        .expect("Failed to reply to RPC");
+                }
+            }
+        }
+        None
+    }
+
+    async fn handle_supervisor_evt(
+        &self,
+        _myself: ActorCell,
+        message: SupervisionEvent,
+        state: &Self::State,
+    ) -> Option<Self::State> {
+        match message {
+            SupervisionEvent::ActorPanicked(dead_actor, panic_msg)
+                if dead_actor.get_id() == state.leaf_actor.get_id() =>
+            {
+                println!(
+                    "MidLevelActor: {:?} panicked with '{}'",
+                    dead_actor, panic_msg
+                );
+
+                panic!(
+                    "MidLevelActor: Mid-level actor panicking because Leaf actor panicked with '{}'",
+                    panic_msg
+                );
+            }
+            other => {
+                println!("MidLevelActor: recieved supervisor event '{}'", other);
+            }
+        }
+        None
+    }
+}
+
+// ============================== Root Actor Definition ============================== //
+
+struct RootActor;
+
+struct RootActorState {
+    mid_level_actor: ActorCell,
+}
+
+enum RootActorMessage {
+    GetMidLevel(RpcReplyPort<ActorCell>),
+}
+
+#[async_trait::async_trait]
+impl ActorHandler for RootActor {
+    type Msg = RootActorMessage;
+    type State = RootActorState;
+
+    async fn pre_start(&self, myself: ActorCell) -> Self::State {
+        println!("RootActor: Started {:?}", myself);
+        let (mid_level_actor, _) = Actor::spawn_linked(Some("mid-level"), MidLevelActor, myself)
+            .await
+            .expect("Failed to spawn mid-level actor");
+        RootActorState { mid_level_actor }
+    }
+
+    // async fn post_start(&self, myself: ActorCell, state: &Self::State) -> Option<Self::State> {
+    //     None
+    // }
+
+    // async fn post_stop(&self, myself: ActorCell, state: Self::State) -> Self::State {
+    //     state
+    // }
+
+    async fn handle(
+        &self,
+        _myself: ActorCell,
+        message: Self::Msg,
+        state: &Self::State,
+    ) -> Option<Self::State> {
+        match message {
+            RootActorMessage::GetMidLevel(reply) => {
+                if !reply.is_closed() {
+                    reply
+                        .send(state.mid_level_actor.clone())
+                        .expect("Failed to reply to RPC");
+                }
+            }
+        }
+        None
+    }
+
+    async fn handle_supervisor_evt(
+        &self,
+        myself: ActorCell,
+        message: SupervisionEvent,
+        state: &Self::State,
+    ) -> Option<Self::State> {
+        match message {
+            SupervisionEvent::ActorPanicked(dead_actor, panic_msg)
+                if dead_actor.get_id() == state.mid_level_actor.get_id() =>
+            {
+                println!("RootActor: {:?} panicked with '{}'", dead_actor, panic_msg);
+
+                println!("RootActor: Terminating root actor, all my kids are dead!");
+                myself.stop(Some("Everyone died :(".to_string()));
+            }
+            other => {
+                println!("RootActor: recieved supervisor event '{}'", other);
+            }
+        }
+        None
+    }
+}
diff --git a/ractor/src/actor/actor_cell.rs b/ractor/src/actor/actor_cell.rs
index 073f88f..f3a1c16 100644
--- a/ractor/src/actor/actor_cell.rs
+++ b/ractor/src/actor/actor_cell.rs
@@ -3,7 +3,11 @@
 // This source code is licensed under both the MIT license found in the
 // LICENSE-MIT file in the root directory of this source tree.
 
-//! A reference counted actor which can be passed around as needed
+//! [ActorCell] is reference counted actor which can be passed around as needed
+//!
+//! This module contains all the functionality around the [ActorCell], including
+//! the internal properties, ports, states, etc. [ActorCell] is the basic primitive
+//! for references to a given actor and its communication channels
 
 use std::sync::atomic::{AtomicU8, Ordering};
 use std::sync::Arc;
@@ -19,7 +23,7 @@ use crate::port::{
     BoundedInputPort, BoundedInputPortReceiver, InputPort, InputPortReceiver, RpcReplyPort,
 };
 use crate::rpc::{self, CallResult};
-use crate::{ActorHandler, ActorId, Message};
+use crate::{ActorHandler, ActorId, ActorName, Message, SpawnErr};
 
 /// [ActorStatus] represents the status of an actor
 #[derive(Debug, Clone, Eq, PartialEq, Copy)]
@@ -54,10 +58,84 @@ pub(crate) struct ActorPortSet {
     pub(crate) message_rx: InputPortReceiver<BoxedMessage>,
 }
 
+pub(crate) enum ActorPortMessage {
+    Signal(Signal),
+    Stop(StopMessage),
+    Supervision(SupervisionEvent),
+    Message(BoxedMessage),
+}
+
+impl ActorPortSet {
+    /// Run a future beside the signal port, so that
+    /// the signal port can terminate the async work
+    ///
+    /// * `future` - The future to execute
+    ///
+    /// Returns [Ok(`TState`)] when the future completes without
+    /// signal interruption, [Err(Signal)] in the event the
+    /// signal interrupts the async work.
+    pub async fn run_with_signal<TState>(
+        &mut self,
+        future: impl futures::Future<Output = TState>,
+    ) -> Result<TState, Signal>
+    where
+        TState: crate::State,
+    {
+        tokio::select! {
+            // Biased ensures that we poll the ports in the order they appear, giving
+            // priority to our message reception operations. See:
+            // https://docs.rs/tokio/latest/tokio/macro.select.html#fairness
+            // for more information
+            biased;
+
+            // supervision or message processing work
+            // can be interrupted by the signal port receiving
+            // a kill signal
+            signal = self.signal_rx.recv() => {
+                Err(signal.unwrap_or(Signal::Kill))
+            }
+            new_state = future => {
+                Ok(new_state)
+            }
+        }
+    }
+
+    /// List to the input ports in priority. The priority of listening for messages is
+    /// 1. Signal port
+    /// 2. Stop port
+    /// 3. Supervision message port
+    /// 4. General message port
+    ///
+    /// Returns [Ok(ActorPortMessage)] on a successful message reception, [MessagingErr]
+    /// in the event any of the channels is closed.
+    pub async fn listen_in_priority(&mut self) -> Result<ActorPortMessage, MessagingErr> {
+        tokio::select! {
+            // Biased ensures that we poll the ports in the order they appear, giving
+            // priority to our message reception operations. See:
+            // https://docs.rs/tokio/latest/tokio/macro.select.html#fairness
+            // for more information
+            biased;
+
+            signal = self.signal_rx.recv() => {
+                signal.map(ActorPortMessage::Signal).ok_or(MessagingErr::ChannelClosed)
+            }
+            stop = self.stop_rx.recv() => {
+                stop.map(ActorPortMessage::Stop).ok_or(MessagingErr::ChannelClosed)
+            }
+            supervision = self.supervisor_rx.recv() => {
+                supervision.map(ActorPortMessage::Supervision).ok_or(MessagingErr::ChannelClosed)
+            }
+            message = self.message_rx.recv() => {
+                message.map(ActorPortMessage::Message).ok_or(MessagingErr::ChannelClosed)
+            }
+        }
+    }
+}
+
 /// The inner-properties of an Actor
 struct ActorProperties {
     id: ActorId,
-    name: Option<String>,
+    name: Option<ActorName>,
     status: Arc<AtomicU8>,
     signal: BoundedInputPort<Signal>,
     stop: BoundedInputPort<StopMessage>,
@@ -65,9 +143,10 @@ struct ActorProperties {
     message: InputPort<BoxedMessage>,
     tree: SupervisionTree,
 }
+
 impl ActorProperties {
     pub fn new(
-        name: Option<String>,
+        name: Option<ActorName>,
     ) -> (
         Self,
         BoundedInputPortReceiver<Signal>,
@@ -147,25 +226,44 @@ impl std::fmt::Debug for ActorCell {
     }
 }
 
+impl Drop for ActorCell {
+    fn drop(&mut self) {
+        if let (Some(name), pid) = (self.get_name(), self.get_id()) {
+            let count = Arc::strong_count(&self.inner);
+            if count <= 2 && crate::registry::is_enrolled(name, pid) {
+                // there's 2 references left
+                // 1. This reference which is being dropped and
+                // 2. The reference in the registry so drop it from there which will complete the teardown of this actor
+                crate::registry::unenroll(name);
+            }
+        }
+    }
+}
+
 impl ActorCell {
     /// Construct a new [ActorCell] pointing to an [super::Actor] and return the message reception channels as a [ActorPortSet]
     ///
     /// * `name` - Optional name for the actor
     ///
     /// Returns a tuple [(ActorCell, ActorPortSet)] to bootstrap the [Actor]
-    pub(crate) fn new(name: Option<String>) -> (Self, ActorPortSet) {
+    pub(crate) fn new(name: Option<ActorName>) -> Result<(Self, ActorPortSet), SpawnErr> {
         let (props, rx1, rx2, rx3, rx4) = ActorProperties::new(name);
-        (
-            Self {
-                inner: Arc::new(props),
-            },
+        let cell = Self {
+            inner: Arc::new(props),
+        };
+        if let Some(r_name) = name {
+            crate::registry::enroll(r_name, cell.clone())?;
+        }
+
+        Ok((
+            cell,
             ActorPortSet {
                 signal_rx: rx1,
                 stop_rx: rx2,
                 supervisor_rx: rx3,
                 message_rx: rx4,
             },
-        )
+        ))
     }
 
     /// Retrieve the [super::Actor]'s unique identifier [ActorId]
@@ -174,8 +272,8 @@ impl ActorCell {
     }
 
     /// Retrieve the [super::Actor]'s name
-    pub fn get_name(&self) -> Option<String> {
-        self.inner.name.clone()
+    pub fn get_name(&self) -> Option<ActorName> {
+        self.inner.name
     }
 
     /// Retrieve the current status of an [super::Actor]
diff --git a/ractor/src/actor/errors.rs b/ractor/src/actor/errors.rs
index d365265..edf4be2 100644
--- a/ractor/src/actor/errors.rs
+++ b/ractor/src/actor/errors.rs
@@ -7,8 +7,10 @@
 
 use std::fmt::Display;
 
+use crate::ActorName;
+
 /// Spawn errors starting an actor
-#[derive(Debug)]
+#[derive(Debug, Eq, PartialEq)]
 pub enum SpawnErr {
     /// Actor panic'd during startup
     StartupPanic(String),
@@ -16,6 +18,8 @@ pub enum SpawnErr {
     StartupCancelled,
     /// An actor cannot be started > 1 time
     ActorAlreadyStarted,
+    /// The named actor is already registered in the registry
+    ActorAlreadyRegistered(ActorName),
 }
 
 impl Display for SpawnErr {
@@ -33,6 +37,23 @@ impl Display for SpawnErr {
             Self::ActorAlreadyStarted => {
                 write!(f, "Actor cannot be re-started more than once")
             }
+            Self::ActorAlreadyRegistered(actor_name) => {
+                write!(
+                    f,
+                    "Actor '{}' is already registered in the actor registry",
+                    actor_name
+                )
+            }
+        }
+    }
+}
+
+impl From<crate::registry::ActorRegistryErr> for SpawnErr {
+    fn from(value: crate::registry::ActorRegistryErr) -> Self {
+        match value {
+            crate::registry::ActorRegistryErr::AlreadyRegistered(actor_name) => {
+                SpawnErr::ActorAlreadyRegistered(actor_name)
+            }
         }
     }
 }
diff --git a/ractor/src/actor/messages.rs b/ractor/src/actor/messages.rs
index 1e5ed23..e283223 100644
--- a/ractor/src/actor/messages.rs
+++ b/ractor/src/actor/messages.rs
@@ -3,10 +3,15 @@
 // This source code is licensed under both the MIT license found in the
 // LICENSE-MIT file in the root directory of this source tree.
 
-//! Messages which are built-in for the actor
+//! Messages which are built-in for `ractor`'s processing routines
+//!
+//! Additionally contains definitions for [BoxedMessage] and [BoxedState]
+//! which are used to handle strongly-typed messages and states in a
+//! generic way without having to know the strong type in the underlying framework
 
 use std::any::Any;
 use std::fmt::Debug;
+use std::sync::Arc;
 
 use crate::{Message, State};
 
@@ -54,33 +59,43 @@ impl BoxedMessage {
 
 /// A "boxed" state denoting a strong-type state
 /// but generic so it can be passed around without type
-/// constraints
-#[derive(Debug)]
+/// constraints.
+///
+/// It is a shared [Arc] to the last [crate::ActorHandler::State]
+/// that the actor reported. This means, that the state shouldn't
+/// be mutated once the actor is dead, it should be generally
+/// immutable since a shared [Arc] will be sent to every supervisor
 pub struct BoxedState {
     /// The state value
-    pub state: Box<dyn Any + Send>,
+    pub state: Box<dyn Any + Send + Sync + 'static>,
+}
+
+impl Debug for BoxedState {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "BoxedState")
+    }
 }
 
 impl BoxedState {
     /// Create a new [BoxedState] from a strongly-typed state
     pub fn new<T>(msg: T) -> Self
     where
-        T: Any + State,
+        T: State,
     {
         Self {
-            state: Box::new(msg),
+            state: Box::new(Arc::new(msg)),
         }
     }
 
     /// Try and take the resulting type via cloning, such that we don't
     /// consume the value
-    pub fn take<T>(&self) -> Result<T, BoxedDowncastErr>
+    pub fn take<T>(&self) -> Result<Arc<T>, BoxedDowncastErr>
     where
-        T: Any + State,
+        T: State,
     {
         let state_ref = &self.state;
-        if state_ref.is::<T>() {
-            Ok(state_ref.downcast_ref::<T>().cloned().unwrap())
+        if state_ref.is::<Arc<T>>() {
+            Ok(state_ref.downcast_ref::<Arc<T>>().cloned().unwrap())
         } else {
             Err(BoxedDowncastErr)
         }
@@ -101,19 +116,64 @@ pub(crate) enum StopMessage {
     Reason(String),
 }
 
+impl Debug for StopMessage {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "Stop message: {}", self)
+    }
+}
+
+impl std::fmt::Display for StopMessage {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Stop => write!(f, "Stop"),
+            Self::Reason(reason) => write!(f, "Stop (reason = {})", reason),
+        }
+    }
+}
+
 /// A supervision event from the supervision tree
-#[derive(Debug)]
 pub enum SupervisionEvent {
     /// An actor was started
     ActorStarted(super::actor_cell::ActorCell),
     /// An actor terminated. In the event it shutdown cleanly (i.e. didn't panic or get
     /// signaled) we capture the last state of the actor which can be used to re-build an actor
-    /// should the need arise
-    ActorTerminated(super::actor_cell::ActorCell, Option<BoxedState>),
+    /// should the need arise. Includes an optional "exit reason" if it could be captured
+    /// and was provided
+    ActorTerminated(
+        super::actor_cell::ActorCell,
+        Option<BoxedState>,
+        Option<String>,
+    ),
     /// An actor panicked
     ActorPanicked(super::actor_cell::ActorCell, String),
 }
 
+impl Debug for SupervisionEvent {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "Supervision event: {}", self)
+    }
+}
+
+impl std::fmt::Display for SupervisionEvent {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            SupervisionEvent::ActorStarted(actor) => {
+                write!(f, "Started actor {:?}", actor)
+            }
+            SupervisionEvent::ActorTerminated(actor, _, reason) => {
+                if let Some(r) = reason {
+                    write!(f, "Stopped actor {:?} (reason = {})", actor, r)
+                } else {
+                    write!(f, "Stopped actor {:?}", actor)
+                }
+            }
+            SupervisionEvent::ActorPanicked(actor, panic_msg) => {
+                write!(f, "Actor panicked {:?} - {}", actor, panic_msg)
+            }
+        }
+    }
+}
+
 impl SupervisionEvent {
     /// A custom "clone" for [SupervisionEvent]s, because they hold a handle to a [BoxedState]
     /// which can't directly implement the clone trait, therefore the [SupervisionEvent]
@@ -125,13 +185,17 @@ impl SupervisionEvent {
     {
         match self {
             Self::ActorStarted(actor) => Ok(Self::ActorStarted(actor.clone())),
-            Self::ActorTerminated(actor, maybe_state) => {
+            Self::ActorTerminated(actor, maybe_state, maybe_exit_reason) => {
                 let cloned_maybe_state = match maybe_state {
                     Some(state) => Some(state.take::<TState>()?),
                     _ => None,
                 }
                 .map(BoxedState::new);
-                Ok(Self::ActorTerminated(actor.clone(), cloned_maybe_state))
+                Ok(Self::ActorTerminated(
+                    actor.clone(),
+                    cloned_maybe_state,
+                    maybe_exit_reason.clone(),
+                ))
             }
             Self::ActorPanicked(actor, message) => {
                 Ok(Self::ActorPanicked(actor.clone(), message.clone()))
@@ -141,8 +205,24 @@ impl SupervisionEvent {
 }
 
 /// A signal message which takes priority above all else
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 pub enum Signal {
     /// Terminate the agent, cancelling all async work immediately
     Kill,
 }
+
+impl Debug for Signal {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "Signal: {}", self)
+    }
+}
+
+impl std::fmt::Display for Signal {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Kill => {
+                write!(f, "killed")
+            }
+        }
+    }
+}
diff --git a/ractor/src/actor/mod.rs b/ractor/src/actor/mod.rs
index 4f0cf82..d6cad37 100644
--- a/ractor/src/actor/mod.rs
+++ b/ractor/src/actor/mod.rs
@@ -23,7 +23,7 @@ pub mod supervision;
 #[cfg(test)]
 mod tests;
 
-use crate::{Message, State};
+use crate::{ActorName, Message, State};
 use actor_cell::{ActorCell, ActorPortSet, ActorStatus};
 use errors::{ActorErr, MessagingErr, SpawnErr};
 
@@ -162,7 +162,7 @@ pub trait ActorHandler: Sized + Sync + Send + 'static {
 pub struct Actor<TMsg, TState, THandler>
 where
     TMsg: Message,
-    TState: State + Clone,
+    TState: State,
     THandler: ActorHandler<Msg = TMsg, State = TState>,
 {
     base: ActorCell,
@@ -172,7 +172,7 @@ where
 impl<TMsg, TState, THandler> Actor<TMsg, TState, THandler>
 where
     TMsg: Message,
-    TState: State + Clone,
+    TState: State,
     THandler: ActorHandler<Msg = TMsg, State = TState>,
 {
     /// Spawn an actor, which is unsupervised, automatically starting the actor
@@ -184,10 +184,10 @@ where
     /// along with the join handle which will complete when the actor terminates. Returns [Err(SpawnErr)] if
     /// the actor failed to start
     pub async fn spawn(
-        name: Option<String>,
+        name: Option<ActorName>,
         handler: THandler,
     ) -> Result<(ActorCell, JoinHandle<()>), SpawnErr> {
-        let (actor, ports) = Self::new(name, handler);
+        let (actor, ports) = Self::new(name, handler)?;
         actor.start(ports, None).await
     }
 
@@ -201,11 +201,11 @@ where
     /// along with the join handle which will complete when the actor terminates. Returns [Err(SpawnErr)] if
     /// the actor failed to start
     pub async fn spawn_linked(
-        name: Option<String>,
+        name: Option<ActorName>,
         handler: THandler,
         supervisor: ActorCell,
     ) -> Result<(ActorCell, JoinHandle<()>), SpawnErr> {
-        let (actor, ports) = Self::new(name, handler);
+        let (actor, ports) = Self::new(name, handler)?;
         actor.start(ports, Some(supervisor)).await
     }
 
@@ -215,15 +215,15 @@ where
     /// * `handler` The [ActorHandler] defining the logic for this actor
     ///
     /// Returns A tuple [(Actor, ActorPortSet)] to be passed to the `start` function of [Actor]
-    fn new(name: Option<String>, handler: THandler) -> (Self, ActorPortSet) {
-        let (actor_cell, ports) = ActorCell::new(name);
-        (
+    fn new(name: Option<ActorName>, handler: THandler) -> Result<(Self, ActorPortSet), SpawnErr> {
+        let (actor_cell, ports) = ActorCell::new(name)?;
+        Ok((
             Self {
                 base: actor_cell,
                 handler: Arc::new(handler),
             },
             ports,
-        )
+        ))
     }
 
     /// Start the actor immediately, optionally linking to a parent actor (supervision tree)
@@ -266,14 +266,17 @@ where
                 match Self::processing_loop(ports, state, self.handler.clone(), self.base.clone())
                     .await
                 {
-                    Ok(last_state) => SupervisionEvent::ActorTerminated(
+                    Ok((last_state, exit_reason)) => SupervisionEvent::ActorTerminated(
                         myself.clone(),
                         Some(BoxedState::new(last_state)),
+                        exit_reason,
                     ),
                     Err(actor_err) => match actor_err {
-                        ActorErr::Cancelled => {
-                            SupervisionEvent::ActorTerminated(myself.clone(), None)
-                        }
+                        ActorErr::Cancelled => SupervisionEvent::ActorTerminated(
+                            myself.clone(),
+                            None,
+                            Some("killed".to_string()),
+                        ),
                         ActorErr::Panic(msg) => {
                             SupervisionEvent::ActorPanicked(myself.clone(), msg)
                         }
@@ -301,9 +304,9 @@ where
         state: TState,
         handler: Arc<THandler>,
         myself: ActorCell,
-    ) -> Result<TState, ActorErr> {
+    ) -> Result<(TState, Option<String>), ActorErr> {
         // perform the post-start, with supervision enabled
-        let mut state = Self::do_post_start(myself.clone(), handler.clone(), state).await?;
+        let state = Self::do_post_start(myself.clone(), handler.clone(), state).await?;
 
         myself.set_status(ActorStatus::Running);
         myself.notify_supervisors::<THandler>(SupervisionEvent::ActorStarted(myself.clone()));
@@ -312,34 +315,42 @@ where
         let myself_clone = myself.clone();
         let handler_clone = handler.clone();
 
-        let last_state = Arc::new(tokio::sync::RwLock::new(state.clone()));
+        // let last_state = Arc::new(tokio::sync::RwLock::new(state.clone()));
 
-        let last_state_clone = last_state.clone();
-        let handle = tokio::spawn(async move {
+        // let last_state_clone = last_state.clone();
+        let handle: JoinHandle<(TState, Option<String>)> = tokio::spawn(async move {
             let mut ports = ports;
-            while let (n_state, false) = Self::process_message(
-                myself_clone.clone(),
-                state,
-                handler_clone.clone(),
-                &mut ports,
-            )
-            .await
-            {
-                state = n_state;
-                *(last_state_clone.write().await) = state.clone();
+            let mut p_state = state;
+            // the message processing loop. If we get and exit flag, try and capture the exit reason if there
+            // is one
+            loop {
+                let (n_state, should_exit, maybe_exit_reason) = Self::process_message(
+                    myself_clone.clone(),
+                    p_state,
+                    handler_clone.clone(),
+                    &mut ports,
+                )
+                .await;
+                p_state = n_state;
+                // processing loop exit
+                if should_exit {
+                    return (p_state, maybe_exit_reason);
+                }
             }
         });
 
-        match handle_panickable(handle).await {
-            PanickableResult::Ok(_) => (),
-            PanickableResult::Cancelled => return Err(ActorErr::Cancelled),
-            PanickableResult::Panic(msg) => return Err(ActorErr::Panic(msg)),
-        }
+        let loop_done = match handle_panickable(handle).await {
+            PanickableResult::Ok(r) => Ok(r),
+            PanickableResult::Cancelled => Err(ActorErr::Cancelled),
+            PanickableResult::Panic(msg) => Err(ActorErr::Panic(msg)),
+        };
 
         myself.set_status(ActorStatus::Stopping);
 
-        let deref_state = last_state.read().await;
-        Self::do_post_stop(myself, handler, deref_state.clone()).await
+        let (last_state, exit_reason) = loop_done?;
+
+        let last_state = Self::do_post_stop(myself, handler, last_state).await?;
+        Ok((last_state, exit_reason))
     }
 
     /// Process a message, returning the "new" state (if changed)
@@ -357,55 +368,59 @@ where
         state: TState,
         handler: Arc<THandler>,
         ports: &mut ActorPortSet,
-    ) -> (TState, bool) {
-        tokio::select! {
-            signal = ports.signal_rx.recv() => {
-                Self::handle_signal(myself, signal.unwrap_or(Signal::Kill)).await;
-                (state, true)
-            },
-            _stop = ports.stop_rx.recv() => {
-                // TODO: pass the "stop reason" somewhere
-                (state, true)
-            },
-            supervision = ports.supervisor_rx.recv() => {
-                if let Ok(signal) = ports.signal_rx.try_recv() {
-                    Self::handle_signal(myself, signal).await;
-                    (state, true)
-                } else if let Ok(_stop) = ports.stop_rx.try_recv() {
-                    // TODO: pass the stop reason up
-                    (state, true)
-                } else {
-                    match supervision {
-                        None => (state, true),
-                        Some(s) => Self::handle_supervision_message(myself.clone(), ports, state, handler.clone(), s).await
+    ) -> (TState, bool, Option<String>) {
+        match ports.listen_in_priority().await {
+            Ok(actor_port_message) => match actor_port_message {
+                actor_cell::ActorPortMessage::Signal(signal) => {
+                    (state, true, Self::handle_signal(myself, signal).await)
+                }
+                actor_cell::ActorPortMessage::Stop(stop_message) => {
+                    let exit_reason = match stop_message {
+                        StopMessage::Stop => None,
+                        StopMessage::Reason(reason) => Some(reason),
+                    };
+                    (state, true, exit_reason)
+                }
+                actor_cell::ActorPortMessage::Supervision(supervision) => {
+                    let new_state_future = Self::handle_supervision_message(
+                        myself.clone(),
+                        &state,
+                        handler.clone(),
+                        supervision,
+                    );
+                    let new_state = ports.run_with_signal(new_state_future).await;
+                    match new_state {
+                        Ok(s) => (s.unwrap_or(state), false, None),
+                        Err(signal) => (state, true, Self::handle_signal(myself, signal).await),
                     }
                 }
-            }
-            message = ports.message_rx.recv() => {
-                // double check signal and stop to make sure they have priority
-                if let Ok(signal) = ports.signal_rx.try_recv() {
-                    Self::handle_signal(myself, signal).await;
-                    (state, true)
-                } else if let Ok(_stop) = ports.stop_rx.try_recv() {
-                    // TODO: pass the stop reason up
-                    (state, true)
-                } else {
-                    match message {
-                        None => (state, true),
-                        Some(m) => Self::handle_message(myself, ports, state, handler, m).await,
+                actor_cell::ActorPortMessage::Message(msg) => {
+                    let new_state_future =
+                        Self::handle_message(myself.clone(), &state, handler, msg);
+                    let new_state = ports.run_with_signal(new_state_future).await;
+                    match new_state {
+                        Ok(s) => (s.unwrap_or(state), false, None),
+                        Err(signal) => (state, true, Self::handle_signal(myself, signal).await),
                     }
                 }
+            },
+            Err(MessagingErr::ChannelClosed) => {
+                // one of the channels is closed, this means
+                // the receiver was dropped and in this case
+                // we should always die. Therefore we simply return
+                // the state and the flag to terminate
+                (state, true, None)
             }
         }
     }
 
     async fn handle_message(
         myself: ActorCell,
-        ports: &mut ActorPortSet,
-        state: TState,
+        state: &TState,
         handler: Arc<THandler>,
         mut msg: BoxedMessage,
-    ) -> (TState, bool) {
+    ) -> Option<TState> {
+        // panic in order to kill the actor
         let typed_msg = match msg.take() {
             Ok(m) => m,
             Err(_) => {
@@ -415,44 +430,26 @@ where
                 )
             }
         };
-        // NOTE: We listen for the signal port again during the processing of async work in order
-        // to "cancel" any pending work should a signal be received immediately
-        tokio::select! {
-            signal = ports.signal_rx.recv() => {
-                Self::handle_signal(myself, signal.unwrap_or(Signal::Kill)).await;
-                (state, true)
-            }
-            maybe_new_state = handler.handle(myself.clone(), typed_msg, &state) => {
-                (maybe_new_state.unwrap_or(state), false)
-            }
-        }
+
+        handler.handle(myself, typed_msg, state).await
     }
 
-    async fn handle_signal(myself: ActorCell, signal: Signal) {
+    async fn handle_signal(myself: ActorCell, signal: Signal) -> Option<String> {
         match &signal {
             Signal::Kill => {
                 myself.terminate();
             }
         }
+        Some(signal.to_string())
     }
 
     async fn handle_supervision_message(
         myself: ActorCell,
-        ports: &mut ActorPortSet,
-        state: TState,
+        state: &TState,
         handler: Arc<THandler>,
         message: SupervisionEvent,
-    ) -> (TState, bool) {
-        let me_clone = myself.clone();
-        tokio::select! {
-            signal = ports.signal_rx.recv() => {
-                Self::handle_signal(me_clone, signal.unwrap_or(Signal::Kill)).await;
-                (state, true)
-            }
-            maybe_new_state = handler.handle_supervisor_evt(myself, message, &state) => {
-                (maybe_new_state.unwrap_or(state), false)
-            }
-        }
+    ) -> Option<TState> {
+        handler.handle_supervisor_evt(myself, message, state).await
     }
 
     async fn do_pre_start(myself: ActorCell, handler: Arc<THandler>) -> Result<TState, SpawnErr> {
@@ -476,16 +473,12 @@ where
         handler: Arc<THandler>,
         state: TState,
     ) -> Result<TState, ActorErr> {
-        // TODO: Not a fan of cloning here :/ but the async move moves ownership of "state" into
-        // the async block
-        let original_state = state.clone();
         let post_start_result = handle_panickable(tokio::spawn(async move {
-            handler.post_start(myself, &state).await
+            handler.post_start(myself, &state).await.unwrap_or(state)
         }))
         .await;
         match post_start_result {
-            PanickableResult::Ok(Some(new_state)) => Ok(new_state),
-            PanickableResult::Ok(None) => Ok(original_state),
+            PanickableResult::Ok(new_state) => Ok(new_state),
             PanickableResult::Cancelled => Err(ActorErr::Cancelled),
             PanickableResult::Panic(panic_information) => Err(ActorErr::Panic(format!(
                 "Actor panicked in post_start with '{}'",
diff --git a/ractor/src/actor/supervision.rs b/ractor/src/actor/supervision.rs
index 8ab1183..ddc36d4 100644
--- a/ractor/src/actor/supervision.rs
+++ b/ractor/src/actor/supervision.rs
@@ -10,6 +10,10 @@
 //! when a child actor starts, stops, or panics (when possible). The supervisor can then decide
 //! how to handle the event. Should it restart the actor, leave it dead, potentially die itself
 //! notifying the supervisor's supervisor? That's up to the implementation of the [super::ActorHandler]
+//!
+//! This is currently an initial implementation of [Erlang supervisors](https://www.erlang.org/doc/man/supervisor.html)
+//! which will be expanded upon as the library develops. Next in line is likely supervision strategies
+//! for automatic restart routines.
 
 use std::sync::Arc;
 
diff --git a/ractor/src/lib.rs b/ractor/src/lib.rs
index a869c45..83d0c6c 100644
--- a/ractor/src/lib.rs
+++ b/ractor/src/lib.rs
@@ -136,6 +136,7 @@
 #![warn(unused_crate_dependencies)]
 #![cfg_attr(docsrs, feature(doc_cfg))]
 
+use std::any::Any;
 use std::sync::atomic::AtomicU64;
 
 /// An actor's globally unique identifier
@@ -143,8 +144,12 @@ pub type ActorId = u64;
 /// The global id allocator for actors
 pub static ACTOR_ID_ALLOCATOR: AtomicU64 = AtomicU64::new(0u64);
 
+/// An actor's `atom()` similar name
+pub type ActorName = &'static str;
+
 pub mod actor;
 pub mod port;
+pub mod registry;
 pub mod rpc;
 pub mod time;
 
@@ -155,13 +160,27 @@ pub use actor::messages::{Signal, SupervisionEvent};
 pub use actor::{Actor, ActorHandler};
 pub use port::{InputPort, RpcReplyPort};
 
-/// Message type for an actor
+/// Message type for an actor. Generally an enum
+/// which muxes the various types of inner-messages the actor
+/// supports
+///
+/// ## Example
+///
+/// ```rust
+/// pub enum MyMessage {
+///     /// Record the name to the actor state
+///     RecordName(String),
+///     /// Print the recorded name from the state to command line
+///     PrintName,
+/// }
+/// ```
 pub trait Message: Send + 'static {}
 impl<T: Send + 'static> Message for T {}
 
-/// Represents the state of an actor
-pub trait State: Clone + Sync + Send + 'static {}
-impl<T: Clone + Sync + Send + 'static> State for T {}
+/// Represents the state of an actor. Must be safe
+/// to send between threads
+pub trait State: Any + Sync + Send + 'static {}
+impl<T: Any + Sync + Send + 'static> State for T {}
 
 /// Error types which can result from Ractor processes
 #[derive(Debug)]
diff --git a/ractor/src/port/mod.rs b/ractor/src/port/mod.rs
index 156e327..aa52723 100644
--- a/ractor/src/port/mod.rs
+++ b/ractor/src/port/mod.rs
@@ -4,6 +4,11 @@
 // LICENSE-MIT file in the root directory of this source tree.
 
 //! Port implementations for signaling and reception of messages in the Ractor environment
+//!
+//! Most of the ports we utilize are direct aliases of [tokio]'s channels
+//! (in the `sync` feature of the crate), however there are some helpful wrappers
+//! and utilities to make working with mailbox processing in `ractor` easier in
+//! the actor framework.
 
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
diff --git a/ractor/src/registry/mod.rs b/ractor/src/registry/mod.rs
new file mode 100644
index 0000000..02c21d3
--- /dev/null
+++ b/ractor/src/registry/mod.rs
@@ -0,0 +1,101 @@
+// Copyright (c) Sean Lawlor
+//
+// This source code is licensed under both the MIT license found in the
+// LICENSE-MIT file in the root directory of this source tree.
+
+//! Represents an actor registry.
+//!
+//! It allows unique naming of actors via `'static &str` (not strings)
+//! so it works more like a Erlang `atom()`
+//!
+//! Actors are automatically registered into the global registry, if they
+//! provide a name, upon construction.Actors are also
+//! automatically unenrolled from the registry upon being dropped, therefore freeing
+//! the name for subsequent registration.
+//!
+//! You can then retrieve actors by name with [try_get]. Note: this
+//! function only returns the [ActorCell] reference to the actor, it
+//! additionally requires knowledge of the [crate::ActorHandler] in order
+//! to send messages to it (since you need to know the message type)
+//! or agents will runtime panic on message reception, and supervision
+//! processes would need to restart the actors.
+//!
+//! ## Example
+//!
+//! ```rust
+//! let maybe_actor = ractor::registry::try_get("my_actor");
+//! if let Some(actor) = maybe_actor {
+//!     // send a message, or interact with the actor
+//!     // but you'll need to know the actor's strong type
+//! }
+//! ```
+
+use std::sync::Arc;
+
+use dashmap::{mapref::entry::Entry, DashMap};
+use once_cell::sync::OnceCell;
+
+use crate::{ActorCell, ActorId, ActorName};
+
+#[cfg(test)]
+mod tests;
+
+/// Errors involving the [crate::registry]'s actor registry
+pub enum ActorRegistryErr {
+    /// Actor already registered
+    AlreadyRegistered(ActorName),
+}
+
+/// The name'd actor registry
+static ACTOR_REGISTRY: OnceCell<Arc<DashMap<ActorName, ActorCell>>> = OnceCell::new();
+
+/// Retrieve the named actor registry handle
+fn get_actor_registry() -> Arc<DashMap<ActorName, ActorCell>> {
+    let reg = ACTOR_REGISTRY.get_or_init(|| Arc::new(DashMap::new()));
+    reg.clone()
+}
+
+/// Put an actor into the registry
+pub(crate) fn enroll(name: ActorName, actor: ActorCell) -> Result<(), ActorRegistryErr> {
+    let reg = get_actor_registry();
+    let entry = reg.entry(name);
+
+    match entry {
+        Entry::Occupied(_) => Err(ActorRegistryErr::AlreadyRegistered(name)),
+        Entry::Vacant(vacancy) => {
+            vacancy.insert(actor);
+            Ok(())
+        }
+    }
+}
+
+/// Check if an actor name is enrolled in the registry
+///
+/// * `name` - The actor's name
+/// * `pid` - the ID of the actor in question
+pub(crate) fn is_enrolled(name: ActorName, pid: ActorId) -> bool {
+    match get_actor_registry().entry(name) {
+        Entry::Occupied(actor) => actor.get().get_id() == pid,
+        _ => false,
+    }
+}
+
+/// Remove an actor from the registry given it's actor name
+pub(crate) fn unenroll(name: ActorName) {
+    let reg = get_actor_registry();
+    let entry = reg.entry(name);
+    if let Entry::Occupied(actor) = entry {
+        let _ = actor.remove();
+    }
+}
+
+/// Try and retrieve an actor from the registry
+///
+/// * `name` - The name of the [ActorCell] to try and retrieve
+///
+/// Returns: Some(actor) on successful identification of an actor, None if
+/// actor not registered
+pub fn try_get(name: ActorName) -> Option<ActorCell> {
+    let reg = get_actor_registry();
+    reg.get(&name).map(|item| item.value().clone())
+}
diff --git a/ractor/src/rpc/call_result.rs b/ractor/src/rpc/call_result.rs
index d5a9347..5a475ca 100644
--- a/ractor/src/rpc/call_result.rs
+++ b/ractor/src/rpc/call_result.rs
@@ -3,7 +3,8 @@
 // This source code is licensed under both the MIT license found in the
 // LICENSE-MIT file in the root directory of this source tree.
 
-//! This module contains the remote procedure call [CallResult] structure and supported operations
+//! This module contains the remote procedure call's [CallResult] structure
+//! and supported operations
 
 /// The result from a [crate::rpc::call] operation
 #[derive(Debug, Eq, PartialEq)]
diff --git a/ractor/src/rpc/mod.rs b/ractor/src/rpc/mod.rs
index e1c91ff..e4c9384 100644
--- a/ractor/src/rpc/mod.rs
+++ b/ractor/src/rpc/mod.rs
@@ -3,8 +3,14 @@
 // This source code is licensed under both the MIT license found in the
 // LICENSE-MIT file in the root directory of this source tree.
 
-//! Helpers for remote-procedure calls
+//! Remote procedure calls (RPC) are helpful communication primitives to communicate with actors
+//!
+//! There are generally 2 kinds of RPCs, cast and call, and their definition comes from the
+//! standard [Erlang `gen_server`](https://www.erlang.org/doc/man/gen_server.html#cast-2).
+//! The tl;dr is that `cast` is an send without waiting on a reply while `call` is expecting
+//! a reply from the actor being communicated with.
 
+use futures::future::join_all;
 use tokio::sync::oneshot;
 use tokio::task::JoinHandle;
 use tokio::time::{self, Duration};
@@ -67,6 +73,60 @@ where
     })
 }
 
+/// Sends an asynchronous request to the specified actors, building a one-time
+/// use reply channel for each actor and awaiting the results with the
+/// specified timeout
+///
+/// * `actors` - A reference to the group of [ActorCell]s to communicate with
+/// * `msg_builder` - The [FnOnce] to construct the message
+/// * `timeout_option` - An optional [Duration] which represents the amount of
+/// time until the operation times out
+///
+/// Returns [Ok(`Vec<CallResult<TReply>>>`)] upon successful initial sending with the reply from
+/// the [crate::Actor]s, [Err(MessagingErr)] if the initial send operation failed
+pub async fn multi_call<TActor, TReply, TMsgBuilder>(
+    actors: &[ActorCell],
+    msg_builder: TMsgBuilder,
+    timeout_option: Option<Duration>,
+) -> Result<Vec<CallResult<TReply>>, MessagingErr>
+where
+    TActor: ActorHandler,
+    TMsgBuilder: Fn(RpcReplyPort<TReply>) -> TActor::Msg,
+{
+    let mut rx_ports = Vec::with_capacity(actors.len());
+    // send to all actors
+    for actor in actors {
+        let (tx, rx) = oneshot::channel();
+        actor.send_message::<TActor>(msg_builder(tx.into()))?;
+        rx_ports.push(rx);
+    }
+
+    let results: Vec<_> = async move {
+        if let Some(duration) = timeout_option {
+            join_all(rx_ports.into_iter().map(|rx| async {
+                match tokio::time::timeout(duration, rx).await {
+                    Ok(Ok(result)) => CallResult::Success(result),
+                    Ok(Err(_send_err)) => CallResult::SenderError,
+                    Err(_) => CallResult::Timeout,
+                }
+            }))
+            .await
+        } else {
+            join_all(rx_ports.into_iter().map(|rx| async {
+                match rx.await {
+                    Ok(result) => CallResult::Success(result),
+                    Err(_send_err) => CallResult::SenderError,
+                }
+            }))
+            .await
+        }
+    }
+    .await;
+
+    // wait for the replies
+    Ok(results)
+}
+
 /// Send a message asynchronously to another actor, waiting in a new task for the reply
 /// and then forwaring the reply to a followup-actor. If this [CallResult] from the first
 /// actor is not success, the forward is not sent.
diff --git a/ractor/src/time/mod.rs b/ractor/src/time/mod.rs
index bbac6a7..102b9c0 100644
--- a/ractor/src/time/mod.rs
+++ b/ractor/src/time/mod.rs
@@ -4,6 +4,15 @@
 // LICENSE-MIT file in the root directory of this source tree.
 
 //! Timers for sending messages to actors periodically
+//!
+//! The methodology of timers in `ractor` are based on [Erlang's `timer` module](https://www.erlang.org/doc/man/timer.html).
+//! We aren't supporting all timer functions, as many of them don't make sense but we
+//! support the relevant ones for `ractor`. In short
+//!
+//! 1. Send on a period
+//! 2. Send after a delay
+//! 3. Stop after a delay
+//! 4. Kill after a delay
 
 use tokio::{task::JoinHandle, time::Duration};
 


Test patch:
diff --git a/ractor/src/actor/tests/supervisor.rs b/ractor/src/actor/tests/supervisor.rs
index b382e0f..b8a19cd 100644
--- a/ractor/src/actor/tests/supervisor.rs
+++ b/ractor/src/actor/tests/supervisor.rs
@@ -319,9 +319,7 @@ async fn test_supervision_panic_in_supervisor_handle() {
 
     let flag = Arc::new(AtomicU64::new(0));
 
-    let (supervisor, supervisor_ports) = Actor::new(None, Supervisor { flag: flag.clone() });
-    let (supervisor_ref, s_handle) = supervisor
-        .start(supervisor_ports, None)
+    let (supervisor_ref, s_handle) = Actor::spawn(None, Supervisor { flag: flag.clone() })
         .await
         .expect("Supervisor panicked on startup");
 
diff --git a/ractor/src/registry/tests.rs b/ractor/src/registry/tests.rs
new file mode 100644
index 0000000..d437c95
--- /dev/null
+++ b/ractor/src/registry/tests.rs
@@ -0,0 +1,94 @@
+// Copyright (c) Sean Lawlor
+//
+// This source code is licensed under both the MIT license found in the
+// LICENSE-MIT file in the root directory of this source tree.
+
+//! Tests on the actor registry
+
+use crate::{Actor, ActorHandler, SpawnErr};
+
+#[tokio::test]
+async fn test_basic_registation() {
+    struct EmptyActor;
+
+    #[async_trait::async_trait]
+    impl ActorHandler for EmptyActor {
+        type Msg = ();
+
+        type State = ();
+
+        async fn pre_start(&self, _this_actor: crate::ActorCell) -> Self::State {}
+    }
+
+    let (actor, handle) = Actor::spawn(Some("my_actor"), EmptyActor)
+        .await
+        .expect("Actor failed to start");
+
+    assert!(crate::registry::try_get("my_actor").is_some());
+
+    actor.stop(None);
+    handle.await.expect("Failed to clean stop the actor");
+}
+
+#[tokio::test]
+async fn test_duplicate_registration() {
+    struct EmptyActor;
+
+    #[async_trait::async_trait]
+    impl ActorHandler for EmptyActor {
+        type Msg = ();
+
+        type State = ();
+
+        async fn pre_start(&self, _this_actor: crate::ActorCell) -> Self::State {}
+    }
+
+    let (actor, handle) = Actor::spawn(Some("my_second_actor"), EmptyActor)
+        .await
+        .expect("Actor failed to start");
+
+    assert!(crate::registry::try_get("my_second_actor").is_some());
+
+    let second_actor = Actor::spawn(Some("my_second_actor"), EmptyActor).await;
+    // fails to spawn the second actor due to name err
+    assert!(matches!(
+        second_actor,
+        Err(SpawnErr::ActorAlreadyRegistered(_))
+    ));
+
+    // make sure the first actor is still registered
+    assert!(crate::registry::try_get("my_second_actor").is_some());
+
+    actor.stop(None);
+    handle.await.expect("Failed to clean stop the actor");
+}
+
+#[tokio::test]
+async fn test_actor_registry_unenrollment() {
+    struct EmptyActor;
+
+    #[async_trait::async_trait]
+    impl ActorHandler for EmptyActor {
+        type Msg = ();
+
+        type State = ();
+
+        async fn pre_start(&self, _this_actor: crate::ActorCell) -> Self::State {}
+    }
+
+    let (actor, handle) = Actor::spawn(Some("unenrollment"), EmptyActor)
+        .await
+        .expect("Actor failed to start");
+
+    assert!(crate::registry::try_get("unenrollment").is_some());
+
+    // stop the actor and wait for its death
+    actor.stop(None);
+    handle.await.expect("Failed to wait for agent stop");
+
+    // drop the actor ref's
+    drop(actor);
+
+    // the actor was automatically removed
+    assert!(crate::registry::try_get("unenrollment").is_none());
+}
diff --git a/ractor/src/rpc/tests.rs b/ractor/src/rpc/tests.rs
index 9fa4d32..c96f8f5 100644
--- a/ractor/src/rpc/tests.rs
+++ b/ractor/src/rpc/tests.rs
@@ -246,3 +246,5 @@ async fn test_rpc_call_forwarding() {
     forwarder_handle.await.expect("Actor stopped with err");
     worker_handle.await.expect("Actor stopped with err");
 }
+
+// TODO: test multi_call