Repository: slawlor/ractor
Pull request #25 (2023-01-13 15:36:18)

Adding more complex examples
Some more complex example actor patterns, such as multiple-actors inter-communicating, delayed message processing for FSM style handling, etc

Additionally some documentation cleanups


Patch:
diff --git a/.github/codecov.yml b/.github/codecov.yml
deleted file mode 100644
index 52258a4..0000000
--- a/.github/codecov.yml
+++ /dev/null
@@ -1,3 +0,0 @@
-ignore:
-  - "ractor-playground"
-  - "xtask"
\ No newline at end of file
diff --git a/README.md b/README.md
index 8bb1cc2..91342c5 100644
--- a/README.md
+++ b/README.md
@@ -9,6 +9,7 @@ A pure-Rust actor framework. Inspired from [Erlang's `gen_server`](https://www.e
 [<img alt="docs.rs" src="https://img.shields.io/badge/docs.rs-ractor-66c2a5?style=for-the-badge&labelColor=555555&logo=docs.rs" height="20">](https://docs.rs/ractor)
 [![CI/main](https://github.com/slawlor/repl/actions/workflows/ci.yaml/badge.svg?branch=main)](https://github.com/slawlor/repl/actions/workflows/ci.yaml)
 [![codecov](https://codecov.io/gh/slawlor/ractor/branch/main/graph/badge.svg?token=61AGYYPWBA)](https://codecov.io/gh/slawlor/ractor)
+![Downloads](https://img.shields.io/crates/d/ractor.svg)
 
 ## About
 
diff --git a/codecov.yml b/codecov.yml
new file mode 100644
index 0000000..b5d6274
--- /dev/null
+++ b/codecov.yml
@@ -0,0 +1,6 @@
+codecov:
+  require_ci_to_pass: false
+
+ignore:
+  - "ractor-playground"
+  - "xtask"
\ No newline at end of file
diff --git a/ractor/Cargo.toml b/ractor/Cargo.toml
index 6a85523..4005c55 100644
--- a/ractor/Cargo.toml
+++ b/ractor/Cargo.toml
@@ -22,6 +22,7 @@ tokio = { version = "1", features = ["rt", "time", "sync", "macros"]}
 [dev-dependencies]
 criterion = "0.3"
 function_name = "0.3"
+rand = "0.8"
 tokio = { version = "1", features = ["rt", "time", "sync", "macros", "rt-multi-thread"] }
 
 [[bench]]
diff --git a/ractor/examples/monte_carlo.rs b/ractor/examples/monte_carlo.rs
new file mode 100644
index 0000000..17ce461
--- /dev/null
+++ b/ractor/examples/monte_carlo.rs
@@ -0,0 +1,187 @@
+// 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.
+
+//! This is an example of a parallel processing implementation of a Monte-Carlo simulation
+//! The simulation is of a basic gambling game adapted from this page:
+//! https://towardsdatascience.com/the-house-always-wins-monte-carlo-simulation-eb82787da2a3
+//!
+//! Demonstrates parallel work and supervision
+//!
+//! Run this example with
+//!
+//! ```bash
+//! cargo run --example monte_carlo
+//! ```
+
+use std::collections::HashMap;
+
+use ractor::{Actor, ActorHandler, ActorId, ActorRef};
+use rand::{thread_rng, Rng};
+
+// ================== Player Actor ================== //
+
+struct GameState {
+    /// The player's current funds. Funds are allowed to be negative since the player
+    /// can potentially lose more money than they started with.
+    funds: i64,
+    /// How much money the player wagers per turn.
+    wager: u32,
+    /// The total number of game rounds that will be played.
+    total_rounds: u32,
+
+    current_round: u32,
+    results_vec: Vec<i64>,
+}
+
+impl Default for GameState {
+    fn default() -> Self {
+        Self {
+            funds: 10_000,
+            wager: 100,
+            total_rounds: 100,
+            current_round: 1,
+            results_vec: vec![],
+        }
+    }
+}
+
+impl GameState {
+    /// This function performs a dice roll according to the rules of the simple gambling game.
+    /// On average, the player will win their roll 49 out of 100 times, resulting in a house edge
+    /// of 2%
+    fn roll_dice() -> bool {
+        let mut rng = thread_rng();
+        matches!(rng.gen_range(0..101), x if x > 51)
+    }
+}
+
+struct Game;
+
+#[async_trait::async_trait]
+impl ActorHandler for Game {
+    type Msg = ActorRef<GameManager>;
+
+    type State = GameState;
+
+    async fn pre_start(&self, _myself: ActorRef<Self>) -> Self::State {
+        GameState::default()
+    }
+
+    async fn handle(&self, myself: ActorRef<Self>, message: Self::Msg, state: &mut Self::State) {
+        if state.current_round <= state.total_rounds {
+            state.current_round += 1;
+            match Self::State::roll_dice() {
+                true => state.funds += state.wager as i64,
+                false => state.funds -= state.wager as i64,
+            }
+            state.results_vec.push(state.funds);
+            let _ = myself.cast(message);
+        } else {
+            // Now that the game is finished, the results of the game need to be reported
+            // to the `GameManager`.
+            let _ = message.cast(GameManagerMessage {
+                id: myself.get_id(),
+                results: state.results_vec.clone(),
+            });
+            // Because the `GameManager` is monitoring this actor we can stop this actor
+            myself.stop(None);
+        }
+    }
+}
+
+// ================== Manager Actor ================== //
+
+struct GameManager {
+    num_games: u32,
+}
+
+struct GameManagerMessage {
+    id: ActorId,
+    results: Vec<i64>,
+}
+
+struct GameManagerState {
+    /// The number of games that have been played so far.
+    games_finished: u32,
+    /// The total number of games that are to be played.
+    total_games: u32,
+    /// The results of each finished game, keyed by `Game` actor ID
+    results: HashMap<ActorId, Vec<i64>>,
+}
+
+impl GameManagerState {
+    fn new(total_games: u32) -> Self {
+        Self {
+            games_finished: 0,
+            total_games,
+            results: HashMap::new(),
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl ActorHandler for GameManager {
+    type Msg = GameManagerMessage;
+
+    type State = GameManagerState;
+
+    async fn pre_start(&self, myself: ActorRef<Self>) -> Self::State {
+        // This is the first code that will run in the actor. It spawns the Game actors,
+        // registers them to its monitoring list, then sends them a message indicating
+        // that they should start their games.
+
+        let game_conditions = GameState::default();
+        println!("Starting funds: ${}", game_conditions.funds);
+        println!("Wager per round: ${}", game_conditions.wager);
+        println!("Rounds per game: {}", game_conditions.total_rounds);
+        println!("Running simulations...");
+        for _ in 0..self.num_games {
+            let (actor, _) = Actor::spawn_linked(None, Game, myself.clone().into())
+                .await
+                .expect("Failed to start game");
+            let _ = actor.cast(myself.clone());
+        }
+
+        GameManagerState::new(self.num_games)
+    }
+
+    async fn handle(&self, myself: ActorRef<Self>, message: Self::Msg, state: &mut Self::State) {
+        state.results.insert(message.id, message.results);
+
+        state.games_finished += 1;
+        if state.games_finished >= state.total_games {
+            // Each vec of results contains the entire history of a game for every time that
+            // the dice was rolled. Instead of printing out all of that data, we will simply
+            // print the average of the funds that the player had at the end of each game.
+            let average_funds = state
+                .results
+                .values()
+                .map(|v| v.last().unwrap())
+                .sum::<i64>()
+                / state.total_games as i64;
+
+            println!("Simulations ran: {}", state.results.len());
+            println!("Final average funds: ${}", average_funds);
+
+            myself.stop(None);
+        }
+    }
+}
+
+const NUM_GAMES: u32 = 100;
+
+#[tokio::main]
+async fn main() {
+    // create the supervisor
+    let manager = GameManager {
+        num_games: NUM_GAMES,
+    };
+    // spawn it off and wait for it to complete/exit
+    let (_actor, handle) = Actor::spawn(None, manager)
+        .await
+        .expect("Failed to start game manager");
+
+    handle.await.unwrap();
+}
diff --git a/ractor/examples/philosophers.rs b/ractor/examples/philosophers.rs
new file mode 100644
index 0000000..c5c29b9
--- /dev/null
+++ b/ractor/examples/philosophers.rs
@@ -0,0 +1,510 @@
+// 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 implementation of the Chandy & Misra solution to the classic finite state machine (FSM)
+//! concurrency problem known as [Dining Philosophers]
+//! (https://en.wikipedia.org/wiki/Dining_philosophers_problem) problem using `ractor`.
+//!
+//! Utilizes:
+//! * Multiple actors, inter-communicating
+//! * RPCs
+//! * Mutable state transitions
+//!
+//! Run this example with
+//!
+//! ```bash
+//! cargo run --example philosophers
+//! ```
+
+use std::collections::{HashMap, VecDeque};
+
+use ractor::{Actor, ActorHandler, ActorId, ActorName, ActorRef, RpcReplyPort};
+use tokio::time::{Duration, Instant};
+
+// ============================ Fork Actor ============================ //
+
+enum ForkMessage {
+    /// Request the fork be sent to a philospher
+    RequestFork(ActorRef<Philosopher>),
+    /// Mark the fork as currently being used
+    UsingFork(ActorId),
+    /// Sent to a fork to indicate that it was put down and no longer is in use. This will
+    /// allow the fork to be sent to the next user.
+    PutForkDown(ActorId),
+}
+
+struct ForkState {
+    /// Flag to identify if the fork is clean or not
+    clean: bool,
+    /// The actor who currently owns the fork
+    owned_by: Option<ActorRef<Philosopher>>,
+    // A backlog of messages which get queue'd up in a state transition
+    backlog: VecDeque<ForkMessage>,
+}
+
+struct Fork;
+
+impl Fork {
+    fn handle_internal(
+        &self,
+        myself: &ActorRef<Self>,
+        message: ForkMessage,
+        state: &mut ForkState,
+    ) -> Option<ForkMessage> {
+        match &message {
+            ForkMessage::RequestFork(who) => {
+                match &state.owned_by {
+                    Some(owner) => {
+                        if !state.clean {
+                            let _ = owner.cast(PhilosopherMessage::GiveUpFork(myself.get_id()));
+                        }
+                        // there's already an owner, backlog this message in priority
+                        return Some(message);
+                    }
+                    None => {
+                        // give the fork to the requester
+                        let _ = who.cast(PhilosopherMessage::ReceiveFork(myself.get_id()));
+                        // set ownership
+                        state.owned_by = Some(who.clone());
+                    }
+                }
+            }
+            ForkMessage::UsingFork(who) => match &state.owned_by {
+                Some(owner) if owner.get_id() == *who => {
+                    state.clean = false;
+                }
+                Some(other_owner) => {
+                    println!(
+                        "ERROR Received `UsingFork` from {:?}. Real owner is {:?}",
+                        who,
+                        other_owner.get_name().unwrap()
+                    );
+                }
+                None => {
+                    println!(
+                        "ERROR Received `UsingFork` from {:?}. Real owner is `None`",
+                        who
+                    );
+                }
+            },
+            ForkMessage::PutForkDown(who) => match &state.owned_by {
+                Some(owner) if owner.get_id() == *who => {
+                    state.owned_by = None;
+                    state.clean = true;
+                }
+                Some(other_owner) => {
+                    println!(
+                        "ERROR Received `PutForkDown` from {:?}. Real owner is {:?}",
+                        who,
+                        other_owner.get_name().unwrap()
+                    );
+                }
+                None => {
+                    println!(
+                        "ERROR Received `PutForkDown` from {:?}. Real owner is `None`",
+                        who
+                    );
+                }
+            },
+        }
+        None
+    }
+}
+
+#[async_trait::async_trait]
+impl ActorHandler for Fork {
+    type Msg = ForkMessage;
+    type State = ForkState;
+    async fn pre_start(&self, _myself: ActorRef<Self>) -> Self::State {
+        Self::State {
+            clean: false,
+            owned_by: None,
+            backlog: VecDeque::new(),
+        }
+    }
+
+    async fn handle(&self, myself: ActorRef<Self>, message: Self::Msg, state: &mut Self::State) {
+        let mut maybe_unhandled = self.handle_internal(&myself, message, state);
+        if let Some(message) = maybe_unhandled {
+            state.backlog.push_back(message);
+        } else {
+            // we handled the message, check the queue for any work to dequeue and handle
+            while !state.backlog.is_empty() && maybe_unhandled.is_none() {
+                let head = state.backlog.pop_front().unwrap();
+                maybe_unhandled = self.handle_internal(&myself, head, state);
+            }
+            // put the first unhandled msg back to the front of the queue
+            if let Some(msg) = maybe_unhandled {
+                state.backlog.push_front(msg);
+            }
+        }
+    }
+}
+
+// ============================ Philosopher Actor ============================ //
+
+#[derive(PartialEq, Eq)]
+enum PhilosopherMode {
+    /// The philosopher is thinking
+    Thinking,
+    /// The philosopher is hungry and waiting for one of the forks
+    Hungry,
+    /// The philosopher is eating
+    Eating,
+}
+
+struct PhilosophersFork {
+    /// The pointer to the fork actor
+    fork: ActorRef<Fork>,
+    /// Does this philosopher currently have this fork?
+    has: bool,
+    /// Has the philosopher requested this fork?
+    requested: bool,
+}
+
+#[derive(Clone, Debug)]
+struct PhilosopherMetrics {
+    /// The number of state changes that have occurred.
+    state_change_count: u16,
+    /// The number of times a Philosopher failed to eat because he didnt have both forks.
+    failed_to_eat: u16,
+    /// The time that the Philosopher spent thinking.
+    time_thinking: Duration,
+    /// The time that the Philosopher spent hungry.
+    time_hungry: Duration,
+    /// The time that the Philosopher spent eating.
+    time_eating: Duration,
+}
+
+struct PhilosopherState {
+    /// The current mode/state the philosopher is in
+    mode: PhilosopherMode,
+    /// The fork to the left of the Philosopher
+    left: PhilosophersFork,
+    /// The fork to the right of the Philosopher
+    right: PhilosophersFork,
+    /// The last time the philosopher's state changed. Tracking time eating, etc
+    last_state_change: Instant,
+    /// The metrics of this actor
+    metrics: PhilosopherMetrics,
+}
+
+impl PhilosopherState {
+    fn new(left: ActorRef<Fork>, right: ActorRef<Fork>) -> Self {
+        Self {
+            mode: PhilosopherMode::Thinking,
+            left: PhilosophersFork {
+                fork: left,
+                has: false,
+                requested: false,
+            },
+            right: PhilosophersFork {
+                fork: right,
+                has: false,
+                requested: false,
+            },
+            last_state_change: Instant::now(),
+            metrics: PhilosopherMetrics {
+                state_change_count: 0,
+                failed_to_eat: 0,
+                time_thinking: Duration::from_micros(0),
+                time_hungry: Duration::from_micros(0),
+                time_eating: Duration::from_micros(0),
+            },
+        }
+    }
+}
+
+enum PhilosopherMessage {
+    /// Command to stop eating. Note that since the `StopEating` message is sent as a scheduled message
+    /// it may arrive after the philosopher has already changed state. For this reason we track
+    /// the state change count and compare it with the number in the message.
+    StopEating(u16),
+    /// Command to stop eating. Note that since the `BecomeHungry` message is sent as a scheduled message
+    /// it may arrive after the philosopher has already changed state. For this reason we track
+    /// the state change count and compare it with the number in the message.
+    BecomeHungry(u16),
+    /// Instructs the philosopher to give up the fork
+    GiveUpFork(ActorId),
+    /// Instructs the philosopher they've received the specified fork
+    ReceiveFork(ActorId),
+    SendMetrics(RpcReplyPort<PhilosopherMetrics>),
+}
+
+struct Philosopher {
+    time_slice: Duration,
+    left: ActorRef<Fork>,
+    right: ActorRef<Fork>,
+}
+
+impl Philosopher {
+    /// Helper method to set the internal state to begin thinking
+    fn begin_thinking(&self, myself: &ActorRef<Self>, state: &mut PhilosopherState) {
+        state.mode = PhilosopherMode::Thinking;
+        state.metrics.state_change_count += 1;
+        state.metrics.time_eating += Instant::elapsed(&state.last_state_change);
+        state.last_state_change = Instant::now();
+
+        // schedule become hungry after the thinking time has elapsed
+        let _ = myself.send_after(
+            self.time_slice,
+            PhilosopherMessage::BecomeHungry(state.metrics.state_change_count),
+        );
+    }
+
+    /// Helper command to set the internal state to begin eating
+    fn begin_eating(&self, myself: &ActorRef<Self>, state: &mut PhilosopherState) {
+        state.metrics.time_hungry += Instant::elapsed(&state.last_state_change);
+        state.last_state_change = Instant::now();
+        state.mode = PhilosopherMode::Eating;
+        state.metrics.state_change_count += 1;
+
+        // Now that we are eating we will tell the fork that we are using it,
+        // thus marking the fork as dirty.
+        let _ = state
+            .left
+            .fork
+            .cast(ForkMessage::UsingFork(myself.get_id()));
+        let _ = state
+            .right
+            .fork
+            .cast(ForkMessage::UsingFork(myself.get_id()));
+
+        // schedule stop eating after the eating time has elapsed
+        let _ = myself.send_after(
+            self.time_slice,
+            PhilosopherMessage::StopEating(state.metrics.state_change_count),
+        );
+    }
+
+    /// Helper command to request any forks which are missing
+    fn request_missing_forks(&self, myself: &ActorRef<Self>, state: &mut PhilosopherState) {
+        if !state.left.has && !state.left.requested {
+            state.left.requested = true;
+            let _ = state
+                .left
+                .fork
+                .cast(ForkMessage::RequestFork(myself.clone()));
+        }
+        if !state.right.has && !state.right.requested {
+            state.right.requested = true;
+            let _ = state
+                .right
+                .fork
+                .cast(ForkMessage::RequestFork(myself.clone()));
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl ActorHandler for Philosopher {
+    type Msg = PhilosopherMessage;
+    type State = PhilosopherState;
+    async fn pre_start(&self, myself: ActorRef<Self>) -> Self::State {
+        // initialize the simulation by making the philosopher's hungry
+        let _ = myself.cast(Self::Msg::BecomeHungry(0));
+        Self::State::new(self.left.clone(), self.right.clone())
+    }
+
+    async fn handle(&self, myself: ActorRef<Self>, message: Self::Msg, state: &mut Self::State) {
+        match message {
+            PhilosopherMessage::SendMetrics(reply) => {
+                let _ = reply.send(state.metrics.clone());
+            }
+            PhilosopherMessage::StopEating(state_id) => {
+                // Processes a command to stop eating.
+                if state.metrics.state_change_count == state_id
+                    && state.mode == PhilosopherMode::Eating
+                {
+                    self.begin_thinking(&myself, state);
+                }
+            }
+            PhilosopherMessage::BecomeHungry(state_id) => {
+                // The philosopher is being instructed to get hungry which will cause them to ask for the
+                // forks to eat.
+                if state.metrics.state_change_count == state_id {
+                    if state.left.has && state.right.has {
+                        // we have both forks, starting eating
+                        self.begin_eating(&myself, state);
+                    } else {
+                        // we're missing some forks, maybe request the forks we need?
+                        match state.mode {
+                            PhilosopherMode::Thinking => {
+                                state.metrics.time_thinking +=
+                                    Instant::elapsed(&state.last_state_change);
+                                state.last_state_change = Instant::now();
+                                state.mode = PhilosopherMode::Hungry;
+                                state.metrics.state_change_count += 1;
+                                self.request_missing_forks(&myself, state);
+                            }
+                            PhilosopherMode::Hungry => {
+                                println!(
+                                    "ERROR: {} Got `BecomeHungry` while hungry!",
+                                    myself.get_name().unwrap()
+                                );
+                            }
+                            PhilosopherMode::Eating => {
+                                println!(
+                                    "ERROR: {} Got `BecomeHungry` while eating!",
+                                    myself.get_name().unwrap()
+                                );
+                            }
+                        }
+                    }
+                }
+            }
+            PhilosopherMessage::GiveUpFork(fork) => {
+                // Processes a command to a philosopher to give up a fork. Note that this can be received
+                // when the philosopher is in any state since the philosopher will not put down a fork
+                // unless he is asked to. A philosopher can be eating, stop eating and start thinking
+                // and then start eating again if no one asked for his forks. The fork actor is the only
+                // actor sending this message and it will only do so if the fork is dirty.
+                if state.left.fork.get_id() == fork {
+                    if state.left.has {
+                        state.left.has = false;
+                        let _ = state
+                            .left
+                            .fork
+                            .cast(ForkMessage::PutForkDown(myself.get_id()));
+                    }
+                } else if state.right.fork.get_id() == fork {
+                    if state.right.has {
+                        state.right.has = false;
+                        let _ = state
+                            .right
+                            .fork
+                            .cast(ForkMessage::PutForkDown(myself.get_id()));
+                    }
+                } else {
+                    println!(
+                        "ERROR: {} received a `GiveUpFork` from an unknown fork!",
+                        myself.get_name().unwrap()
+                    );
+                }
+                match state.mode {
+                    PhilosopherMode::Hungry => {
+                        state.metrics.failed_to_eat += 1;
+                        self.begin_thinking(&myself, state);
+                    }
+                    PhilosopherMode::Eating => {
+                        self.begin_thinking(&myself, state);
+                    }
+                    _ => {
+                        // already thinking
+                    }
+                }
+            }
+            PhilosopherMessage::ReceiveFork(fork) => {
+                // The philosopher received a fork. Once they have both forks they can start eating.
+                // Otherwise they have to wait for the other fork to begin eating.
+                if state.left.fork.get_id() == fork {
+                    state.left.has = true;
+                    state.left.requested = false;
+                } else if state.right.fork.get_id() == fork {
+                    state.right.has = true;
+                    state.right.requested = false;
+                } else {
+                    println!(
+                        "ERROR: {} received a `ReceiveFork` from an unknown fork!",
+                        myself.get_name().unwrap()
+                    );
+                }
+
+                // if we have both forks, we can start eating
+                if state.left.has && state.right.has {
+                    self.begin_eating(&myself, state);
+                }
+            }
+        }
+    }
+}
+
+#[tokio::main]
+async fn main() {
+    // TODO: move configuration to CLAP args
+    let time_slice = Duration::from_millis(10);
+    let run_time = Duration::from_secs(5);
+
+    let philosopher_names = vec![
+        "Confucius",
+        "Descartes",
+        "Benjamin Franklin",
+        "Socrates",
+        "Aristotle",
+        "Plato",
+        "John Locke",
+        "Nietzsche",
+        "Karl Marx",
+        "Pythagoras",
+        "Montesquieu",
+    ];
+    let mut forks = Vec::with_capacity(philosopher_names.len());
+    let mut philosophers = Vec::with_capacity(philosopher_names.len());
+    let mut all_handles = Vec::new();
+
+    let mut results: HashMap<ActorName, Option<PhilosopherMetrics>> =
+        HashMap::with_capacity(philosopher_names.len());
+
+    // create the forks
+    for _i in 0..philosopher_names.len() {
+        let (fork, handle) = Actor::spawn(None, Fork)
+            .await
+            .expect("Failed to create fork!");
+        forks.push(fork);
+        all_handles.push(handle);
+    }
+
+    // Spawn the philosopher actors clockwise from top of the table
+    for left in 0..philosopher_names.len() {
+        let right = if left == 0 {
+            philosopher_names.len() - 1
+        } else {
+            left - 1
+        };
+        let p = Philosopher {
+            time_slice,
+            left: forks[left].clone(),
+            right: forks[right].clone(),
+        };
+        let (philosopher, handle) = Actor::spawn(Some(philosopher_names[left]), p)
+            .await
+            .expect("Failed to create philosopher!");
+        results.insert(philosopher_names[left], None);
+        philosophers.push(philosopher);
+        all_handles.push(handle);
+    }
+
+    // wait for the simulation to end
+    tokio::time::sleep(run_time).await;
+    // collect the metrics from the philosophers, and they'll stop after reporting metrics
+    for philosopher in philosophers.iter() {
+        let metrics = philosopher
+            .call(
+                PhilosopherMessage::SendMetrics,
+                Some(Duration::from_millis(50)),
+            )
+            .await
+            .expect("Failed to send message to philosopher")
+            .expect("Call result to philosopher wasn't successful");
+        results.insert(philosopher.get_name().unwrap(), Some(metrics));
+    }
+
+    // cleanup forks & philosophers
+    for fork in forks {
+        fork.stop(None);
+    }
+    for philosopher in philosophers {
+        philosopher.stop(None);
+    }
+
+    // wait for everything to shut down
+    let _ = futures::future::join_all(all_handles).await;
+
+    // print metrics
+    println!("Simulation results");
+    for (who, metric) in results {
+        println!("{}: {:?}", who, metric);
+    }
+}
diff --git a/ractor/src/actor/mod.rs b/ractor/src/actor/mod.rs
index cf73edc..bacd884 100644
--- a/ractor/src/actor/mod.rs
+++ b/ractor/src/actor/mod.rs
@@ -6,8 +6,8 @@
 //! This module contains the basic building blocks of an actor.
 //!
 //! They are:
-//! [ActorHandler] : The internal logic for how an actor behaves
-//! [Actor] : Management structure processing the message handler, signals, and supervision events in a loop
+//! [ActorHandler]: The behavior definition for an actor's internal processing logic + state management
+//! [Actor]: Management structure processing the message handler, signals, and supervision events in a loop
 
 use std::{panic::AssertUnwindSafe, sync::Arc};
 
diff --git a/ractor/src/lib.rs b/ractor/src/lib.rs
index 0819e35..6d074be 100644
--- a/ractor/src/lib.rs
+++ b/ractor/src/lib.rs
@@ -137,10 +137,10 @@
 
 use std::any::Any;
 
-/// An actor's `atom()` similar name
+/// An actor's name, equivalent to an [Erlang `atom()`](https://www.erlang.org/doc/reference_manual/data_types.html#atom)
 pub type ActorName = &'static str;
 
-/// A process group's name, equivalent to an Erlang `atom()`
+/// A process group's name, equivalent to an [Erlang `atom()`](https://www.erlang.org/doc/reference_manual/data_types.html#atom)
 pub type GroupName = &'static str;
 
 pub mod actor;
@@ -153,6 +153,8 @@ pub mod time;
 
 #[cfg(test)]
 use criterion as _;
+#[cfg(test)]
+use rand as _;
 
 // WIP
 // #[cfg(feature = "remote")]
@@ -184,51 +186,6 @@ pub trait Message: Any + Send + 'static {}
 impl<T: Any + Send + 'static> Message for T {}
 
 /// Represents the state of an actor. Must be safe
-/// to send between threads
+/// to send between threads (same bounds as a [Message])
 pub trait State: Message {}
 impl<T: Message> State for T {}
-
-/// Error types which can result from Ractor processes
-#[derive(Debug)]
-pub enum RactorErr {
-    /// An error occurred spawning
-    Spawn(SpawnErr),
-    /// An error occurred in messaging (sending/receiving)
-    Messaging(MessagingErr),
-    /// An actor encountered an error while processing (canceled or panicked)
-    Actor(ActorErr),
-}
-
-impl From<SpawnErr> for RactorErr {
-    fn from(value: SpawnErr) -> Self {
-        RactorErr::Spawn(value)
-    }
-}
-
-impl From<MessagingErr> for RactorErr {
-    fn from(value: MessagingErr) -> Self {
-        RactorErr::Messaging(value)
-    }
-}
-
-impl From<ActorErr> for RactorErr {
-    fn from(value: ActorErr) -> Self {
-        RactorErr::Actor(value)
-    }
-}
-
-impl std::fmt::Display for RactorErr {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match self {
-            Self::Actor(actor_err) => {
-                write!(f, "{}", actor_err)
-            }
-            Self::Messaging(messaging_err) => {
-                write!(f, "{}", messaging_err)
-            }
-            Self::Spawn(spawn_err) => {
-                write!(f, "{}", spawn_err)
-            }
-        }
-    }
-}
diff --git a/ractor/src/time/mod.rs b/ractor/src/time/mod.rs
index b789f92..834f79e 100644
--- a/ractor/src/time/mod.rs
+++ b/ractor/src/time/mod.rs
@@ -61,18 +61,17 @@ where
 /// Returns: The [JoinHandle<Result<(), MessagingErr>>] which represents the backgrounded work.
 /// Awaiting the handle will yield the result of the send operation. Can be safely ignored to
 /// "fire and forget"
-pub fn send_after<TActor, F>(
+pub fn send_after<TActor>(
     period: Duration,
     actor: ActorCell,
-    msg: F,
+    msg: TActor::Msg,
 ) -> JoinHandle<Result<(), MessagingErr>>
 where
     TActor: ActorHandler,
-    F: Fn() -> TActor::Msg + Send + 'static,
 {
     tokio::spawn(async move {
         tokio::time::sleep(period).await;
-        actor.send_message::<TActor>(msg())
+        actor.send_message::<TActor>(msg)
     })
 }
 
@@ -120,11 +119,12 @@ where
     }
 
     /// Alias of [send_after]
-    pub fn send_after<F>(&self, period: Duration, msg: F) -> JoinHandle<Result<(), MessagingErr>>
-    where
-        F: Fn() -> TActor::Msg + Send + 'static,
-    {
-        send_after::<TActor, F>(period, self.get_cell(), msg)
+    pub fn send_after(
+        &self,
+        period: Duration,
+        msg: TActor::Msg,
+    ) -> JoinHandle<Result<(), MessagingErr>> {
+        send_after::<TActor>(period, self.get_cell(), msg)
     }
 
     /// Alias of [exit_after]


Test patch:
diff --git a/ractor/src/time/tests.rs b/ractor/src/time/tests.rs
index 8512c63..3dcd784 100644
--- a/ractor/src/time/tests.rs
+++ b/ractor/src/time/tests.rs
@@ -108,10 +108,10 @@ async fn test_send_after() {
     .await
     .expect("Failed to create test actor");
 
-    let send_after_handle = crate::time::send_after::<TestActor, _>(
+    let send_after_handle = crate::time::send_after::<TestActor>(
         Duration::from_millis(10),
         actor_ref.clone().into(),
-        || (),
+        (),
     );
 
     // even though the timer is started, we should be in a "pause" state for 10ms,