[15/n] Pass monarch tensors to actor endpoints, part 1

Pull Request resolved: #518

This makes it possible to send a monarch tensor to an actor endpoint that is defiend over the same proc mesh as the tensor.

The send is done locally so the actor can do work with the tensors.

The stream actor in the tensor engine is sent a SendResultOfActorCall message, which it will forward to the actor, binding to the message the real local tensors that were passed as arguments.

The stream actor that owns the tensor waits for called the actor to finish since the actor 'owns' the tensor through the duration of the call.

Known limitation: this message to the actor can go out of order w.r.t to other messages sent from the owner of the tensor engine because the real message is being sent from the stream actor. The next PR will fix this limitation by sending _both_ the tensor engine and the actor a message at the same time. The actor will get a 'wait for SendResultOfActorCall' message, at which point it will stop processing any messages except for the SentResultOfActorCall message it is suppose to be waiting for. This way the correct order is preserved from the perspective of the tensor engine stream and the actor.
ghstack-source-id: 296136314
@exported-using-ghexport

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

Author: zdevito

monarch_extension/src/convert.rs

@@ -176,6 +176,35 @@ impl<'a> MessageParser<'a> {
     fn parseWorkerMessageList(&self, name: &str) -> PyResult<Vec<WorkerMessage>> {
         self.attr(name)?.try_iter()?.map(|x| convert(x?)).collect()
     }
+    #[allow(non_snake_case)]
+    fn parseLocalStateList(&self, name: &str) -> PyResult<Vec<worker::LocalState>> {
+        let local_state_list = self.attr(name)?;
+        let mut result = Vec::new();
+
+        // Get the PyMailbox class for type checking
+        let mailbox_class = self
+            .current
+            .py()
+            .import("monarch._rust_bindings.monarch_hyperactor.mailbox")?
+            .getattr("Mailbox")?;
+
+        for item in local_state_list.try_iter()? {
+            let item = item?;
+            // Check if it's a Ref by trying to extract it
+            if let Ok(ref_val) = create_ref(item.clone()) {
+                result.push(worker::LocalState::Ref(ref_val));
+            } else if item.is_instance(&mailbox_class)? {
+                // It's a PyMailbox instance
+                result.push(worker::LocalState::Mailbox);
+            } else {
+                return Err(PyValueError::new_err(format!(
+                    "Expected Ref or Mailbox in local_state, got: {}",
+                    item.get_type().name()?
+                )));
+            }
+        }
+        Ok(result)
+    }
     fn parse_error_reason(&self, name: &str) -> PyResult<Option<(Option<ActorId>, String)>> {
         let err = self.attr(name)?;
         if err.is_none() {
@@ -432,6 +461,20 @@ fn create_map(py: Python) -> HashMap<u64, FnType> {
             p.parseWorkerMessageList("commands")?,
         ))
     });
+    m.insert(key("SendResultOfActorCall"), |p| {
+        Ok(WorkerMessage::SendResultOfActorCall(
+            worker::ActorCallParams {
+                seq: p.parseSeq("seq")?,
+                actor: p.parse("actor")?,
+                index: p.parse("actor_index")?,
+                method: p.parse("method")?,
+                args_kwargs_tuple: p.parse("args_kwargs_tuple")?,
+                local_state: p.parseLocalStateList("local_state")?,
+                mutates: p.parseRefList("mutates")?,
+                stream: p.parseStreamRef("stream")?,
+            },
+        ))
+    });
     m
 }
 

monarch_extension/src/tensor_worker.rs

@@ -1372,6 +1372,7 @@ pub(crate) fn worker_message_to_py(py: Python<'_>, message: &WorkerMessage) -> P
         }
         WorkerMessage::SetRefUnitTestsOnly { .. } => unimplemented!(),
         WorkerMessage::GetRefUnitTestsOnly { .. } => unimplemented!(),
+        WorkerMessage::SendResultOfActorCall { .. } => unimplemented!(),
     }
 }
 

monarch_hyperactor/src/actor.rs

