iHore/vLLM
8
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[Config] Register the fork's knobs by cache identity; lint; gpu test marker

Browse Source 
- Every fork knob is registered in envs.py and classified as a compile/capture
  factor or runtime-only; envs.py documents that capture identity and AOT
  cache identity are different things.
- ruff findings cleared on lines this fork authored; upstream lines untouched.
- A `gpu` pytest marker so the GPU-free tier can be selected, not just skipped.
- Duplicate knob-table entries from two branches reconciled onto one
  off|warn|raise normaliser.

Co-Authored-By: Hinata <hinata@idihore.id>
Signed-off-by: MOVZX <movzx@yahoo.com>
This commit is contained in:
MOVZX
2026-10-09 19:47:45 +07:00
co-authored by Hinata
parent 31ad7e113f
commit f1e5437a3a
6 changed files with 358 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+1
View File
@@ -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",
]
vllm/envs.py
+212 -7
View File
@@ -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",
}
vllm/v1/core/sched/e2_alloc_log.py
+115 -33
View File
@@ -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=<path>).
"""Debug-only scheduler allocation log for KV-capacity probes (off unless
VLLM_E2_SCHED_LOG=<path>).
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)
vllm/v1/ple_offload/test_fault.py
+5 -1
View File
@@ -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,
)
vllm/v1/worker/guard_telemetry.py
+19 -18
View File
@@ -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
vllm_ple_test_fault/__init__.py
+6 -9
View File
@@ -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: