diff --git a/pyproject.toml b/pyproject.toml index 890dd6bdb7..f1a90cdd9d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -132,6 +132,7 @@ markers = [ "cpu_test: mark test as CPU-only test", "split: run this test as part of a split", "distributed: run this test only in distributed GPU tests", + "gpu: requires a CUDA device; auto-skipped when none is present, and deselectable as a tier with -m 'not gpu'", "optional: optional tests that are automatically skipped, include --optional to run them", ] diff --git a/vllm/envs.py b/vllm/envs.py index 160acb7009..82af839ae3 100755 --- a/vllm/envs.py +++ b/vllm/envs.py @@ -310,6 +310,132 @@ if TYPE_CHECKING: # AOT cache on it; the layer itself still reads os.environ directly. VLLM_HOST_EMBED_TABLE: bool = False VLLM_PLE_OFFLOAD_READY_TIMEOUT: float = 600.0 + # --- Flash-Next fork knobs (v2.2) -------------------------------- + # Every name below was read straight from os.environ at its use site, so + # envs.validate_environ() logged it as unknown on every boot that set one + # and compile_factors() could not see it. Registering makes the hash + # complete; the read sites are unchanged (they still parse and fail closed + # themselves, which registration must not weaken). Runtime-only entries are + # listed in compile_factors().ignored_factors with their reason. + # + # TWO IDENTITIES, NOT ONE. "Read once at startup" + # settles neither of them, so every entry below is labelled with both: + # + # AOT identity -- does the value change the torch.compile / AOT + # artifact? This is the ONLY identity compile_factors() + # keys. It protects against reusing another process's + # cached artifact. + # CAPTURE identity -- does the value change what a cudagraph capture bakes + # in (ops present, launch constants, shapes)? Nothing + # in this tree keys anything on capture identity: the + # env is read before capture, in-process, so capture + # identity is protected only by "do not change the + # value between capture and replay in one process". + # + # So a knob that is a capture factor but NOT an AOT factor gets no + # protection from being registered here. Registering it is a contract + # statement and a cross-process guard, not a mechanism. Where the two + # identities are not both established by reading the code, the knob is + # hashed anyway: a spurious cache miss costs a recompile, a missed one costs + # a wrong graph. + # + # COMPILE/CAPTURE FACTORS (hashed): + # AOT: yes. CAPTURE: yes. PLE table is FP8-with-global-scale -> + # Qwen4ExpPLEFp8EmbeddingMethod instead of the checkpoint's method: + # different weight dtype and a different embedding forward. + VLLM_PLE_FP8_GLOBAL_SCALE: bool = False + # CAPTURE: yes, established -- clip counting allocates persistent per-layer + # int64 counters and its increment ops are CAPTURED into the cudagraph (see + # the graph-safety note in models/qwen4_exp/nvidia/qsa.py); unset, those ops + # are absent. AOT: not established (the QSA forward is a custom op); hashed + # conservatively. + VLLM_QSA_KV_CLIP_COUNT: str = "" + # CAPTURE: n/a -- collection requires --enforce-eager, so an enabled run + # does not capture at all. AOT: adds absmax reduction ops to the attention + # forward; hashed conservatively. + VLLM_QSA_KV_COLLECT: str = "" + # CAPTURE: yes -- the resolved value is a launch constant baked into the + # graph at capture (csrc/libtorch_stable/persistent_topk.cuh:143-148). + # AOT: not established. Hashed conservatively. + # NOT A PYTHON READ SITE: the value is resolved once per process by + # exact_medium_enabled() in csrc/libtorch_stable/topk.cu:30-36, which is the + # single source of truth. This entry is a raw passthrough so nothing can + # mistake it for a second parser; do not use envs.MERLIN_TOPK_EXACT_MEDIUM + # to decide behaviour. Registered with the medium-path exact restart + # so the cache-factor contract is complete. + MERLIN_TOPK_EXACT_MEDIUM: str | None = None + # + # RUNTIME-ONLY (registered for validate_environ + a single cached read): + # Static K/V scale sidecar: fill_()ed into existing fp32 device buffers + # after load and before capture, and the kernel does tl.load(k_scale_ptr), + # so the value is never a constexpr and the graph is identical. + VLLM_QSA_KV_SCALES: str = "" + # Guard telemetry: mode raise|warn|off and the reader poll interval in + # seconds. counter() returns a real tensor in every mode, so the Triton + # arguments and the captured graph do not depend on either. + VLLM_E2_GUARD_MODE: str = "off" + VLLM_E2_GUARD_INTERVAL: str = "0.05" + # Debug allocator telemetry (poll seconds + output directory) and the + # scheduler allocation log path. + VLLM_E2_MEMSTATS: str = "0" + VLLM_E2_MEMSTATS_DIR: str = "" + VLLM_E2_SCHED_LOG: str = "" + # O(live blocks) host-side ownership assertion in the scheduler, which is + # plain Python and never compiled. + VLLM_E2_ZERO_OWNERSHIP_CHECK: bool = False + # PLE offload child placement: rewrites CUDA_VISIBLE_DEVICES for the child + # only (CUDA_VISIBLE_DEVICES is itself an ignored factor). + VLLM_PLE_OFFLOAD_HOME_DEVICE: str = "" + # PLE timeouts / deadlines, in seconds. + VLLM_PLE_OFFLOAD_WAIT_TIMEOUT: float = 120.0 + VLLM_PLE_PROGRESS_DEADLINE_S: float = 300.0 + VLLM_PLE_STARTUP_DEADLINE_S: float = 1800.0 + # Test-only PLE fault injection: cut spec, marker log, bootstrap gate. + # AOT: no. CAPTURE: no. Read once at import by the vllm_ple_test_fault + # singleton, but the process that reads them is a real engine, so + # validate_environ sees them. + VLLM_PLE_TEST_FAULT: str | None = None + VLLM_PLE_TEST_FAULT_GATE: str | None = None + VLLM_PLE_TEST_FAULT_LOG: str | None = None + # + # --- Knobs registered ahead of their read sites (v2.2) ----------- + # Registered ahead of their code so the MERGED tree is complete and + # validate_environ is quiet from the first boot (a + # cross-change knob table). Each mirrors its read site's + # parse; if a read site changes, the entry here must be reconciled. + # (Both read sites below are now live.) + # AOT: no. CAPTURE: no -- strictly host-side and outside capture. + # off|warn|raise (1 = warn), normalised by _e2_diagnostic_mode like the + # rest of the family. The mamba ownership check moved this gate off a plain + # boolean; reconciled here. + # Read site returns with the deferred PLE transport. + VLLM_E2_PLE_FLAG_CHECK: str = "off" + # AOT: no. CAPTURE: no -- strictly host-side. off|warn|raise (1 = warn). + # the mamba slot-ownership check, v1/worker/mamba_utils.py:74. + VLLM_E2_MAMBA_SLOT_OWNERSHIP: str = "off" + # + # --- TODO: knobs that are NOT in this tree at all ---------- + # The tie-determinism toggles and the BSM pin live in a separate + # patch stack, not in v2.0.1 and not in this tree: + # + # MERLIN_FULL_K MERLIN_QSA_SORT MERLIN_TIE_DET MERLIN_TIE_RECENT + # MERLIN_PIN_BSM + # + # They are therefore UNAUDITED: nothing in this repository shows what they + # change, and this file will not classify code it cannot read. The rule they + # fall under is the standing one -- an os.environ (or std::getenv) knob that + # changes a compiled or captured artifact is not in the cache key unless it + # is registered here. From what is known of them (Marlin split-K, a + # persistent_topk ordering change and a cutoff tie-break) they are very + # likely capture factors and possibly AOT factors, which is exactly why a + # guess is not good enough. + # + # WHEN THE PATCHES LAND, for each name: register it below, label both + # identities from its actual read site, hash it unless exclusion is proven, + # and delete it from this block. tests/test_envs.py holds a test that fails + # if a name is registered while still listed here, so the classification + # cannot be skipped silently. + # ---------------------------------------------------------------------- VLLM_LOG_MODEL_INSPECTION: bool = False VLLM_DEBUG_MFU_METRICS: bool = False VLLM_WEIGHT_OFFLOADING_DISABLE_PIN_MEMORY: bool = False @@ -2153,6 +2279,43 @@ environment_variables: dict[str, Callable[[], Any]] = { "VLLM_PLE_OFFLOAD_READY_TIMEOUT": lambda: float( os.getenv("VLLM_PLE_OFFLOAD_READY_TIMEOUT", "600") ), + # --- Flash-Next fork knobs (v2.2); see the TYPE_CHECKING block for + # which of these are compile/capture factors and which are ignored. + # Compile/capture factors. + "VLLM_PLE_FP8_GLOBAL_SCALE": lambda: ( + os.getenv("VLLM_PLE_FP8_GLOBAL_SCALE", "") == "1" + ), + "VLLM_QSA_KV_CLIP_COUNT": lambda: os.getenv("VLLM_QSA_KV_CLIP_COUNT", "").strip(), + "VLLM_QSA_KV_COLLECT": lambda: os.getenv("VLLM_QSA_KV_COLLECT", "").strip(), + # Capture-configuration factor, read in C++; raw passthrough on purpose. + "MERLIN_TOPK_EXACT_MEDIUM": lambda: os.getenv("MERLIN_TOPK_EXACT_MEDIUM", None), + # Runtime only; all of the following are in compile_factors.ignored_factors. + "VLLM_QSA_KV_SCALES": lambda: os.getenv("VLLM_QSA_KV_SCALES", "").strip(), + "VLLM_E2_GUARD_MODE": lambda: ( + os.getenv("VLLM_E2_GUARD_MODE", "off").strip().lower() + ), + "VLLM_E2_GUARD_INTERVAL": lambda: os.getenv("VLLM_E2_GUARD_INTERVAL", "0.05"), + "VLLM_E2_MEMSTATS": lambda: os.getenv("VLLM_E2_MEMSTATS", "0"), + "VLLM_E2_MEMSTATS_DIR": lambda: os.getenv("VLLM_E2_MEMSTATS_DIR", ""), + "VLLM_E2_SCHED_LOG": lambda: os.getenv("VLLM_E2_SCHED_LOG", ""), + "VLLM_E2_ZERO_OWNERSHIP_CHECK": lambda: ( + os.getenv("VLLM_E2_ZERO_OWNERSHIP_CHECK", "") == "1" + ), + "VLLM_PLE_OFFLOAD_HOME_DEVICE": lambda: os.getenv( + "VLLM_PLE_OFFLOAD_HOME_DEVICE", "" + ).strip(), + "VLLM_PLE_OFFLOAD_WAIT_TIMEOUT": lambda: float( + os.getenv("VLLM_PLE_OFFLOAD_WAIT_TIMEOUT", "120") + ), + "VLLM_PLE_PROGRESS_DEADLINE_S": lambda: float( + os.getenv("VLLM_PLE_PROGRESS_DEADLINE_S", "300") + ), + "VLLM_PLE_STARTUP_DEADLINE_S": lambda: float( + os.getenv("VLLM_PLE_STARTUP_DEADLINE_S", "1800") + ), + "VLLM_PLE_TEST_FAULT": lambda: os.getenv("VLLM_PLE_TEST_FAULT", None), + "VLLM_PLE_TEST_FAULT_GATE": lambda: os.getenv("VLLM_PLE_TEST_FAULT_GATE", None), + "VLLM_PLE_TEST_FAULT_LOG": lambda: os.getenv("VLLM_PLE_TEST_FAULT_LOG", None), # Log model inspection after loading. # If enabled, logs a transformers-style hierarchical view of the model # with quantization methods and attention backends. @@ -2381,13 +2544,6 @@ def compile_factors() -> dict[str, object]: "VLLM_DEBUG_DUMP_PATH", "VLLM_PORT", "VLLM_CACHE_ROOT", - # PLE host-side diagnostics: both gate Python that runs outside any - # capture and neither adds a node to a compiled graph, so hashing them - # would invalidate every compile cache the moment a diagnostic boot is - # taken -- which is exactly when the cache is wanted. Remove from this - # set if either ever grows a device-side component. - "VLLM_E2_PLE_FLAG_CHECK", - "VLLM_E2_MAMBA_SLOT_OWNERSHIP", # Runtime memory-plan persistence; does not affect compiled graphs. "VLLM_ENABLE_STARTUP_PLAN", # Location-only derived paths: where a cache/config directory lives @@ -2464,6 +2620,55 @@ def compile_factors() -> dict[str, object]: "CUDA_VISIBLE_DEVICES", "NO_COLOR", "VLLM_SKIP_VERSION_SUFFIX", + # --- Flash-Next fork knobs that are runtime-only (v2.2) ------ + # Registering a name in environment_variables makes it a compile + # factor by default (this function is a deny-list), so every fork knob + # that cannot change a compiled or captured artifact is listed here + # with its reason. Hashing them would invalidate the whole AOT cache on + # a log path, a timeout or a guard-mode flip. + # + # Static QSA K/V scales: fill_()ed into existing fp32 device buffers + # after load and before capture; the kernel reads them with + # tl.load(k_scale_ptr), never as a constexpr. + "VLLM_QSA_KV_SCALES", + # Guard telemetry: counter() returns a real [3] int64 view in every + # mode, so the Triton arguments and the captured graph are identical; + # the mode only gates the side reader thread and the check_pending + # raise, and the interval is that thread's poll cadence. + "VLLM_E2_GUARD_MODE", + "VLLM_E2_GUARD_INTERVAL", + # Debug telemetry cadence and output locations. + "VLLM_E2_MEMSTATS", + "VLLM_E2_MEMSTATS_DIR", + "VLLM_E2_SCHED_LOG", + # Host-side scheduler assertion; the scheduler is never compiled. + "VLLM_E2_ZERO_OWNERSHIP_CHECK", + # Placement only: rewrites CUDA_VISIBLE_DEVICES for the PLE offload + # child, and CUDA_VISIBLE_DEVICES is already ignored above. + "VLLM_PLE_OFFLOAD_HOME_DEVICE", + # Timeouts and deadlines, in seconds. VLLM_PLE_OFFLOAD_READY_TIMEOUT + # was registered in the v2 host-embed commit and left hashed by + # oversight: serve-v2.sh sets it to 1800, so every boot that changed it + # missed the AOT cache for no reason. + "VLLM_PLE_OFFLOAD_READY_TIMEOUT", + "VLLM_PLE_OFFLOAD_WAIT_TIMEOUT", + "VLLM_PLE_PROGRESS_DEADLINE_S", + "VLLM_PLE_STARTUP_DEADLINE_S", + # Test-only fault injection: a cut spec, a marker log path and a + # bootstrap gate file. None of them reaches a kernel or a graph. + "VLLM_PLE_TEST_FAULT", + "VLLM_PLE_TEST_FAULT_GATE", + "VLLM_PLE_TEST_FAULT_LOG", + # Diagnostic knobs, both strictly host-side and outside capture, so + # neither AOT nor capture identity depends on them. Hashing + # them would invalidate every compile cache the moment a diagnostic + # boot is taken -- exactly when the cache is wanted. Remove from this + # set if either ever grows a device-side component. + # MERLIN_TOPK_EXACT_MEDIUM is deliberately NOT here: it is a capture + # configuration factor and is hashed conservatively. + "VLLM_E2_PLE_FLAG_CHECK", + "VLLM_E2_MAMBA_SLOT_OWNERSHIP", + # ------------------------------------------------------------------ "FORCE_COLOR", } diff --git a/vllm/v1/core/sched/e2_alloc_log.py b/vllm/v1/core/sched/e2_alloc_log.py index 0efcbd74f0..6fa2354feb 100644 --- a/vllm/v1/core/sched/e2_alloc_log.py +++ b/vllm/v1/core/sched/e2_alloc_log.py @@ -1,23 +1,45 @@ # SPDX-License-Identifier: Apache-2.0 -"""Debug-only scheduler allocation log for KV capacity probes (off unless VLLM_E2_SCHED_LOG=). +"""Debug-only scheduler allocation log for KV-capacity probes (off unless +VLLM_E2_SCHED_LOG=). -One JSON line per scheduler update while any request exists, and one "fin" line per request as it is freed: - {"t": wall, "m": monotonic, "step": n, "total": pool blocks, "free": free blocks, "running": r, "waiting": w, +One JSON line per scheduler update while any request exists, and one "fin" line +per request as it is freed: + {"t": wall, "m": monotonic, "step": n, "total": pool blocks, + "free": free blocks, "running": r, "waiting": w, "tot": tokens scheduled this step, - "reqs": [[id, prompt, computed, output, cached, preemptions, spec_scheduled, status, [blocks per group], + "reqs": [[id, prompt, computed, output, cached, preemptions, + spec_scheduled, status, [blocks per group], tokens scheduled for this request this step]]} - {"t": wall, "m": monotonic, "fin": [id, prompt, computed, output, cached, preemptions, status, [blocks per group]]} -Blocks are counted per KV-cache group, null blocks excluded. A logging failure writes an "error" line and never -raises into the scheduler. Unset = the scheduler only pays one env check per update. + {"t": wall, "m": monotonic, + "fin": [id, prompt, computed, output, cached, preemptions, status, + [blocks per group]]} +Blocks are counted per KV-cache group, null blocks excluded. A logging failure +writes an "error" line and never raises into the scheduler. Unset = the +scheduler only pays one env check per update. """ + import json import os import time _PATH = os.environ.get("VLLM_E2_SCHED_LOG", "") -_state = {"f": None, "step": 0, "flush": 0.0, "wrapped": False, "errors": 0, "checked": False} -REQ_FIELDS = ("request_id", "num_prompt_tokens", "num_computed_tokens", "num_output_tokens", "num_preemptions", - "status", "prefill_stats") +_state = { + "f": None, + "step": 0, + "flush": 0.0, + "wrapped": False, + "errors": 0, + "checked": False, +} +REQ_FIELDS = ( + "request_id", + "num_prompt_tokens", + "num_computed_tokens", + "num_output_tokens", + "num_preemptions", + "status", + "prefill_stats", +) def enabled() -> bool: @@ -25,8 +47,10 @@ def enabled() -> bool: def _blocks(sched, rid): - return [sum(1 for b in m.req_to_blocks.get(rid, ()) if not b.is_null) - for m in sched.kv_cache_manager.coordinator.single_type_managers] + return [ + sum(1 for b in m.req_to_blocks.get(rid, ()) if not b.is_null) + for m in sched.kv_cache_manager.coordinator.single_type_managers + ] def _status(r): @@ -34,28 +58,41 @@ def _status(r): def _cached(r): - """Prefix-cache hit for this request. vLLM keeps it in `request.prefill_stats` (PrefillStats.num_cached_tokens = - local + external); the output processor takes those stats after the first prefill (`take_prefill_stats` sets the - attribute to None), so -1 means "no longer knowable from the scheduler", never a silent zero.""" + """Prefix-cache hit for this request. vLLM keeps it in + `request.prefill_stats` (PrefillStats.num_cached_tokens = local + + external); the output processor takes those stats after the first prefill + (`take_prefill_stats` sets the attribute to None), so -1 means "no longer + knowable from the scheduler", never a silent zero.""" ps = getattr(r, "prefill_stats", None) return int(getattr(ps, "num_cached_tokens", -1)) if ps is not None else -1 def _contract(r): - """Once per process: every field this log reads must exist on the real Request object. A missing field is written - once as a schema_error record — the evaluator fails closed on it — instead of turning every row into an error.""" + """Once per process: every field this log reads must exist on the real Request + object. A missing field is written once as a schema_error record — the evaluator + fails closed on it — instead of turning every row into an error.""" if _state["checked"]: return _state["checked"] = True missing = [f for f in REQ_FIELDS if not hasattr(r, f)] if missing: - _write({"t": round(time.time(), 4), "m": round(time.monotonic(), 4), "schema_error": missing}, force=True) + _write( + { + "t": round(time.time(), 4), + "m": round(time.monotonic(), 4), + "schema_error": missing, + }, + force=True, + ) def _write(rec, force=False): f = _state["f"] if f is None: - f = _state["f"] = open(_PATH, "a", buffering=1 << 16) + # noqa SIM115: one append handle for the process lifetime, kept in + # _state and flushed on a cadence; a context manager would reopen the + # log on every scheduler update. + f = _state["f"] = open(_PATH, "a", buffering=1 << 16) # noqa: SIM115 f.write(json.dumps(rec, separators=(",", ":")) + "\n") if force or rec["t"] - _state["flush"] > 0.5: f.flush() @@ -65,10 +102,19 @@ def _write(rec, force=False): def _error(e): _state["errors"] += 1 if _state["errors"] <= 20: - try: - _write({"t": round(time.time(), 4), "m": round(time.monotonic(), 4), - "error": f"{type(e).__name__}: {e}"[:300]}, force=True) + try: # noqa: SIM105 + _write( + { + "t": round(time.time(), 4), + "m": round(time.monotonic(), 4), + "error": f"{type(e).__name__}: {e}"[:300], + }, + force=True, + ) except Exception: # noqa: BLE001 + # SIM105: this module imports json, os and time only, so that an + # enabled log costs the scheduler nothing it can see; the + # last-resort writer does not earn a contextlib import. pass @@ -77,10 +123,23 @@ def _wrap_free(sched): def _free_request(request, *args, **kwargs): try: - _write({"t": round(time.time(), 4), "m": round(time.monotonic(), 4), - "fin": [request.request_id, request.num_prompt_tokens, request.num_computed_tokens, - request.num_output_tokens, _cached(request), request.num_preemptions, - _status(request), _blocks(sched, request.request_id)]}, force=True) + _write( + { + "t": round(time.time(), 4), + "m": round(time.monotonic(), 4), + "fin": [ + request.request_id, + request.num_prompt_tokens, + request.num_computed_tokens, + request.num_output_tokens, + _cached(request), + request.num_preemptions, + _status(request), + _blocks(sched, request.request_id), + ], + }, + force=True, + ) except Exception as e: # noqa: BLE001 - the debug log must never break the scheduler _error(e) return orig(request, *args, **kwargs) @@ -100,12 +159,35 @@ def log_step(sched, scheduler_output) -> None: pool = sched.kv_cache_manager.block_pool nsched = getattr(scheduler_output, "num_scheduled_tokens", None) or {} _contract(next(iter(sched.requests.values()))) - rows = [[rid, r.num_prompt_tokens, r.num_computed_tokens, r.num_output_tokens, _cached(r), - r.num_preemptions, len(spec.get(rid, ())), _status(r), _blocks(sched, rid), int(nsched.get(rid, 0))] - for rid, r in sched.requests.items()] - _write({"t": round(time.time(), 4), "m": round(time.monotonic(), 4), "step": _state["step"], - "total": pool.num_gpu_blocks, "free": pool.get_num_free_blocks(), - "running": len(sched.running), "waiting": len(sched.waiting), - "tot": int(getattr(scheduler_output, "total_num_scheduled_tokens", 0) or 0), "reqs": rows}) + rows = [ + [ + rid, + r.num_prompt_tokens, + r.num_computed_tokens, + r.num_output_tokens, + _cached(r), + r.num_preemptions, + len(spec.get(rid, ())), + _status(r), + _blocks(sched, rid), + int(nsched.get(rid, 0)), + ] + for rid, r in sched.requests.items() + ] + _write( + { + "t": round(time.time(), 4), + "m": round(time.monotonic(), 4), + "step": _state["step"], + "total": pool.num_gpu_blocks, + "free": pool.get_num_free_blocks(), + "running": len(sched.running), + "waiting": len(sched.waiting), + "tot": int( + getattr(scheduler_output, "total_num_scheduled_tokens", 0) or 0 + ), + "reqs": rows, + } + ) except Exception as e: # noqa: BLE001 _error(e) diff --git a/vllm/v1/ple_offload/test_fault.py b/vllm/v1/ple_offload/test_fault.py index 3546a1f089..9ae74f7205 100644 --- a/vllm/v1/ple_offload/test_fault.py +++ b/vllm/v1/ple_offload/test_fault.py @@ -2,8 +2,12 @@ # SPDX-FileCopyrightText: Copyright contributors to the vLLM project """Compatibility exports for the top-level test-only fault singleton.""" +from vllm_ple_test_fault import ( + _FAULT as _FAULT, +) from vllm_ple_test_fault import ( # noqa: F401 FaultSpec as FaultSpec, - _FAULT as _FAULT, +) +from vllm_ple_test_fault import ( _parse_fault_spec as _parse_fault_spec, ) diff --git a/vllm/v1/worker/guard_telemetry.py b/vllm/v1/worker/guard_telemetry.py index 60fb094d18..3c5ebaf3ab 100644 --- a/vllm/v1/worker/guard_telemetry.py +++ b/vllm/v1/worker/guard_telemetry.py @@ -121,9 +121,7 @@ SITE_NAMES = { _MODE = os.environ.get("VLLM_E2_GUARD_MODE", "off").strip().lower() if _MODE not in ("raise", "warn", "off"): # fail closed: a set-but-broken gate must not silently disarm the instrument - raise ValueError( - f"VLLM_E2_GUARD_MODE must be raise|warn|off, got {_MODE!r}" - ) + raise ValueError(f"VLLM_E2_GUARD_MODE must be raise|warn|off, got {_MODE!r}") def _parse_interval(text: str) -> float: @@ -188,7 +186,7 @@ class NonBlockingReader: """ def __init__(self) -> None: - self._streams: dict[torch.device, "torch.cuda.Stream"] = {} + self._streams: dict[torch.device, torch.cuda.Stream] = {} # key -> [device_tensor, pinned_host_tensor | None, event | None, inflight] self._jobs: dict[object, list] = {} self._lock = threading.Lock() @@ -213,7 +211,9 @@ class NonBlockingReader: if dev not in self._streams: with torch.cuda.device(dev): self._streams[dev] = torch.cuda.Stream(device=dev) - host = torch.empty(tensor.shape, dtype=tensor.dtype, device="cpu", pin_memory=True) + host = torch.empty( + tensor.shape, dtype=tensor.dtype, device="cpu", pin_memory=True + ) with torch.cuda.device(dev): event = torch.cuda.Event() self._jobs[key] = [tensor, host, event, False] @@ -280,7 +280,8 @@ class NonBlockingReader: _nb_reader = NonBlockingReader() -# device -> the copy in which a count increase was first seen (reported on the next tick) +# device -> the copy in which a count increase was first seen +# (reported on the next tick) _confirm_pending: dict[torch.device, list] = {} # Sites that arm the pending raise even under VLLM_E2_GUARD_MODE=warn. warn is @@ -389,12 +390,11 @@ def start_reader() -> None: if _reader_started or _MODE == "off": return _reader_started = True - threading.Thread( - target=_reader, name="bounds-guard-reader", daemon=True - ).start() + threading.Thread(target=_reader, name="bounds-guard-reader", daemon=True).start() logger.info( "bounds-guard reader started (mode=%s, interval=%.3gs).", - _MODE, _INTERVAL, + _MODE, + _INTERVAL, ) @@ -459,7 +459,11 @@ def _poll_once() -> None: # by which time the maxima of that hit are written). The fatal # decision is count-based either way; only its report waits a tick. pend = _confirm_pending.get(device) - rising = [site for site, (n, _r, _v) in enumerate(rows) if n > _seen.get((device, site), 0)] + rising = [ + site + for site, (n, _r, _v) in enumerate(rows) + if n > _seen.get((device, site), 0) + ] if pend is None: if rising: _confirm_pending[device] = rows @@ -572,21 +576,18 @@ def drain(*, raise_on_fatal: bool = False) -> dict[str, list[int]]: _report_rows(device, rows) for site, (n_masked, _r, _v) in enumerate(rows): if n_masked and (_MODE == "raise" or site in _ALWAYS_FATAL_SITES): - fatal.append( - f"{SITE_NAMES.get(site, site)} on {device}: {n_masked}" - ) + fatal.append(f"{SITE_NAMES.get(site, site)} on {device}: {n_masked}") snap = snapshot() if fatal: logger.error( "bounds-guard FINAL DRAIN: fatal-site hits %s (snapshot %s)", - fatal, snap, + fatal, + snap, ) if raise_on_fatal: raise RuntimeError(f"bounds-guard FINAL DRAIN: fatal-site hits {fatal}") else: - logger.info( - "bounds-guard FINAL DRAIN: no fatal-site hits (snapshot %s)", snap - ) + logger.info("bounds-guard FINAL DRAIN: no fatal-site hits (snapshot %s)", snap) return snap diff --git a/vllm_ple_test_fault/__init__.py b/vllm_ple_test_fault/__init__.py index e8fe835ad3..41830e02c4 100644 --- a/vllm_ple_test_fault/__init__.py +++ b/vllm_ple_test_fault/__init__.py @@ -145,16 +145,12 @@ class FaultSpec: if bytes_sent is not None or frame_size is None: return False suffix = ( - f" bytes_written={int(bytes_written)}" - f" frame_size={int(frame_size)}" + f" bytes_written={int(bytes_written)} frame_size={int(frame_size)}" ) elif bytes_sent is not None: if frame_size is None: return False - suffix = ( - f" bytes_sent={int(bytes_sent)}" - f" frame_size={int(frame_size)}" - ) + suffix = f" bytes_sent={int(bytes_sent)} frame_size={int(frame_size)}" elif frame_size is not None: return False message = ( @@ -306,7 +302,9 @@ def _parse_fault_spec(value: str | None) -> FaultSpec | None: if value is None: return None if not value or value != value.strip(): - raise ValueError(f"{_ENV_NAME}: spec must be non-empty with no outer whitespace") + raise ValueError( + f"{_ENV_NAME}: spec must be non-empty with no outer whitespace" + ) fields: dict[str, str] = {} for item in value.split(","): @@ -355,8 +353,7 @@ def _parse_fault_spec(value: str | None) -> FaultSpec | None: if role in {"offload", "bootstrap"} and not has_any_rank: raise ValueError( - f"{_ENV_NAME}: {role} cut {cut!r} requires rank=any; " - "it has no TP/PP rank" + f"{_ENV_NAME}: {role} cut {cut!r} requires rank=any; it has no TP/PP rank" ) if role == "gpu": if pp_rank is not None and pp_rank != 0: