Repository: slawlor/ractor
Pull request #17 (2023-01-10 01:24:45)

Adding PG groups
Erlang's process groups are named groups of collections of actors. They provide easy to use reference material for actors which may have "dispatching" purposes, in that you may want to send to 1, all, a region, etc.

This change also includes making state mutable. We've removed a lot of extra tasks and migrated to using the future's crate `catch_unwind` which saves a lot of state synchronization code laying around.


Patch:
diff --git a/ractor-playground/src/ping_pong.rs b/ractor-playground/src/ping_pong.rs
index 94ed3c6..487963f 100644
--- a/ractor-playground/src/ping_pong.rs
+++ b/ractor-playground/src/ping_pong.rs
@@ -45,36 +45,24 @@ impl ActorHandler for PingPong {
         0u8
     }
 
-    async fn post_start(
-        &self,
-        _this_actor: ActorRef<Self>,
-        _state: &Self::State,
-    ) -> Option<Self::State> {
+    async fn post_start(&self, _this_actor: ActorRef<Self>, _state: &mut Self::State) {
         println!("post_start called");
-        None
     }
 
     /// Invoked after an actor has been stopped.
-    async fn post_stop(&self, _this_actor: ActorRef<Self>, _state: Self::State) -> Self::State {
+    async fn post_stop(&self, _this_actor: ActorRef<Self>, _state: &mut Self::State) {
         println!("post_stop called");
-        _state
     }
 
-    async fn handle(
-        &self,
-        myself: ActorRef<Self>,
-        message: Self::Msg,
-        state: &Self::State,
-    ) -> Option<Self::State> {
+    async fn handle(&self, myself: ActorRef<Self>, message: Self::Msg, state: &mut Self::State) {
         if *state < 10u8 {
             message.print();
             myself.send_message(message.next()).unwrap();
-            Some(*state + 1)
+            *state += 1;
         } else {
             println!();
             myself.stop(None);
             // don't send another message, rather stop the agent after 10 iterations
-            None
         }
     }
 }
diff --git a/ractor/Cargo.toml b/ractor/Cargo.toml
index a231e25..822b1d5 100644
--- a/ractor/Cargo.toml
+++ b/ractor/Cargo.toml
@@ -1,16 +1,16 @@
 [package]
 name = "ractor"
-version = "0.2.0"
-authors = ["Sean Lawlor <seanlawlor@fb.com>"]
+version = "0.3.0"
+authors = ["Sean Lawlor", "Evan Au", "Dillon George"]
 description = "A actor framework for Rust"
+documentation = "https://docs.rs/ractor"
 license = "MIT"
 edition = "2018"
 keywords = ["actor", "ractor"]
 repository = "https://github.com/slawlor/ractor"
 readme = "../README.md"
-license-file = "../LICENSE"
 homepage = "https://github.com/slawlor/ractor"
-categories = ["actor"]
+categories = ["actor", "erlang"]
 
 [dependencies]
 async-trait = "0.1"
@@ -20,4 +20,5 @@ once_cell = "1"
 tokio = { version = "1.23", features = ["rt", "time", "sync", "macros"]}
 
 [dev-dependencies]
+function_name = "0.3"
 tokio = { version = "1.23", features = ["rt", "time", "sync", "macros", "rt-multi-thread"] }
\ No newline at end of file
diff --git a/ractor/examples/counter.rs b/ractor/examples/counter.rs
index d213014..8c0045a 100644
--- a/ractor/examples/counter.rs
+++ b/ractor/examples/counter.rs
@@ -41,24 +41,18 @@ impl ActorHandler for Counter {
         CounterState { count: 0 }
     }
 
-    async fn handle(
-        &self,
-        _myself: ActorRef<Self>,
-        message: Self::Msg,
-        state: &Self::State,
-    ) -> Option<Self::State> {
+    async fn handle(&self, _myself: ActorRef<Self>, message: Self::Msg, state: &mut Self::State) {
         match message {
-            CounterMessage::Increment(how_much) => Some(CounterState {
-                count: state.count + how_much,
-            }),
-            CounterMessage::Decrement(how_much) => Some(CounterState {
-                count: state.count - how_much,
-            }),
+            CounterMessage::Increment(how_much) => {
+                state.count += how_much;
+            }
+            CounterMessage::Decrement(how_much) => {
+                state.count -= how_much;
+            }
             CounterMessage::Retrieve(reply_port) => {
                 if !reply_port.is_closed() {
                     reply_port.send(state.count).unwrap();
                 }
-                None
             }
         }
     }
diff --git a/ractor/examples/output_port.rs b/ractor/examples/output_port.rs
index b0f071b..46bc301 100644
--- a/ractor/examples/output_port.rs
+++ b/ractor/examples/output_port.rs
@@ -34,19 +34,13 @@ impl ActorHandler for Publisher {
 
     async fn pre_start(&self, _myself: ActorRef<Self>) -> Self::State {}
 
-    async fn handle(
-        &self,
-        _myself: ActorRef<Self>,
-        message: Self::Msg,
-        _state: &Self::State,
-    ) -> Option<Self::State> {
+    async fn handle(&self, _myself: ActorRef<Self>, message: Self::Msg, _state: &mut Self::State) {
         match message {
             Self::Msg::Publish(msg) => {
                 println!("Publishing {}", msg);
                 self.output.send(format!("Published: {}", msg));
             }
         }
-        None
     }
 }
 
@@ -64,12 +58,7 @@ impl ActorHandler for Subscriber {
 
     async fn pre_start(&self, _myself: ActorRef<Self>) -> Self::State {}
 
-    async fn handle(
-        &self,
-        myself: ActorRef<Self>,
-        message: Self::Msg,
-        _state: &Self::State,
-    ) -> Option<Self::State> {
+    async fn handle(&self, myself: ActorRef<Self>, message: Self::Msg, _state: &mut Self::State) {
         match message {
             Self::Msg::Published(msg) => {
                 println!(
@@ -78,7 +67,6 @@ impl ActorHandler for Subscriber {
                 );
             }
         }
-        None
     }
 }
 
diff --git a/ractor/examples/ping_pong.rs b/ractor/examples/ping_pong.rs
index 2b71a05..956471d 100644
--- a/ractor/examples/ping_pong.rs
+++ b/ractor/examples/ping_pong.rs
@@ -53,20 +53,14 @@ impl ActorHandler for PingPong {
         0u8
     }
 
-    async fn handle(
-        &self,
-        myself: ActorRef<Self>,
-        message: Self::Msg,
-        state: &Self::State,
-    ) -> Option<Self::State> {
+    async fn handle(&self, myself: ActorRef<Self>, message: Self::Msg, state: &mut Self::State) {
         if *state < 10u8 {
             message.print();
             myself.send_message(message.next()).unwrap();
-            Some(*state + 1)
+            *state += 1;
         } else {
             println!();
             myself.stop(None);
-            None
         }
     }
 }
diff --git a/ractor/examples/supervisor.rs b/ractor/examples/supervisor.rs
index 1e2d8ac..16548cd 100644
--- a/ractor/examples/supervisor.rs
+++ b/ractor/examples/supervisor.rs
@@ -81,12 +81,7 @@ impl ActorHandler for LeafActor {
     //     state
     // }
 
-    async fn handle(
-        &self,
-        _myself: ActorRef<Self>,
-        message: Self::Msg,
-        _state: &Self::State,
-    ) -> Option<Self::State> {
+    async fn handle(&self, _myself: ActorRef<Self>, message: Self::Msg, _state: &mut Self::State) {
         match message {
             Self::Msg::Boom => {
                 panic!("LeafActor: Oh crap!");
@@ -95,7 +90,6 @@ impl ActorHandler for LeafActor {
                 println!("LeafActor: No-op!");
             }
         }
-        None
     }
 
     // async fn handle_supervisor_evt(
@@ -140,12 +134,7 @@ impl ActorHandler for MidLevelActor {
     //     state
     // }
 
-    async fn handle(
-        &self,
-        _myself: ActorRef<Self>,
-        message: Self::Msg,
-        state: &Self::State,
-    ) -> Option<Self::State> {
+    async fn handle(&self, _myself: ActorRef<Self>, message: Self::Msg, state: &mut Self::State) {
         match message {
             MidLevelActorMessage::GetLeaf(reply) => {
                 if !reply.is_closed() {
@@ -155,15 +144,14 @@ impl ActorHandler for MidLevelActor {
                 }
             }
         }
-        None
     }
 
     async fn handle_supervisor_evt(
         &self,
         _myself: ActorRef<Self>,
         message: SupervisionEvent,
-        state: &Self::State,
-    ) -> Option<Self::State> {
+        state: &mut Self::State,
+    ) {
         match message {
             SupervisionEvent::ActorPanicked(dead_actor, panic_msg)
                 if dead_actor.get_id() == state.leaf_actor.get_id() =>
@@ -182,7 +170,6 @@ impl ActorHandler for MidLevelActor {
                 println!("MidLevelActor: recieved supervisor event '{}'", other);
             }
         }
-        None
     }
 }
 
@@ -220,12 +207,7 @@ impl ActorHandler for RootActor {
     //     state
     // }
 
-    async fn handle(
-        &self,
-        _myself: ActorRef<Self>,
-        message: Self::Msg,
-        state: &Self::State,
-    ) -> Option<Self::State> {
+    async fn handle(&self, _myself: ActorRef<Self>, message: Self::Msg, state: &mut Self::State) {
         match message {
             RootActorMessage::GetMidLevel(reply) => {
                 if !reply.is_closed() {
@@ -235,15 +217,14 @@ impl ActorHandler for RootActor {
                 }
             }
         }
-        None
     }
 
     async fn handle_supervisor_evt(
         &self,
         myself: ActorRef<Self>,
         message: SupervisionEvent,
-        state: &Self::State,
-    ) -> Option<Self::State> {
+        state: &mut Self::State,
+    ) {
         match message {
             SupervisionEvent::ActorPanicked(dead_actor, panic_msg)
                 if dead_actor.get_id() == state.mid_level_actor.get_id() =>
@@ -257,6 +238,5 @@ impl ActorHandler for RootActor {
                 println!("RootActor: recieved supervisor event '{}'", other);
             }
         }
-        None
     }
 }
diff --git a/ractor/src/actor/actor_cell/actor_ref.rs b/ractor/src/actor/actor_cell/actor_ref.rs
index d8185fc..3f9f902 100644
--- a/ractor/src/actor/actor_cell/actor_ref.rs
+++ b/ractor/src/actor/actor_cell/actor_ref.rs
@@ -8,7 +8,7 @@
 use std::any::TypeId;
 use std::marker::PhantomData;
 
-use crate::{ActorHandler, ActorId, ActorName, ActorStatus, MessagingErr, SupervisionEvent};
+use crate::{ActorHandler, ActorName, MessagingErr, SupervisionEvent};
 
 use super::ActorCell;
 
@@ -85,61 +85,6 @@ where
         self.inner.clone()
     }
 
-    /// Retrieve the [crate::Actor]'s unique identifier [ActorId]
-    pub fn get_id(&self) -> ActorId {
-        self.inner.get_id()
-    }
-
-    /// Retrieve the [crate::Actor]'s name
-    pub fn get_name(&self) -> Option<ActorName> {
-        self.inner.get_name()
-    }
-
-    /// Retrieve the current status of an [crate::Actor]
-    ///
-    /// Returns the [crate::Actor]'s current [ActorStatus]
-    pub fn get_status(&self) -> ActorStatus {
-        self.inner.get_status()
-    }
-
-    /// Set the status of the [crate::Actor]
-    ///
-    /// * `status` - The [ActorStatus] to set
-    pub(crate) fn set_status(&self, status: ActorStatus) {
-        self.inner.set_status(status)
-    }
-
-    /// Terminate this [crate::Actor] and all it's children
-    pub(crate) fn terminate(&self) {
-        self.inner.terminate();
-    }
-
-    /// Link this [crate::Actor] to the supervisor
-    ///
-    /// * `supervisor` - The supervisor to link this [crate::Actor] to
-    pub fn link(&self, supervisor: ActorCell) {
-        self.inner.link(supervisor);
-    }
-
-    /// Unlink this [crate::Actor] from the supervisor
-    ///
-    /// * `supervisor` - The supervisor to unlink this [crate::Actor] from
-    pub fn unlink(&self, supervisor: ActorCell) {
-        self.inner.unlink(supervisor)
-    }
-
-    /// Kill this [crate::Actor] forcefully (terminates async work)
-    pub fn kill(&self) {
-        self.inner.kill();
-    }
-
-    /// Stop this [crate::Actor] gracefully (stopping message processing)
-    ///
-    /// * `reason` - An optional static string reason why the stop is occurring
-    pub fn stop(&self, reason: Option<String>) {
-        self.inner.stop(reason);
-    }
-
     /// Send a strongly-typed message, constructing the boxed message on the fly
     ///
     /// * `message` - The message to send
@@ -149,19 +94,6 @@ where
         self.inner.send_message::<TActor>(message)
     }
 
-    /// Send a supervisor event to the supervisory port
-    ///
-    /// * `message` - The [SupervisionEvent] to send to the supervisory port
-    ///
-    /// Returns [Ok(())] on successful message send, [Err(MessagingErr)] otherwise
-    #[cfg(test)]
-    pub(crate) fn send_supervisor_evt(
-        &self,
-        message: SupervisionEvent,
-    ) -> Result<(), MessagingErr> {
-        self.inner.send_supervisor_evt(message)
-    }
-
     /// Notify the supervisors that a supervision event occurred
     ///
     /// * `evt` - The event to send to this [crate::Actor]'s supervisors
diff --git a/ractor/src/actor/actor_cell/mod.rs b/ractor/src/actor/actor_cell/mod.rs
index ecceb89..e4f3d7c 100644
--- a/ractor/src/actor/actor_cell/mod.rs
+++ b/ractor/src/actor/actor_cell/mod.rs
@@ -148,26 +148,12 @@ 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]
+    /// Returns a tuple [(ActorCell, ActorPortSet)] to bootstrap the [crate::Actor]
     pub(crate) fn new<TActor>(name: Option<ActorName>) -> Result<(Self, ActorPortSet), SpawnErr>
     where
         TActor: ActorHandler,
@@ -208,10 +194,24 @@ impl ActorCell {
         self.inner.get_status()
     }
 
-    /// Set the status of the [super::Actor]
+    /// Set the status of the [super::Actor]. If the status is set to
+    /// [ActorStatus::Stopping] or [ActorStatus::Stopped] the actor
+    /// will also be unenrolled from both the named registry ([crate::registry])
+    /// and the PG groups ([crate::pg]) if it's enrolled in any
     ///
     /// * `status` - The [ActorStatus] to set
     pub(crate) fn set_status(&self, status: ActorStatus) {
+        // The actor is shut down
+        if status == ActorStatus::Stopped || status == ActorStatus::Stopping {
+            // If it's enrolled in the registry, remove it
+            if let Some(name) = self.get_name() {
+                crate::registry::unenroll(name);
+            }
+            // Leave all + stop monitoring pg groups (if any)
+            crate::pg::demonitor_all(self.get_id());
+            crate::pg::leave_all(self.get_id());
+        }
+
         self.inner.set_status(status)
     }
 
@@ -296,9 +296,15 @@ impl ActorCell {
         self.inner.tree.notify_supervisors::<TActor>(evt)
     }
 
-    /// Test utility to retrieve a clone of the underlying supervision tree
+    // ================== Test Utilities ================== //
+
+    #[cfg(test)]
+    pub(crate) fn get_num_children(&self) -> usize {
+        self.inner.tree.get_num_children()
+    }
+
     #[cfg(test)]
-    pub(crate) fn get_tree(&self) -> super::supervision::SupervisionTree {
-        self.inner.tree.clone()
+    pub(crate) fn get_num_parents(&self) -> usize {
+        self.inner.tree.get_num_parents()
     }
 }
diff --git a/ractor/src/actor/messages.rs b/ractor/src/actor/messages.rs
index e283223..0e0ef3a 100644
--- a/ractor/src/actor/messages.rs
+++ b/ractor/src/actor/messages.rs
@@ -146,6 +146,9 @@ pub enum SupervisionEvent {
     ),
     /// An actor panicked
     ActorPanicked(super::actor_cell::ActorCell, String),
+
+    /// A subscribed process group changed
+    ProcessGroupChanged(crate::pg::GroupChangeMessage),
 }
 
 impl Debug for SupervisionEvent {
@@ -170,6 +173,9 @@ impl std::fmt::Display for SupervisionEvent {
             SupervisionEvent::ActorPanicked(actor, panic_msg) => {
                 write!(f, "Actor panicked {:?} - {}", actor, panic_msg)
             }
+            SupervisionEvent::ProcessGroupChanged(change) => {
+                write!(f, "Process group {} changed", change.get_group())
+            }
         }
     }
 }
@@ -200,6 +206,7 @@ impl SupervisionEvent {
             Self::ActorPanicked(actor, message) => {
                 Ok(Self::ActorPanicked(actor.clone(), message.clone()))
             }
+            Self::ProcessGroupChanged(change) => Ok(Self::ProcessGroupChanged(change.clone())),
         }
     }
 }
diff --git a/ractor/src/actor/mod.rs b/ractor/src/actor/mod.rs
index abbff7f..bb42cd4 100644
--- a/ractor/src/actor/mod.rs
+++ b/ractor/src/actor/mod.rs
@@ -9,8 +9,9 @@
 //! [ActorHandler] : The internal logic for how an actor behaves
 //! [Actor] : Management structure processing the message handler, signals, and supervision events in a loop
 
-use std::sync::Arc;
+use std::{panic::AssertUnwindSafe, sync::Arc};
 
+use futures::TryFutureExt;
 use tokio::task::JoinHandle;
 
 pub mod messages;
@@ -37,29 +38,6 @@ pub(crate) fn get_panic_string(e: Box<dyn std::any::Any + Send>) -> String {
     }
 }
 
-enum PanickableResult<T> {
-    Ok(T),
-    Cancelled,
-    Panic(String),
-}
-
-async fn handle_panickable<TResult>(
-    handle: tokio::task::JoinHandle<TResult>,
-) -> PanickableResult<TResult> {
-    match handle.await {
-        Ok(t) => PanickableResult::Ok(t),
-        Err(maybe_panic) => {
-            if maybe_panic.is_panic() {
-                let panic_message = get_panic_string(maybe_panic.into_panic());
-                PanickableResult::Panic(panic_message)
-            } else {
-                // task cancelled, just notify of exit
-                PanickableResult::Cancelled
-            }
-        }
-    }
-}
-
 /// [ActorHandler] is the "business logic" of an Actor. It defines the
 /// Message type, State type, and all processing logic for the actor
 #[async_trait::async_trait]
@@ -76,7 +54,8 @@ pub trait ActorHandler: Sized + Sync + Send + 'static {
     /// performed here hence why it returns the initial state.
     ///
     /// Panics in `pre_start` do not invoke the
-    /// supervision strategy and the actor will be terminated.
+    /// supervision strategy and the actor won't be started. [Actor]::`spawn`
+    /// will return an error to the caller
     ///
     /// * `myself` - A handle to the [ActorCell] representing this actor
     ///
@@ -91,64 +70,48 @@ pub trait ActorHandler: Sized + Sync + Send + 'static {
     /// Panics in `post_start` follow the supervision strategy.
     ///
     /// * `myself` - A handle to the [ActorCell] representing this actor
-    /// * `state` - A reference to the internal actor's state
-    ///
-    /// Returns [Some(ActorHandler::State)] if the state should be updated with a new value, [None] otherwise
+    /// * `state` - A mutable reference to the internal actor's state
     #[allow(unused_variables)]
-    async fn post_start(&self, myself: ActorRef<Self>, state: &Self::State) -> Option<Self::State> {
-        None
-    }
+    async fn post_start(&self, myself: ActorRef<Self>, state: &mut Self::State) {}
 
-    /// Invoked after an actor has been stopped.
+    /// Invoked after an actor has been stopped to perform final cleanup. In the
+    /// event the actor is terminated with [Signal::Kill] or has self-panicked,
+    /// `post_stop` won't be called.
     ///
-    /// * `myself` - A handle to the [ActorCell] representing this actor
-    /// * `state` - A reference to the internal actor's state
+    /// Panics in `post_stop` follow the supervision strategy.
     ///
-    /// Returns: The last [ActorHandler::State] (after doing any teardown) which might be necessary to re-create the actor
+    /// * `myself` - A handle to the [ActorCell] representing this actor
+    /// * `state` - A mutable reference to the internal actor's last known state
     #[allow(unused_variables)]
-    async fn post_stop(&self, myself: ActorRef<Self>, state: Self::State) -> Self::State {
-        state
-    }
+    async fn post_stop(&self, myself: ActorRef<Self>, state: &mut Self::State) {}
 
-    /// Handle the incoming message in a basic event processing loop. Unhandled panic's will be captured and
-    /// treated as actor death in the supervision tree
+    /// Handle the incoming message from the event processing loop. Unhandled panicks will be
+    /// captured and sent to the supervisor(s)
     ///
     /// * `myself` - A handle to the [ActorCell] representing this actor
     /// * `message` - The message to process
-    /// * `state` - A reference to the internal actor's state
-    ///
-    /// Returns [Some(ActorHandler::State)] if the state should be updated with a new value, [None] otherwise
+    /// * `state` - A mutable reference to the internal actor's state
     #[allow(unused_variables)]
-    async fn handle(
-        &self,
-        myself: ActorRef<Self>,
-        message: Self::Msg,
-        state: &Self::State,
-    ) -> Option<Self::State> {
-        None
-    }
+    async fn handle(&self, myself: ActorRef<Self>, message: Self::Msg, state: &mut Self::State) {}
 
-    /// Handle the incoming supervision event. Unhandled panic's will captured and treated as actor death in
-    /// the supervision tree
+    /// Handle the incoming supervision event. Unhandled panicks will captured and
+    /// sent the the supervisor(s)
     ///
     /// * `myself` - A handle to the [ActorCell] representing this actor
     /// * `message` - The message to process
-    /// * `state` - A reference to the internal actor's state
-    ///
-    /// Returns [Some(ActorHandler::State)] if the state should be updated with a new value, [None] otherwise
+    /// * `state` - A mutable reference to the internal actor's state
     #[allow(unused_variables)]
     async fn handle_supervisor_evt(
         &self,
         myself: ActorRef<Self>,
         message: SupervisionEvent,
-        state: &Self::State,
-    ) -> Option<Self::State> {
-        None
+        state: &mut Self::State,
+    ) {
     }
 }
 
 /// [Actor] is a struct which represents the actor. This struct is consumed by the
-/// `start` operation, but results in an [ActorCell] to communicate and operate with
+/// `start` operation, but results in an [ActorRef] to communicate and operate with
 pub struct Actor<TMsg, TState, THandler>
 where
     TMsg: Message,
@@ -170,7 +133,7 @@ where
     /// * `name`: A name to give the actor. Useful for global referencing or debug printing
     /// * `handler` The [ActorHandler] defining the logic for this actor
     ///
-    /// Returns a [Ok((ActorCell, JoinHandle<()>))] upon successful start, denoting the actor reference
+    /// Returns a [Ok((ActorRef, JoinHandle<()>))] upon successful start, denoting the actor reference
     /// 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(
@@ -187,7 +150,7 @@ where
     /// * `handler` The [ActorHandler] defining the logic for this actor
     /// * `supervisor`: The [ActorCell] which is to become the supervisor (parent) of this actor
     ///
-    /// Returns a [Ok((ActorCell, JoinHandle<()>))] upon successful start, denoting the actor reference
+    /// Returns a [Ok((ActorRef, JoinHandle<()>))] upon successful start, denoting the actor reference
     /// 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(
@@ -225,7 +188,7 @@ where
     /// * `ports` - The [ActorPortSet] for this actor
     /// * `supervisor` - The optional [ActorCell] representing the supervisor of this actor
     ///
-    /// Returns a [Ok((ActorCell, JoinHandle<()>))] upon successful start, denoting the actor reference
+    /// Returns a [Ok((ActorRef, JoinHandle<()>))] upon successful start, denoting the actor reference
     /// along with the join handle which will complete when the actor terminates. Returns [Err(SpawnErr)] if
     /// the actor failed to start
     async fn start(
@@ -241,37 +204,41 @@ where
         self.base.set_status(ActorStatus::Starting);
 
         // Perform the pre-start routine, crashing immediately if we fail to start
-        let state = Self::do_pre_start(self.base.clone(), self.handler.clone()).await?;
+        let mut state = Self::do_pre_start(self.base.clone(), self.handler.clone()).await?;
 
         // setup supervision
         if let Some(sup) = &supervisor {
             sup.link(self.base.clone().into());
         }
 
-        // run the processing loop, capturing panic's
+        // run the processing loop, backgrounding the work
         let myself = self.base.clone();
         let myself_ret = self.base.clone();
         let handle = tokio::spawn(async move {
-            let evt =
-                match Self::processing_loop(ports, state, self.handler.clone(), self.base.clone())
-                    .await
-                {
-                    Ok((last_state, exit_reason)) => SupervisionEvent::ActorTerminated(
+            let evt = match Self::processing_loop(
+                ports,
+                &mut state,
+                self.handler.clone(),
+                self.base.clone(),
+            )
+            .await
+            {
+                Ok(exit_reason) => SupervisionEvent::ActorTerminated(
+                    myself.clone().into(),
+                    Some(BoxedState::new(state)),
+                    exit_reason,
+                ),
+                Err(actor_err) => match actor_err {
+                    ActorErr::Cancelled => SupervisionEvent::ActorTerminated(
                         myself.clone().into(),
-                        Some(BoxedState::new(last_state)),
-                        exit_reason,
+                        None,
+                        Some("killed".to_string()),
                     ),
-                    Err(actor_err) => match actor_err {
-                        ActorErr::Cancelled => SupervisionEvent::ActorTerminated(
-                            myself.clone().into(),
-                            None,
-                            Some("killed".to_string()),
-                        ),
-                        ActorErr::Panic(msg) => {
-                            SupervisionEvent::ActorPanicked(myself.clone().into(), msg)
-                        }
-                    },
-                };
+                    ActorErr::Panic(msg) => {
+                        SupervisionEvent::ActorPanicked(myself.clone().into(), msg)
+                    }
+                },
+            };
 
             // terminate children
             myself.terminate();
@@ -279,8 +246,10 @@ where
             // notify supervisors of the actor's death
             myself.notify_supervisors(evt);
 
+            // set status to stopped
             myself.set_status(ActorStatus::Stopped);
-            // signal received or process exited cleanly, we should already have "handled" the signal, so we can just terminate
+
+            // unlink superisors
             if let Some(sup) = supervisor {
                 sup.unlink(myself.clone().into());
             }
@@ -291,62 +260,57 @@ where
 
     async fn processing_loop(
         ports: ActorPortSet,
-        state: TState,
+        state: &mut TState,
         handler: Arc<THandler>,
         myself: ActorRef<THandler>,
-    ) -> Result<(TState, Option<String>), ActorErr> {
+    ) -> Result<Option<String>, ActorErr> {
         // perform the post-start, with supervision enabled
-        let state = Self::do_post_start(myself.clone(), handler.clone(), state).await?;
+        Self::do_post_start(myself.clone(), handler.clone(), state).await?;
 
         myself.set_status(ActorStatus::Running);
         myself.notify_supervisors(SupervisionEvent::ActorStarted(myself.clone().into()));
 
-        // let mut last_state = state.clone();
         let myself_clone = myself.clone();
         let handler_clone = handler.clone();
 
-        // let last_state = Arc::new(tokio::sync::RwLock::new(state.clone()));
-
-        // let last_state_clone = last_state.clone();
-        let handle: JoinHandle<(TState, Option<String>)> = tokio::spawn(async move {
+        let future = async move {
             let mut ports = ports;
-            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(
+                let (should_exit, maybe_exit_reason) = Self::process_message(
                     myself_clone.clone(),
-                    p_state,
+                    state,
                     handler_clone.clone(),
                     &mut ports,
                 )
                 .await;
-                p_state = n_state;
                 // processing loop exit
                 if should_exit {
-                    return (p_state, maybe_exit_reason);
+                    return (state, maybe_exit_reason);
                 }
             }
-        });
-
-        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)),
         };
+        // capture any panicks in this future and convert to an ActorErr
+        let loop_done = futures::FutureExt::catch_unwind(AssertUnwindSafe(future))
+            .map_err(|err| ActorErr::Panic(get_panic_string(err)))
+            .await;
 
+        // set status to stopping
         myself.set_status(ActorStatus::Stopping);
 
-        let (last_state, exit_reason) = loop_done?;
+        let (nstate, exit_reason) = loop_done?;
+
+        // if we didn't exit in error mode, call `post_stop`
+        Self::do_post_stop(myself, handler, nstate).await?;
 
-        let last_state = Self::do_post_stop(myself, handler, last_state).await?;
-        Ok((last_state, exit_reason))
+        Ok(exit_reason)
     }
 
     /// Process a message, returning the "new" state (if changed)
     /// along with optionally whether we were signaled mid-processing or not
     ///
-    /// * `myself` - The current [ActorCell]
+    /// * `myself` - The current [ActorRef]
     /// * `state` - The current [ActorHandler::State] object
     /// * `handler` - Pointer to the [ActorHandler] definition
     /// * `ports` - The mutable [ActorPortSet] which are the message ports for this actor
@@ -355,65 +319,65 @@ where
     /// loop is done
     async fn process_message(
         myself: ActorRef<THandler>,
-        state: TState,
+        state: &mut TState,
         handler: Arc<THandler>,
         ports: &mut ActorPortSet,
-    ) -> (TState, bool, Option<String>) {
+    ) -> (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)
+                    (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)
+                    (true, exit_reason)
                 }
                 actor_cell::ActorPortMessage::Supervision(supervision) => {
                     let new_state_future = Self::handle_supervision_message(
                         myself.clone(),
-                        &state,
+                        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),
+                        Ok(()) => (false, None),
+                        Err(signal) => (true, Self::handle_signal(myself, signal).await),
                     }
                 }
                 actor_cell::ActorPortMessage::Message(msg) => {
                     let new_state_future =
-                        Self::handle_message(myself.clone(), &state, handler, msg);
+                        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),
+                        Ok(()) => (false, None),
+                        Err(signal) => (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)
+                // we should always die. Therefore we flag
+                // to terminate
+                (true, None)
             }
             Err(MessagingErr::InvalidActorType) => {
                 // not possible. Treat like a channel closed
-                (state, true, None)
+                (true, None)
             }
         }
     }
 
     async fn handle_message(
         myself: ActorRef<THandler>,
-        state: &TState,
+        state: &mut TState,
         handler: Arc<THandler>,
         mut msg: BoxedMessage,
-    ) -> Option<TState> {
+    ) {
         // panic in order to kill the actor
         let typed_msg = match msg.take() {
             Ok(m) => m,
@@ -439,10 +403,10 @@ where
 
     async fn handle_supervision_message(
         myself: ActorRef<THandler>,
-        state: &TState,
+        state: &mut TState,
         handler: Arc<THandler>,
         message: SupervisionEvent,
-    ) -> Option<TState> {
+    ) {
         handler.handle_supervisor_evt(myself, message, state).await
     }
 
@@ -450,56 +414,46 @@ where
         myself: ActorRef<THandler>,
         handler: Arc<THandler>,
     ) -> Result<TState, SpawnErr> {
-        let start_result =
-            handle_panickable(tokio::spawn(async move { handler.pre_start(myself).await })).await;
-        match start_result {
-            PanickableResult::Ok(state) => {
-                // intitialize the state
-                Ok(state)
-            }
-            PanickableResult::Cancelled => Err(SpawnErr::StartupCancelled),
-            PanickableResult::Panic(panic_information) => Err(SpawnErr::StartupPanic(format!(
-                "Actor panicked during pre_start with '{}'",
-                panic_information
-            ))),
-        }
+        let future = handler.pre_start(myself);
+        futures::FutureExt::catch_unwind(AssertUnwindSafe(future))
+            .await
+            .map_err(|err| {
+                SpawnErr::StartupPanic(format!(
+                    "Actor panicked in pre_start with '{}'",
+                    get_panic_string(err)
+                ))
+            })
     }
 
     async fn do_post_start(
         myself: ActorRef<THandler>,
         handler: Arc<THandler>,
-        state: TState,
-    ) -> Result<TState, ActorErr> {
-        let post_start_result = handle_panickable(tokio::spawn(async move {
-            handler.post_start(myself, &state).await.unwrap_or(state)
-        }))
-        .await;
-        match post_start_result {
-            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 '{}'",
-                panic_information
-            ))),
-        }
+        state: &mut TState,
+    ) -> Result<(), ActorErr> {
+        let future = handler.post_start(myself, state);
+        futures::FutureExt::catch_unwind(AssertUnwindSafe(future))
+            .await
+            .map_err(|err| {
+                ActorErr::Panic(format!(
+                    "Actor panicked in post_start with '{}'",
+                    get_panic_string(err)
+                ))
+            })
     }
 
     async fn do_post_stop(
         myself: ActorRef<THandler>,
         handler: Arc<THandler>,
-        state: TState,
-    ) -> Result<TState, ActorErr> {
-        let post_stop_result = handle_panickable(tokio::spawn(async move {
-            handler.post_stop(myself, state).await
-        }))
-        .await;
-        match post_stop_result {
-            PanickableResult::Ok(last_state) => Ok(last_state),
-            PanickableResult::Cancelled => Err(ActorErr::Cancelled),
-            PanickableResult::Panic(panic_information) => Err(ActorErr::Panic(format!(
-                "Actor panicked in post_start with '{}'",
-                panic_information
-            ))),
-        }
+        state: &mut TState,
+    ) -> Result<(), ActorErr> {
+        let future = handler.post_stop(myself, state);
+        futures::FutureExt::catch_unwind(AssertUnwindSafe(future))
+            .await
+            .map_err(|err| {
+                ActorErr::Panic(format!(
+                    "Actor panicked in post_stop with '{}'",
+                    get_panic_string(err)
+                ))
+            })
     }
 }
