From 7576c4ceb98c2219a3f54b295d27be8b5fed3e31 Mon Sep 17 00:00:00 2001 From: Hongsheng Liu Date: Thu, 10 Sep 2026 17:56:27 +0800 Subject: [PATCH] [Bugfix] Bound HWR domain initialization lock waits (#7128) Signed-off-by: Hongsheng Liu --- docs/design/feature/host_weight_runtime.md | 11 +- docs/design/module/host_weight_runtime.md | 12 +- .../test_runtime_resolution.py | 108 ++++++++++++++++++ .../host_weight_runtime/filesystem/store.py | 10 +- vllm_omni/host_weight_runtime/runtime.py | 2 +- 5 files changed, 134 insertions(+), 9 deletions(-) diff --git a/docs/design/feature/host_weight_runtime.md b/docs/design/feature/host_weight_runtime.md index a9bb1b0b8..ac7bc7b3f 100644 --- a/docs/design/feature/host_weight_runtime.md +++ b/docs/design/feature/host_weight_runtime.md @@ -313,8 +313,15 @@ One process per exact identity owns a build; other workers wait and then acquire leases for the published artifact. Publication is invisible until all payloads and metadata are validated, hashed, fsynced, and atomically renamed. -`coordination_timeout_seconds` bounds filesystem lock acquisition. It does not -cancel synchronous validation, a producer that has already started, or atomic +`coordination_timeout_seconds` bounds domain-initialization and lookup/build +lock acquisition. Store construction and each later resolution or publication +operation have separate budgets from the same wait policy, rather than one +end-to-end startup deadline. A domain-init timeout follows the retryable domain +failure policy below. After contention ends, a fresh runtime construction can +retry initialization; the timed-out runtime retains its original failure. + +The coordination budget does not cancel filesystem I/O, synchronous validation, +a producer that has already started, or atomic publication. A hung in-process producer therefore blocks its owning process and must be handled by external process supervision. Enforceable producer cancellation requires a future process-isolated producer contract. diff --git a/docs/design/module/host_weight_runtime.md b/docs/design/module/host_weight_runtime.md index ac92711fc..c97561667 100644 --- a/docs/design/module/host_weight_runtime.md +++ b/docs/design/module/host_weight_runtime.md @@ -136,8 +136,16 @@ so only when it is marked retryable; unsupported capabilities, identity collisions, producer failures, and publication failures remain visible as startup failures. -`coordination_timeout_seconds` bounds acquisition of lookup and build locks; it -is not a hard wall-clock deadline for in-process validation, production, or +`coordination_timeout_seconds` bounds acquisition of the domain-initialization +lock and lookup/build locks. Store construction and each later resolution or +publication operation receive separate coordination budgets from the same +`WaitPolicy`; they do not share an end-to-end startup deadline. A domain-init +lock timeout is a retryable domain failure: preferred mode may fall back and +required mode fails. Retrying initialization requires constructing a fresh +runtime after contention ends; a waiter never cancels or unlocks the owner. + +The coordination budget is not a hard wall-clock deadline for filesystem I/O, +in-process validation, production, or atomic publication. Once this V1 implementation becomes the producer, the synchronous producer and publication run to completion. A hung producer blocks its owning process and requires external process supervision; enforceable diff --git a/tests/host_weight_runtime/test_runtime_resolution.py b/tests/host_weight_runtime/test_runtime_resolution.py index 21bab37de..cd2de2816 100644 --- a/tests/host_weight_runtime/test_runtime_resolution.py +++ b/tests/host_weight_runtime/test_runtime_resolution.py @@ -4,6 +4,10 @@ from __future__ import annotations +import multiprocessing as mp +import threading +import time +from multiprocessing.connection import Connection from pathlib import Path from typing import cast @@ -15,6 +19,7 @@ from vllm_omni.host_weight_runtime import ( CanonicalJson, ComponentIdentity, FailureCode, + HostWeightError, HostWeightFailure, HostWeightLease, HostWeightLeaseCarrier, @@ -40,12 +45,14 @@ from vllm_omni.host_weight_runtime import ( StoreStatus, TensorWriteSpec, ValidationLevel, + WaitPolicy, WeightArtifactIdentity, WeightProductionSpec, WeightRepresentation, WeightSourceIdentity, ) from vllm_omni.host_weight_runtime.filesystem import FilesystemHostWeightStore, detect_storage_class +from vllm_omni.host_weight_runtime.filesystem.locks import FileLock, lock_is_active pytestmark = [pytest.mark.core_model, pytest.mark.cpu] @@ -518,6 +525,107 @@ def test_retryable_store_initialization_failure_obeys_runtime_mode( assert resolution.report.attempts[0].failure.retryable +def _initialize_contended_runtime(config: HostWeightRuntimeConfig, output: Connection) -> None: + try: + output.send("starting") + started = time.monotonic() + runtime = HostWeightRuntime.from_config(config) + resolution = runtime.resolve(_identity()) + try: + output.send((time.monotonic() - started, resolution.report)) + finally: + if resolution.lease is not None: + resolution.lease.close() + finally: + output.close() + + +@pytest.mark.parametrize( + ("mode", "expected"), + [ + (RuntimeMode.PREFERRED, ResolutionOutcome.CANONICAL_FALLBACK), + (RuntimeMode.REQUIRED, ResolutionOutcome.FAILED), + ], +) +def test_domain_lock_timeout_and_fresh_runtime_recovery( + tmp_path: Path, mode: RuntimeMode, expected: ResolutionOutcome +) -> None: + config = HostWeightRuntimeConfig(mode=mode, domain=_domain(tmp_path), wait=WaitPolicy(0.1)) + initial = HostWeightRuntime.from_config(config) + built = initial.resolve(_identity(), producer=CountingProducer(_identity())) + assert built.lease is not None + built.lease.close() + metadata = {name: (tmp_path / name).read_bytes() for name in ("domain.json", "domain-policy.json")} + lock_path = tmp_path / "locks" / "domain-init.lock" + ctx = mp.get_context("spawn") + reader, writer = ctx.Pipe(duplex=False) + process = ctx.Process(target=_initialize_contended_runtime, args=(config, writer)) + + with FileLock(lock_path, exclusive=True, deadline=None): + process.start() + writer.close() + try: + # Import/bootstrap is outside the short coordination budget. + assert reader.poll(60), "waiter did not start" + assert reader.recv() == "starting" + assert reader.poll(5), "domain initialization did not respect the coordination timeout" + elapsed, report = reader.recv() + assert 0.1 <= elapsed < 5 + assert report.outcome is expected + failure = report.attempts[0].failure + assert failure is not None + assert failure.stage is ResolutionStage.DOMAIN + assert failure.code is FailureCode.DOMAIN_UNAVAILABLE + assert failure.retryable + assert "timed out waiting for" in failure.message + assert "domain-init.lock" in failure.message + assert lock_is_active(lock_path), "waiter must not unlock the owner" + process.join(5) + assert process.exitcode == 0 + finally: + if process.is_alive(): + process.kill() + process.join(5) + process.close() + reader.close() + + assert {name: (tmp_path / name).read_bytes() for name in metadata} == metadata + recovered = HostWeightRuntime.from_config(config).resolve(_identity()) + assert recovered.report.outcome is ResolutionOutcome.LOCAL_HIT + assert recovered.lease is not None + recovered.lease.close() + + +def test_direct_store_waits_for_domain_lock_released_before_deadline(tmp_path: Path) -> None: + config = HostWeightRuntimeConfig(mode=RuntimeMode.PREFERRED, domain=_domain(tmp_path)) + assert config.domain is not None + initial = HostWeightRuntime.from_config(config) + assert initial.store is not None + lock_path = tmp_path / "locks" / "domain-init.lock" + with FileLock(lock_path, exclusive=True, deadline=None) as owner: + release = threading.Timer(0.1, owner.close) + release.start() + try: + store = FilesystemHostWeightStore(config.domain, config.capacity, config.integrity, wait=WaitPolicy(5)) + finally: + release.join(5) + assert store.root == tmp_path + + +def test_direct_store_domain_timeout_is_typed(tmp_path: Path) -> None: + config = HostWeightRuntimeConfig(mode=RuntimeMode.PREFERRED, domain=_domain(tmp_path)) + assert config.domain is not None + HostWeightRuntime.from_config(config) + lock_path = tmp_path / "locks" / "domain-init.lock" + with FileLock(lock_path, exclusive=True, deadline=None): + with pytest.raises(HostWeightError) as caught: + FilesystemHostWeightStore(config.domain, config.capacity, config.integrity, wait=WaitPolicy(0.01)) + assert caught.value.failure.code is FailureCode.DOMAIN_UNAVAILABLE + assert caught.value.failure.retryable + assert lock_is_active(lock_path) + assert not lock_is_active(lock_path) + + @pytest.mark.parametrize( ("stage", "code"), [ diff --git a/vllm_omni/host_weight_runtime/filesystem/store.py b/vllm_omni/host_weight_runtime/filesystem/store.py index 2b58b77e8..ec6b49f3c 100644 --- a/vllm_omni/host_weight_runtime/filesystem/store.py +++ b/vllm_omni/host_weight_runtime/filesystem/store.py @@ -23,7 +23,7 @@ import regex as re import torch from safetensors import SafetensorError, safe_open -from ..config import CapacityPolicy, IntegrityPolicy, StorageClass, StorageDomainPolicy, ValidationLevel +from ..config import CapacityPolicy, IntegrityPolicy, StorageClass, StorageDomainPolicy, ValidationLevel, WaitPolicy from ..errors import FailureCode, HostWeightError, HostWeightFailure, ResolutionStage from ..identity import CanonicalJson, WeightArtifactIdentity, canonical_json from ..inspection import ( @@ -429,6 +429,8 @@ class FilesystemHostWeightStore: domain: StorageDomainPolicy, capacity: CapacityPolicy, integrity: IntegrityPolicy, + *, + wait: WaitPolicy = WaitPolicy(), ) -> None: self.domain_policy = domain self.capacity_policy = capacity @@ -464,7 +466,7 @@ class FilesystemHostWeightStore: ) ) try: - self._initialize_domain() + self._initialize_domain(deadline=time.monotonic() + wait.coordination_timeout_seconds) except HostWeightError: raise except OSError as exc: @@ -485,7 +487,7 @@ class FilesystemHostWeightStore: ) ) from exc - def _initialize_domain(self) -> None: + def _initialize_domain(self, *, deadline: float) -> None: try: self.filesystem_type = detect_filesystem_type(self.root) detected_storage_class = _storage_class_for_filesystem_type(self.filesystem_type) @@ -545,7 +547,7 @@ class FilesystemHostWeightStore: ) ) - with FileLock(self.locks_dir / "domain-init.lock", exclusive=True, deadline=None): + with FileLock(self.locks_dir / "domain-init.lock", exclusive=True, deadline=deadline): domain_path = self.root / _DOMAIN_FILE if domain_path.exists(): domain_metadata = _read_json_file(domain_path) diff --git a/vllm_omni/host_weight_runtime/runtime.py b/vllm_omni/host_weight_runtime/runtime.py index fbc726669..17ababce9 100644 --- a/vllm_omni/host_weight_runtime/runtime.py +++ b/vllm_omni/host_weight_runtime/runtime.py @@ -70,7 +70,7 @@ class HostWeightRuntime: from .filesystem import FilesystemHostWeightStore # noqa: PLC0415 try: - store = FilesystemHostWeightStore(config.domain, config.capacity, config.integrity) + store = FilesystemHostWeightStore(config.domain, config.capacity, config.integrity, wait=config.wait) except HostWeightError as exc: return cls( config,