simple pub/sub example (#838)

Summary:
Pull Request resolved: #838

This is a simple example to demonstrate how to implement pub/sub channels in Monarch.

This example does not yet handle unsubscribes (through undeliverable message notifications). This will come as a follow-up.
ghstack-source-id: 302711477

Reviewed By: highker

Differential Revision: D80140121

fbshipit-source-id: d778bcac0c7bafb2f54670629127e6fff7af71cc

Author: mariusae

hyperactor/Cargo.toml

@@ -1,4 +1,4 @@
-# @generated by autocargo from //monarch/hyperactor:[channel_benchmarks,hyperactor,hyperactor-example-derive]
+# @generated by autocargo from //monarch/hyperactor:[channel_benchmarks,hyperactor,hyperactor-example-derive,hyperactor-example-stream]
 
 [package]
 name = "hyperactor"
@@ -17,6 +17,10 @@ path = "benches/main.rs"
 name = "hyperactor_example_derive"
 path = "example/derive.rs"
 
+[[bin]]
+name = "hyperactor_example_stream"
+path = "example/stream.rs"
+
 [dependencies]
 anyhow = "1.0.98"
 async-trait = "0.1.86"

hyperactor/example/stream.rs

@@ -0,0 +1,95 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ * All rights reserved.
+ *
+ * This source code is licensed under the BSD-style license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+//! This example implements a basic pub-sub pattern in Hyperactor.
+
+use std::time::Duration;
+
+use async_trait::async_trait;
+use hyperactor::Actor;
+use hyperactor::ActorHandle;
+use hyperactor::Context;
+use hyperactor::Handler;
+use hyperactor::Instance;
+use hyperactor::Named;
+use hyperactor::PortRef;
+use hyperactor::proc::Proc;
+use serde::Deserialize;
+use serde::Serialize;
+
+#[derive(Debug, Actor, Default)]
+struct CounterActor {
+    subscribers: Vec<PortRef<u64>>,
+    n: u64,
+}
+
+#[derive(Serialize, Deserialize, Debug, Named)]
+struct Subscribe(PortRef<u64>);
+
+#[async_trait]
+impl Handler<Subscribe> for CounterActor {
+    async fn handle(
+        &mut self,
+        cx: &Context<Self>,
+        subscriber: Subscribe,
+    ) -> Result<(), anyhow::Error> {
+        self.subscribers.push(subscriber.0);
+        for port in &self.subscribers {
+            port.send(cx, self.n)?;
+        }
+        self.n += 1;
+        Ok(())
+    }
+}
+
+#[derive(Debug)]
+struct CountClient {
+    counter: PortRef<Subscribe>,
+}
+
+#[async_trait]
+impl Actor for CountClient {
+    // Where to send subscribe messages.
+    type Params = PortRef<Subscribe>;
+
+    async fn new(counter: PortRef<Subscribe>) -> Result<Self, anyhow::Error> {
+        Ok(Self { counter })
+    }
+
+    async fn init(&mut self, this: &Instance<Self>) -> Result<(), anyhow::Error> {
+        // Subscribe to the counter on initialization. We give it our u64 port to report
+        // messages back to.
+        self.counter.send(this, Subscribe(this.port().bind()))?;
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl Handler<u64> for CountClient {
+    async fn handle(&mut self, cx: &Context<Self>, count: u64) -> Result<(), anyhow::Error> {
+        eprintln!("{}: count: {}", cx.self_id(), count);
+        Ok(())
+    }
+}
+
+#[tokio::main]
+async fn main() {
+    let proc = Proc::local();
+
+    let counter_actor: ActorHandle<CounterActor> = proc.spawn("counter", ()).await.unwrap();
+
+    for i in 0..10 {
+        // Spawn new "countees". Every time each subscribes, the counter broadcasts
+        // the count to everyone.
+        let _countee_actor: ActorHandle<CountClient> = proc
+            .spawn(&format!("countee_{}", i), counter_actor.port().bind())
+            .await
+            .unwrap();
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+}