diff --git a/ractor/src/actor/supervision.rs b/ractor/src/actor/supervision.rs
index ddc36d4..3665b0b 100644
--- a/ractor/src/actor/supervision.rs
+++ b/ractor/src/actor/supervision.rs
@@ -15,18 +15,16 @@
 //! which will be expanded upon as the library develops. Next in line is likely supervision strategies
 //! for automatic restart routines.
 
-use std::sync::Arc;
-
 use dashmap::DashMap;
 
 use super::{actor_cell::ActorCell, messages::SupervisionEvent};
 use crate::{ActorHandler, ActorId};
 
 /// A supervision tree
-#[derive(Clone, Default)]
+#[derive(Default)]
 pub struct SupervisionTree {
-    children: Arc<DashMap<ActorId, ActorCell>>,
-    parents: Arc<DashMap<ActorId, ActorCell>>,
+    children: DashMap<ActorId, ActorCell>,
+    parents: DashMap<ActorId, ActorCell>,
 }
 
 impl SupervisionTree {
diff --git a/ractor/src/actor_id.rs b/ractor/src/actor_id.rs
index 56ba4f8..cf386fc 100644
--- a/ractor/src/actor_id.rs
+++ b/ractor/src/actor_id.rs
@@ -18,6 +18,15 @@ pub enum ActorId {
     Remote(u64, u64),
 }
 
+impl ActorId {
+    /// Determine if this actor id is a local or remote actor
+    ///
+    /// Returns [true] if it is a local actor, [false] otherwise
+    pub fn is_local(&self) -> bool {
+        matches!(self, ActorId::Local(_))
+    }
+}
+
 impl Display for ActorId {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
diff --git a/ractor/src/lib.rs b/ractor/src/lib.rs
index 28166ff..4b3cfb9 100644
--- a/ractor/src/lib.rs
+++ b/ractor/src/lib.rs
@@ -74,16 +74,15 @@
 //!         &self,
 //!         myself: ActorRef<Self>,
 //!         message: Self::Msg,
-//!         state: &Self::State,
-//!     ) -> Option<Self::State> {
+//!         state: &mut Self::State,
+//!     ) {
 //!         if *state < 10u8 {
 //!             message.print();
 //!             myself.send_message(message.next()).unwrap();
-//!             Some(*state + 1)
+//!             *state += 1;
 //!         } else {
 //!             myself.stop(None);
 //!             // don't send another message, rather stop the agent after 10 iterations
-//!             None
 //!         }
 //!     }
 //! }
@@ -141,8 +140,12 @@ use std::any::Any;
 /// An actor's `atom()` similar name
 pub type ActorName = &'static str;
 
+/// A process group's name, equivalent to an Erlang `atom()`
+pub type GroupName = &'static str;
+
 pub mod actor;
 pub mod actor_id;
+pub mod pg;
 pub mod port;
 pub mod registry;
 pub mod rpc;
diff --git a/ractor/src/pg/mod.rs b/ractor/src/pg/mod.rs
new file mode 100644
index 0000000..9d8229f
--- /dev/null
+++ b/ractor/src/pg/mod.rs
@@ -0,0 +1,242 @@
+// 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.
+
+//! Process groups (PG)
+//!
+//! Inspired from [Erlang's `pg` module](https://www.erlang.org/doc/man/pg.html)
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use dashmap::mapref::entry::Entry::{Occupied, Vacant};
+use dashmap::DashMap;
+
+use once_cell::sync::OnceCell;
+
+use crate::{ActorCell, ActorId, GroupName, SupervisionEvent};
+
+#[cfg(test)]
+mod tests;
+
+/// Represents a change in group or scope membership
+#[derive(Clone)]
+pub enum GroupChangeMessage {
+    /// Some actors joined a group
+    Join(GroupName, Vec<ActorCell>),
+    /// Some actors left a group
+    Leave(GroupName, Vec<ActorCell>),
+}
+
+impl GroupChangeMessage {
+    /// Retrieve the group that changed
+    pub fn get_group(&self) -> GroupName {
+        match self {
+            Self::Join(name, _) => name,
+            Self::Leave(name, _) => name,
+        }
+    }
+}
+
+struct PgState {
+    map: Arc<DashMap<GroupName, HashMap<ActorId, ActorCell>>>,
+    listeners: Arc<DashMap<GroupName, Vec<ActorCell>>>,
+}
+
+static PG_MONITOR: OnceCell<PgState> = OnceCell::new();
+
+fn get_monitor<'a>() -> &'a PgState {
+    PG_MONITOR.get_or_init(|| PgState {
+        map: Arc::new(DashMap::new()),
+        listeners: Arc::new(DashMap::new()),
+    })
+}
+
+/// Join actors to the group `group`
+///
+/// * `group` - The staticly named group. Will be created if first actors to join
+/// * `actors` - Thie list of [crate::Actor]s to add to the group
+pub fn join(group: GroupName, actors: Vec<ActorCell>) {
+    let monitor = get_monitor();
+    // insert into the monitor group
+    match monitor.map.entry(group) {
+        Occupied(mut occupied) => {
+            let oref = occupied.get_mut();
+            for actor in actors.iter() {
+                oref.insert(actor.get_id(), actor.clone());
+            }
+        }
+        Vacant(vacancy) => {
+            let map = actors
+                .iter()
+                .map(|a| (a.get_id(), a.clone()))
+                .collect::<HashMap<_, _>>();
+            vacancy.insert(map);
+        }
+    }
+    // notify supervisors
+    if let Some(listeners) = monitor.listeners.get(&group) {
+        for listener in listeners.value() {
+            let _ = listener.send_supervisor_evt(SupervisionEvent::ProcessGroupChanged(
+                GroupChangeMessage::Join(group, actors.clone()),
+            ));
+        }
+    }
+}
+
+/// Leaves the specified [crate::Actor]s from the PG group
+///
+/// * `group` - The staticly named group
+/// * `actors` - Thie list of actors to remove from the group
+pub fn leave(group: GroupName, actors: Vec<ActorCell>) {
+    let monitor = get_monitor();
+    match monitor.map.entry(group) {
+        Vacant(_) => {}
+        Occupied(mut occupied) => {
+            let mut_ref = occupied.get_mut();
+            for actor in actors.iter() {
+                mut_ref.remove(&actor.get_id());
+            }
+            // the group is empty, remove it
+            if mut_ref.is_empty() {
+                occupied.remove();
+            }
+            if let Some(listeners) = monitor.listeners.get(&group) {
+                for listener in listeners.value() {
+                    let _ = listener.send_supervisor_evt(SupervisionEvent::ProcessGroupChanged(
+                        GroupChangeMessage::Leave(group, actors.clone()),
+                    ));
+                }
+            }
+        }
+    }
+}
+
+/// Leave all groups for a specific [ActorId].
+/// Used only during actor shutdown
+pub(crate) fn leave_all(actor: ActorId) {
+    let pg_monitor = get_monitor();
+    let map = pg_monitor.map.clone();
+
+    let mut empty_groups = vec![];
+    let mut removal_events = HashMap::new();
+
+    for mut kv in map.iter_mut() {
+        if let Some(actor_cell) = kv.value_mut().remove(&actor) {
+            removal_events.insert(*kv.key(), actor_cell);
+        }
+        if kv.value().is_empty() {
+            empty_groups.push(*kv.key());
+        }
+    }
+
+    // notify the listeners
+    let all_listeners = pg_monitor.listeners.clone();
+    for (group, cell) in removal_events.into_iter() {
+        if let Some(this_listeners) = all_listeners.get(&group) {
+            this_listeners.iter().for_each(|listener| {
+                let _ = listener.send_supervisor_evt(SupervisionEvent::ProcessGroupChanged(
+                    GroupChangeMessage::Leave(group, vec![cell.clone()]),
+                ));
+            });
+        }
+    }
+
+    // Cleanup empty groups
+    for group in empty_groups {
+        map.remove(group);
+    }
+}
+
+/// Returns all the actors running on the local node in the group `group`.
+///
+/// * `group_name` - Either a statically named group or scope
+pub fn get_local_members(group_name: GroupName) -> Vec<ActorCell> {
+    let monitor = get_monitor();
+    if let Some(actors) = monitor.map.get(&group_name) {
+        actors
+            .value()
+            .values()
+            .filter(|a| a.get_id().is_local())
+            .cloned()
+            .collect::<Vec<_>>()
+    } else {
+        vec![]
+    }
+}
+
+/// Returns all the actors running on any node in the group `group`.
+///
+/// * `group_name` - Either a statically named group or scope
+///
+/// Returns [Vec<_>] with the associated actors
+pub fn get_members(group_name: GroupName) -> Vec<ActorCell> {
+    let monitor = get_monitor();
+    if let Some(actors) = monitor.map.get(&group_name) {
+        actors.value().values().cloned().collect::<Vec<_>>()
+    } else {
+        vec![]
+    }
+}
+
+/// Return a list of all known groups
+///
+/// Returns a [Vec<_>] representing all the registered group names
+pub fn which_groups() -> Vec<GroupName> {
+    let monitor = get_monitor();
+    monitor
+        .map
+        .iter()
+        .map(|kvp| *(kvp.key()))
+        .collect::<Vec<_>>()
+}
+
+/// Subscribes the provided [crate::Actor] to the scope or group for updates
+///
+/// * `group_name` - The scope or group to monitor
+/// * `actor` - The [ActorCell] representing who will receive updates
+pub fn monitor(group_name: GroupName, actor: ActorCell) {
+    let monitor = get_monitor();
+    match monitor.listeners.entry(group_name) {
+        Occupied(mut occupied) => occupied.get_mut().push(actor),
+        Vacant(vacancy) => {
+            vacancy.insert(vec![actor]);
+        }
+    }
+}
+
+/// Unsubscribes the provided [crate::Actor] from the scope or group for updates
+///
+/// * `group_name` - The scope or group to monitor
+/// * `actor` - The [ActorCell] representing who will receive updates
+pub fn demonitor(group_name: GroupName, actor: ActorId) {
+    let monitor = get_monitor();
+    if let Occupied(mut entry) = monitor.listeners.entry(group_name) {
+        let mut_ref = entry.get_mut();
+        mut_ref.retain(|a| a.get_id() != actor);
+        if mut_ref.is_empty() {
+            entry.remove();
+        }
+    }
+}
+
+/// Remove the specified [ActorId] from monitoring all groups it might be.
+/// Used only during actor shutdown
+pub(crate) fn demonitor_all(actor: ActorId) {
+    let monitor = get_monitor();
+    let mut empty_groups = vec![];
+
+    for mut kvp in monitor.listeners.iter_mut() {
+        let v = kvp.value_mut();
+        v.retain(|v| v.get_id() != actor);
+        if v.is_empty() {
+            empty_groups.push(*kvp.key());
+        }
+    }
+
+    // cleanup empty listener groups
+    for empty_group in empty_groups {
+        monitor.listeners.remove(&empty_group);
+    }
+}
diff --git a/ractor/src/port/output/mod.rs b/ractor/src/port/output/mod.rs
index 0df99fe..ffafc18 100644
--- a/ractor/src/port/output/mod.rs
+++ b/ractor/src/port/output/mod.rs
@@ -74,16 +74,7 @@ where
         let mut subs = self.subscriptions.write().unwrap();
 
         // filter out dead subscriptions, since they're no longer valid
-        let dead_subs = subs
-            .iter()
-            .enumerate()
-            .rev()
-            .filter(|(_, sub)| sub.is_dead())
-            .map(|(a, _)| a)
-            .collect::<Vec<_>>();
-        for dead_sub in dead_subs {
-            subs.remove(dead_sub);
-        }
+        subs.retain(|sub| !sub.is_dead());
 
         let sub = OutputPortSubscription::new::<TMsg, F, TReceiver>(
             self.tx.subscribe(),
diff --git a/ractor/src/registry/mod.rs b/ractor/src/registry/mod.rs
index 02c21d3..5c291ac 100644
--- a/ractor/src/registry/mod.rs
+++ b/ractor/src/registry/mod.rs
@@ -23,19 +23,22 @@
 //! ## 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
+//! async fn test() {
+//!     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 dashmap::mapref::entry::Entry::{Occupied, Vacant};
+use dashmap::DashMap;
 use once_cell::sync::OnceCell;
 
-use crate::{ActorCell, ActorId, ActorName};
+use crate::{ActorCell, ActorName};
 
 #[cfg(test)]
 mod tests;
@@ -50,42 +53,25 @@ pub enum ActorRegistryErr {
 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()
+fn get_actor_registry<'a>() -> &'a Arc<DashMap<ActorName, ActorCell>> {
+    ACTOR_REGISTRY.get_or_init(|| Arc::new(DashMap::new()))
 }
 
 /// 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) => {
+    match get_actor_registry().entry(name) {
+        Occupied(_) => Err(ActorRegistryErr::AlreadyRegistered(name)),
+        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();
+    if let Some(reg) = ACTOR_REGISTRY.get() {
+        let _ = reg.remove(&name);
     }
 }
 
@@ -97,5 +83,5 @@ pub(crate) fn unenroll(name: ActorName) {
 /// actor not registered
 pub fn try_get(name: ActorName) -> Option<ActorCell> {
     let reg = get_actor_registry();
-    reg.get(&name).map(|item| item.value().clone())
+    reg.get(&name).map(|v| v.value().clone())
 }
diff --git a/ractor/src/rpc/call_result.rs b/ractor/src/rpc/call_result.rs
index 5a475ca..8b87b16 100644
--- a/ractor/src/rpc/call_result.rs
+++ b/ractor/src/rpc/call_result.rs
@@ -48,8 +48,14 @@ impl<T> CallResult<T> {
     pub fn expect(self, msg: &'static str) -> T {
         match self {
             Self::Success(t) => t,
-            Self::Timeout => panic!("{}", msg),
-            Self::SenderError => panic!("{}", msg),
+            Self::Timeout => panic!(
+                "{} - called CallResult::<T>::expect()  on a `Timeout` value",
+                msg
+            ),
+            Self::SenderError => panic!(
+                "{} - called CallResult::<T>::expect() on a `SenderError` value",
+                msg
+            ),
         }
     }
 


Test patch:
diff --git a/ractor/src/actor/tests/mod.rs b/ractor/src/actor/tests/mod.rs
index f31e1e1..217112d 100644
--- a/ractor/src/actor/tests/mod.rs
+++ b/ractor/src/actor/tests/mod.rs
@@ -55,11 +55,10 @@ async fn test_stop_higher_priority_over_messages() {
             &self,
             _myself: ActorRef<Self>,
             _message: Self::Msg,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+            _state: &mut Self::State,
+        ) {
             self.counter.fetch_add(1, Ordering::Relaxed);
             tokio::time::sleep(Duration::from_millis(100)).await;
-            None
         }
     }
 
@@ -119,10 +118,9 @@ async fn test_kill_terminates_work() {
             &self,
             _myself: ActorRef<Self>,
             _message: Self::Msg,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+            _state: &mut Self::State,
+        ) {
             tokio::time::sleep(Duration::from_secs(10)).await;
-            None
         }
     }
 
@@ -158,10 +156,9 @@ async fn test_stop_does_not_terminate_async_work() {
             &self,
             _myself: ActorRef<Self>,
             _message: Self::Msg,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+            _state: &mut Self::State,
+        ) {
             tokio::time::sleep(Duration::from_millis(100)).await;
-            None
         }
     }
 
@@ -204,10 +201,9 @@ async fn test_kill_terminates_supervision_work() {
             &self,
             _myself: ActorRef<Self>,
             _message: SupervisionEvent,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+            _state: &mut Self::State,
+        ) {
             tokio::time::sleep(Duration::from_millis(100)).await;
-            None
         }
     }
 
diff --git a/ractor/src/actor/tests/supervisor.rs b/ractor/src/actor/tests/supervisor.rs
index 76c8d6f..dc1397c 100644
--- a/ractor/src/actor/tests/supervisor.rs
+++ b/ractor/src/actor/tests/supervisor.rs
@@ -26,11 +26,7 @@ async fn test_supervision_panic_in_post_startup() {
         type Msg = ();
         type State = ();
         async fn pre_start(&self, _this_actor: ActorRef<Self>) -> Self::State {}
-        async fn post_start(
-            &self,
-            _this_actor: ActorRef<Self>,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+        async fn post_start(&self, _this_actor: ActorRef<Self>, _state: &mut Self::State) {
             panic!("Boom");
         }
     }
@@ -44,17 +40,16 @@ async fn test_supervision_panic_in_post_startup() {
             &self,
             _this_actor: ActorRef<Self>,
             _message: Self::Msg,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
-            None
+            _state: &mut Self::State,
+        ) {
         }
 
         async fn handle_supervisor_evt(
             &self,
             this_actor: ActorRef<Self>,
             message: SupervisionEvent,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+            _state: &mut Self::State,
+        ) {
             println!("Supervisor event received {:?}", message);
 
             // check that the panic was captured
@@ -63,8 +58,6 @@ async fn test_supervision_panic_in_post_startup() {
                     .store(dead_actor.get_id().get_pid(), Ordering::Relaxed);
                 this_actor.stop(None);
             }
-
-            None
         }
     }
 
@@ -74,8 +67,7 @@ async fn test_supervision_panic_in_post_startup() {
         .await
         .expect("Supervisor panicked on startup");
 
-    let supervisor_cell: ActorCell = supervisor_ref.into();
-    let supervisor_tree = supervisor_cell.get_tree();
+    let supervisor_cell: ActorCell = supervisor_ref.clone().into();
 
     let (child_ref, c_handle) = Actor::spawn_linked(None, Child, supervisor_cell)
         .await
@@ -86,7 +78,7 @@ async fn test_supervision_panic_in_post_startup() {
     assert_eq!(child_ref.get_id().get_pid(), flag.load(Ordering::Relaxed));
 
     // supervisor relationship cleaned up correctly
-    assert_eq!(0, supervisor_tree.get_num_children());
+    assert_eq!(0, supervisor_ref.get_num_children());
 }
 
 #[tokio::test]
@@ -105,8 +97,8 @@ async fn test_supervision_panic_in_handle() {
             &self,
             _this_actor: ActorRef<Self>,
             _message: Self::Msg,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+            _state: &mut Self::State,
+        ) {
             panic!("Boom");
         }
     }
@@ -120,17 +112,16 @@ async fn test_supervision_panic_in_handle() {
             &self,
             _this_actor: ActorRef<Self>,
             _message: Self::Msg,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
-            None
+            _state: &mut Self::State,
+        ) {
         }
 
         async fn handle_supervisor_evt(
             &self,
             this_actor: ActorRef<Self>,
             message: SupervisionEvent,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+            _state: &mut Self::State,
+        ) {
             println!("Supervisor event received {:?}", message);
 
             // check that the panic was captured
@@ -139,8 +130,6 @@ async fn test_supervision_panic_in_handle() {
                     .store(dead_actor.get_id().get_pid(), Ordering::Relaxed);
                 this_actor.stop(None);
             }
-
-            None
         }
     }
 
@@ -150,17 +139,15 @@ async fn test_supervision_panic_in_handle() {
         .await
         .expect("Supervisor panicked on startup");
 
-    let supervisor_cell: ActorCell = supervisor_ref.into();
-    let supervisor_tree = supervisor_cell.get_tree();
+    let supervisor_cell: ActorCell = supervisor_ref.clone().into();
 
     let (child_ref, c_handle) = Actor::spawn_linked(None, Child, supervisor_cell)
         .await
         .expect("Child panicked on startup");
 
     // check that the supervision is wired up correctly
-    assert_eq!(1, supervisor_tree.get_num_children());
-    let child_cell: ActorCell = child_ref.clone().into();
-    assert_eq!(1, child_cell.get_tree().get_num_parents());
+    assert_eq!(1, supervisor_ref.get_num_children());
+    assert_eq!(1, child_ref.get_num_parents());
 
     // trigger the child failure
     child_ref.send_message(()).expect("Failed to send message");
@@ -170,7 +157,7 @@ async fn test_supervision_panic_in_handle() {
     assert_eq!(child_ref.get_id().get_pid(), flag.load(Ordering::Relaxed));
 
     // supervisor relationship cleaned up correctly
-    assert_eq!(0, supervisor_tree.get_num_children());
+    assert_eq!(0, supervisor_ref.get_num_children());
 }
 
 #[tokio::test]
@@ -188,7 +175,7 @@ async fn test_supervision_panic_in_post_stop() {
             // trigger stop, which starts shutdown
             this_actor.stop(None);
         }
-        async fn post_stop(&self, _this_actor: ActorRef<Self>, _state: Self::State) -> Self::State {
+        async fn post_stop(&self, _this_actor: ActorRef<Self>, _state: &mut Self::State) {
             panic!("Boom");
         }
     }
@@ -202,17 +189,16 @@ async fn test_supervision_panic_in_post_stop() {
             &self,
             _this_actor: ActorRef<Self>,
             _message: Self::Msg,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
-            None
+            _state: &mut Self::State,
+        ) {
         }
 
         async fn handle_supervisor_evt(
             &self,
             this_actor: ActorRef<Self>,
             message: SupervisionEvent,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+            _state: &mut Self::State,
+        ) {
             println!("Supervisor event received {:?}", message);
 
             // check that the panic was captured
@@ -221,8 +207,6 @@ async fn test_supervision_panic_in_post_stop() {
                     .store(dead_actor.get_id().get_pid(), Ordering::Relaxed);
                 this_actor.stop(None);
             }
-
-            None
         }
     }
 
@@ -232,8 +216,7 @@ async fn test_supervision_panic_in_post_stop() {
         .await
         .expect("Supervisor panicked on startup");
 
-    let supervisor_cell: ActorCell = supervisor_ref.into();
-    let supervisor_tree = supervisor_cell.get_tree();
+    let supervisor_cell: ActorCell = supervisor_ref.clone().into();
 
     let (child_ref, c_handle) = Actor::spawn_linked(None, Child, supervisor_cell)
         .await
@@ -244,7 +227,7 @@ async fn test_supervision_panic_in_post_stop() {
     assert_eq!(child_ref.get_id().get_pid(), flag.load(Ordering::Relaxed));
 
     // supervisor relationship cleaned up correctly
-    assert_eq!(0, supervisor_tree.get_num_children());
+    assert_eq!(0, supervisor_ref.get_num_children());
 }
 
 /// Test that a panic in the supervisor's handling propogates to
@@ -266,8 +249,8 @@ async fn test_supervision_panic_in_supervisor_handle() {
             &self,
             _this_actor: ActorRef<Self>,
             _message: Self::Msg,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+            _state: &mut Self::State,
+        ) {
             panic!("Boom!");
         }
     }
@@ -281,12 +264,11 @@ async fn test_supervision_panic_in_supervisor_handle() {
             &self,
             _this_actor: ActorRef<Self>,
             _message: SupervisionEvent,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+            _state: &mut Self::State,
+        ) {
             if let SupervisionEvent::ActorPanicked(_child, _msg) = _message {
                 panic!("Boom again!");
             }
-            None
         }
     }
 
