Add native v1 castings (#1304)

Summary:
Pull Request resolved: #1304

This diff adds native v1 casting implementation, instead of piggybacking v0's implementation. Specifically, this diff:

1. Add V1 handlers in comm actors, i.e. `CastMessageV1` and `ForwardMessageV1`.
2. On the v1 ActorMesh side, use `Sequencers` to generate seq numbers, and send to comm actors through `CastMessageV1`.
3. Update several direct send call sites to use `ActorRef::seq_send`, so seq numbers are always assigned.

Note that we still need to wait for all `&impl cap::CanSend` callsites to be updated with `&impl context::Actor` before we can turn on the native v1 casting in production, since we need to ensure all `direct send`s are also getting the sequence number assigned.

Differential Revision: D82537988

Author: pzhan9

hyperactor/src/actor.rs

@@ -1085,7 +1085,7 @@ mod tests {
 
     // Returning the sequence number assigned to the message.
     #[derive(Debug)]
-    #[hyperactor::export(handlers = [String])]
+    #[hyperactor::export(handlers = [String, Callback])]
     struct GetSeqActor(PortRef<(String, SeqInfo)>);
 
     #[async_trait]
@@ -1111,6 +1111,28 @@ mod tests {
         }
     }
 
+    // Unlike Handler<String>, where the sender provides the string message
+    // directly, in Hanlder<Callback>, sender needs to provide a port, and
+    // handler will reply that port with its own callback port. Then sender can
+    // send the string message through thsi callback port.
+    #[derive(Clone, Debug, Serialize, Deserialize, Named)]
+    struct Callback(PortRef<PortRef<String>>);
+
+    #[async_trait]
+    impl Handler<Callback> for GetSeqActor {
+        async fn handle(
+            &mut self,
+            cx: &Context<Self>,
+            message: Callback,
+        ) -> Result<(), anyhow::Error> {
+            let (handle, mut receiver) = cx.open_port::<String>();
+            let callback_ref = handle.bind();
+            message.0.send(cx, callback_ref).unwrap();
+            let msg = receiver.recv().await.unwrap();
+            self.handle(cx, msg).await
+        }
+    }
+
     #[async_timed_test(timeout_secs = 30)]
     async fn test_sequencing_actor_handle_basic() {
         let proc = Proc::local();
@@ -1151,6 +1173,37 @@ mod tests {
         }
     }
 
+    // Verify that we can pass port refs between sender and destination actors
+    // back and forward, and send messages through them without being deadlocked.
+    #[async_timed_test(timeout_secs = 30)]
+    async fn test_sequencing_actor_handle_callback() {
+        let config = config::global::lock();
+        let _guard = config.override_key(config::ENABLE_DEST_ACTOR_REORDERING_BUFFER, true);
+
+        let proc = Proc::local();
+        let (client, _) = proc.instance("client").unwrap();
+        let (tx, mut rx) = client.open_port();
+
+        let actor_handle = proc
+            .spawn::<GetSeqActor>("get_seq", tx.bind())
+            .await
+            .unwrap();
+        let actor_ref: ActorRef<GetSeqActor> = actor_handle.bind();
+
+        let (callback_tx, mut callback_rx) = client.open_port();
+        actor_ref
+            .send(&client, Callback(callback_tx.bind()))
+            .unwrap();
+        let msg_port_ref = callback_rx.recv().await.unwrap();
+        msg_port_ref.send(&client, "finally".to_string()).unwrap();
+
+        let session_id = client.sequencer().session_id();
+        assert_eq!(
+            rx.recv().await.unwrap(),
+            ("finally".to_string(), SeqInfo { session_id, seq: 1 })
+        );
+    }
+
     // Adding a delay before sending the destination proc. Useful for tests
     // requiring latency injection.
     #[derive(Debug)]
@@ -1254,7 +1307,7 @@ mod tests {
 
         // By disabling the actor side re-ordering buffer, the mssages will
         // be processed in the same order as they sent out.
-        let _guard = config.override_key(config::ENABLE_CLIENT_SEQ_ASSIGNMENT, false);
+        let _guard = config.override_key(config::ENABLE_DEST_ACTOR_REORDERING_BUFFER, false);
         assert_out_of_order_delivery(
             vec![("second".to_string(), 2), ("first".to_string(), 1)],
             latency_plan.clone(),
@@ -1263,7 +1316,7 @@ mod tests {
 
         // By enabling the actor side re-ordering buffer, the mssages will
         // be re-ordered before being processed.
-        let _guard = config.override_key(config::ENABLE_CLIENT_SEQ_ASSIGNMENT, true);
+        let _guard = config.override_key(config::ENABLE_DEST_ACTOR_REORDERING_BUFFER, true);
         assert_out_of_order_delivery(
             vec![("first".to_string(), 1), ("second".to_string(), 2)],
             latency_plan.clone(),
@@ -1281,7 +1334,7 @@ mod tests {
 
         // By enabling the actor side re-ordering buffer, the mssages will
         // be re-ordered before being processed.
-        let _guard = config.override_key(config::ENABLE_CLIENT_SEQ_ASSIGNMENT, true);
+        let _guard = config.override_key(config::ENABLE_DEST_ACTOR_REORDERING_BUFFER, true);
         let expected = (1..10000)
             .map(|i| (format!("msg{i}"), i))
             .collect::<Vec<_>>();

hyperactor/src/config.rs

@@ -110,7 +110,12 @@ declare_attrs! {
     pub attr MESSAGE_LATENCY_SAMPLING_RATE: f32 = 0.01;
 
     /// Whether to enable client sequence assignment.
-    pub attr ENABLE_CLIENT_SEQ_ASSIGNMENT: bool = false;
+    @meta(CONFIG_ENV_VAR = "HYPERACTOR_ENABLE_DEST_ACTOR_REORDERING_BUFFER".to_string())
+    pub attr ENABLE_DEST_ACTOR_REORDERING_BUFFER: bool = false;
+
+    /// Whether to use native v1 casting in v1 ActorMesh.
+    @meta(CONFIG_ENV_VAR = "HYPERACTOR_ENABLE_NATIVE_V1_CASTING".to_string())
+    pub attr ENABLE_NATIVE_V1_CASTING: bool = false;
 
     /// Timeout for [`Host::spawn`] to await proc readiness.
     ///

hyperactor/src/proc.rs

@@ -946,7 +946,7 @@ impl<A: Actor> Instance<A> {
         let mailbox = Mailbox::new(actor_id.clone(), BoxedMailboxSender::new(proc.downgrade()));
         let (work_tx, work_rx) = ordered_channel(
             actor_id.to_string(),
-            config::global::get(config::ENABLE_CLIENT_SEQ_ASSIGNMENT),
+            config::global::get(config::ENABLE_DEST_ACTOR_REORDERING_BUFFER),
         );
         let ports: Arc<Ports<A>> = Arc::new(Ports::new(mailbox.clone(), work_tx));
         proc.state().proc_muxer.bind_mailbox(mailbox.clone());

hyperactor/src/reference.rs

@@ -899,9 +899,21 @@ impl PortId {
         cx: &impl context::Actor,
         serialized: Serialized,
         mut headers: Attrs,
+    ) {
+        self.send_with_headers_with_option(cx, serialized, headers, true);
+    }
+
+    /// Similar to [`PortId::send_with_headers`], but allows the caller to
+    /// decide whether to set the sequence info header with this method.
+    pub fn send_with_headers_with_option(
+        &self,
+        cx: &impl context::Actor,
+        serialized: Serialized,
+        mut headers: Attrs,
+        set_seq_info: bool,
     ) {
         crate::mailbox::headers::set_send_timestamp(&mut headers);
-        if self.is_actor_port() {
+        if set_seq_info && self.is_actor_port() {
             // This method is infallible so is okay to assign the sequence number
             // without worrying about rollback.
             let sequencer = cx.instance().sequencer();

hyperactor_mesh/Cargo.toml

@@ -81,9 +81,11 @@ tokio-stream = { version = "0.1.17", features = ["fs", "io-util", "net", "signal
 tokio-util = { version = "0.7.15", features = ["full"] }
 tracing = { version = "0.1.41", features = ["attributes", "valuable"] }
 tracing-subscriber = { version = "0.3.20", features = ["chrono", "env-filter", "json", "local-time", "parking_lot", "registry"] }
+uuid = { version = "1.2", features = ["serde", "v4", "v5", "v6", "v7", "v8"] }
 
 [dev-dependencies]
 bytes = { version = "1.10", features = ["serde"] }
+fastrand = "2.1.1"
 itertools = "0.14.0"
 maplit = "1.0"
 proptest = "1.5"

hyperactor_mesh/src/actor_mesh.rs

@@ -1092,17 +1092,17 @@ mod tests {
                 let actor_mesh: RootActorMesh<TestActor> = mesh.spawn("test", &()).await.unwrap();
                 let actor_ref = actor_mesh.get(0).unwrap();
                 let mut headers = Attrs::new();
-                set_cast_info_on_headers(&mut headers, extent.point_of_rank(0).unwrap(), mesh.client().self_id().clone());
+                set_cast_info_on_headers(&mut headers, extent.point_of_rank(0).unwrap(), mesh.client().self_id().clone(), None);
                 actor_ref.send_with_headers(mesh.client(), headers.clone(), GetRank(true, reply_port.clone())).unwrap();
                 assert_eq!(0, reply_port_receiver.recv().await.unwrap());
 
-                set_cast_info_on_headers(&mut headers, extent.point_of_rank(1).unwrap(), mesh.client().self_id().clone());
+                set_cast_info_on_headers(&mut headers, extent.point_of_rank(1).unwrap(), mesh.client().self_id().clone(), None);
                 actor_ref.port()
                     .send_with_headers(mesh.client(), headers.clone(), GetRank(true, reply_port.clone()))
                     .unwrap();
                 assert_eq!(1, reply_port_receiver.recv().await.unwrap());
 
-                set_cast_info_on_headers(&mut headers, extent.point_of_rank(2).unwrap(), mesh.client().self_id().clone());
+                set_cast_info_on_headers(&mut headers, extent.point_of_rank(2).unwrap(), mesh.client().self_id().clone(), None);
                 actor_ref.actor_id()
                     .port_id(GetRank::port())
                     .send_with_headers(

hyperactor_mesh/src/bootstrap.rs

@@ -3322,7 +3322,7 @@ mod tests {
         actor_mesh
             .cast(&instance, testactor::GetActorId(port.bind()))
             .unwrap();
-        let got_id = rx.recv().await.unwrap();
+        let (got_id, _seq) = rx.recv().await.unwrap();
         assert_eq!(
             got_id,
             actor_mesh.values().next().unwrap().actor_id().clone()