@@ -174,10 +174,10 @@ impl PickledMessageClientActor {
 #[pyclass(frozen, module = "monarch._rust_bindings.monarch_hyperactor.actor")]
 #[derive(Default, Clone, Serialize, Deserialize, Named, PartialEq)]
 pub struct PythonMessage {
-    pub(crate) method: String,
-    pub(crate) message: ByteBuf,
-    response_port: Option<EitherPortRef>,
-    rank: Option<usize>,
+    pub method: String,
+    pub message: ByteBuf,
+    pub response_port: Option<EitherPortRef>,
+    pub rank: Option<usize>,
 }
 
 impl PythonMessage {
@@ -288,7 +288,7 @@ impl PythonActorHandle {
         PythonMessage { cast = true },
     ],
 )]
-pub(super) struct PythonActor {
+pub struct PythonActor {
     /// The Python object that we delegate message handling to. An instance of
     /// `monarch.actor_mesh._Actor`.
     pub(super) actor: PyObject,
@@ -398,16 +398,30 @@ impl PanicFlag {
 }
 
 #[async_trait]
-impl Handler<PythonMessage> for PythonActor {
-    async fn handle(&mut self, cx: &Context<Self>, message: PythonMessage) -> anyhow::Result<()> {
-        let mailbox = PyMailbox {
+impl Handler<LocalPythonMessage> for PythonActor {
+    async fn handle(
+        &mut self,
+        cx: &Context<Self>,
+        message: LocalPythonMessage,
+    ) -> anyhow::Result<()> {
+        let mailbox: PyMailbox = PyMailbox {
             inner: cx.mailbox_for_py().clone(),
         };
         // Create a channel for signaling panics in async endpoints.
         // See [Panics in async endpoints].
         let (sender, receiver) = oneshot::channel();
 
         let future = Python::with_gil(|py| -> Result<_, SerializablePyErr> {
+            let mailbox = mailbox.into_bound_py_any(py).unwrap();
+            let local_state: Option<Vec<Bound<'_, PyAny>>> = message.local_state.map(|state| {
+                state
+                    .into_iter()
+                    .map(|e| match e {
+                        LocalState::Mailbox => mailbox.clone(),
+                        LocalState::PyObject(obj) => obj.into_bound_py_any(py).unwrap(),
+                    })
+                    .collect()
+            });
             let (rank, shape) = cx.cast_info();
             let awaitable = self.actor.call_method(
                 py,
@@ -416,10 +430,11 @@ impl Handler<PythonMessage> for PythonActor {
                     mailbox,
                     rank,
                     PyShape::from(shape),
-                    message,
+                    message.message,
                     PanicFlag {
                         sender: Some(sender),
                     },
+                    local_state,
                 ),
                 None,
             )?;
@@ -438,6 +453,20 @@ impl Handler<PythonMessage> for PythonActor {
     }
 }
 
+#[async_trait]
+impl Handler<PythonMessage> for PythonActor {
+    async fn handle(&mut self, cx: &Context<Self>, message: PythonMessage) -> anyhow::Result<()> {
+        self.handle(
+            cx,
+            LocalPythonMessage {
+                message,
+                local_state: None,
+            },
+        )
+        .await
+    }
+}
+
 /// Helper struct to make a Python future passable in an actor message.
 ///
 /// Also so that we don't have to write this massive type signature everywhere
@@ -537,6 +566,17 @@ impl AsyncEndpointInvocationHandler for AsyncEndpointTask {
         Ok(())
     }
 }
+#[derive(Debug)]
+pub enum LocalState {
+    Mailbox,
+    PyObject(PyObject),
+}
+
+#[derive(Debug)]
+pub struct LocalPythonMessage {
+    pub message: PythonMessage,
+    pub local_state: Option<Vec<LocalState>>,
+}
 
 pub fn register_python_bindings(hyperactor_mod: &Bound<'_, PyModule>) -> PyResult<()> {
     hyperactor_mod.add_class::<PickledMessage>()?;

monarch_messages/src/worker.rs

@@ -402,6 +402,36 @@ pub struct CallFunctionParams {
     pub remote_process_groups: Vec<Ref>,
 }
 
+/// The local state that has to be restored into the python_message during
+/// its unpickling.
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub enum LocalState {
+    Ref(Ref),
+    Mailbox,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct ActorCallParams {
+    pub seq: Seq,
+    /// The actor to call is (proc_id of stream worker, 'actor', 'index')
+    pub actor: String,
+    pub index: usize,
+
+    // method name to call
+    pub method: String,
+    // pickled arguments, that will need to be patched with local state
+    pub args_kwargs_tuple: Vec<u8>,
+    /// Referenceable objects to pass to the actor,
+    /// these will be put into the PythonMessage
+    /// during its unpickling. Because unpickling also needs to
+    /// restore mailboxes, we also have to keep track of which
+    /// members should just be the mailbox object.
+    pub local_state: Vec<LocalState>,
+
+    /// Tensors that will be mutated by the call.
+    pub mutates: Vec<Ref>,
+    pub stream: StreamRef,
+}
 /// Type of reduction for [`WorkerMessage::Reduce`].
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum Reduction {
@@ -642,21 +672,30 @@ pub enum WorkerMessage {
 
     /// First use of the borrow on the receiving stream. This is a marker for
     /// synchronization.
-    BorrowFirstUse { borrow: u64 },
+    BorrowFirstUse {
+        borrow: u64,
+    },
 
     /// Last use of the borrow on the receiving stream. This is a marker for
     /// synchronization.
-    BorrowLastUse { borrow: u64 },
+    BorrowLastUse {
+        borrow: u64,
+    },
 
     /// Drop the borrow and free the resources associated with it.
-    BorrowDrop { borrow: u64 },
+    BorrowDrop {
+        borrow: u64,
+    },
 
     /// Delete these refs from the worker state.
     DeleteRefs(Vec<Ref>),
 
     /// A [`ControllerMessage::Status`] will be send to the controller
     /// when all streams have processed all the message sent before this one.
-    RequestStatus { seq: Seq, controller: bool },
+    RequestStatus {
+        seq: Seq,
+        controller: bool,
+    },
 
     /// Perform a reduction operation, using an efficient communication backend.
     /// Only NCCL is supported for now.
@@ -758,6 +797,7 @@ pub enum WorkerMessage {
         stream: StreamRef,
     },
 
+    SendResultOfActorCall(ActorCallParams),
     PipeRecv {
         seq: Seq,
         /// Result refs.

monarch_simulator/src/worker.rs

@@ -311,6 +311,14 @@ impl WorkerMessageHandler for WorkerActor {
         Ok(())
     }
 
+    async fn send_result_of_actor_call(
+        &mut self,
+        cx: &hyperactor::Context<Self>,
+        params: ActorCallParams,
+    ) -> Result<()> {
+        bail!("unimplemented: send_result_of_actor_call");
+    }
+
     async fn command_group(
         &mut self,
         cx: &hyperactor::Context<Self>,

monarch_tensor_worker/src/lib.rs

@@ -71,6 +71,7 @@ use monarch_messages::controller::ControllerActor;
 use monarch_messages::controller::ControllerMessageClient;
 use monarch_messages::controller::Seq;
 use monarch_messages::wire_value::WireValue;
+use monarch_messages::worker::ActorCallParams;
 use monarch_messages::worker::CallFunctionError;
 use monarch_messages::worker::CallFunctionParams;
 use monarch_messages::worker::Factory;
@@ -859,6 +860,17 @@ impl WorkerMessageHandler for WorkerActor {
             .await
     }
 
+    async fn send_result_of_actor_call(
+        &mut self,
+        cx: &hyperactor::Context<Self>,
+        params: ActorCallParams,
+    ) -> Result<()> {
+        let stream = self.try_get_stream(params.stream)?;
+        stream
+            .send_result_of_actor_call(cx, cx.self_id().clone(), params)
+            .await?;
+        Ok(())
+    }
     async fn split_comm(
         &mut self,
         cx: &hyperactor::Context<Self>,

monarch_tensor_worker/src/stream.rs

@@ -36,12 +36,17 @@ use hyperactor::mailbox::Mailbox;
 use hyperactor::mailbox::OncePortHandle;
 use hyperactor::mailbox::PortReceiver;
 use hyperactor::proc::Proc;
+use monarch_hyperactor::actor::LocalPythonMessage;
+use monarch_hyperactor::actor::PythonActor;
 use monarch_hyperactor::actor::PythonMessage;
+use monarch_hyperactor::mailbox::EitherPortRef;
 use monarch_messages::controller::ControllerMessageClient;
 use monarch_messages::controller::Seq;
 use monarch_messages::controller::WorkerError;
+use monarch_messages::worker::ActorCallParams;
 use monarch_messages::worker::CallFunctionError;
 use monarch_messages::worker::CallFunctionParams;
+use monarch_messages::worker::LocalState;
 use monarch_messages::worker::StreamRef;
 use monarch_types::PyTree;
 use monarch_types::SerializablePyErr;
@@ -233,6 +238,8 @@ pub enum StreamMessage {
     ),
 
     GetTensorRefUnitTestsOnly(Ref, #[reply] OncePortHandle<Option<TensorCellResult>>),
+
+    SendResultOfActorCall(ActorId, ActorCallParams),
 }
 
 impl StreamMessage {
@@ -1666,6 +1673,75 @@ impl StreamMessageHandler for StreamActor {
         Ok(())
     }
 
+    async fn send_result_of_actor_call(
+        &mut self,
+        cx: &Context<Self>,
+        worker_actor_id: ActorId,
+        params: ActorCallParams,
+    ) -> anyhow::Result<()> {
+        // TODO: handle mutates
+        let actor_id = ActorId(cx.proc().proc_id().clone(), params.actor, params.index);
+        let actor_ref: ActorRef<PythonActor> = ActorRef::attest(actor_id);
+        let actor_handle = actor_ref.downcast_handle(cx).unwrap();
+        let (send, recv) = cx.open_once_port();
+        let send = send.bind();
+        let send = EitherPortRef::Once(send.into());
+        let message = PythonMessage {
+            method: params.method,
+            message: params.args_kwargs_tuple.into(),
+            response_port: Some(send),
+            rank: None,
+        };
+        let local_state: Result<Vec<monarch_hyperactor::actor::LocalState>> =
+            Python::with_gil(|py| {
+                params
+                    .local_state
+                    .into_iter()
+                    .map(|elem| match elem {
+                        LocalState::Mailbox => Ok(monarch_hyperactor::actor::LocalState::Mailbox),
+                        // SAFETY: python is gonna make unsafe copies of this stuff anyway
+                        LocalState::Ref(r) => unsafe {
+                            let x = self.ref_to_rvalue(&r)?.try_to_object_unsafe(py)?.into();
+                            Ok(monarch_hyperactor::actor::LocalState::PyObject(x))
+                        },
+                    })
+                    .collect()
+            });
+        // including making the PyMailbox
+        let message = LocalPythonMessage {
+            message,
+            local_state: Some(local_state?),
+        };
+        actor_handle.send(message)?;
+        let result = recv.recv().await?.with_rank(worker_actor_id.rank());
+        if result.method == "exception" {
+            // If result has "exception" as its kind, then
+            // we need to unpickle and turn it into a WorkerError
+            // and call remote_function_failed otherwise the
+            // controller assumes the object is correct and doesn't handle
+            // dependency tracking correctly.
+            let err = Python::with_gil(|py| -> Result<WorkerError, SerializablePyErr> {
+                let err = py
+                    .import("pickle")
+                    .unwrap()
+                    .call_method1("loads", (result.message.into_vec(),))?;
+                Ok(WorkerError {
+                    worker_actor_id,
+                    backtrace: err.to_string(),
+                })
+            })?;
+            self.controller_actor
+                .remote_function_failed(cx, params.seq, err)
+                .await?;
+        } else {
+            let result = Serialized::serialize(&result).unwrap();
+            self.controller_actor
+                .fetch_result(cx, params.seq, Ok(result))
+                .await?;
+        }
+        Ok(())
+    }
+
     async fn set_value(
         &mut self,
         cx: &Context<Self>,