@@ -299,17 +281,16 @@ async fn test_supervision_panic_in_supervisor_handle() {
             &self,
             _this_actor: ActorRef<Self>,
             _message: Self::Msg,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
-            None
+            _state: &mut Self::State,
+        ) {
         }
 
         async fn handle_supervisor_evt(
             &self,
             this_actor: ActorRef<Self>,
             message: SupervisionEvent,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+            _state: &mut Self::State,
+        ) {
             println!("Supervisor event received {:?}", message);
 
             // check that the panic was captured
@@ -318,8 +299,6 @@ async fn test_supervision_panic_in_supervisor_handle() {
                     .store(dead_actor.get_id().get_pid(), Ordering::Relaxed);
                 this_actor.stop(None);
             }
-
-            None
         }
     }
 
@@ -329,15 +308,13 @@ async fn test_supervision_panic_in_supervisor_handle() {
         .await
         .expect("Supervisor panicked on startup");
 
-    let supervisor_cell: ActorCell = supervisor_ref.into();
-    let supervisor_tree = supervisor_cell.get_tree();
+    let supervisor_cell: ActorCell = supervisor_ref.clone().into();
 
     let (midpoint_ref, m_handle) = Actor::spawn_linked(None, Midpoint, supervisor_cell)
         .await
         .expect("Midpoint actor failed to startup");
 
     let midpoint_cell: ActorCell = midpoint_ref.clone().into();
-    let midpoint_tree = midpoint_cell.get_tree();
     let midpoint_ref_clone = midpoint_ref.clone();
 
     let (child_ref, c_handle) = Actor::spawn_linked(None, Child, midpoint_cell)
@@ -345,11 +322,10 @@ async fn test_supervision_panic_in_supervisor_handle() {
         .expect("Child panicked on startup");
 
     // check that the supervision is wired up correctly
-    assert_eq!(1, supervisor_tree.get_num_children());
-    assert_eq!(1, midpoint_tree.get_num_children());
-    assert_eq!(1, midpoint_tree.get_num_parents());
-    let child_cell: ActorCell = child_ref.clone().into();
-    assert_eq!(1, child_cell.get_tree().get_num_parents());
+    assert_eq!(1, supervisor_ref.get_num_children());
+    assert_eq!(1, midpoint_ref.get_num_children());
+    assert_eq!(1, midpoint_ref.get_num_parents());
+    assert_eq!(1, child_ref.get_num_parents());
 
     // trigger the child failure
     child_ref.send_message(()).expect("Failed to send message");
@@ -368,7 +344,7 @@ async fn test_supervision_panic_in_supervisor_handle() {
     );
 
     // supervisor relationship cleaned up correctly
-    assert_eq!(0, supervisor_tree.get_num_children());
+    assert_eq!(0, supervisor_ref.get_num_children());
 }
 
 #[tokio::test]
@@ -392,11 +368,10 @@ async fn test_killing_a_supervisor_terminates_children() {
             &self,
             _this_actor: ActorRef<Self>,
             _message: Self::Msg,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+            _state: &mut Self::State,
+        ) {
             // stop the supervisor, which starts the supervision shutdown of children
             _this_actor.stop(None);
-            None
         }
     }
 
