[hyperactor] make all variants of spawn sync

Building on the previous diff, separating out remote from local instantiation, this diff implements synchronous spawns throughout. This means we can always spawn an actor in a nonblocking way, regardless of context.

Spawns should also become infallible, instead relying on supervision to handle errors.

Differential Revision: [D87608082](https://our.internmc.facebook.com/intern/diff/D87608082/)

**NOTE FOR REVIEWERS**: This PR has internal Meta-specific changes or comments, please review them on [Phabricator](https://our.internmc.facebook.com/intern/diff/D87608082/)!

ghstack-source-id: 324888978
Pull Request resolved: #1968

Author: mariusae

hyperactor/src/actor.rs

@@ -98,8 +98,8 @@ pub trait Actor: Sized + Send + Debug + 'static {
 
     /// Spawn a child actor, given a spawning capability (usually given by [`Instance`]).
     /// The spawned actor will be supervised by the parent (spawning) actor.
-    async fn spawn(self, cx: &impl context::Actor) -> anyhow::Result<ActorHandle<Self>> {
-        cx.instance().spawn(self).await
+    fn spawn(self, cx: &impl context::Actor) -> anyhow::Result<ActorHandle<Self>> {
+        cx.instance().spawn(self)
     }
 
     /// Spawns this actor in a detached state, handling its messages
@@ -108,8 +108,8 @@ pub trait Actor: Sized + Send + Debug + 'static {
     ///
     /// Actors spawned through `spawn_detached` are not attached to a supervision
     /// hierarchy, and not managed by a [`Proc`].
-    async fn spawn_detached(self) -> Result<ActorHandle<Self>, anyhow::Error> {
-        Proc::local().spawn("anon", self).await
+    fn spawn_detached(self) -> Result<ActorHandle<Self>, anyhow::Error> {
+        Proc::local().spawn("anon", self)
     }
 
     /// This method is used by the runtime to spawn the actor server. It can be
@@ -260,7 +260,7 @@ pub trait RemoteSpawn: Actor + Referable + Binds<Self> {
         Box::pin(async move {
             let params = bincode::deserialize(&serialized_params)?;
             let actor = Self::new(params).await?;
-            let handle = proc.spawn(&name, actor).await?;
+            let handle = proc.spawn(&name, actor)?;
             // We return only the ActorId, not a typed ActorRef.
             // Callers that hold this ID can interact with the actor
             // only via the serialized/opaque messaging path, which
@@ -792,7 +792,7 @@ mod tests {
         let client = proc.attach("client").unwrap();
         let (tx, mut rx) = client.open_port();
         let actor = EchoActor(tx.bind());
-        let handle = proc.spawn::<EchoActor>("echo", actor).await.unwrap();
+        let handle = proc.spawn::<EchoActor>("echo", actor).unwrap();
         handle.send(123u64).unwrap();
         handle.drain_and_stop().unwrap();
         handle.await;
@@ -808,14 +808,8 @@ mod tests {
 
         let ping_actor = PingPongActor::new(Some(undeliverable_msg_tx.bind()), None, None);
         let pong_actor = PingPongActor::new(Some(undeliverable_msg_tx.bind()), None, None);
-        let ping_handle = proc
-            .spawn::<PingPongActor>("ping", ping_actor)
-            .await
-            .unwrap();
-        let pong_handle = proc
-            .spawn::<PingPongActor>("pong", pong_actor)
-            .await
-            .unwrap();
+        let ping_handle = proc.spawn::<PingPongActor>("ping", ping_actor).unwrap();
+        let pong_handle = proc.spawn::<PingPongActor>("pong", pong_actor).unwrap();
 
         let (local_port, local_receiver) = client.open_once_port();
 
@@ -842,14 +836,8 @@ mod tests {
             PingPongActor::new(Some(undeliverable_msg_tx.bind()), Some(error_ttl), None);
         let pong_actor =
             PingPongActor::new(Some(undeliverable_msg_tx.bind()), Some(error_ttl), None);
-        let ping_handle = proc
-            .spawn::<PingPongActor>("ping", ping_actor)
-            .await
-            .unwrap();
-        let pong_handle = proc
-            .spawn::<PingPongActor>("pong", pong_actor)
-            .await
-            .unwrap();
+        let ping_handle = proc.spawn::<PingPongActor>("ping", ping_actor).unwrap();
+        let pong_handle = proc.spawn::<PingPongActor>("pong", pong_actor).unwrap();
 
         let (local_port, local_receiver) = client.open_once_port();
 
@@ -895,7 +883,7 @@ mod tests {
     async fn test_init() {
         let proc = Proc::local();
         let actor = InitActor(false);
-        let handle = proc.spawn::<InitActor>("init", actor).await.unwrap();
+        let handle = proc.spawn::<InitActor>("init", actor).unwrap();
         let client = proc.attach("client").unwrap();
 
         let (port, receiver) = client.open_once_port();
@@ -954,7 +942,7 @@ mod tests {
             let proc = Proc::local();
             let values: MultiValues = Arc::new(Mutex::new((0, "".to_string())));
             let actor = MultiActor(values.clone());
-            let handle = proc.spawn::<MultiActor>("myactor", actor).await.unwrap();
+            let handle = proc.spawn::<MultiActor>("myactor", actor).unwrap();
             let (client, client_handle) = proc.instance("client").unwrap();
             Self {
                 proc,
@@ -1072,7 +1060,7 @@ mod tests {
         // Just test that we can round-trip the handle through a downcast.
 
         let proc = Proc::local();
-        let handle = proc.spawn("nothing", NothingActor).await.unwrap();
+        let handle = proc.spawn("nothing", NothingActor).unwrap();
         let cell = handle.cell();
 
         // Invalid actor doesn't succeed.

hyperactor/src/host.rs

@@ -1220,7 +1220,7 @@ mod tests {
     #[tokio::test]
     async fn test_basic() {
         let proc_manager =
-            LocalProcManager::new(|proc: Proc| async move { proc.spawn::<()>("agent", ()).await });
+            LocalProcManager::new(|proc: Proc| async move { proc.spawn::<()>("agent", ()) });
         let procs = Arc::clone(&proc_manager.procs);
         let (mut host, _handle) =
             Host::serve(proc_manager, ChannelAddr::any(ChannelTransport::Local))

hyperactor/src/mailbox.rs

@@ -3125,7 +3125,7 @@ mod tests {
         let mut proc = Proc::new(proc_id.clone(), proc_forwarder);
         ProcSupervisionCoordinator::set(&proc).await.unwrap();
 
-        let foo = proc.spawn("foo", Foo).await.unwrap();
+        let foo = proc.spawn("foo", Foo).unwrap();
         let return_handle = foo.port::<Undeliverable<MessageEnvelope>>();
         let message = MessageEnvelope::new(
             foo.actor_id().clone(),

hyperactor/src/proc.rs

@@ -507,12 +507,7 @@ impl Proc {
 
     /// Spawn a named (root) actor on this proc. The name of the actor must be
     /// unique.
-    #[hyperactor::observe_result("Proc")]
-    pub async fn spawn<A: Actor>(
-        &self,
-        name: &str,
-        actor: A,
-    ) -> Result<ActorHandle<A>, anyhow::Error> {
+    pub fn spawn<A: Actor>(&self, name: &str, actor: A) -> Result<ActorHandle<A>, anyhow::Error> {
         let actor_id = self.allocate_root_id(name)?;
         let span = tracing::span!(
             Level::INFO,
@@ -532,10 +527,7 @@ impl Proc {
             .ledger
             .insert(actor_id.clone(), instance.inner.cell.downgrade())?;
 
-        Ok(instance
-            .start(actor, actor_loop_receivers.take().unwrap(), work_rx)
-            .instrument(span)
-            .await)
+        Ok(instance.start(actor, actor_loop_receivers.take().unwrap(), work_rx))
     }
 
     /// Create and return an actor instance and its corresponding handle. This allows actors to be
@@ -584,17 +576,15 @@ impl Proc {
     ///
     /// When spawn_child returns, the child has an associated cell and is linked
     /// with its parent.
-    async fn spawn_child<A: Actor>(
+    fn spawn_child<A: Actor>(
         &self,
         parent: InstanceCell,
         actor: A,
     ) -> Result<ActorHandle<A>, anyhow::Error> {
         let actor_id = self.allocate_child_id(parent.actor_id())?;
         let (instance, mut actor_loop_receivers, work_rx) =
             Instance::new(self.clone(), actor_id, false, Some(parent.clone()));
-        Ok(instance
-            .start(actor, actor_loop_receivers.take().unwrap(), work_rx)
-            .await)
+        Ok(instance.start(actor, actor_loop_receivers.take().unwrap(), work_rx))
     }
 
     /// Call `abort` on the `JoinHandle` associated with the given
@@ -1166,8 +1156,7 @@ impl<A: Actor> Instance<A> {
 
     /// Start an A-typed actor onto this instance with the provided params. When spawn returns,
     /// the actor has been linked with its parent, if it has one.
-    #[hyperactor::instrument_infallible(fields(actor_id=self.inner.cell.actor_id().clone().to_string(), actor_name=self.inner.cell.actor_id().name()))]
-    async fn start(
+    fn start(
         self,
         actor: A,
         actor_loop_receivers: (PortReceiver<Signal>, PortReceiver<ActorSupervisionEvent>),
@@ -1540,11 +1529,8 @@ impl<A: Actor> Instance<A> {
     }
 
     /// Spawn on child on this instance. Currently used only by cap::CanSpawn.
-    pub(crate) async fn spawn<C: Actor>(&self, actor: C) -> anyhow::Result<ActorHandle<C>> {
-        self.inner
-            .proc
-            .spawn_child(self.inner.cell.clone(), actor)
-            .await
+    pub(crate) fn spawn<C: Actor>(&self, actor: C) -> anyhow::Result<ActorHandle<C>> {
+        self.inner.proc.spawn_child(self.inner.cell.clone(), actor)
     }
 
     /// Create a new direct child instance.
@@ -2282,7 +2268,7 @@ mod tests {
             cx: &crate::Context<Self>,
             reply: oneshot::Sender<ActorHandle<TestActor>>,
         ) -> Result<(), anyhow::Error> {
-            let handle = TestActor::default().spawn(cx).await?;
+            let handle = TestActor::default().spawn(cx)?;
             reply.send(handle).unwrap();
             Ok(())
         }
@@ -2292,7 +2278,7 @@ mod tests {
     #[async_timed_test(timeout_secs = 30)]
     async fn test_spawn_actor() {
         let proc = Proc::local();
-        let handle = proc.spawn("test", TestActor::default()).await.unwrap();
+        let handle = proc.spawn("test", TestActor::default()).unwrap();
 
         // Check on the join handle.
         assert!(logs_contain(
@@ -2343,11 +2329,9 @@ mod tests {
         let proc = Proc::local();
         let first = proc
             .spawn::<TestActor>("first", TestActor::default())
-            .await
             .unwrap();
         let second = proc
             .spawn::<TestActor>("second", TestActor::default())
-            .await
             .unwrap();
         let (tx, rx) = oneshot::channel::<()>();
         let reply_message = TestActorMessage::Reply(tx);
@@ -2387,12 +2371,10 @@ mod tests {
 
         let target_actor = proc
             .spawn::<TestActor>("target", TestActor::default())
-            .await
             .unwrap();
         let target_actor_ref = target_actor.bind();
         let lookup_actor = proc
             .spawn::<LookupTestActor>("lookup", LookupTestActor::default())
-            .await
             .unwrap();
 
         assert!(
@@ -2453,7 +2435,6 @@ mod tests {
 
         let first = proc
             .spawn::<TestActor>("first", TestActor::default())
-            .await
             .unwrap();
         let second = TestActor::spawn_child(&first).await;
         let third = TestActor::spawn_child(&second).await;
@@ -2524,7 +2505,6 @@ mod tests {
 
         let root = proc
             .spawn::<TestActor>("root", TestActor::default())
-            .await
             .unwrap();
         let root_1 = TestActor::spawn_child(&root).await;
         let root_2 = TestActor::spawn_child(&root).await;
@@ -2548,7 +2528,6 @@ mod tests {
 
         let root = proc
             .spawn::<TestActor>("root", TestActor::default())
-            .await
             .unwrap();
         let root_1 = TestActor::spawn_child(&root).await;
         let root_2 = TestActor::spawn_child(&root).await;
@@ -2589,10 +2568,7 @@ mod tests {
         let proc = Proc::local();
 
         // Add the 1st root. This root will remain active until the end of the test.
-        let root: ActorHandle<TestActor> = proc
-            .spawn::<TestActor>("root", TestActor::default())
-            .await
-            .unwrap();
+        let root: ActorHandle<TestActor> = proc.spawn("root", TestActor::default()).unwrap();
         wait_until_idle(&root).await;
         {
             let snapshot = proc.state().ledger.snapshot();
@@ -2606,10 +2582,8 @@ mod tests {
         }
 
         // Add the 2nd root.
-        let another_root: ActorHandle<TestActor> = proc
-            .spawn::<TestActor>("another_root", TestActor::default())
-            .await
-            .unwrap();
+        let another_root: ActorHandle<TestActor> =
+            proc.spawn("another_root", TestActor::default()).unwrap();
         wait_until_idle(&another_root).await;
         {
             let snapshot = proc.state().ledger.snapshot();
@@ -2844,7 +2818,7 @@ mod tests {
         let proc = Proc::local();
         let state = Arc::new(AtomicUsize::new(0));
         let actor = TestActor(state.clone());
-        let handle = proc.spawn::<TestActor>("test", actor).await.unwrap();
+        let handle = proc.spawn::<TestActor>("test", actor).unwrap();
         let client = proc.attach("client").unwrap();
         let (tx, rx) = client.open_once_port();
         handle.send(tx).unwrap();
@@ -2868,10 +2842,7 @@ mod tests {
         ProcSupervisionCoordinator::set(&proc).await.unwrap();
 
         let (client, _handle) = proc.instance("client").unwrap();
-        let actor_handle = proc
-            .spawn::<TestActor>("test", TestActor::default())
-            .await
-            .unwrap();
+        let actor_handle = proc.spawn("test", TestActor::default()).unwrap();
         actor_handle
             .panic(&client, "some random failure".to_string())
             .await
@@ -2895,6 +2866,8 @@ mod tests {
 
     #[async_timed_test(timeout_secs = 30)]
     async fn test_local_supervision_propagation() {
+        hyperactor_telemetry::initialize_logging_for_test();
+
         #[derive(Debug)]
         struct TestActor(Arc<AtomicBool>, bool);
 
@@ -2943,7 +2916,6 @@ mod tests {
 
         let root = proc
             .spawn::<TestActor>("root", TestActor(root_state.clone(), false))
-            .await
             .unwrap();
         let root_1 = proc
             .spawn_child::<TestActor>(
@@ -2953,32 +2925,27 @@ mod tests {
                     true, /* set true so children's event stops here */
                 ),
             )
-            .await
             .unwrap();
         let root_1_1 = proc
             .spawn_child::<TestActor>(
                 root_1.cell().clone(),
                 TestActor(root_1_1_state.clone(), false),
             )
-            .await
             .unwrap();
         let root_1_1_1 = proc
             .spawn_child::<TestActor>(
                 root_1_1.cell().clone(),
                 TestActor(root_1_1_1_state.clone(), false),
             )
-            .await
             .unwrap();
         let root_2 = proc
             .spawn_child::<TestActor>(root.cell().clone(), TestActor(root_2_state.clone(), false))
-            .await
             .unwrap();
         let root_2_1 = proc
             .spawn_child::<TestActor>(
                 root_2.cell().clone(),
                 TestActor(root_2_1_state.clone(), false),
             )
-            .await
             .unwrap();
 
         // fail `root_1_1_1`, the supervision msg should be propagated to
@@ -3030,7 +2997,7 @@ mod tests {
 
         let (instance, handle) = proc.instance("my_test_actor").unwrap();
 
-        let child_actor = TestActor::default().spawn(&instance).await.unwrap();
+        let child_actor = TestActor::default().spawn(&instance).unwrap();
 
         let (port, mut receiver) = instance.open_port();
         child_actor
@@ -3061,10 +3028,7 @@ mod tests {
             // Intentionally not setting a proc supervison coordinator. This
             // should cause the process to terminate.
             // ProcSupervisionCoordinator::set(&proc).await.unwrap();
-            let root = proc
-                .spawn::<TestActor>("root", TestActor::default())
-                .await
-                .unwrap();
+            let root = proc.spawn("root", TestActor::default()).unwrap();
             let (client, _handle) = proc.instance("client").unwrap();
             root.fail(&client, anyhow::anyhow!("some random failure"))
                 .await
@@ -3159,7 +3123,7 @@ mod tests {
         }
 
         trace_and_block(async {
-            let handle = LoggingActor::default().spawn_detached().await.unwrap();
+            let handle = LoggingActor::default().spawn_detached().unwrap();
             handle.send("hello world".to_string()).unwrap();
             handle.send("hello world again".to_string()).unwrap();
             handle.send(123u64).unwrap();

hyperactor/src/test_utils/proc_supervison.rs

@@ -43,9 +43,7 @@ impl ProcSupervisionCoordinator {
     pub async fn set(proc: &Proc) -> Result<ReportedEvent, anyhow::Error> {
         let state = ReportedEvent::new();
         let actor = ProcSupervisionCoordinator(state.clone());
-        let coordinator = proc
-            .spawn::<ProcSupervisionCoordinator>("coordinator", actor)
-            .await?;
+        let coordinator = proc.spawn::<ProcSupervisionCoordinator>("coordinator", actor)?;
         proc.set_supervision_coordinator(coordinator.port())?;
         Ok(state)
     }