[hyperactor] make all variants of spawn sync

Pull Request resolved: #1968

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.
ghstack-source-id: 325005475

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/)!

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>),
@@ -1531,11 +1520,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.
@@ -2273,7 +2259,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(())
         }
@@ -2283,7 +2269,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(
@@ -2334,11 +2320,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);
@@ -2378,12 +2362,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!(
@@ -2444,7 +2426,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;
@@ -2515,7 +2496,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;
@@ -2539,7 +2519,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;
@@ -2580,10 +2559,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();
@@ -2597,10 +2573,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();
@@ -2837,7 +2811,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();
@@ -2861,10 +2835,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
@@ -2888,6 +2859,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);
 
@@ -2936,7 +2909,6 @@ mod tests {
 
         let root = proc
             .spawn::<TestActor>("root", TestActor(root_state.clone(), false))
-            .await
             .unwrap();
         let root_1 = proc
             .spawn_child::<TestActor>(
@@ -2946,32 +2918,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
@@ -3023,7 +2990,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
@@ -3054,10 +3021,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
@@ -3152,7 +3116,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)
     }