@@ -404,15 +379,13 @@ async fn test_killing_a_supervisor_terminates_children() {
         .await
         .expect("Supervisor panicked on startup");
 
-    let supervisor_tree = supervisor_ref.get_tree();
-
     let (child_ref, c_handle) = Actor::spawn_linked(None, Child, supervisor_ref.clone().into())
         .await
         .expect("Child panicked on startup");
 
     // check that the supervision is wired up correctly
-    assert_eq!(1, supervisor_tree.get_num_children());
-    assert_eq!(1, child_ref.clone().get_tree().get_num_parents());
+    assert_eq!(1, supervisor_ref.get_num_children());
+    assert_eq!(1, child_ref.get_num_parents());
 
     // initate the shutdown of the supervisor
     supervisor_ref
diff --git a/ractor/src/pg/tests.rs b/ractor/src/pg/tests.rs
new file mode 100644
index 0000000..1b2d8d5
--- /dev/null
+++ b/ractor/src/pg/tests.rs
@@ -0,0 +1,272 @@
+// 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.
+
+use std::sync::atomic::{AtomicU8, Ordering};
+use std::sync::Arc;
+
+use ::function_name::named;
+use tokio::time::Duration;
+
+use crate::{Actor, ActorHandler, GroupName, SupervisionEvent};
+
+use crate::pg;
+
+struct TestActor;
+
+#[async_trait::async_trait]
+impl ActorHandler for TestActor {
+    type Msg = ();
+
+    type State = ();
+
+    async fn pre_start(&self, _this_actor: crate::ActorRef<Self>) -> Self::State {}
+}
+
+#[named]
+#[tokio::test]
+async fn test_basic_group() {
+    let (actor, handle) = Actor::spawn(None, TestActor)
+        .await
+        .expect("Failed to spawn test actor");
+
+    let group = function_name!();
+
+    // join the group
+    pg::join(group, vec![actor.clone().into()]);
+
+    let members = pg::get_members(group);
+    assert_eq!(1, members.len());
+
+    // Cleanup
+    actor.stop(None);
+    handle.await.expect("Actor cleanup failed");
+}
+
+#[named]
+#[tokio::test]
+async fn test_multiple_members_in_group() {
+    let group = function_name!();
+
+    let mut actors = vec![];
+    let mut handles = vec![];
+    for _ in 0..10 {
+        let (actor, handle) = Actor::spawn(None, TestActor)
+            .await
+            .expect("Failed to spawn test actor");
+        actors.push(actor);
+        handles.push(handle);
+    }
+
+    // join the group
+    pg::join(
+        group,
+        actors
+            .iter()
+            .map(|aref| aref.clone().get_cell())
+            .collect::<Vec<_>>(),
+    );
+
+    let members = pg::get_members(group);
+    assert_eq!(10, members.len());
+
+    // Cleanup
+    for actor in actors {
+        actor.stop(None);
+    }
+    for handle in handles.into_iter() {
+        handle.await.expect("Actor cleanup failed");
+    }
+}
+
+#[named]
+#[tokio::test]
+async fn test_multiple_groups() {
+    let group_a = concat!(function_name!(), "_a");
+    let group_b = concat!(function_name!(), "_b");
+
+    let mut actors = vec![];
+    let mut handles = vec![];
+    for _ in 0..10 {
+        let (actor, handle) = Actor::spawn(None, TestActor)
+            .await
+            .expect("Failed to spawn test actor");
+        actors.push(actor);
+        handles.push(handle);
+    }
+
+    // setup group_a and group_b
+    let these_actors = actors[0..5]
+        .iter()
+        .map(|a| a.clone().get_cell())
+        .collect::<Vec<_>>();
+    pg::join(group_a, these_actors);
+
+    let these_actors = actors[5..10]
+        .iter()
+        .map(|a| a.clone().get_cell())
+        .collect::<Vec<_>>();
+    pg::join(group_b, these_actors);
+
+    let members = pg::get_members(group_a);
+    assert_eq!(5, members.len());
+
+    let members = pg::get_members(group_b);
+    assert_eq!(5, members.len());
+
+    // Cleanup
+    for actor in actors {
+        actor.stop(None);
+    }
+    for handle in handles.into_iter() {
+        handle.await.expect("Actor cleanup failed");
+    }
+}
+
+#[named]
+#[tokio::test]
+async fn test_actor_leaves_pg_group_on_shutdown() {
+    let (actor, handle) = Actor::spawn(None, TestActor)
+        .await
+        .expect("Failed to spawn test actor");
+
+    let group = function_name!();
+
+    // join the group
+    pg::join(group, vec![actor.clone().into()]);
+
+    let members = pg::get_members(group);
+    assert_eq!(1, members.len());
+
+    // Cleanup
+    actor.stop(None);
+    handle.await.expect("Actor cleanup failed");
+    drop(actor);
+
+    let members = pg::get_members(group);
+    assert_eq!(0, members.len());
+}
+
+#[named]
+#[tokio::test]
+async fn test_actor_leaves_pg_group_manually() {
+    let group = function_name!();
+
+    let (actor, handle) = Actor::spawn(None, TestActor)
+        .await
+        .expect("Failed to spawn test actor");
+
+    // join the group (create on first use)
+    pg::join(group, vec![actor.clone().into()]);
+
+    // the group was created and is present
+    let groups = pg::which_groups();
+    assert!(groups.contains(&group));
+
+    let members = pg::get_members(group);
+    assert_eq!(1, members.len());
+
+    // leave the group
+    pg::leave(group, vec![actor.clone().into()]);
+
+    // pif-paf-poof the group is gone!
+    let groups = pg::which_groups();
+    assert!(!groups.contains(&group));
+
+    // members comes back empty
+    let members = pg::get_members(group);
+    assert_eq!(0, members.len());
+
+    // Cleanup
+    actor.stop(None);
+    handle.await.expect("Actor cleanup failed");
+}
+
+#[named]
+#[tokio::test]
+async fn test_pg_monitoring() {
+    let group = function_name!();
+
+    let counter = Arc::new(AtomicU8::new(0u8));
+
+    struct AutoJoinActor {
+        pg_group: GroupName,
+    }
+
+    #[async_trait::async_trait]
+    impl ActorHandler for AutoJoinActor {
+        type Msg = ();
+
+        type State = ();
+
+        async fn pre_start(&self, myself: crate::ActorRef<Self>) -> Self::State {
+            pg::join(self.pg_group, vec![myself.into()]);
+        }
+    }
+
+    struct NotificationMonitor {
+        pg_group: GroupName,
+        counter: Arc<AtomicU8>,
+    }
+
+    #[async_trait::async_trait]
+    impl ActorHandler for NotificationMonitor {
+        type Msg = ();
+
+        type State = ();
+
+        async fn pre_start(&self, myself: crate::ActorRef<Self>) -> Self::State {
+            pg::monitor(self.pg_group, myself.into());
+        }
+
+        async fn handle_supervisor_evt(
+            &self,
+            _myself: crate::ActorRef<Self>,
+            message: SupervisionEvent,
+            _state: &mut Self::State,
+        ) {
+            if let SupervisionEvent::ProcessGroupChanged(change) = message {
+                match change {
+                    pg::GroupChangeMessage::Join(_which, who) => {
+                        self.counter.fetch_add(who.len() as u8, Ordering::Relaxed);
+                    }
+                    pg::GroupChangeMessage::Leave(_which, who) => {
+                        self.counter.fetch_sub(who.len() as u8, Ordering::Relaxed);
+                    }
+                }
+            }
+        }
+    }
+    let (monitor_actor, monitor_handle) = Actor::spawn(
+        None,
+        NotificationMonitor {
+            pg_group: group,
+            counter: counter.clone(),
+        },
+    )
+    .await
+    .expect("Failed to start monitor actor");
+
+    // this actor's startup should "monitor" for PG changes
+    let (test_actor, test_handle) = Actor::spawn(None, AutoJoinActor { pg_group: group })
+        .await
+        .expect("Failed to start test actor");
+
+    // the monitor is notified async, so we need to wait a tiny bit
+    tokio::time::sleep(Duration::from_millis(100)).await;
+    assert_eq!(1, counter.load(Ordering::Relaxed));
+
+    // kill the pg member
+    test_actor.stop(None);
+    test_handle.await.expect("Actor cleanup failed");
+    // it should have notified that it's unsubscribed
+    assert_eq!(0, counter.load(Ordering::Relaxed));
+
+    // cleanup
+    monitor_actor.stop(None);
+    monitor_handle.await.expect("Actor cleanup failed");
+}
+
+// TODO: Tests to add
+// 1. Local vs remote members (can't test until we have proper remoting)
diff --git a/ractor/src/port/output/tests.rs b/ractor/src/port/output/tests.rs
index b97cbfd..f2ffd7d 100644
--- a/ractor/src/port/output/tests.rs
+++ b/ractor/src/port/output/tests.rs
@@ -34,8 +34,8 @@ async fn test_single_forward() {
             &self,
             myself: ActorRef<Self>,
             message: Self::Msg,
-            state: &Self::State,
-        ) -> Option<Self::State> {
+            state: &mut Self::State,
+        ) {
             println!("Test actor received a message");
             match message {
                 Self::Msg::Stop => {
@@ -44,7 +44,7 @@ async fn test_single_forward() {
                     }
                 }
             }
-            Some(*state + 1)
+            *state += 1;
         }
     }
 
