Cleanup dead python future code (meta-pytorch#917)

Summary:
Pull Request resolved: meta-pytorch#917
ghstack-source-id: 304156747

Reviewed By: mariusae

Differential Revision: D80132819

fbshipit-source-id: 32faf67511efa541a1990bf19bb4270181b92f87

Author: zdevito

python/monarch/_src/actor/actor_mesh.py

@@ -124,7 +124,6 @@ class MonarchContext:
     mailbox: Mailbox
     proc_id: str
     point: Point
-    send_queue: Tuple[Optional["Shared[Any]"], int]
     controller_controller: Optional["_ControllerController"]
     proc_mesh: Optional["ProcMesh"]  # actually this is a ProcMeshRef under the hood
 
@@ -134,9 +133,7 @@ def get() -> "MonarchContext":
         if c is None:
             mb = Mailbox.root_client_mailbox()
             proc_id = mb.actor_id.proc_id
-            c = MonarchContext(
-                mb, proc_id, Point(0, singleton_shape), (None, 0), None, None
-            )
+            c = MonarchContext(mb, proc_id, Point(0, singleton_shape), None, None)
             _context.set(c)
         return c
 
@@ -353,76 +350,6 @@ async def task():
         return PythonTask.from_coroutine(task())
 
 
-class SharedProtocolAdapter:
-    def __init__(self, inner: "Shared[ActorMeshProtocol]", supervise: bool):
-        self._inner = inner
-        self._supervise = supervise
-
-    def cast(
-        self,
-        message: PythonMessage,
-        selection: str,
-        mailbox: Mailbox,
-    ) -> None:
-        ctx = MonarchContext.get()
-        last, count = ctx.send_queue
-
-        async def task():
-            if last is not None:
-                await last
-            try:
-                self._inner.__await__()
-                inner = await self._inner
-                inner.cast(message, selection, mailbox)
-            except Exception as e:
-                match message.kind:
-                    case CallMethod(response_port=port) if port is not None:
-                        Port(port, mailbox, 0).exception(e)
-
-        ctx.send_queue = (PythonTask.from_coroutine(task()).spawn(), count + 1)
-
-    def new_with_shape(self, shape: Shape) -> "SharedProtocolAdapter":
-        async def task():
-            inner = await self._inner
-            return inner.new_with_shape(shape)
-
-        return SharedProtocolAdapter(PythonTask.from_coroutine(task()).spawn(), False)
-
-    def supervision_event(self) -> "Optional[Shared[Exception]]":
-        if not self._supervise:
-            return None
-
-        async def task():
-            inner = await self._inner
-            return await inner.supervision_event()
-
-        return PythonTask.from_coroutine(task()).spawn()
-
-    def stop(self) -> "PythonTask[None]":
-        async def task():
-            await (await self._inner).stop()
-
-        return PythonTask.from_coroutine(task())
-
-    @staticmethod
-    def _restore(inner: "ActorMeshProtocol") -> "ActorMeshProtocol":
-        return inner
-
-    def __reduce_ex__(self, protocol):
-        # blocking here means that we cannot send messages that contain actor
-        # references from the tokio event loop
-        if is_tokio_thread():
-            raise NotImplementedError(
-                "Cannot send actor references from a coroutine on the tokio event loop."
-                "To fix this we have to either make it psosible for pickling to defer this work,"
-                "or resolve the actor id without blocking."
-            )
-        return SharedProtocolAdapter._restore, (self._inner.block_on(),)
-
-    async def initialized(self):
-        await self._inner
-
-
 class ActorEndpoint(Endpoint[P, R]):
     def __init__(
         self,
@@ -783,7 +710,6 @@ async def handle(
                     mailbox,
                     mailbox.actor_id.proc_id,
                     Point(rank, shape),
-                    (None, 0),
                     None,
                     None,
                 )