Delete the rollout health monitor ahead of its rewrite

miles/ray/rollout/inference_controller.py

@@ -1,4 +1,3 @@
-import asyncio
 import logging
 from dataclasses import dataclass
 
@@ -12,7 +11,6 @@
 from miles.ray.rollout.router_manager import start_session_server
 from miles.ray.rollout.server_cell import get_cell_indexer_of_id_map
 from miles.ray.utils import Lock
-from miles.utils.health_monitor import RolloutHealthMonitor
 
 
 logger = logging.getLogger(__name__)
@@ -33,33 +31,20 @@ def __init__(self, args, pg):
         self.rollout_id = -1
         self.eval_fleet = EvalFleet(args, srv=self.servers["eval"]) if args.eval_num_gpus > 0 else None
 
-        # TODO will be replaced by full ft, thus temporarily leave it without modifications
-        self._health_monitors = []
-        self._rollout_ft_enabled = self.args.use_fault_tolerance and "rollout" in self.args.ft_components
-        self._ci_fault_injection_pending = False
-        if not self.args.debug_train_only and self._rollout_ft_enabled:
-            for srv in self.servers.values():
-                for group in srv.server_groups:
-                    monitor = RolloutHealthMonitor(group, args)
-                    monitor.start()
-                    self._health_monitors.append(monitor)
-            self._ci_fault_injection_pending = self.args.ci_test
-
     # -------------------------- rollout lifecycle hooks -----------------------------
 
     async def prepare_rollout(self, rollout_id):
         self.rollout_id = rollout_id
-        self._health_monitoring_resume()
+        await self.health_monitoring_resume()
         if self.args.ci_test and self._rollout_ft_enabled and rollout_id >= 2:
             await self._try_ci_fault_injection()
         dashboard_hooks.register_engines(self.servers)
 
     async def prepare_eval(self):
-        self._health_monitoring_resume()
+        await self.health_monitoring_resume()
 
     async def dispose(self):
-        for monitor in self._health_monitors:
-            monitor.stop()
+        pass
 
     # -------------------------- offload/onload -----------------------------
 
@@ -176,12 +161,21 @@ async def check_weights(
     # -------------------------- utils -----------------------------
 
     async def health_monitoring_pause(self) -> None:
-        for monitor in self._health_monitors:
-            monitor.pause()
+        self._assert_rollout_fault_tolerance_is_unsupported()
 
-    def _health_monitoring_resume(self) -> None:
-        for monitor in self._health_monitors:
-            monitor.resume()
+    async def health_monitoring_resume(self) -> None:
+        self._assert_rollout_fault_tolerance_is_unsupported()
+
+    @property
+    def _rollout_ft_enabled(self) -> bool:
+        return self.args.use_fault_tolerance and "rollout" in self.args.ft_components
+
+    def _assert_rollout_fault_tolerance_is_unsupported(self) -> None:
+        if not self.args.debug_train_only and self._rollout_ft_enabled:
+            raise NotImplementedError(
+                "rollout fault tolerance is being rebuilt; health monitoring must pause before "
+                "get_updatable_engines_and_lock snapshots the engines"
+            )
 
     @property
     def _server(self) -> RolloutServer | None:
@@ -190,31 +184,8 @@ def _server(self) -> RolloutServer | None:
             return None
         return next(iter(self.servers.values()))
 
-    # TODO will be replaced by full ft, thus temporarily leave it without modifications
     async def _try_ci_fault_injection(self):
-        """Try to inject fault during generate (when health monitor is running)."""
-        if not self._ci_fault_injection_pending:
-            return
-
-        # Only inject fault once
-        self._ci_fault_injection_pending = False
-
-        if (
-            self._server
-            and self._server.server_groups[0].all_engines
-            and self._server.server_groups[0].all_engines[0].is_allocated
-        ):
-            logger.info("CI Fault Injection: Simulating crash on engine 0 during generate")
-            try:
-                # This will cause the ray actor to exit
-                self._server.server_groups[0].all_engines[0].actor_handle.simulate_crash.remote()
-                # Wait for health monitor to detect the crash and mark engine as None
-                # health_check_interval + health_check_timeout + buffer
-                wait_time = self.args.rollout_health_check_interval + self.args.rollout_health_check_timeout + 5
-                logger.info(f"CI Fault Injection: Waiting {wait_time}s for health monitor to detect crash")
-                await asyncio.sleep(wait_time)
-            except Exception as e:
-                logger.warning(f"CI Fault Injection failed: {e}")
+        raise NotImplementedError("rollout fault injection is being rebuilt with rollout fault tolerance")
 
 
 @dataclass(frozen=True)

miles/ray/rollout/server_group.py

@@ -169,17 +169,11 @@ def start_engines(
         ]
         return init_handles, new_engine_indices
 
-    # There are two callers, only one of them will exist in a running system
-    # 1. For new callers (InferenceController.stop_cell, main thread, async),
-    #    deliberately make this function non-async here to avoid introducing two states
-    #    like "stopping (but not stopped)" vs "stopped", since single-thread async code will not yield
-    #    without an await point
-    #    it has the drawback of freezing the whole async thread, which may be avoided later by
-    #    moving `shutdown` mainly to local code
-    # 2. For legacy callers (RolloutHealthMonitor, another thread, sync)
-    #    it is still unsafe to be called in another thread
-    #    because engine may be observed as non-stopped while being shutdown,
-    #    but that is same as the original code
+    # Called from InferenceController.stop_cell (main thread, async): deliberately non-async here
+    # to avoid introducing two states like "stopping (but not stopped)" vs "stopped", since
+    # single-thread async code will not yield without an await point
+    # it has the drawback of freezing the whole async thread, which may be avoided later by
+    # moving `shutdown` mainly to local code
     def stop_engines(self, engine_indices: list[int]):
         logger.info(f"Killing server {engine_indices=}...")
         for i in engine_indices:

miles/utils/arguments.py

@@ -3671,7 +3671,7 @@ def _maybe_apply_dumper_overrides(args) -> None:
         return
 
     if args.use_fault_tolerance:
-        logger.info("Dumper mode: disabling --use-fault-tolerance to suppress RolloutHealthMonitor heartbeats")
+        logger.info("Dumper mode: disabling --use-fault-tolerance to suppress fault tolerance heartbeats")
         args.use_fault_tolerance = False
 
     logger.info("Dumper mode: all heartbeat mechanisms disabled")

miles/utils/ft_utils/health_checker.py

@@ -83,8 +83,7 @@ def resume(self) -> None: ...
 class SimpleHealthChecker(BaseHealthChecker):
     """Periodic async health checker. Calls *check_fn*; reports result via *on_result*.
 
-    After each ``resume()``, waits ``first_wait`` seconds before the first check
-    (matching ``RolloutHealthMonitor._need_first_wait`` semantics).
+    After each ``resume()``, waits ``first_wait`` seconds before the first check.
     """
 
     def __init__(

tests/fast/ray/rollout/conftest.py

@@ -94,6 +94,7 @@ def make_args(**overrides: Any) -> Namespace:
         # offload / fault tolerance
         offload_rollout=False,
         use_fault_tolerance=False,
+        ft_components=[],
         rollout_health_check_interval=10.0,
         rollout_health_check_timeout=30.0,
         # checkpoint / data source