@@ -90,8 +90,8 @@ async fn test_50_receivers() {
             &self,
             myself: ActorRef<Self>,
             message: Self::Msg,
-            state: &Self::State,
-        ) -> Option<Self::State> {
+            state: &mut Self::State,
+        ) {
             println!("Test actor received a message");
             match message {
                 Self::Msg::Stop => {
@@ -100,7 +100,7 @@ async fn test_50_receivers() {
                     }
                 }
             }
-            Some(*state + 1)
+            *state += 1;
         }
     }
 
diff --git a/ractor/src/registry/tests.rs b/ractor/src/registry/tests.rs
index fa4696e..eb240cb 100644
--- a/ractor/src/registry/tests.rs
+++ b/ractor/src/registry/tests.rs
@@ -5,6 +5,8 @@
 
 //! Tests on the actor registry
 
+use tokio::time::Duration;
+
 use crate::{Actor, ActorHandler, SpawnErr};
 
 #[tokio::test]
@@ -89,6 +91,9 @@ async fn test_actor_registry_unenrollment() {
     // drop the actor ref's
     drop(actor);
 
+    // unenrollment is a cast operation, so it's not immediate. wait for cleanup
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
     // 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 6910015..999b1b2 100644
--- a/ractor/src/rpc/tests.rs
+++ b/ractor/src/rpc/tests.rs
@@ -33,10 +33,9 @@ async fn test_rpc_cast() {
             &self,
             _this_actor: ActorRef<Self>,
             _message: Self::Msg,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+            _state: &mut Self::State,
+        ) {
             self.counter.fetch_add(1u8, Ordering::Relaxed);
-            None
         }
     }
 
