stop becomes a future (#833)

Summary:
Pull Request resolved: #833

Now it works with synchronous uses as well.
ghstack-source-id: 302569853
exported-using-ghexport

Reviewed By: highker, mariusae

Differential Revision: D80017456

fbshipit-source-id: 8ab207c4fffdf5d58b59ce81c271e3947d40bc5c

Author: zdevito

monarch_hyperactor/src/actor_mesh.rs

@@ -238,9 +238,9 @@ impl PythonActorMesh {
         Err(self.pickling_err())
     }
 
-    fn stop<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
+    fn stop<'py>(&self) -> PyResult<PyPythonTask> {
         let actor_mesh = self.inner.clone();
-        crate::runtime::future_into_py(py, async move {
+        PyPythonTask::new(async move {
             let actor_mesh = actor_mesh
                 .take()
                 .await

python/monarch/_src/actor/actor_mesh.py

@@ -506,8 +506,7 @@ async def task():
         return PythonTask.from_coroutine(task()).spawn()
 
     async def stop(self) -> None:
-        inner = await Future(coro=self._inner.task())
-        await inner.stop()
+        await (await self._inner).stop()
 
     @staticmethod
     def _restore(inner: ActorMeshProtocol) -> ActorMeshProtocol:
@@ -1183,8 +1182,8 @@ def _new_with_shape(self, shape: Shape) -> "ActorMesh[T]":
     def __repr__(self) -> str:
         return f"ActorMesh(class={self._class}, shape={self._shape}), inner={type(self._inner)})"
 
-    async def stop(self):
-        await self._inner.stop()
+    def stop(self) -> "Future[None]":
+        return Future(coro=self._inner.stop())
 
     @property
     def initialized(self) -> Future[None]: