Add cx: context::Actor to ActorHandle

Differential Revision: D83533433

Author: pzhan9

controller/src/lib.rs

@@ -1678,14 +1678,17 @@ mod tests {
         // Join the world.
         server_handle
             .system_actor_handle()
-            .send(SystemMessage::Join {
-                proc_id: local_proc_id.clone(),
-                world_id,
-                proc_message_port: local_proc_message_port.bind(),
-                proc_addr: local_proc_addr,
-                labels: HashMap::new(),
-                lifecycle_mode: ProcLifecycleMode::ManagedBySystem,
-            })
+            .send(
+                &client_mailbox,
+                SystemMessage::Join {
+                    proc_id: local_proc_id.clone(),
+                    world_id,
+                    proc_message_port: local_proc_message_port.bind(),
+                    proc_addr: local_proc_addr,
+                    labels: HashMap::new(),
+                    lifecycle_mode: ProcLifecycleMode::ManagedBySystem,
+                },
+            )
             .unwrap();
 
         assert_matches!(
@@ -1779,14 +1782,17 @@ mod tests {
         // Join the world.
         server_handle
             .system_actor_handle()
-            .send(SystemMessage::Join {
-                proc_id: local_proc_id.clone(),
-                world_id,
-                proc_message_port: local_proc_message_port.bind(),
-                proc_addr: local_proc_addr,
-                labels: HashMap::new(),
-                lifecycle_mode: ProcLifecycleMode::ManagedBySystem,
-            })
+            .send(
+                &client_mailbox,
+                SystemMessage::Join {
+                    proc_id: local_proc_id.clone(),
+                    world_id,
+                    proc_message_port: local_proc_message_port.bind(),
+                    proc_addr: local_proc_addr,
+                    labels: HashMap::new(),
+                    lifecycle_mode: ProcLifecycleMode::ManagedBySystem,
+                },
+            )
             .unwrap();
 
         assert_matches!(

hyperactor/src/actor.rs

@@ -622,7 +622,12 @@ impl<A: Actor> ActorHandle<A> {
 
     /// Send a message to the actor. Messages sent through the handle
     /// are always queued in process, and do not require serialization.
-    pub fn send<M: Message>(&self, message: M) -> Result<(), MailboxSenderError>
+    pub fn send<M: Message>(
+        &self,
+        // TODO(pzhang): use this parameter to generate sequence number.
+        _cx: &impl context::Actor,
+        message: M,
+    ) -> Result<(), MailboxSenderError>
     where
         A: Handler<M>,
     {
@@ -753,10 +758,10 @@ mod tests {
     #[tokio::test]
     async fn test_server_basic() {
         let proc = Proc::local();
-        let client = proc.attach("client").unwrap();
+        let (client, _) = proc.instance("client").unwrap();
         let (tx, mut rx) = client.open_port();
         let handle = proc.spawn::<EchoActor>("echo", tx.bind()).await.unwrap();
-        handle.send(123u64).unwrap();
+        handle.send(&client, 123u64).unwrap();
         handle.drain_and_stop().unwrap();
         handle.await;
 
@@ -766,7 +771,7 @@ mod tests {
     #[tokio::test]
     async fn test_ping_pong() {
         let proc = Proc::local();
-        let client = proc.attach("client").unwrap();
+        let (client, _) = proc.instance("client").unwrap();
         let (undeliverable_msg_tx, _) = client.open_port();
 
         let ping_pong_actor_params =
@@ -783,7 +788,10 @@ mod tests {
         let (local_port, local_receiver) = client.open_once_port();
 
         ping_handle
-            .send(PingPongMessage(10, pong_handle.bind(), local_port.bind()))
+            .send(
+                &client,
+                PingPongMessage(10, pong_handle.bind(), local_port.bind()),
+            )
             .unwrap();
 
         assert!(local_receiver.recv().await.unwrap());
@@ -792,7 +800,7 @@ mod tests {
     #[tokio::test]
     async fn test_ping_pong_on_handler_error() {
         let proc = Proc::local();
-        let client = proc.attach("client").unwrap();
+        let (client, _) = proc.instance("client").unwrap();
         let (undeliverable_msg_tx, _) = client.open_port();
 
         // Need to set a supervison coordinator for this Proc because there will
@@ -814,11 +822,14 @@ mod tests {
         let (local_port, local_receiver) = client.open_once_port();
 
         ping_handle
-            .send(PingPongMessage(
-                error_ttl + 1, // will encounter an error at TTL=66
-                pong_handle.bind(),
-                local_port.bind(),
-            ))
+            .send(
+                &client,
+                PingPongMessage(
+                    error_ttl + 1, // will encounter an error at TTL=66
+                    pong_handle.bind(),
+                    local_port.bind(),
+                ),
+            )
             .unwrap();
 
         // TODO: Fix this receiver hanging issue in T200423722.
@@ -861,10 +872,10 @@ mod tests {
     async fn test_init() {
         let proc = Proc::local();
         let handle = proc.spawn::<InitActor>("init", ()).await.unwrap();
-        let client = proc.attach("client").unwrap();
+        let (client, _) = proc.instance("client").unwrap();
 
         let (port, receiver) = client.open_once_port();
-        handle.send(port).unwrap();
+        handle.send(&client, port).unwrap();
         assert!(receiver.recv().await.unwrap());
 
         handle.drain_and_stop().unwrap();
@@ -946,12 +957,12 @@ mod tests {
             M: RemoteMessage,
             MultiActor: Handler<M>,
         {
-            self.handle.send(message).unwrap()
+            self.handle.send(&self.client, message).unwrap()
         }
 
         async fn sync(&self) {
             let (port, done) = self.client.open_once_port::<bool>();
-            self.handle.send(port).unwrap();
+            self.handle.send(&self.client, port).unwrap();
             assert!(done.recv().await.unwrap());
         }