@@ -83,8 +82,8 @@ async fn test_rpc_call() {
             &self,
             _this_actor: ActorRef<Self>,
             message: Self::Msg,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+            _state: &mut Self::State,
+        ) {
             match message {
                 Self::Msg::TestRpc(reply) => {
                     // An error sending means no one is listening anymore (the receiver was dropped),
@@ -94,7 +93,6 @@ async fn test_rpc_call() {
                     }
                 }
             }
-            None
         }
     }
 
@@ -141,8 +139,8 @@ async fn test_rpc_call_forwarding() {
             &self,
             _this_actor: ActorRef<Self>,
             message: Self::Msg,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+            _state: &mut Self::State,
+        ) {
             match message {
                 Self::Msg::TestRpc(reply) => {
                     // An error sending means no one is listening anymore (the receiver was dropped),
@@ -152,7 +150,6 @@ async fn test_rpc_call_forwarding() {
                     }
                 }
             }
-            None
         }
     }
 
@@ -177,15 +174,14 @@ async fn test_rpc_call_forwarding() {
             &self,
             _this_actor: ActorRef<Self>,
             message: Self::Msg,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+            _state: &mut Self::State,
+        ) {
             match message {
                 Self::Msg::ForwardResult(s) if s == *"howdy" => {
                     self.counter.fetch_add(1, Ordering::Relaxed);
                 }
                 _ => {}
             }
-            None
         }
     }
 
