propagate display name for process failures (#1911)

Summary:
Pull Request resolved: #1911

Looks like this now:

```
The actor <root>.<tests.test_supervision_hierarchy.Nest top> and all its descendants have failed.
This occurred because the actor <root>.<tests.test_supervision_hierarchy.Nest top>.<tests.test_supervision_hierarchy.Lambda nested{'a_dim': 0/1}> was running on a process which failed.
The error was:
 process failure: Killed(sig=9)
```

Kinda janky because supervision code is messy. When we clean up supervision code, we should make sure the owning actor knows the friendly names of all of its children (e.g. has a map from actor/proc id to their fully qualified display names). A parent always knows what the name will be before it spawns it.
ghstack-source-id: 324165921
exported-using-ghexport

Reviewed By: dulinriley

Differential Revision: D87284093

fbshipit-source-id: e32a32c5a138b3b2c7bb28711d164e0cca7864b2

Author: zdevito

monarch_hyperactor/src/actor_mesh.rs

@@ -79,7 +79,11 @@ pub(crate) trait ActorMeshProtocol: Send + Sync {
     /// will provide "supervision_event" with events.
     /// The default implementation does nothing, and it is not required that
     /// it has to be called before supervision_event.
-    fn start_supervision(&self, _instance: &PyInstance) -> PyResult<()> {
+    fn start_supervision(
+        &self,
+        _instance: &PyInstance,
+        _supervision_display_name: String,
+    ) -> PyResult<()> {
         Ok(())
     }
 
@@ -156,8 +160,13 @@ impl PythonActorMesh {
         self.inner.supervision_event(instance)
     }
 
-    fn start_supervision(&self, instance: &PyInstance) -> PyResult<()> {
-        self.inner.start_supervision(instance)
+    fn start_supervision(
+        &self,
+        instance: &PyInstance,
+        supervision_display_name: String,
+    ) -> PyResult<()> {
+        self.inner
+            .start_supervision(instance, supervision_display_name)
     }
 
     fn stop(&self, instance: &PyInstance) -> PyResult<PyPythonTask> {
@@ -387,8 +396,12 @@ impl PythonActorMeshImpl {
     fn supervision_event(&self, instance: &PyInstance) -> PyResult<Option<PyShared>> {
         ActorMeshProtocol::supervision_event(self, instance)
     }
-    fn start_supervision(&self, instance: &PyInstance) -> PyResult<()> {
-        ActorMeshProtocol::start_supervision(self, instance)
+    fn start_supervision(
+        &self,
+        instance: &PyInstance,
+        supervision_display_name: String,
+    ) -> PyResult<()> {
+        ActorMeshProtocol::start_supervision(self, instance, supervision_display_name)
     }
 
     fn stop(&self, instance: &PyInstance) -> PyResult<PyPythonTask> {
@@ -543,6 +556,8 @@ impl Drop for PythonActorMeshImpl {
         self.monitor.abort();
     }
 }
+
+#[derive(Debug)]
 struct ClonePyErr {
     inner: PyErr,
 }
@@ -669,18 +684,28 @@ impl ActorMeshProtocol for AsyncActorMesh {
         if !self.supervised {
             return Ok(None);
         }
-        let mesh = self.mesh.clone();
         let instance = Python::with_gil(|_py| instance.clone());
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        let mesh = self.mesh.clone();
+        self.push(async move {
+            if tx.send(mesh.await.unwrap()).is_err() {
+                panic!("oneshot failed");
+            }
+        });
         PyPythonTask::new(async move {
-            let mut event = mesh.await?.supervision_event(&instance)?.unwrap();
+            let mut event = rx.await.unwrap().supervision_event(&instance)?.unwrap();
             event.task()?.take_task()?.await
         })
         // This task must be aborted to run the Drop for the inner PyShared, in
         // case that one is also abortable.
         .map(|mut x| x.spawn_abortable().map(Some))?
     }
 
-    fn start_supervision(&self, instance: &PyInstance) -> PyResult<()> {
+    fn start_supervision(
+        &self,
+        instance: &PyInstance,
+        supervision_display_name: String,
+    ) -> PyResult<()> {
         if !self.supervised {
             return Ok(());
         }
@@ -689,7 +714,8 @@ impl ActorMeshProtocol for AsyncActorMesh {
         self.push(async move {
             let mesh = mesh.await;
             if let Ok(mesh) = mesh {
-                mesh.start_supervision(&instance).unwrap();
+                mesh.start_supervision(&instance, supervision_display_name)
+                    .unwrap();
             }
         });
         Ok(())

monarch_hyperactor/src/context.rs

@@ -131,7 +131,7 @@ impl PyInstance {
     }
 
     #[getter]
-    fn actor_id(&self) -> PyActorId {
+    pub fn actor_id(&self) -> PyActorId {
         self.inner.self_id().clone().into()
     }
 }

monarch_hyperactor/src/v1/actor_mesh.rs

@@ -178,7 +178,12 @@ impl PythonActorMeshImpl {
         }
     }
 
-    fn make_monitor<F>(&self, instance: PyInstance, unhandled: F) -> SupervisionMonitor
+    fn make_monitor<F>(
+        &self,
+        instance: PyInstance,
+        unhandled: F,
+        supervision_display_name: String,
+    ) -> SupervisionMonitor
     where
         F: Fn(MeshFailure) + Send + 'static,
     {
@@ -190,6 +195,7 @@ impl PythonActorMeshImpl {
                 inner.health_state.clone(),
                 true,
                 unhandled,
+                supervision_display_name,
             ),
             // Ref meshes send no message, they are only used to generate
             // the v0 style exception.
@@ -199,6 +205,7 @@ impl PythonActorMeshImpl {
                 inner.health_state.clone(),
                 false,
                 unhandled,
+                supervision_display_name,
             ),
         }
     }
@@ -211,14 +218,19 @@ impl PythonActorMeshImpl {
         instance: &PyInstance,
         monitor: &Arc<Mutex<Option<SupervisionMonitor>>>,
         unhandled: F,
+        supervision_display_name: Option<String>,
     ) -> watch::Receiver<Option<PyErr>>
     where
         F: Fn(MeshFailure) + Send + 'static,
     {
         let mut guard = monitor.lock().unwrap();
         guard.get_or_insert_with(move || {
             let instance = Python::with_gil(|_py| instance.clone());
-            self.make_monitor(instance, unhandled)
+            self.make_monitor(
+                instance,
+                unhandled,
+                supervision_display_name.unwrap_or_default(),
+            )
         });
         let monitor = guard.as_ref().unwrap();
         monitor.receiver.clone()
@@ -296,6 +308,7 @@ impl PythonActorMeshImpl {
         health_state: Arc<RootHealthState>,
         is_owned: bool,
         unhandled: F,
+        supervision_display_name: String,
     ) -> SupervisionMonitor
     where
         F: Fn(MeshFailure) + Send + 'static,
@@ -324,6 +337,7 @@ impl PythonActorMeshImpl {
                         time_between_checks,
                         sender,
                         canceled,
+                        supervision_display_name.clone(),
                     )
                     .await;
                 }
@@ -345,6 +359,7 @@ impl PythonActorMeshImpl {
                         time_between_checks,
                         sender,
                         canceled,
+                        supervision_display_name,
                     )
                     .await;
                 }
@@ -492,6 +507,7 @@ async fn actor_states_monitor<A, F>(
     time_between_checks: tokio::time::Duration,
     sender: watch::Sender<Option<PyErr>>,
     canceled: CancellationToken,
+    supervision_display_name: String,
 ) where
     A: Actor + RemotableActor + Referable,
     A::Params: RemoteMessage,
@@ -554,24 +570,29 @@ async fn actor_states_monitor<A, F>(
                         status
                     ))),
                 };
-
+                let display_name = if !point.is_empty() {
+                    let coords_display = point.format_as_dict();
+                    if let Some(pos) = supervision_display_name.rfind('>') {
+                        format!(
+                            "{}{}{}",
+                            &supervision_display_name[..pos],
+                            coords_display,
+                            &supervision_display_name[pos..]
+                        )
+                    } else {
+                        format!("{}{}", supervision_display_name, coords_display)
+                    }
+                } else {
+                    supervision_display_name.clone()
+                };
                 send_state_change(
                     point.rank(),
                     ActorSupervisionEvent::new(
                         // Attribute this to the monitored actor, even if the underlying
                         // cause is a proc_failure. We propagate the cause explicitly.
                         mesh.get(point.rank()).unwrap().actor_id().clone(),
-                        None,
+                        Some(format!("{} was running on a process which", display_name)),
                         actor_status,
-                        // ActorStatus::Failed(ActorErrorKind::Generic(format!(
-                        //     "process failure: {}",
-                        //     state
-                        //         .state
-                        //         .and_then(|state| state.proc_status)
-                        //         .unwrap_or_else(|| ProcStatus::Failed {
-                        //             reason: "unknown".to_string()
-                        //         })
-                        // ))),
                         None,
                     ),
                     mesh.name(),
@@ -593,7 +614,7 @@ async fn actor_states_monitor<A, F>(
                 0,
                 ActorSupervisionEvent::new(
                     cx.instance().self_id().clone(),
-                    None,
+                    Some(supervision_display_name.clone()),
                     ActorStatus::generic_failure(format!(
                         "unable to query for actor states: {:?}",
                         e
@@ -720,7 +741,7 @@ impl ActorMeshProtocol for PythonActorMeshImpl {
         // Make a clone so each endpoint can get the same supervision events.
         let unhandled = self.get_unhandled(instance);
         let monitor = self.monitor().clone();
-        let mut receiver = self.supervision_receiver(instance, &monitor, unhandled);
+        let mut receiver = self.supervision_receiver(instance, &monitor, unhandled, None);
         PyPythonTask::new(async move {
             receiver.changed().await.map_err(|e| {
                 PyValueError::new_err(format!("Waiting for supervision event change: {}", e))
@@ -744,11 +765,20 @@ impl ActorMeshProtocol for PythonActorMeshImpl {
         .map(Some)
     }
 
-    fn start_supervision(&self, instance: &PyInstance) -> PyResult<()> {
+    fn start_supervision(
+        &self,
+        instance: &PyInstance,
+        supervision_display_name: String,
+    ) -> PyResult<()> {
         // Fetch the receiver once, this will initialize the monitor task.
         let unhandled = self.get_unhandled(instance);
         let monitor = self.monitor().clone();
-        self.supervision_receiver(instance, &monitor, unhandled);
+        self.supervision_receiver(
+            instance,
+            &monitor,
+            unhandled,
+            Some(supervision_display_name),
+        );
         Ok(())
     }
 
@@ -827,7 +857,11 @@ impl ActorMeshProtocol for ActorMeshRef<PythonActor> {
         ))
     }
 
-    fn start_supervision(&self, _instance: &PyInstance) -> PyResult<()> {
+    fn start_supervision(
+        &self,
+        _instance: &PyInstance,
+        _supervision_display_name: String,
+    ) -> PyResult<()> {
         Err(PyErr::new::<PyNotImplementedError, _>(
             "This should never be called on ActorMeshRef directly",
         ))

ndslice/src/view.rs

@@ -769,6 +769,31 @@ impl Point {
     pub fn is_empty(&self) -> bool {
         self.extent.len() == 0
     }
+
+    /// Formats the coordinates of this Point as a string suitable for display names.
+    /// Returns a string in the format: "{'label': coord/size, 'label': coord/size, ...}"
+    ///
+    /// # Examples
+    /// ```
+    /// use ndslice::extent;
+    ///
+    /// let ext = extent!(x = 2, y = 3);
+    /// let point = ext.point(vec![1, 2]).unwrap();
+    /// assert_eq!(point.format_as_dict(), "{'x': 1/2, 'y': 2/3}");
+    /// ```
+    pub fn format_as_dict(&self) -> String {
+        format!(
+            "{{{}}}",
+            self.extent()
+                .labels()
+                .iter()
+                .zip(self.coords_iter())
+                .zip(self.extent().sizes())
+                .map(|((label, coord), size)| format!("'{}': {}/{}", label, coord, size))
+                .collect::<Vec<_>>()
+                .join(", ")
+        )
+    }
 }
 
 /// Formats a `Point` as a comma-separated list of per-axis

python/monarch/_rust_bindings/monarch_hyperactor/actor_mesh.pyi

@@ -31,7 +31,9 @@ class ActorMeshProtocol(Protocol):
         self, instance: Instance
     ) -> "Optional[Shared[Exception]]": ...
     # Starts supervision monitoring for future uses of "supervision_event".
-    def start_supervision(self, instance: Instance) -> None: ...
+    def start_supervision(
+        self, instance: Instance, supervision_display_name: str
+    ) -> None: ...
     def stop(self, instance: Instance) -> PythonTask[None]: ...
     def initialized(self) -> PythonTask[None]: ...
 

python/monarch/_src/actor/actor_mesh.py

@@ -468,7 +468,9 @@ def new_with_region(self, region: Region) -> "ActorMeshProtocol":
     def supervision_event(self, instance: HyInstance) -> "Optional[Shared[Exception]]":
         return None
 
-    def start_supervision(self, instance: HyInstance) -> None:
+    def start_supervision(
+        self, instance: HyInstance, supervision_display_name: str
+    ) -> None:
         return None
 
     def stop(self, instance: HyInstance) -> "PythonTask[None]":
@@ -1255,10 +1257,6 @@ def __init__(
         self._inner: "ActorMeshProtocol" = inner
         self._shape = shape
         self._proc_mesh = proc_mesh
-        # We don't start the supervision polling loop until the first call to
-        # supervision_event, which needs an Instance. Initialize here so events
-        # can be collected even without any endpoints being awaited.
-        self._inner.start_supervision(context().actor_instance._as_rust())
 
         async_endpoints = []
         sync_endpoints = []
@@ -1347,6 +1345,15 @@ def _create(
     ) -> "ActorMesh[T]":
         mesh = cls(Class, actor_mesh, shape, proc_mesh)
 
+        # We don't start the supervision polling loop until the first call to
+        # supervision_event, which needs an Instance. Initialize here so events
+        # can be collected even without any endpoints being awaited.
+        instance = context().actor_instance
+        supervision_display_name = (
+            f"{str(instance)}.<{Class.__module__}.{Class.__name__} {name}>"
+        )
+        mesh._inner.start_supervision(instance._as_rust(), supervision_display_name)
+
         async def null_func(*_args: Iterable[Any], **_kwargs: Dict[str, Any]) -> None:
             return None
 

python/tests/test_supervision_hierarchy.py

@@ -27,7 +27,9 @@ def run(self, func: Callable[[], T]) -> T:
 
 class Nest(Actor):
     def __init__(self):
-        self.nest = this_host().spawn_procs().spawn("nested", Lambda)
+        self.nest = (
+            this_host().spawn_procs(per_host={"a_dim": 1}).spawn("nested", Lambda)
+        )
 
     # pyre-ignore[56]
     @endpoint
@@ -112,10 +114,10 @@ def test_proc_failure():
     If a proc dies, the client should receive an unhandled fault.
     """
     with FaultCapture() as capture:
-        actor = this_host().spawn_procs().spawn("actor", Nest)
+        actor = this_host().spawn_procs().spawn("top", Nest)
         actor.kill_nest.call_one().get()
 
-    capture.assert_fault_occurred()
+    capture.assert_fault_occurred("nested{'a_dim': 0/1}")
 
 
 def test_nested_mesh_kills_actor_actor_error():
@@ -129,5 +131,5 @@ def test_nested_mesh_kills_actor_actor_error():
         actor.nested.call_one(error).get()
         print("ERRORED THE ACTOR")
     capture.assert_fault_occurred(
-        "actor <root>.<tests.test_supervision_hierarchy.Nest actor>.<tests.test_supervision_hierarchy.Lambda nested> failed"
+        "actor <root>.<tests.test_supervision_hierarchy.Nest actor>.<tests.test_supervision_hierarchy.Lambda nested{'a_dim': 0/1}> failed"
     )