09 · Lock、Event 与线程退出

        """Stop the health monitor thread completely. Called during dispose."""
        if not self._thread:
            return

        logger.info("Stopping RolloutHealthMonitor...")
        self._stop_event.set()
        if timeout is None:
            timeout = self._check_timeout + 5
        self._thread.join(timeout=timeout)
        if self._thread.is_alive():
            raise TimeoutError(f"Rollout health monitor did not terminate within {timeout:.1f}s")
        else:
            logger.info("RolloutHealthMonitor stopped.")

        self._thread = None

    def _heartbeat_loop(self):
        interval = max(0.1, getattr(self.args, "rollout_queue_lease_seconds", 300) / 3)
        while not self._stop.wait(interval):
            with self._lock:
                leases = list(self._leases.values())
            try:
                for offset in range(0, len(leases), 128):
                    batch = leases[offset : offset + 128]