diff --git a/ractor/src/time/tests.rs b/ractor/src/time/tests.rs
index 1191372..8512c63 100644
--- a/ractor/src/time/tests.rs
+++ b/ractor/src/time/tests.rs
@@ -33,11 +33,10 @@ async fn test_intervals() {
             &self,
             _this_actor: ActorRef<Self>,
             _message: Self::Msg,
-            state: &Self::State,
-        ) -> Option<Self::State> {
+            state: &mut Self::State,
+        ) {
             // stop the supervisor, which starts the supervision shutdown of children
             state.fetch_add(1, Ordering::Relaxed);
-            None
         }
     }
 
@@ -64,7 +63,7 @@ async fn test_intervals() {
     // kill the actor
     actor_ref.stop(None);
 
-    tokio::time::sleep(Duration::from_millis(20)).await;
+    tokio::time::sleep(Duration::from_millis(50)).await;
     // make sure the actor is dead + the interval handle doesn't send again
     assert!(interval_handle.is_finished());
     assert!(actor_handle.is_finished());
@@ -93,11 +92,10 @@ async fn test_send_after() {
             &self,
             _this_actor: ActorRef<Self>,
             _message: Self::Msg,
-            state: &Self::State,
-        ) -> Option<Self::State> {
+            state: &mut Self::State,
+        ) {
             // stop the supervisor, which starts the supervision shutdown of children
             state.fetch_add(1, Ordering::Relaxed);
-            None
         }
     }
 
@@ -170,10 +168,9 @@ async fn test_kill_after() {
             &self,
             _myself: ActorRef<Self>,
             _message: Self::Msg,
-            _state: &Self::State,
-        ) -> Option<Self::State> {
+            _state: &mut Self::State,
+        ) {
             tokio::time::sleep(Duration::from_millis(100)).await;
-            None
         }
     }