Remove USE_STANDIN_ACTOR_MESH and any dead code along with that (#1273)

pzhan9 · facebook-github-bot · commit daa561d0c3c0 · 2025-09-19T21:12:33.000-07:00
Summary:

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

Differential Revision: D82691879
diff --git a/monarch_hyperactor/src/mailbox.rs b/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 {
diff --git a/monarch_hyperactor/src/proc_mesh.rs b/monarch_hyperactor/src/proc_mesh.rs
@@ -49,7 +49,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::PyRegion;
 use crate::supervision::SupervisionError;
 use crate::supervision::Unhealthy;
@@ -340,7 +339,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 {
@@ -368,18 +366,11 @@ 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(async move {
+        let r = PythonActorMesh::new(async move {
                 let meshimpl: Box<dyn ActorMeshProtocol> = meshimpl.await?;
                 Ok(meshimpl)
             });
-            Python::with_gil(|py| r.into_py_any(py))
-        }
+        Python::with_gil(|py| r.into_py_any(py))
     }
 
     // User can call this to monitor the proc mesh events. This will override
diff --git a/python/monarch/_rust_bindings/monarch_hyperactor/actor_mesh.pyi b/python/monarch/_rust_bindings/monarch_hyperactor/actor_mesh.pyi
@@ -17,7 +17,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(
@@ -35,34 +35,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
diff --git a/python/monarch/_rust_bindings/monarch_hyperactor/mailbox.pyi b/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.
diff --git a/python/monarch/_rust_bindings/monarch_hyperactor/proc_mesh.pyi b/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.context import Instance
@@ -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:
         """
diff --git a/python/monarch/_src/actor/actor_mesh.py b/python/monarch/_src/actor/actor_mesh.py
@@ -13,7 +13,6 @@
 import inspect
 import itertools
 import logging
-import random
 from abc import abstractproperty
 
 from dataclasses import dataclass
@@ -48,16 +47,11 @@
     PythonMessage,
     PythonMessageKind,
 )
-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.context import Instance as HyInstance
 from monarch._rust_bindings.monarch_hyperactor.mailbox import (
     Mailbox,
-    OncePortReceiver as HyOncePortReceiver,  # noqa: F401
     OncePortRef,
-    PortReceiver as HyPortReceiver,  # noqa: F401
     PortRef,
     UndeliverableMessageEnvelope,
 )
@@ -71,8 +65,6 @@
     Region,
     Shape,
 )
-from monarch._rust_bindings.monarch_hyperactor.supervision import SupervisionError
-
 from monarch._rust_bindings.monarch_hyperactor.value_mesh import (
     ValueMesh as HyValueMesh,
 )
@@ -251,10 +243,6 @@ def set(debug_context: "DebugContext") -> None:
 R = TypeVar("R")
 A = TypeVar("A")
 
-# keep this load balancing deterministic, but
-# equally distributed.
-_load_balancing_seed = random.Random(4)
-
 
 class _SingletonActorAdapator:
     def __init__(self, inner: ActorId, region: Optional[Region] = None) -> None:
@@ -287,159 +275,6 @@ async def empty():
         return PythonTask.from_coroutine(empty())
 
 
-# standin class for whatever is the serializable python object we use
-# to name an actor mesh. Hacked up today because ActorMesh
-# isn't plumbed to non-clients
-class _ActorMeshRefImpl:
-    def __init__(
-        self,
-        mailbox: Mailbox,
-        hy_actor_mesh: Optional[PythonActorMeshImpl],
-        proc_mesh: "Optional[ProcMesh]",
-        shape: Shape,
-        actor_ids: List[ActorId],
-    ) -> None:
-        self._mailbox = mailbox
-        self._actor_mesh = hy_actor_mesh
-        # actor meshes do not have a way to look this up at the moment,
-        # so we fake it here
-        self._proc_mesh = proc_mesh
-        self._shape = shape
-        self._please_replace_me_actor_ids = actor_ids
-
-    @staticmethod
-    def from_hyperactor_mesh(
-        mailbox: Mailbox,
-        shape: Shape,
-        hy_actor_mesh: PythonActorMeshImpl,
-        proc_mesh: "ProcMesh",
-    ) -> "_ActorMeshRefImpl":
-        return _ActorMeshRefImpl(
-            mailbox,
-            hy_actor_mesh,
-            proc_mesh,
-            shape,
-            [cast(ActorId, hy_actor_mesh.get(i)) for i in range(len(shape))],
-        )
-
-    def __getstate__(
-        self,
-    ) -> Tuple[Shape, List[ActorId], Mailbox]:
-        return self._shape, self._please_replace_me_actor_ids, self._mailbox
-
-    def __setstate__(
-        self,
-        state: Tuple[Shape, List[ActorId], Mailbox],
-    ) -> None:
-        self._actor_mesh = None
-        self._shape, self._please_replace_me_actor_ids, self._mailbox = state
-
-    def _check_state(self) -> None:
-        # This is temporary until we have real cast integration here. We need to actively check
-        # supervision error here is because all communication is done through direct mailbox sending
-        # and not through comm actor casting.
-        # TODO: remove this when casting integration is done.
-        if self._actor_mesh is not None:
-            if self._actor_mesh.stopped:
-                raise SupervisionError(
-                    "actor mesh is unhealthy with reason: actor mesh is stopped due to proc mesh shutdown. "
-                    "`PythonActorMesh` has already been stopped."
-                )
-
-            event = self._actor_mesh.get_supervision_event()
-            if event is not None:
-                raise SupervisionError(f"actor mesh is unhealthy with reason: {event}")
-
-    def cast(
-        self,
-        message: PythonMessage,
-        selection: str,
-        instance: HyInstance,
-    ) -> None:
-        self._check_state()
-
-        # TODO: use the actual actor mesh when available. We cannot currently use it
-        # directly because we risk bifurcating the message delivery paths from the same
-        # client, since slicing the mesh will produce a reference, which calls actors
-        # directly. The reason these paths are bifurcated is that actor meshes will
-        # use multicasting, while direct actor comms do not. Separately we need to decide
-        # whether actor meshes are ordered with actor references.
-        #
-        # The fix is to provide a first-class reference into Python, and always call "cast"
-        # on it, including for load balanced requests.
-        if selection == "choose":
-            idx = _load_balancing_seed.randrange(len(self._shape))
-            actor_rank = self._shape.ndslice[idx]
-            self._mailbox.post(self._please_replace_me_actor_ids[actor_rank], message)
-        elif selection == "all":
-            # replace me with actual remote actor mesh
-            call_shape = Shape(
-                self._shape.labels, NDSlice.new_row_major(self._shape.ndslice.sizes)
-            )
-            for i, rank in enumerate(self._shape.ranks()):
-                self._mailbox.post_cast(
-                    self._please_replace_me_actor_ids[rank],
-                    i,
-                    call_shape,
-                    message,
-                )
-        elif isinstance(selection, int):
-            try:
-                self._mailbox.post(
-                    self._please_replace_me_actor_ids[selection], message
-                )
-            except IndexError:
-                raise IndexError(
-                    f"Tried to send to an out-of-range rank {selection}: "
-                    f"mesh has {len(self._please_replace_me_actor_ids)} elements."
-                )
-        else:
-            raise ValueError(f"invalid selection: {selection}")
-
-    def __len__(self) -> int:
-        return len(self._shape)
-
-    @property
-    def _name_pid(self):
-        actor_id0 = self._please_replace_me_actor_ids[0]
-        return actor_id0.actor_name, actor_id0.pid
-
-    @property
-    def shape(self) -> Shape:
-        return self._shape
-
-    @property
-    def proc_mesh(self) -> Optional["ProcMesh"]:
-        return self._proc_mesh
-
-    def new_with_region(self, region: Region) -> "_ActorMeshRefImpl":
-        return _ActorMeshRefImpl(
-            self._mailbox,
-            None,
-            None,
-            region.as_shape(),
-            self._please_replace_me_actor_ids,
-        )
-
-    def supervision_event(self) -> "Optional[Shared[Exception]]":
-        if self._actor_mesh is None:
-            return None
-        return self._actor_mesh.supervision_event()
-
-    def stop(self) -> PythonTask[None]:
-        async def task():
-            if self._actor_mesh is not None:
-                self._actor_mesh.stop()
-
-        return PythonTask.from_coroutine(task())
-
-    def initialized(self) -> PythonTask[None]:
-        async def task():
-            pass
-
-        return PythonTask.from_coroutine(task())
-
-
 class ActorEndpoint(Endpoint[P, R]):
     def __init__(
         self,
@@ -1148,7 +983,7 @@ def _endpoint(
     def _create(
         cls,
         Class: Type[T],
-        actor_mesh: "PythonActorMesh | PythonActorMeshImpl",
+        actor_mesh: "PythonActorMesh",
         mailbox: Mailbox,
         shape: Shape,
         proc_mesh: "ProcMesh",
@@ -1158,11 +993,6 @@ def _create(
         *args: Any,
         **kwargs: Any,
     ) -> "ActorMesh[T]":
-        if isinstance(actor_mesh, PythonActorMeshImpl):
-            actor_mesh = _ActorMeshRefImpl.from_hyperactor_mesh(
-                mailbox, shape, actor_mesh, proc_mesh
-            )
-
         mesh = cls(Class, actor_mesh, shape, proc_mesh)
 
         async def null_func(*_args: Iterable[Any], **_kwargs: Dict[str, Any]) -> None:
diff --git a/python/monarch/_src/actor/endpoint.py b/python/monarch/_src/actor/endpoint.py
diff --git a/python/monarch/_src/actor/proc_mesh.py b/python/monarch/_src/actor/proc_mesh.py
