Adding CI workflow for code-coverage along with benchmarks

This PR adds benchmarks for
1. Actor spawning
2. Actor timing to handle first message + shutdown
3. Actor timing to handle 100K messages

Additionally we're adding new CI for (1) code coverage and (2) running the benchmarks automatically

.cargo.config

@@ -0,0 +1,2 @@
+[alias]
+xtask = "run --package xtask --"

.github/workflows/ci.yaml

@@ -88,4 +88,21 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: doc
-          args: --lib -r
+          args: --lib -r
+
+  benches:
+    name: benches
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@main
+      - name: Install rust
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: stable
+          override: true
+
+      - name: Run benches
+        uses: actions-rs/cargo@v1
+        with:
+          command: bench
+

.github/workflows/coverage.yml

@@ -0,0 +1,45 @@
+name: Code Coverage
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    types: [opened, reopened, synchronize]
+
+jobs:
+  coverage:
+    name: Coverage using xtask
+    strategy:
+      matrix:
+        os: [ubuntu-latest]
+        rust: [stable]
+    runs-on: ${{ matrix.os }}
+    steps:
+      - name: Checkout sources
+        uses: actions/checkout@v2
+
+      - name: Install stable toolchain
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: ${{ matrix.rust }}
+          override: true
+          components: llvm-tools-preview
+
+      - uses: Swatinem/rust-cache@v1
+
+      - name: Download grcov
+        run: |
+          mkdir -p "${HOME}/.local/bin"
+          curl -sL https://github.com/mozilla/grcov/releases/download/v0.8.10/grcov-x86_64-unknown-linux-gnu.tar.bz2 | tar jxf - -C "${HOME}/.local/bin"
+          echo "$HOME/.local/bin" >> $GITHUB_PATH
+      - name: Run xtask coverage
+        uses: actions-rs/cargo@v1
+        with:
+          command: run
+          args: --bin xtask coverage
+
+
+      - name: Upload to codecov.io
+        uses: codecov/codecov-action@v3
+        with:
+          files: coverage/*.lcov

.github/workflows/publish.yaml

@@ -25,4 +25,5 @@ jobs:
     - name: Publish crate ractor
       run: cargo publish --manifest-path Cargo.toml -p ractor
       env:
-        CARGO_REGISTRY_TOKEN: ${{ secrets.crates_io_token }}
+        CARGO_REGISTRY_TOKEN: ${{ secrets.crates_io_token }}
+

.gitignore

@@ -10,4 +10,9 @@ Cargo.lock
 **/*.rs.bk
 
 # Remove all target compilation folders from all sub-folders as well
