De-future spawning actors (#813)

Summary:
Pull Request resolved: #813

This commit switches actor spawning to no longer return a Future. ActorMesh becomes a DeprecatedNotAFuture to keep existing code working.

The implementation delays the send part of endpoints until the actor has spawned. I expect this to have some performance impact, which needs to be measured before we land this.

However, this implementation is not what we expect to actually use: the ability to spawn actors/return a ref immediately will be made part of the rust runtime where the overhead will be substantially less.
ghstack-source-id: 302498028
exported-using-ghexport

Reviewed By: mariusae

Differential Revision: D79938417

fbshipit-source-id: f2606bc82087d6aff4f31f826cbad80615cc8982

Author: zdevito

hyperactor_mesh/src/comm/multicast.rs

@@ -254,6 +254,7 @@ pub trait CastInfo {
     /// we represent it as the only member of a 0-dimensonal cast shape,
     /// which is the same as a singleton.
     fn cast_info(&self) -> (usize, Shape);
+    fn sender(&self) -> &ActorId;
 }
 
 impl<A: Actor> CastInfo for Context<'_, A> {
@@ -265,4 +266,9 @@ impl<A: Actor> CastInfo for Context<'_, A> {
             _ => panic!("Expected either both rank and shape or neither"),
         }
     }
+    fn sender(&self) -> &ActorId {
+        self.headers()
+            .get(CAST_ORIGINATING_SENDER)
+            .expect("has sender header")
+    }
 }

monarch_hyperactor/src/pytokio.rs

@@ -149,8 +149,13 @@ fn send_result(
 
 #[pymethods]
 impl PyPythonTask {
-    fn block_on(&mut self, py: Python<'_>) -> PyResult<PyObject> {
-        signal_safe_block_on(py, self.take_task()?)?
+    fn block_on(mut slf: PyRefMut<PyPythonTask>, py: Python<'_>) -> PyResult<PyObject> {
+        let task = slf.take_task()?;
+        // mutable references to python objects must be dropped before calling
+        // signal_safe_block_on. It will release the GIL, and any other thread
+        // trying to access slf will throw.
+        drop(slf);
+        signal_safe_block_on(py, task)?
     }
 
     fn spawn(&mut self) -> PyResult<PyShared> {
@@ -293,9 +298,13 @@ impl PyShared {
         let task = self.task()?;
         Ok(PythonTaskAwaitIterator::new(task.into_py_any(py)?))
     }
-    pub fn block_on(&mut self, py: Python<'_>) -> PyResult<PyObject> {
-        let mut task = self.task()?;
-        task.block_on(py)
+    pub fn block_on(mut slf: PyRefMut<PyShared>, py: Python<'_>) -> PyResult<PyObject> {
+        let task = slf.task()?.take_task()?;
+        // mutable references to python objects must be dropped before calling
+        // signal_safe_block_on. It will release the GIL, and any other thread
+        // trying to access slf will throw.
+        drop(slf);
+        signal_safe_block_on(py, task)?
     }
 }