| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | """End-to-end tests for the elastic fault-tolerance framework. |
| 4 | |
| 5 | Requires nixl_ep FT hardware; gated behind ``has_nixl_ep()``. |
| 6 | """ |
| 7 | |
| 8 | import contextlib |
| 9 | import os |
| 10 | import threading |
| 11 | import time |
| 12 | from concurrent.futures import ThreadPoolExecutor |
| 13 | from typing import Any |
| 14 | |
| 15 | import psutil |
| 16 | import pytest |
| 17 | import requests |
| 18 | |
| 19 | from tests.utils import RemoteOpenAIServer, multi_gpu_test |
| 20 | from vllm.utils.import_utils import has_nixl_ep |
| 21 | |
| 22 | MODEL_NAME = os.getenv("MODEL_NAME", "ibm-research/PowerMoE-3b") |
| 23 | DP_SIZE = 2 |
| 24 | |
| 25 | # Fault-detection timeout budget: |
| 26 | # - CPU: Gloo DP allreduce timeout (30s) detects the dead peer. |
| 27 | # - nixl_ep: kernel masks the dead rank after Buffer's default timeout_ms=30000 (30s). |
| 28 | # - Deadline (45s): slowest fallback (30s) + margin. |
| 29 | CPU_DISTRIBUTED_TIMEOUT_S = 30 |
| 30 | FAULT_DETECTION_DEADLINE_S = 45 |
| 31 | |
| 32 | |
| 33 | # Patches ``gpu.dp_utils.sync_cudagraph_and_dp_padding`` to raise on ``rank`` at |
| 34 | # a chosen step. Gated on VLLM_FT_TEST_INJECT_FAULT. |
| 35 | _FAULT_INJECT_SITECUSTOMIZE = """\ |
| 36 | import builtins |
| 37 | import os |
| 38 | import sys |
| 39 | |
| 40 | _SPEC = os.environ.get("VLLM_FT_TEST_INJECT_FAULT") |
| 41 | _MODULE = "vllm.v1.worker.gpu.dp_utils" |
| 42 | _ATTR = "sync_cudagraph_and_dp_padding" |
| 43 | |
| 44 | if _SPEC: |
| 45 | _f = dict(kv.split("=", 1) for kv in _SPEC.split(",")) |
| 46 | _RANK, _STEP = int(_f["rank"]), int(_f["step"]) |
| 47 | _steps = [0] |
| 48 | |
| 49 | def _patch(m): |
| 50 | import inspect |
| 51 | _orig = getattr(m, _ATTR) |
| 52 | _sig = inspect.signature(_orig) |
| 53 | def _wrapped(*args, **kwargs): |
| 54 | result = _orig(*args, **kwargs) |
| 55 | bound = _sig.bind(*args, **kwargs) |
| 56 | bound.apply_defaults() |
| 57 | dp_rank = bound.arguments.get("dp_rank") |
| 58 | if dp_rank == _RANK: |
| 59 | _steps[0] += 1 |
| 60 | if _steps[0] == _STEP: |
| 61 | raise RuntimeError( |
| 62 | "FT test fault injection (rank=%d step=%d)" % (_RANK, _STEP) |
| 63 | ) |
| 64 | return result |
| 65 | |
| 66 | setattr(m, _ATTR, _wrapped) |
| 67 | |
| 68 | _real_import = builtins.__import__ |
| 69 | |
| 70 | def _hook(name, *a, **k): |
| 71 | module = _real_import(name, *a, **k) |
| 72 | m = sys.modules.get(_MODULE) |
| 73 | # During vLLM's circular import the module lands in sys.modules before |
| 74 | # its functions are defined; hasattr guards against patching too early. |
| 75 | if ( |
| 76 | m is not None |
| 77 | and hasattr(m, _ATTR) |
| 78 | and not getattr(m, "_ft_patched", False) |
| 79 | ): |
| 80 | m._ft_patched = True |
| 81 | _patch(m) |
| 82 | return module |
| 83 | |
| 84 | builtins.__import__ = _hook |
| 85 | """ |
| 86 | |
| 87 | |
| 88 | def _install_fault_injection(monkeypatch, tmp_path, rank: int, step: int) -> None: |
| 89 | """Arrange for the DP-sync fn to raise on ``rank`` at serving ``step``. |
| 90 | |
| 91 | Writes a ``sitecustomize.py`` and prepends its dir to PYTHONPATH so every |
| 92 | vLLM subprocess picks it up; the fault spec is read from the environment. |
| 93 | """ |
| 94 | site_dir = tmp_path / "ft_inject" |
| 95 | site_dir.mkdir() |
| 96 | (site_dir / "sitecustomize.py").write_text(_FAULT_INJECT_SITECUSTOMIZE) |
| 97 | existing = os.environ.get("PYTHONPATH", "") |
| 98 | monkeypatch.setenv( |
| 99 | "PYTHONPATH", |
| 100 | str(site_dir) + (os.pathsep + existing if existing else ""), |
| 101 | ) |
| 102 | monkeypatch.setenv("VLLM_FT_TEST_INJECT_FAULT", f"rank={rank},step={step}") |
| 103 | |
| 104 | |
| 105 | def _ft_server_args() -> list[str]: |
| 106 | return [ |
| 107 | "--enforce-eager", |
| 108 | "--dtype", |
| 109 | "bfloat16", |
| 110 | "--max-model-len", |
| 111 | "2048", |
| 112 | "--max-num-seqs", |
| 113 | "128", |
| 114 | "--enable-expert-parallel", |
| 115 | "--all2all-backend", |
| 116 | "nixl_ep", |
| 117 | "--enable-fault-tolerance", |
| 118 | "--cpu-distributed-timeout-seconds", |
| 119 | str(CPU_DISTRIBUTED_TIMEOUT_S), |
| 120 | "--fault-tolerance-config", |
| 121 | '{"engine_recovery_timeout_sec": 120}', |
| 122 | ] |
| 123 | |
| 124 | |
| 125 | def _ft_manager(): |
| 126 | """Build the shared DP+EP fault-tolerant server topology (one engine/server).""" |
| 127 | from tests.v1.distributed.test_external_lb_dp import ExternalLBServerManager |
| 128 | |
| 129 | return ExternalLBServerManager( |
| 130 | MODEL_NAME, |
| 131 | DP_SIZE, |
| 132 | api_server_count=1, # FT requires a single API server per engine |
| 133 | base_server_args=_ft_server_args(), |
| 134 | tp_size=1, |
| 135 | ) |
| 136 | |
| 137 | |
| 138 | def _server_for_rank(servers, rank: int): |
| 139 | """Locate the server for a DP rank.""" |
| 140 | for server, sargs in servers: |
| 141 | if "--data-parallel-rank" in sargs: |
| 142 | idx = sargs.index("--data-parallel-rank") |
| 143 | if int(sargs[idx + 1]) == rank: |
| 144 | return server |
| 145 | raise AssertionError(f"no server found for DP rank {rank}") |
| 146 | |
| 147 | |
| 148 | def _complete(client): |
| 149 | """Issue the one standard completion the tests use everywhere.""" |
| 150 | return client.completions.create( |
| 151 | model=MODEL_NAME, |
| 152 | prompt="Hello, my name is", |
| 153 | max_tokens=5, |
| 154 | temperature=0.0, |
| 155 | timeout=10.0, |
| 156 | ) |
| 157 | |
| 158 | |
| 159 | def _in_parallel(fn, servers) -> list: |
| 160 | """Run ``fn(server)`` for all servers concurrently; return results in order.""" |
| 161 | with ThreadPoolExecutor(max_workers=len(servers)) as ex: |
| 162 | return list(ex.map(fn, servers)) |
| 163 | |
| 164 | |
| 165 | def _get_ft_status(server) -> dict: |
| 166 | resp = requests.get(server.url_for("fault_tolerance/status"), timeout=10) |
| 167 | resp.raise_for_status() |
| 168 | return resp.json() |
| 169 | |
| 170 | |
| 171 | def _assert_serving_and_healthy(servers) -> None: |
| 172 | """Wait until every engine is healthy, then serve one request per server.""" |
| 173 | healthy = _wait_for_engines( |
| 174 | list(servers), match_key="status", match_values={"healthy"} |
| 175 | ) |
| 176 | assert all(healthy), healthy |
| 177 | _in_parallel(lambda s: _complete(s.get_client()), servers) |
| 178 | |
| 179 | |
| 180 | def _apply_ft(server, instruction: str, params: dict | None = None) -> dict: |
| 181 | """POST an FT instruction; assert it is accepted (202) and return the body.""" |
| 182 | resp = requests.post( |
| 183 | server.url_for("fault_tolerance/apply"), |
| 184 | json={"instruction": instruction, "params": params or {}}, |
| 185 | timeout=10, |
| 186 | ) |
| 187 | assert resp.status_code == 202, resp.text |
| 188 | return resp.json() |
| 189 | |
| 190 | |
| 191 | def _kill_worker_process(server) -> None: |
| 192 | """SIGKILL only the worker proc, leaving EngineCore and API server alive.""" |
| 193 | workers = [ |
| 194 | p |
| 195 | for p in psutil.Process(server.proc.pid).children(recursive=True) |
| 196 | if "Worker" in " ".join(p.cmdline()) |
| 197 | ] |
| 198 | assert len(workers) == 1, f"expected 1 worker proc, found: {workers}" |
| 199 | workers[0].kill() |
| 200 | |
| 201 | |
| 202 | def _wait_for_engines( |
| 203 | servers: list[RemoteOpenAIServer], |
| 204 | match_key: str, |
| 205 | match_values: set[str], |
| 206 | deadline_s: int = FAULT_DETECTION_DEADLINE_S, |
| 207 | ) -> list[dict[str, Any] | None]: |
| 208 | """Poll ``/fault_tolerance/status`` until each server's engine status matches. |
| 209 | |
| 210 | A server matches when its engine-status dict has ``match_key`` equal to |
| 211 | one of ``match_values``. Returns one engine-status dict per server. Servers still |
| 212 | unmatched after ``deadline_s`` get None. |
| 213 | """ |
| 214 | results: dict[int, dict[str, Any]] = {} |
| 215 | pending = dict(enumerate(servers)) |
| 216 | start = time.time() |
| 217 | while pending and time.time() - start < deadline_s: |
| 218 | for i, server in list(pending.items()): |
| 219 | with contextlib.suppress(Exception): |
| 220 | for engine_status in _get_ft_status(server)["engines"]: |
| 221 | if engine_status.get(match_key) in match_values: |
| 222 | results[i] = engine_status |
| 223 | del pending[i] |
| 224 | break |
| 225 | if pending: |
| 226 | time.sleep(1.0) |
| 227 | return [results.get(i) for i in range(len(servers))] |
| 228 | |
| 229 | |
| 230 | @contextlib.contextmanager |
| 231 | def _driving(*servers): |
| 232 | """Pump completions at each server in the background for the block's duration. |
| 233 | |
| 234 | Keeps every engine stepping into its failed component so a fault surfaces. |
| 235 | Errors are expected once faulted and are ignored. |
| 236 | """ |
| 237 | stop = threading.Event() |
| 238 | |
| 239 | def _drive(server): |
| 240 | client = server.get_client() |
| 241 | while not stop.is_set(): |
| 242 | with contextlib.suppress(Exception): |
| 243 | _complete(client) |
| 244 | time.sleep(0.2) |
| 245 | |
| 246 | threads = [threading.Thread(target=_drive, args=(s,), daemon=True) for s in servers] |
| 247 | for t in threads: |
| 248 | t.start() |
| 249 | try: |
| 250 | yield |
| 251 | finally: |
| 252 | stop.set() |
| 253 | for t in threads: |
| 254 | t.join(timeout=2) |
| 255 | |
| 256 | |
| 257 | def _wait_for_ft_apply_outcome(server, request_id: str, deadline_s: int) -> str | None: |
| 258 | """Wait until ``/fault_tolerance/status`` records the FT apply outcome.""" |
| 259 | engine_status = _wait_for_engines( |
| 260 | [server], |
| 261 | match_key="last_ft_request_id", |
| 262 | match_values={request_id}, |
| 263 | deadline_s=deadline_s, |
| 264 | )[0] |
| 265 | return engine_status.get("ft_error") if engine_status else None |
| 266 | |
| 267 | |
| 268 | @pytest.mark.skipif(not has_nixl_ep(), reason="Requires nixl_ep all2all backend") |
| 269 | @multi_gpu_test(num_gpus=2) |
| 270 | def test_injected_fault_retry_recovers_all_ranks(monkeypatch, tmp_path): |
| 271 | """An exception injected into the inference path drives full retry recovery. |
| 272 | |
| 273 | Injecting an exception into ``sync_cudagraph_and_dp_padding`` at a chosen |
| 274 | step on rank 1. |
| 275 | |
| 276 | - Rank 1 raises inside the busy loop and goes UNHEALTHY. |
| 277 | - Rank 0 detects the now-absent peer via the communication timeout and also |
| 278 | goes UNHEALTHY. |
| 279 | |
| 280 | Both being UNHEALTHY is the precondition for ``retry``. The fault is patched |
| 281 | into the DP-sync fn from the test (via a generated ``sitecustomize``). |
| 282 | """ |
| 283 | fault_step = int(os.getenv("FT_FAULT_STEP", "50")) |
| 284 | _install_fault_injection(monkeypatch, tmp_path, rank=1, step=fault_step) |
| 285 | |
| 286 | with _ft_manager() as servers: |
| 287 | assert len(servers) == DP_SIZE |
| 288 | rank0 = _server_for_rank(servers, 0) |
| 289 | rank1 = _server_for_rank(servers, 1) |
| 290 | |
| 291 | # 1. Both engines healthy and serving. |
| 292 | _assert_serving_and_healthy((rank0, rank1)) |
| 293 | |
| 294 | # 2. Drive both ranks so rank 1 accumulates execute_model steps and trips |
| 295 | # the injected fault; rank 0 then times out on the DP allreduce. |
| 296 | with _driving(rank0, rank1): |
| 297 | faulted = _wait_for_engines( |
| 298 | [rank0, rank1], match_key="status", match_values={"unhealthy"} |
| 299 | ) |
| 300 | |
| 301 | for rank, engine_status in enumerate(faulted): |
| 302 | assert engine_status is not None, ( |
| 303 | f"rank {rank} did not report UNHEALTHY within " |
| 304 | f"{FAULT_DETECTION_DEADLINE_S}s -- it likely hung" |
| 305 | ) |
| 306 | # The rank that raised carries the fault info from its own exception. |
| 307 | assert faulted[1] is not None |
| 308 | assert faulted[1].get("fault_info"), faulted[1] |
| 309 | |
| 310 | # 3. retry both engines. |
| 311 | for server in (rank0, rank1): |
| 312 | _apply_ft(server, "retry") |
| 313 | |
| 314 | # 4. Recovery completes: both engines return to healthy and serve again. |
| 315 | _assert_serving_and_healthy((rank0, rank1)) |
| 316 | |
| 317 | |
| 318 | @pytest.mark.skipif(not has_nixl_ep(), reason="Requires nixl_ep all2all backend") |
| 319 | @multi_gpu_test(num_gpus=2) |
| 320 | def test_worker_kill_survivor_unhealthy_and_dead_rejects_retry(): |
| 321 | """One worker kill surfaces two status transitions at once. |
| 322 | |
| 323 | SIGKILLing only rank 1's worker leaves both EngineCores alive, so the same |
| 324 | fault is seen two ways: |
| 325 | |
| 326 | - Survivor (rank 0): detects the dead peer via Gloo allreduce / nixl_ep |
| 327 | kernel timeout. Its own executor is fine, so ``on_fault`` marks it |
| 328 | UNHEALTHY with a ``fault_info``. |
| 329 | - Victim (rank 1): detects its own executor failure and marks itself DEAD. |
| 330 | |
| 331 | Recovery is gated on UNHEALTHY: the DEAD engine accepts ``retry`` at the |
| 332 | HTTP layer (202 = background dispatch) but rejects it in the engine, |
| 333 | recording the reason as ``ft_error``. |
| 334 | """ |
| 335 | with _ft_manager() as servers: |
| 336 | assert len(servers) == DP_SIZE |
| 337 | survivor = _server_for_rank(servers, 0) |
| 338 | victim = _server_for_rank(servers, 1) |
| 339 | |
| 340 | # 1. Confirm both engines are healthy and serving. |
| 341 | _assert_serving_and_healthy((survivor, victim)) |
| 342 | |
| 343 | # 2. Kill only the victim's worker; both EngineCores stay alive. |
| 344 | _kill_worker_process(victim) |
| 345 | |
| 346 | # 3. Drive both engines so each keeps stepping into the failed component. |
| 347 | with _driving(survivor, victim): |
| 348 | survivor_faulted, victim_faulted = _wait_for_engines( |
| 349 | [survivor, victim], |
| 350 | match_key="status", |
| 351 | match_values={"dead", "unhealthy"}, |
| 352 | ) |
| 353 | |
| 354 | assert survivor_faulted is not None, ( |
| 355 | "survivor did not report the peer fault within " |
| 356 | f"{FAULT_DETECTION_DEADLINE_S}s -- it likely hung" |
| 357 | ) |
| 358 | # The survivor's own executor is fine, so it must be UNHEALTHY, not DEAD. |
| 359 | assert survivor_faulted["status"] == "unhealthy", survivor_faulted |
| 360 | assert survivor_faulted.get("fault_info"), survivor_faulted |
| 361 | |
| 362 | assert victim_faulted is not None, ( |
| 363 | "victim did not report its worker's death within " |
| 364 | f"{FAULT_DETECTION_DEADLINE_S}s" |
| 365 | ) |
| 366 | assert victim_faulted["status"] == "dead", victim_faulted |
| 367 | |
| 368 | # 4. retry is accepted at the HTTP layer (202 = background dispatch)... |
| 369 | request_id = _apply_ft(victim, "retry")["request_id"] |
| 370 | |
| 371 | # 5. ...but the DEAD engine must reject it: recovery requires UNHEALTHY. |
| 372 | ft_error = _wait_for_ft_apply_outcome( |
| 373 | victim, request_id, FAULT_DETECTION_DEADLINE_S |
| 374 | ) |
| 375 | assert ft_error is not None, ( |
| 376 | "rejection was never recorded in /fault_tolerance/status" |
| 377 | ) |
| 378 | assert "status is DEAD" in ft_error, ft_error |