-**/target/
+**/target/
+
+# Remove code-coverage generated files from git
+debug/
+coverage/
+**/*.profraw

Cargo.toml

@@ -2,5 +2,6 @@
 
 members = [
     "ractor",
-    "ractor-playground"
+    "ractor-playground",
+    "xtask"
 ]

ractor-playground/Cargo.toml

@@ -15,4 +15,4 @@ publish = false
 async-trait = "0.1"
 ractor = { path="../ractor" }
 rustyrepl = "0.1"
-tokio = { version = "1.23", features = ["full"]}
+tokio = { version = "1.23", features = ["full"]}

ractor/Cargo.toml

@@ -20,5 +20,11 @@ once_cell = "1"
 tokio = { version = "1.23", features = ["rt", "time", "sync", "macros"]}
 
 [dev-dependencies]
+criterion = "0.3"
 function_name = "0.3"
-tokio = { version = "1.23", features = ["rt", "time", "sync", "macros", "rt-multi-thread"] }
+tokio = { version = "1.23", features = ["rt", "time", "sync", "macros", "rt-multi-thread"] }
+
+[[bench]]
+name = "actor"
+harness = false
+required-features = []

ractor/benches/actor.rs

@@ -0,0 +1,188 @@
+// 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.
+
+#[macro_use]
+extern crate criterion;
+
+use criterion::{BatchSize, Criterion};
+use ractor::{Actor, ActorHandler, ActorRef};
+
+struct BenchActor;
+
+#[async_trait::async_trait]
+impl ActorHandler for BenchActor {
+    type Msg = ();
+
+    type State = ();
+
+    async fn pre_start(&self, myself: ActorRef<Self>) -> Self::State {
+        let _ = myself.cast(());
+    }
+
+    async fn handle(&self, myself: ActorRef<Self>, _message: Self::Msg, _state: &mut Self::State) {
+        myself.stop(None);
+    }
+}
+
+fn create_actors(c: &mut Criterion) {
+    let small = 100;
+    let large = 10000;
+
+    let id = format!("Creation of {} actors", small);
+    let runtime = tokio::runtime::Builder::new_multi_thread().build().unwrap();
+    c.bench_function(&id, move |b| {
+        b.iter_batched(
+            || {},
+            |()| {
+                runtime.block_on(async move {
+                    let mut handles = vec![];
+                    for _ in 0..small {
+                        let (_, handler) = Actor::spawn(None, BenchActor)
+                            .await
+                            .expect("Failed to create test agent");
+                        handles.push(handler);
+                    }
+                    handles
+                })
+            },
+            BatchSize::PerIteration,
+        );
+    });
+
+    let id = format!("Creation of {} actors", large);
+    let runtime = tokio::runtime::Builder::new_multi_thread().build().unwrap();
+    c.bench_function(&id, move |b| {
+        b.iter_batched(
+            || {},
+            |()| {
+                runtime.block_on(async move {
+                    let mut handles = vec![];
+                    for _ in 0..large {
+                        let (_, handler) = Actor::spawn(None, BenchActor)
+                            .await
+                            .expect("Failed to create test agent");
+                        handles.push(handler);
+                    }
+                    handles
+                })
+            },
+            BatchSize::PerIteration,
+        );
+    });
+}
+
+fn schedule_work(c: &mut Criterion) {
+    let small = 100;
+    let large = 1000;
+
+    let id = format!("Waiting on {} actors to process first message", small);
+    let runtime = tokio::runtime::Builder::new_multi_thread().build().unwrap();
+    c.bench_function(&id, move |b| {
+        b.iter_batched(
+            || {
+                runtime.block_on(async move {
+                    let mut handles = vec![];
+                    for _ in 0..small {
+                        let (_, handler) = Actor::spawn(None, BenchActor)
+                            .await
+                            .expect("Failed to create test agent");
+                        handles.push(handler);
+                    }
+                    handles
+                })
+            },
+            |handles| {
+                runtime.block_on(async move {
+                    let _ = futures::future::join_all(handles).await;
+                })
+            },
+            BatchSize::PerIteration,
+        );
+    });
+
+    let id = format!("Waiting on {} actors to process first message", large);
+    let runtime = tokio::runtime::Builder::new_multi_thread().build().unwrap();
+    c.bench_function(&id, move |b| {
+        b.iter_batched(
+            || {
+                runtime.block_on(async move {
+                    let mut handles = vec![];
+                    for _ in 0..large {
+                        let (_, handler) = Actor::spawn(None, BenchActor)
+                            .await
+                            .expect("Failed to create test agent");
+                        handles.push(handler);
+                    }
+                    handles
+                })
+            },
+            |handles| {
+                runtime.block_on(async move {
+                    let _ = futures::future::join_all(handles).await;
+                })
+            },
+            BatchSize::PerIteration,
+        );
+    });
+}
+
+#[allow(clippy::async_yields_async)]
+fn process_messages(c: &mut Criterion) {
+    const NUM_MSGS: u64 = 100000;
+
+    struct MessagingActor {
+        num_msgs: u64,
+    }
+
+    #[async_trait::async_trait]
+    impl ActorHandler for MessagingActor {
+        type Msg = ();
+
+        type State = u64;
+
+        async fn pre_start(&self, myself: ActorRef<Self>) -> Self::State {
+            let _ = myself.cast(());
+            0u64
+        }
+
+        async fn handle(
+            &self,
+            myself: ActorRef<Self>,
+            _message: Self::Msg,
+            state: &mut Self::State,
+        ) {
+            *state += 1;
+            if *state >= self.num_msgs {
+                myself.stop(None);
+            } else {
+                let _ = myself.cast(());
+            }
+        }
+    }
+
+    let id = format!("Waiting on {} messages to be processed", NUM_MSGS);
+    let runtime = tokio::runtime::Builder::new_multi_thread().build().unwrap();
+    c.bench_function(&id, move |b| {
+        b.iter_batched(
+            || {
+                runtime.block_on(async move {
+                    let (_, handle) = Actor::spawn(None, MessagingActor { num_msgs: NUM_MSGS })
+                        .await
+                        .expect("Failed to create test actor");
+                    handle
+                })
+            },
+            |handle| {
+                runtime.block_on(async move {
+                    let _ = handle.await;
+                })
+            },
+            BatchSize::PerIteration,
+        );
+    });
+}
+
+criterion_group!(actors, create_actors, schedule_work, process_messages);
+criterion_main!(actors);

ractor/examples/supervisor.rs

@@ -58,6 +58,7 @@ async fn main() {
 
 struct LeafActor;
 
+#[derive(Clone)]
 struct LeafActorState {}
 
 enum LeafActorMessage {
@@ -106,6 +107,7 @@ impl ActorHandler for LeafActor {
 
 struct MidLevelActor;
 
+#[derive(Clone)]
 struct MidLevelActorState {
     leaf_actor: ActorRef<LeafActor>,
 }
@@ -177,6 +179,7 @@ impl ActorHandler for MidLevelActor {
 
 struct RootActor;
 
+#[derive(Clone)]
 struct RootActorState {
     mid_level_actor: ActorRef<MidLevelActor>,
 }

ractor/src/actor/actor_cell/actor_ref.rs

@@ -97,8 +97,8 @@ where
     /// Notify the supervisors that a supervision event occurred
     ///
     /// * `evt` - The event to send to this [crate::Actor]'s supervisors
-    pub fn notify_supervisors(&self, evt: SupervisionEvent) {
-        self.inner.notify_supervisors::<TActor>(evt)
+    pub fn notify_supervisor(&self, evt: SupervisionEvent) {
+        self.inner.notify_supervisor::<TActor>(evt)
     }
 
     // ========================== General Actor Operation Aliases ========================== //

ractor/src/actor/actor_cell/mod.rs

@@ -226,23 +226,26 @@ impl ActorCell {
         }
 
         // notify children they should die. They will unlink themselves from the supervisor
-        self.inner.tree.terminate_children();
+        self.inner.tree.terminate_all_children();
     }
 
-    /// Link this [super::Actor] to the supervisor
+    /// Link this [super::Actor] to the provided supervisor
     ///
-    /// * `supervisor` - The supervisor to link this [super::Actor] to
+    /// * `supervisor` - The child to link this [super::Actor] to
     pub fn link(&self, supervisor: ActorCell) {
-        supervisor.inner.tree.insert_parent(self.clone());
-        self.inner.tree.insert_child(supervisor);
+        supervisor.inner.tree.insert_child(self.clone());
+        self.inner.tree.set_supervisor(supervisor);
     }
 
-    /// Unlink this [super::Actor] from the supervisor
+    /// Unlink this [super::Actor] from the supervisor if it's
+    /// currently linked (if self's supervisor is `supervisor`)
     ///
-    /// * `supervisor` - The supervisor to unlink this [super::Actor] from
+    /// * `supervisor` - The child to unlink this [super::Actor] from
     pub fn unlink(&self, supervisor: ActorCell) {
-        supervisor.inner.tree.remove_parent(self.clone());
-        self.inner.tree.remove_child(supervisor);
+        if self.inner.tree.is_child_of(supervisor.get_id()) {
+            supervisor.inner.tree.remove_child(self.get_id());
+            self.inner.tree.clear_supervisor();
+        }
     }
 
     /// Kill this [super::Actor] forcefully (terminates async work)
@@ -289,11 +292,11 @@ impl ActorCell {
     /// Notify the supervisors that a supervision event occurred
     ///
     /// * `evt` - The event to send to this [super::Actor]'s supervisors
-    pub fn notify_supervisors<TActor>(&self, evt: SupervisionEvent)
+    pub fn notify_supervisor<TActor>(&self, evt: SupervisionEvent)
     where
         TActor: ActorHandler,
     {
-        self.inner.tree.notify_supervisors::<TActor>(evt)
+        self.inner.tree.notify_supervisor::<TActor>(evt)
     }
 
     // ================== Test Utilities ================== //