remove controller health monitoring (#2069)

Summary:
Pull Request resolved: #2069

this removes some of the dead code mariusae talks about in D88434964.

recover D88441603. cleanup all the dead controller supervision machinery that was left behind after hyperactor_multiprocess crate was deleted. the supervision check was already stubbed out, `ControllerError` was never constructed, and the controller fields were write-only.

Reviewed By: vidhyav, mariusae

Differential Revision: D88446442

fbshipit-source-id: b61b36d62e8c27dacca180324d15647e7874c64a

Author: shayne-fletcher

monarch_extension/src/client.rs

@@ -11,7 +11,6 @@ use std::sync::Arc;
 use hyperactor::ActorRef;
 use hyperactor::data::Serialized;
 use monarch_hyperactor::ndslice::PySlice;
-use monarch_hyperactor::proc::ControllerError;
 use monarch_hyperactor::proc::InstanceWrapper;
 use monarch_hyperactor::proc::PyActorId;
 use monarch_hyperactor::proc::PyProc;
@@ -450,8 +449,7 @@ impl ClientActor {
 
     /// Attach the client to a controller actor. This will block until the controller responds.
     fn attach(&mut self, py: Python, controller_id: PyActorId) -> PyResult<()> {
-        let mut instance_wrapper = self.instance.blocking_lock();
-        instance_wrapper.set_controller((&controller_id).into());
+        let instance_wrapper = self.instance.blocking_lock();
         let actor_id = instance_wrapper.actor_id().clone();
         let (instance, _handler) = instance_wrapper
             .instance()
@@ -495,44 +493,25 @@ impl ClientActor {
             instance.lock().await.next_message(timeout_msec).await
         })?;
 
-        Python::with_gil(|py| {
-            match result {
-                Ok(Some(ClientMessage::Result { seq, result })) => {
-                    WorkerResponse { seq, result }.into_py_any(py)
-                }
-                Ok(Some(ClientMessage::Log { level, message })) => LogMessage {
-                    level: PyLogLevel::from(level),
-                    message,
-                }
-                .into_py_any(py),
-                Ok(Some(ClientMessage::DebuggerMessage {
-                    debugger_actor_id,
-                    action,
-                })) => DebuggerMessage {
-                    debugger_actor_id: debugger_actor_id.into(),
-                    action,
-                }
-                .into_py_any(py),
-                Ok(None) => PyNone::get(py).into_py_any(py),
-                Err(err) => {
-                    if let Some(ControllerError::Failed(controller_id, err_msg)) =
-                        err.downcast_ref::<ControllerError>()
-                    {
-                        let failure = DeviceFailure {
-                            actor_id: controller_id.clone(),
-                            address: "".to_string(), // Controller is always task 0 for now.
-                            backtrace: err_msg.clone(),
-                        };
-                        WorkerResponse {
-                            seq: Seq::default(),
-                            result: Some(Err(Exception::Failure(failure))),
-                        }
-                        .into_py_any(py)
-                    } else {
-                        Err(PyRuntimeError::new_err(err.to_string()))
-                    }
-                }
+        Python::with_gil(|py| match result {
+            Ok(Some(ClientMessage::Result { seq, result })) => {
+                WorkerResponse { seq, result }.into_py_any(py)
+            }
+            Ok(Some(ClientMessage::Log { level, message })) => LogMessage {
+                level: PyLogLevel::from(level),
+                message,
+            }
+            .into_py_any(py),
+            Ok(Some(ClientMessage::DebuggerMessage {
+                debugger_actor_id,
+                action,
+            })) => DebuggerMessage {
+                debugger_actor_id: debugger_actor_id.into(),
+                action,
             }
+            .into_py_any(py),
+            Ok(None) => PyNone::get(py).into_py_any(py),
+            Err(err) => Err(PyRuntimeError::new_err(err.to_string())),
         })
     }
 

monarch_hyperactor/Cargo.toml

@@ -51,7 +51,6 @@ serde = { version = "1.0.219", features = ["derive", "rc"] }
 serde_bytes = "0.11"
 serde_multipart = { version = "0.0.0", path = "../serde_multipart" }
 tempfile = "3.22"
-thiserror = "2.0.12"
 tokio = { version = "1.47.1", features = ["full", "test-util", "tracing"] }
 tokio-util = { version = "0.7.15", features = ["full"] }
 tracing = { version = "0.1.41", features = ["attributes", "valuable"] }

monarch_hyperactor/src/proc.rs

@@ -10,15 +10,13 @@ use std::hash::DefaultHasher;
 use std::hash::Hash;
 use std::hash::Hasher;
 use std::time::Duration;
-use std::time::SystemTime;
 
 use anyhow::Result;
 use hyperactor::RemoteMessage;
 use hyperactor::actor::Signal;
 use hyperactor::clock::Clock;
 use hyperactor::clock::ClockKind;
 use hyperactor::data::Serialized;
-use hyperactor::mailbox::PortHandle;
 use hyperactor::mailbox::PortReceiver;
 use hyperactor::proc::Instance;
 use hyperactor::proc::Proc;
@@ -32,8 +30,6 @@ use pyo3::exceptions::PyValueError;
 use pyo3::prelude::*;
 use pyo3::types::PyList;
 use pyo3::types::PyType;
-use tokio::sync::OnceCell;
-use tokio::sync::watch;
 
 use crate::actor::PythonActor;
 use crate::actor::PythonActorHandle;
@@ -317,23 +313,10 @@ pub struct InstanceWrapper<M: RemoteMessage> {
     signal_receiver: PortReceiver<Signal>,
     status: InstanceStatus,
 
-    // TODO(T216450632): merge actor.rs and client.rs in monarch_extension
-    signal_port: PortHandle<Signal>,
-    last_controller_status_check: SystemTime,
-    controller_id: OnceCell<ActorId>,
-    controller_error_sender: watch::Sender<String>,
-    controller_error_receiver: watch::Receiver<String>,
     clock: ClockKind,
     actor_id: ActorId,
 }
 
-/// Error that can occur when there is controller supervision error.
-#[derive(thiserror::Error, Debug)]
-pub enum ControllerError {
-    #[error("controller actor {0} failed: {1}")]
-    Failed(ActorId, String),
-}
-
 impl<M: RemoteMessage> InstanceWrapper<M> {
     pub fn new(proc: &PyProc, actor_name: &str) -> Result<Self> {
         InstanceWrapper::new_with_instance_and_clock(
@@ -346,30 +329,20 @@ impl<M: RemoteMessage> InstanceWrapper<M> {
         // TEMPORARY: remove after using fixed message ports.
         let (_message_port, message_receiver) = instance.bind_actor_port::<M>();
 
-        let (signal_port, signal_receiver) = instance.bind_actor_port::<Signal>();
+        let (_signal_port, signal_receiver) = instance.bind_actor_port::<Signal>();
 
-        let (controller_error_sender, controller_error_receiver) = watch::channel("".to_string());
         let actor_id = instance.self_id().clone();
 
         Ok(Self {
             instance,
             message_receiver,
             signal_receiver,
             status: InstanceStatus::Running,
-            signal_port,
-            last_controller_status_check: clock.system_time_now(),
-            controller_id: OnceCell::new(),
-            controller_error_sender,
-            controller_error_receiver,
             clock,
             actor_id,
         })
     }
 
-    pub fn set_controller(&mut self, controller_id: ActorId) {
-        self.controller_id.set(controller_id).unwrap();
-    }
-
     /// Send a message to any actor. It is the responsibility of the caller to ensure the right
     /// payload accepted by the target actor has been serialized and provided to this function.
     pub fn send(&self, actor_id: &PyActorId, message: &PySerialized) -> PyResult<()> {