Remove USE_STANDIN_ACTOR_MESH and any dead code along with that

Summary: We have been using multicast for ~1 month without any issues. It is time to remove this gate.

Differential Revision: D82691879

Author: pzhan9

monarch_hyperactor/src/mailbox.rs

@@ -33,7 +33,6 @@ use hyperactor::mailbox::monitored_return_handle;
 use hyperactor::message::Bind;
 use hyperactor::message::Bindings;
 use hyperactor::message::Unbind;
-use hyperactor_mesh::comm::multicast::set_cast_info_on_headers;
 use monarch_types::PickledPyObject;
 use monarch_types::py_global;
 use pyo3::IntoPyObjectExt;
@@ -52,7 +51,6 @@ use crate::proc::PyActorId;
 use crate::pytokio::PyPythonTask;
 use crate::pytokio::PythonTask;
 use crate::runtime::signal_safe_block_on;
-use crate::shape::PyShape;
 
 #[derive(Clone, Debug)]
 #[pyclass(
@@ -139,37 +137,6 @@ impl PyMailbox {
         Ok(())
     }
 
-    pub(super) fn post_cast(
-        &self,
-        dest: &PyActorId,
-        rank: usize,
-        shape: &PyShape,
-        message: &PythonMessage,
-    ) -> PyResult<()> {
-        let port_id = dest.inner.port_id(PythonMessage::port());
-        let mut headers = Attrs::new();
-        set_cast_info_on_headers(
-            &mut headers,
-            rank,
-            shape.inner.clone(),
-            self.inner.actor_id().clone(),
-        );
-        let message = Serialized::serialize(message).map_err(|err| {
-            PyRuntimeError::new_err(format!(
-                "failed to serialize message ({:?}) to Serialized: {}",
-                message, err
-            ))
-        })?;
-        let envelope =
-            MessageEnvelope::new(self.inner.actor_id().clone(), port_id, message, headers);
-        let return_handle = self
-            .inner
-            .bound_return_handle()
-            .unwrap_or(monitored_return_handle());
-        self.inner.post(envelope, return_handle);
-        Ok(())
-    }
-
     #[getter]
     pub(super) fn actor_id(&self) -> PyActorId {
         PyActorId {

monarch_hyperactor/src/proc_mesh.rs

@@ -12,7 +12,6 @@ use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
 
 use hyperactor::Actor;
-use hyperactor::Mailbox;
 use hyperactor::RemoteMessage;
 use hyperactor::WorldId;
 use hyperactor::actor::RemoteActor;
@@ -48,7 +47,6 @@ use crate::mailbox::PyMailbox;
 use crate::pytokio::PyPythonTask;
 use crate::pytokio::PyShared;
 use crate::pytokio::PythonTask;
-use crate::runtime::get_tokio_runtime;
 use crate::shape::PyShape;
 use crate::supervision::SupervisionError;
 use crate::supervision::Unhealthy;
@@ -339,7 +337,6 @@ impl PyProcMesh {
         proc_mesh: &mut PyShared,
         name: String,
         actor: Py<PyType>,
-        emulated: bool,
     ) -> PyResult<PyObject> {
         let task = proc_mesh.task()?.take_task()?;
         let meshimpl = async move {
@@ -367,15 +364,8 @@ impl PyProcMesh {
                 actor_events,
             ))
         };
-        if emulated {
-            // we give up on doing mesh spawn async for the emulated old version
-            // it is too complicated to make both work.
-            let r = get_tokio_runtime().block_on(meshimpl)?;
-            Python::with_gil(|py| r.into_py_any(py))
-        } else {
-            let r = PythonActorMesh::new(meshimpl);
-            Python::with_gil(|py| r.into_py_any(py))
-        }
+        let r = PythonActorMesh::new(meshimpl);
+        Python::with_gil(|py| r.into_py_any(py))
     }
     // User can call this to monitor the proc mesh events. This will override
     // the default monitor that exits the client on process crash, so user can

python/monarch/_rust_bindings/monarch_hyperactor/actor_mesh.pyi

@@ -22,7 +22,7 @@ from typing_extensions import Self
 
 class ActorMeshProtocol(Protocol):
     """
-    Protocol defining the common interface for actor mesh, mesh ref and _ActorMeshRefImpl.
+    Protocol defining the common interface for actor mesh and mesh ref.
     """
 
     def cast(
@@ -40,34 +40,6 @@ class ActorMeshProtocol(Protocol):
 class PythonActorMesh(ActorMeshProtocol):
     pass
 
-class PythonActorMeshImpl:
-    def get_supervision_event(self) -> ActorSupervisionEvent | None:
-        """
-        Returns supervision event if there is any.
-        """
-        ...
-
-    def get(self, rank: int) -> ActorId | None:
-        """
-        Get the actor id for the actor at the given rank.
-        """
-        ...
-
-    def stop(self) -> PythonTask[None]:
-        """
-        Stop all actors that are part of this mesh.
-        Using this mesh after stop() is called will raise an Exception.
-        """
-        ...
-
-    def supervision_event(self) -> "Optional[Shared[Exception]]": ...
-    @property
-    def stopped(self) -> bool:
-        """
-        If the mesh has been stopped.
-        """
-        ...
-
 @final
 class ActorSupervisionEvent:
     @property

python/monarch/_rust_bindings/monarch_hyperactor/mailbox.pyi

@@ -162,15 +162,6 @@ class Mailbox:
         """
         ...
 
-    def post_cast(
-        self, dest: ActorId, rank: int, shape: Shape, message: PythonMessage
-    ) -> None:
-        """
-        Post a message to the provided actor. It will be handled using the handle_cast
-        endpoint as if the destination was `rank` of `shape`.
-        """
-        ...
-
     def undeliverable_receiver(self) -> UndeliverablePortReceiver:
         """
         Open a port to receive undeliverable messages.

python/monarch/_rust_bindings/monarch_hyperactor/proc_mesh.pyi

@@ -10,10 +10,7 @@ from typing import Any, AsyncIterator, final, Literal, overload, Type, TYPE_CHEC
 
 if TYPE_CHECKING:
     from monarch._rust_bindings.monarch_hyperactor.actor import Actor
-from monarch._rust_bindings.monarch_hyperactor.actor_mesh import (
-    PythonActorMesh,
-    PythonActorMeshImpl,
-)
+from monarch._rust_bindings.monarch_hyperactor.actor_mesh import PythonActorMesh
 
 from monarch._rust_bindings.monarch_hyperactor.alloc import Alloc
 from monarch._rust_bindings.monarch_hyperactor.mailbox import Mailbox
@@ -50,7 +47,7 @@ class ProcMesh:
 
     @staticmethod
     def spawn_async(
-        proc_mesh: Shared["ProcMesh"], name: str, actor: Type["Actor"], emulated: bool
+        proc_mesh: Shared["ProcMesh"], name: str, actor: Type["Actor"]
     ) -> PythonActorMesh: ...
     async def monitor(self) -> ProcMeshMonitor:
         """