[zhou9402]

Ports miles' eval fleet: BaseRolloutFn / RolloutFn*Input contract types, non-blocking eval dispatch with HF snapshot export, a dedicated eval fleet on ServingCluster (--eval-num-gpus), and the CheckpointEvalFn contract with an external-vLLM reference implementation. miles' dead "receiver pools" were not ported.

Source: radixark/miles@ba28991834. Port is additive (new modules + new add_*_arguments() groups); slime-origin files are not restructured so future [Sync] Mirror Slime PRs stay clean. One commit per sub-PR; review each commit separately.

Commits:

• 21ee80b BaseRolloutFn and RolloutFn*Input contract types in rollout base_types
• bacecf1 Non-blocking eval dispatch + HF snapshot export plumbing
• a9930f6 Dedicated eval fleet on ServingCluster + --eval-num-gpus
• 6d9ecec CheckpointEvalFn contract + external vLLM reference implementation

Non-blocking timing e2e needs GPU.

Merge order of the miles-port series: F10 → F02 → F06 → F11a → F11b → F05 → F09 → F04 → F03 → F07 → F08 → F01. train.py / vime/utils/arguments.py are touched by several branches; later ones are rebased after each merge.

🤖 Generated with Claude Code

[read-the-docs-community]

Documentation build overview

📚 vime | 🛠️ Build #35444825 | 📁 Comparing 6d9ecec against latest (d7e40fe)

  🔍 Preview build  

59 files changed · + 1 added · ± 58 modified

+ Added

• advanced/async-eval.html

± Modified

• genindex.html
• library.html
• search.html
• advanced/arch-support-beyond-megatron.html
• advanced/delta-weight-sync.html
• advanced/external-rollout-engines.html
• advanced/fault-tolerance.html
• advanced/low-precision.html
• advanced/megatron-config.html
• advanced/observability.html
• and 48 more...

[gemini-code-assist]

Code Review

This pull request introduces a non-blocking, snapshot-based evaluation dispatch system to prevent evaluation from stalling the training loop. It adds support for a dedicated eval engine fleet or custom CheckpointEvalFn backends, along with comprehensive tests and documentation. The review feedback highlights several critical robustness improvements: preventing indefinite hangs during teardown in close() by adding a timeout to drain(), cleaning up partial directories on failed HF snapshot exports to avoid disk space leaks, wrapping weight version retrieval in a try-except block to handle unresponsive engines gracefully, and catching request exceptions during weight info polling in the external vLLM evaluation function.

[gemini-code-assist]

The close() method is designed to handle a stuck thread gracefully by joining with a 30-second timeout. However, it calls self.drain() first, which blocks the caller thread indefinitely on future.result() if any evaluation task is hanging. This completely bypasses the thread-joining timeout and can cause the main training thread to hang forever during teardown.

To fix this, we should add an optional timeout parameter to drain() and call it with a 30-second timeout inside close(), catching any timeout or other exceptions so that the thread-joining and cleanup logic can still run.

Suggested change

	def drain(self) -> None:
	future = asyncio.run_coroutine_threadsafe(self._dispatcher.drain(), self._loop)
	future.result()
	def close(self) -> bool:
	"""Drain and stop the loop. Returns False when the thread is stuck; in
	that case the eval runner is left alone because it may be mid-RPC."""
	self.drain()
	self._loop.call_soon_threadsafe(self._loop.stop)
	self._thread.join(timeout=30)
	if self._thread.is_alive():
	logger.error("Eval dispatch thread did not stop within 30s; leaving the eval runner undisposed.")
	return False
	if self._on_close is not None:
	self._on_close()
	return True
	def drain(self, timeout: float | None = None) -> None:
	future = asyncio.run_coroutine_threadsafe(self._dispatcher.drain(), self._loop)
	future.result(timeout=timeout)
	def close(self) -> bool:
	"""Drain and stop the loop. Returns False when the thread is stuck; in
	that case the eval runner is left alone because it may be mid-RPC."""
	try:
	self.drain(timeout=30)
	except Exception as e:
	logger.error(f"Timeout or error while draining eval dispatcher: {e}")
	self._loop.call_soon_threadsafe(self._loop.stop)
	self._thread.join(timeout=30)
	if self._thread.is_alive():
	logger.error("Eval dispatch thread did not stop within 30s; leaving the eval runner undisposed.")
	return False
	if self._on_close is not None:
	self._on_close()
	return True

[gemini-code-assist]

If the HF snapshot export fails during _export(), any partially written files or directories in hf_dir are left behind. Since this directory is never added to self._exported, it will never be cleaned up by the garbage collection loop in _retire(). Over time, repeated export failures can leak disk space, which is especially problematic if eval_hf_dir is configured on a RAM disk (tmpfs).

We should clean up the partial export directory on failure using shutil.rmtree(hf_dir, ignore_errors=True).

Suggested change

	try:
	export_time = await self._export(rollout_id, hf_dir)
	except Exception as e:
	logger.error(f"HF snapshot export for eval {rollout_id} failed: {e}")
	await self.rollout_manager.report_eval_skip(rollout_id, "export_failed")
	return
	try:
	export_time = await self._export(rollout_id, hf_dir)
	except Exception as e:
	logger.error(f"HF snapshot export for eval {rollout_id} failed: {e}")
	await self.rollout_manager.report_eval_skip(rollout_id, "export_failed")
	shutil.rmtree(hf_dir, ignore_errors=True)
	return

[gemini-code-assist]

Unlike the update_weights_from_disk and health_generate calls in pin_eval_fleet(), the get_weight_version gather is not wrapped in a try...except block. If any engine becomes unresponsive or raises an exception during this call, it will crash the entire evaluation task instead of returning a clean EvalFleetPin(skip_reason="unhealthy").

We should wrap this gather in a try...except block to match the robustness of the other fleet operations.

Suggested change

	versions = gather([engine.get_weight_version.remote() for engine in engines])
	if any(str(v) != str(weight_version) for v in versions):
	logger.error(f"Eval fleet pin violation: versions {versions} != {weight_version}")
	return EvalFleetPin(skip_reason="pin_violation")
	try:
	versions = gather([engine.get_weight_version.remote() for engine in engines])
	except Exception:
	logger.exception("Eval fleet failed to get weight versions")
	return EvalFleetPin(skip_reason="unhealthy")
	if any(str(v) != str(weight_version) for v in versions):
	logger.error(f"Eval fleet pin violation: versions {versions} != {weight_version}")
	return EvalFleetPin(skip_reason="pin_violation")

[gemini-code-assist]

During weight reloading, the external vLLM server might be temporarily unresponsive or return connection errors. If requests.get() raises a requests.RequestException (such as a connection timeout or connection refused), the loop will immediately crash and fail the evaluation run instead of retrying until the deadline.

We should wrap the requests.get() call in a try...except requests.RequestException: block to ensure robustness during server reloads.

Suggested change

	while time.time() < deadline:
	served = requests.get(f"{self._url}/weight_info", timeout=30).json().get("weight_version")
	if str(served) == str(weight_version):
	return
	time.sleep(5)
	raise EvalSkip("pin_violation")
	while time.time() < deadline:
	try:
	response = requests.get(f"{self._url}/weight_info", timeout=30)
	if response.ok:
	served = response.json().get("weight_version")
	if str(served) == str(weight_version):
	return
	except requests.RequestException:
	pass
	time.sleep(5)
	raise EvalSkip("pin_violation")