@tduncan / Nvidia-Optimization-Stack-VLLM / commits / 0b0bd2b5f6

[Feature] Add fault tolerance framework (simplified) for DP+EP external LB deployments (#44428)

Signed-off-by: fangyuchu <fangyuchu@qq.com> Signed-off-by: a798347923 <2645302020@qq.com> Signed-off-by: TianZhuo <2770730562@qq.com> Signed-off-by: a798347923 <39047817+a798347923@users.noreply.github.com> Signed-off-by: 205150940 <112750056+205150940@users.noreply.github.com> Signed-off-by: w00689259 <wangzhuo66@huawei.com> Signed-off-by: zWaNg3 <37772915+zWaNg3@users.noreply.github.com> Signed-off-by: zWaNg3 <389750525@qq.com> Signed-off-by: yzchang-plus <1078477584@qq.com> Signed-off-by: Jade Zheng <zheng.shoujian@outlook.com> Co-authored-by: zWaNg3 <37772915+zWaNg3@users.noreply.github.com> Co-authored-by: a798347923 <2645302020@qq.com> Co-authored-by: TianZhuo <2770730562@qq.com> Co-authored-by: 205150940 <112750056+205150940@users.noreply.github.com> Co-authored-by: a798347923 <39047817+a798347923@users.noreply.github.com> Co-authored-by: w00689259 <wangzhuo66@huawei.com> Co-authored-by: zWaNg3 <389750525@qq.com> Co-authored-by: yzchang-plus <1078477584@qq.com> Co-authored-by: Jade Zheng <zheng.shoujian@outlook.com>

fangyuchu committed Jul 25, 2026 at 23:49 UTC 0b0bd2b5f6a7f15ef59621efe6e535b54c6348f9
27 files changed +1088 -7
.buildkite/test_areas/fault_tolerance.yaml new
+26
@@ -0,0 +1,26 @@
1 +group: Fault Tolerance
2 +depends_on:
3 + - image-build
4 +steps:
5 +- label: Fault Tolerance E2E (2xH100)
6 + key: fault-tolerance-e2e-2xh100
7 + timeout_in_minutes: 35
8 + device: h100
9 + num_devices: 2
10 + working_dir: "/vllm-workspace/tests"
11 + source_file_dependencies:
12 + - vllm/v1/fault_tolerance/
13 + - vllm/v1/worker/sentinel/
14 + - vllm/entrypoints/serve/fault_tolerance/
15 + - vllm/distributed/elastic_ep/
16 + - vllm/distributed/device_communicators/
17 + - vllm/v1/engine/
18 + - vllm/v1/worker/
19 + - tests/v1/fault_tolerance/
20 + - tests/v1/distributed/test_external_lb_dp.py
21 + commands:
22 + # Base image has no nixl; install it or has_nixl_ep() skips the tests.
23 + - bash /vllm-workspace/.buildkite/scripts/install-kv-connectors.sh
24 + # https://github.com/NVIDIA/nccl/issues/1838
25 + - export NCCL_CUMEM_HOST_ENABLE=0
26 + - pytest -v -s v1/fault_tolerance/test_fault_tolerance_e2e.py
tests/test_config.py
+10
@@ -1547,6 +1547,16 @@ def test_needs_dp_coordination(
1547 assert vllm_config.needs_dp_coordinator == expected_needs_coordinator
1548
1549
1550 +def test_fault_tolerance_requires_single_api_server():
1551 + """Fault tolerance assumes one AsyncMPClient manages all engines, so it
1552 + is incompatible with API server scale-out (_api_process_count > 1)."""
1553 + with pytest.raises(ValueError, match="single API server"):
1554 + ParallelConfig(enable_fault_tolerance=True, _api_process_count=2)
1555 +
1556 + # Single API server (the FT-supported topology) is accepted.
1557 + ParallelConfig(enable_fault_tolerance=True, _api_process_count=1)
1558 +
1559 +
1560 def test_renderer_num_workers_with_mm_cache():
1561 """Disallow renderer_num_workers > 1 when mm processor cache is enabled,
1562 since neither cache type is thread-safe."""
tests/v1/fault_tolerance/__init__.py new
+2
@@ -0,0 +1,2 @@
1 +# SPDX-License-Identifier: Apache-2.0
2 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
tests/v1/fault_tolerance/test_fault_tolerance_e2e.py new
+378
@@ -0,0 +1,378 @@
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
vllm/config/__init__.py
+3
@@ -13,6 +13,7 @@ from vllm.config.device import DeviceConfig
13 from vllm.config.diffusion import DiffusionConfig
14 from vllm.config.ec_manager_config import EncoderCacheManagerConfig
15 from vllm.config.ec_transfer import ECTransferConfig
16 +from vllm.config.fault_tolerance import FaultToleranceConfig
17 from vllm.config.kernel import KernelConfig
18 from vllm.config.kv_events import KVEventsConfig
19 from vllm.config.kv_transfer import KVTransferConfig
@@ -124,6 +125,8 @@ __all__ = [
125 "StructuredOutputsConfig",
126 # From vllm.config.profiler
127 "ProfilerConfig",
128 + # From vllm.config.fault_tolerance
129 + "FaultToleranceConfig",
130 # From vllm.config.utils
131 "ConfigType",
132 "SupportsMetricsInfo",
vllm/config/fault_tolerance.py new
+18
@@ -0,0 +1,18 @@
1 +# SPDX-License-Identifier: Apache-2.0
2 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 +
4 +
5 +from vllm.config.utils import config
6 +
7 +
8 +@config
9 +class FaultToleranceConfig:
10 + """Configuration for fault tolerance."""
11 +
12 + engine_recovery_timeout_sec: int = 120
13 + """Timeout (in seconds) to wait for error handling instructions
14 + before raising an exception. If the EngineCore encounters an
15 + error, it waits up to this many seconds for vLLM to receive
16 + instructions on how to handle the error and then recover from the fault.
17 + If vLLM does not recover during this time, the original error is raised.
18 + """
vllm/config/parallel.py
+19
@@ -13,6 +13,7 @@ from torch.distributed import ProcessGroup, ReduceOp, Store
13 from typing_extensions import Self
14
15 import vllm.envs as envs
16 +from vllm.config.fault_tolerance import FaultToleranceConfig
17 from vllm.config.utils import config
18 from vllm.logger import init_logger
19 from vllm.platforms import current_platform
@@ -22,6 +23,7 @@ if TYPE_CHECKING:
23 from ray.runtime_env import RuntimeEnv
24 from ray.util.placement_group import PlacementGroup
25
26 + from vllm.config.fault_tolerance import FaultToleranceConfig
27 from vllm.v1.executor import Executor
28 else:
29 RuntimeEnv = Any
@@ -393,6 +395,16 @@ class ParallelConfig:
395 should only be set by API server scale-out.
396 """
397
398 + enable_fault_tolerance: bool = False
399 + """Enable fault tolerance for detailed error recovery,
400 + such as scaling down fault DPEngineCore.
401 + """
402 +
403 + fault_tolerance_config: FaultToleranceConfig = Field(
404 + default_factory=FaultToleranceConfig
405 + )
406 + """The configurations for fault tolerance."""
407 +
408 @field_validator("disable_nccl_for_dp_synchronization", mode="wrap")
409 @classmethod
410 def _skip_none_validation(cls, value: Any, handler: Callable) -> Any:
@@ -445,6 +457,13 @@ class ParallelConfig:
457 f"but found: {self._api_process_rank}"
458 )
459
460 + if self.enable_fault_tolerance and self._api_process_count > 1:
461 + raise ValueError(
462 + "Fault tolerance requires a single API server process "
463 + f"(--api-server-count=1), but got {self._api_process_count}. "
464 + "The FT system assumes one AsyncMPClient manages all engines."
465 + )
466 +
467 if self.all2all_backend in ["pplx", "naive"]:
468 logger.warning(
469 "The '%s' all2all backend has been removed. "
vllm/distributed/device_communicators/all2all.py
+26 -1
@@ -8,6 +8,7 @@ import torch
8 import torch.distributed as dist
9
10 import vllm.envs as envs
11 +from vllm.config import get_current_vllm_config
12 from vllm.distributed import get_dp_group, get_ep_group, get_pcp_group
13 from vllm.distributed.utils import StatelessProcessGroup
14 from vllm.forward_context import get_forward_context
@@ -278,7 +279,9 @@ class DeepEPLLAll2AllManager(DeepEPAll2AllManagerBase):
279
280 def __init__(self, cpu_group, tcp_store_group=None):
281 super().__init__(cpu_group, tcp_store_group)
281 - self.support_fault_tolerance = False # TODO: set to True when FT is supported.
282 + self.support_fault_tolerance = (
283 + get_current_vllm_config().parallel_config.enable_fault_tolerance
284 + )
285
286 def _make_all2all_kwargs(
287 self,
@@ -360,6 +363,16 @@ class DeepEPLLAll2AllManager(DeepEPAll2AllManagerBase):
363 has_fault = (current != DeepEPLLAll2AllManager._last_mask).any()
364 return has_fault
365
366 + def clean_buffers(self) -> None:
367 + buf = DeepEPLLAll2AllManager._buffer
368 + if buf is None:
369 + return
370 + buf.get_local_buffer_tensor(dtype=torch.int8, use_rdma_buffer=True).zero_()
371 + torch.accelerator.synchronize()
372 + buf.low_latency_clean_mask_buffer()
373 + torch.accelerator.synchronize()
374 + DeepEPLLAll2AllManager._last_mask = None
375 +
376
377 @dataclass
378 class _NixlEPBufferState:
@@ -565,6 +578,18 @@ class NixlEPAll2AllManager(All2AllManagerBase):
578 has_fault = (current != last).any()
579 return has_fault
580
581 + def clean_buffers(self) -> None:
582 + if NixlEPAll2AllManager._buffer is None:
583 + return
584 + state = NixlEPAll2AllManager._buffer
585 + state.buffer.get_local_buffer_tensor(
586 + dtype=torch.int8, use_rdma_buffer=True
587 + ).zero_()
588 + torch.accelerator.synchronize()
589 + state.buffer.clean_mask_buffer()
590 + torch.accelerator.synchronize()
591 + NixlEPAll2AllManager._last_mask = None
592 +
593
594 class FlashInferNVLinkTwoSidedManager(All2AllManagerBase):
595 """
vllm/distributed/device_communicators/base_device_communicator.py
+21 -1
@@ -105,10 +105,30 @@ class All2AllManagerBase:
105 raise NotImplementedError
106
107 def query_active_mask(self) -> torch.Tensor:
108 + """Return the all2all liveness mask for the EP ranks.
109 +
110 + Returns:
111 + An int32 device tensor where 0 marks a live rank and 1 marks a
112 + masked (dead/unreachable) rank.
113 + """
114 raise NotImplementedError
115
116 def query_fault(self) -> torch.Tensor:
111 - """Returns has_fault scalar."""
117 + """Return a scalar bool tensor, True if a new fault appeared.
118 +
119 + Compares the current mask against the baseline recorded at the last
120 + recovery point.
121 + """
122 + raise NotImplementedError
123 +
124 + def clean_buffers(self) -> None:
125 + """Reset this rank's RDMA buffers and all2all mask state (rank-local).
126 +
127 + Post-fault cleanup: a dispatch/combine that hit a dead peer or timed
128 + out can leave partially-written or stale tokens in the RDMA receive
129 + buffer, so it is zeroed to stop the next forward from reading that
130 + contaminated data.
131 + """
132 raise NotImplementedError
133
134 def set_num_sms(self, num_sms: int):
vllm/engine/arg_utils.py
+31
@@ -42,6 +42,7 @@ from vllm.config import (
42 DiffusionConfig,
43 ECTransferConfig,
44 EPLBConfig,
45 + FaultToleranceConfig,
46 KernelConfig,
47 KVEventsConfig,
48 KVTransferConfig,
@@ -724,6 +725,11 @@ class EngineArgs:
725 optimization_level: OptimizationLevel = VllmConfig.optimization_level
726 performance_mode: PerformanceMode = VllmConfig.performance_mode
727
728 + fault_tolerance_config: FaultToleranceConfig = get_field(
729 + ParallelConfig, "fault_tolerance_config"
730 + )
731 + enable_fault_tolerance: bool = ParallelConfig.enable_fault_tolerance
732 +
733 kv_offloading_size: float | None = CacheConfig.kv_offloading_size
734 kv_offloading_backend: KVOffloadingBackend = CacheConfig.kv_offloading_backend
735 tokens_only: bool = False
@@ -756,6 +762,16 @@ class EngineArgs:
762 self.weight_transfer_config = WeightTransferConfig(
763 **self.weight_transfer_config
764 )
765 + if isinstance(self.fault_tolerance_config, dict):
766 + if not self.enable_fault_tolerance:
767 + logger.warning(
768 + "--fault-tolerance-config was passed. Fault tolerance is being "
769 + "automatically enabled."
770 + )
771 + self.enable_fault_tolerance = True
772 + self.fault_tolerance_config = FaultToleranceConfig(
773 + **self.fault_tolerance_config
774 + )
775 if isinstance(self.ir_op_priority, dict):
776 self.ir_op_priority = IrOpPriorityConfig(**self.ir_op_priority)
777
@@ -1153,6 +1169,12 @@ class EngineArgs:
1169 parallel_group.add_argument(
1170 "--worker-extension-cls", **parallel_kwargs["worker_extension_cls"]
1171 )
1172 + parallel_group.add_argument(
1173 + "--enable-fault-tolerance", **parallel_kwargs["enable_fault_tolerance"]
1174 + )
1175 + parallel_group.add_argument(
1176 + "--fault-tolerance-config", **parallel_kwargs["fault_tolerance_config"]
1177 + )
1178
1179 # KV cache arguments
1180 cache_kwargs = get_kwargs(CacheConfig)
@@ -1619,6 +1641,7 @@ class EngineArgs:
1641 def from_cli_args(cls, args: argparse.Namespace):
1642 # Get the list of attributes of this dataclass.
1643 attrs = [attr.name for attr in dataclasses.fields(cls)]
1644 +
1645 # Set the attributes from the parsed arguments.
1646 engine_args = cls(
1647 **{attr: getattr(args, attr) for attr in attrs if hasattr(args, attr)}
@@ -2024,6 +2047,12 @@ class EngineArgs:
2047 data_parallel_external_lb = (
2048 self.data_parallel_external_lb or self.data_parallel_rank is not None
2049 )
2050 + if self.enable_fault_tolerance and not data_parallel_external_lb:
2051 + raise ValueError(
2052 + "Fault tolerance requires external load balancer mode "
2053 + "(--data-parallel-external-lb or --data-parallel-rank). "
2054 + "Internal LB mode is not supported."
2055 + )
2056 if (
2057 self.data_parallel_size > 1
2058 and data_parallel_external_lb
@@ -2181,6 +2210,8 @@ class EngineArgs:
2210 _api_process_count=self._api_process_count,
2211 _api_process_rank=self._api_process_rank,
2212 assigned_physical_gpu_ids=self._resolve_device_ids(),
2213 + enable_fault_tolerance=self.enable_fault_tolerance,
2214 + fault_tolerance_config=self.fault_tolerance_config,
2215 numa_bind=self.numa_bind,
2216 numa_bind_nodes=self.numa_bind_nodes,
2217 numa_bind_cpus=self.numa_bind_cpus,
vllm/engine/protocol.py
+11
@@ -20,6 +20,7 @@ from vllm.sampling_params import SamplingParams
20 from vllm.tasks import SupportedTask
21 from vllm.v1.engine import EngineCoreRequest
22 from vllm.v1.engine.input_processor import InputProcessor
23 +from vllm.v1.fault_tolerance.utils import FaultToleranceRequest, FaultToleranceResult
24
25 if TYPE_CHECKING:
26 from vllm.v1.engine import PauseMode
@@ -234,6 +235,16 @@ class EngineClient(ABC):
235 """Perform a collective RPC call to the given path."""
236 raise NotImplementedError
237
238 + async def handle_fault(
239 + self, fault_tolerance_request: FaultToleranceRequest
240 + ) -> FaultToleranceResult:
241 + """send fault tolerance instruction to the engine"""
242 + raise NotImplementedError
243 +
244 + async def get_status(self):
245 + """Get fault tolerance status of all engines."""
246 + raise NotImplementedError
247 +
248 async def get_supported_tasks(self) -> tuple[SupportedTask, ...]:
249 """Get supported tasks"""
250 raise NotImplementedError
vllm/entrypoints/openai/api_server.py
+7
@@ -269,6 +269,13 @@ def build_app(
269
270 register_pooling_api_routers(app, supported_tasks, model_config)
271
272 + if args.enable_fault_tolerance:
273 + from vllm.entrypoints.serve.fault_tolerance.api_router import (
274 + register_fault_tolerance_api_router,
275 + )
276 +
277 + register_fault_tolerance_api_router(app)
278 +
279 # Endpoint plugins are attached last so their routes are registered after all core
280 # routers. This runs even for the CPU only render server. A plugin eligible for
281 # the `render` task still gets its routes registered. It receives
vllm/entrypoints/serve/fault_tolerance/__init__.py
vllm/entrypoints/serve/fault_tolerance/api_router.py new
+100
@@ -0,0 +1,100 @@
1 +# SPDX-License-Identifier: Apache-2.0
2 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 +import json
4 +import uuid
5 +from http import HTTPStatus
6 +
7 +from fastapi import APIRouter, BackgroundTasks, Depends, FastAPI, HTTPException, Request
8 +from fastapi.responses import JSONResponse
9 +
10 +from vllm.engine.protocol import EngineClient
11 +from vllm.entrypoints.openai.engine.protocol import ErrorResponse
12 +from vllm.entrypoints.serve.utils.api_utils import validate_json_request
13 +from vllm.logger import init_logger
14 +from vllm.v1.fault_tolerance.utils import FaultToleranceRequest
15 +
16 +logger = init_logger(__name__)
17 +
18 +router = APIRouter()
19 +
20 +_ALLOWED_INSTRUCTIONS = {"retry"}
21 +
22 +
23 +def _validate_payload(body: dict) -> tuple[str, dict]:
24 + if not isinstance(body, dict):
25 + raise HTTPException(400, "Request body must be a JSON object.")
26 + instruction = body.get("instruction")
27 + if not instruction:
28 + raise HTTPException(400, "'instruction' is required.")
29 + if instruction not in _ALLOWED_INSTRUCTIONS:
30 + raise HTTPException(400, f"Invalid instruction: '{instruction}'.")
31 + params = body.get("params", {})
32 + if not isinstance(params, dict):
33 + raise HTTPException(400, "'params' must be an object.")
34 + return instruction, params
35 +
36 +
37 +@router.post(
38 + "/fault_tolerance/apply",
39 + dependencies=[Depends(validate_json_request)],
40 + responses={
41 + HTTPStatus.ACCEPTED.value: {"model": dict},
42 + HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
43 + },
44 +)
45 +async def process_fault_tolerance_instruction(
46 + raw_request: Request, background_tasks: BackgroundTasks
47 +):
48 + try:
49 + body = await raw_request.json()
50 + except json.JSONDecodeError as e:
51 + raise HTTPException(400, "Invalid JSON format") from e
52 +
53 + instruction, params = _validate_payload(body)
54 + ft_request = FaultToleranceRequest(
55 + instruction=instruction,
56 + params=params,
57 + request_id=str(uuid.uuid4()),
58 + )
59 +
60 + client: EngineClient = raw_request.app.state.engine_client
61 + # Recovery runs cross-rank collective ops that only complete once every rank
62 + # has been dispatched. Run it in the background and return immediately so the
63 + # orchestrator can dispatch to all ranks without blocking; completion is
64 + # observed by polling GET /fault_tolerance/status.
65 + background_tasks.add_task(_run_fault_recovery, client, ft_request)
66 + return JSONResponse(
67 + status_code=HTTPStatus.ACCEPTED.value,
68 + content={
69 + "message": "Request accepted; poll /fault_tolerance/status for updates.",
70 + "request_id": ft_request.request_id,
71 + },
72 + background=background_tasks,
73 + )
74 +
75 +
76 +async def _run_fault_recovery(
77 + client: EngineClient, ft_request: FaultToleranceRequest
78 +) -> None:
79 + """Drive recovery to completion after the 202 response is sent."""
80 + try:
81 + result = await client.handle_fault(ft_request)
82 + except Exception:
83 + logger.exception("[FT] Recovery dispatch failed.")
84 + return
85 + if not result.success:
86 + logger.error(
87 + "[FT] Recovery failed for request %s: %s",
88 + ft_request.request_id,
89 + result.reason,
90 + )
91 +
92 +
93 +@router.get("/fault_tolerance/status")
94 +async def get_status(raw_request: Request):
95 + client: EngineClient = raw_request.app.state.engine_client
96 + return JSONResponse(content=await client.get_status())
97 +
98 +
99 +def register_fault_tolerance_api_router(app: FastAPI):
100 + app.include_router(router)
vllm/v1/engine/__init__.py
+8
@@ -31,6 +31,8 @@ FINISH_REASON_STRINGS = ("stop", "length", "abort", "error", "repetition")
31
32 EEP_NOTIFICATION_CALL_ID = -1
33
34 +FT_STATUS_CALL_ID = -2
35 +
36
37 class EEPNotificationType(enum.Enum):
38 NEW_CORE_ENGINES_INIT_READY = "NEW_CORE_ENGINES_INIT_READY"
@@ -282,3 +284,9 @@ class ReconfigureRankType(enum.IntEnum):
284
285 KEEP_CURRENT_RANK = -1
286 SHUTDOWN_CURRENT_RANK = -2
287 +
288 +
289 +class EngineStatusType(enum.IntEnum):
290 + HEALTHY = 0
291 + DEAD = 1
292 + UNHEALTHY = 2
vllm/v1/engine/async_llm.py
+10
@@ -44,6 +44,7 @@ from vllm.v1.engine.input_processor import InputProcessor
44 from vllm.v1.engine.output_processor import OutputProcessor, RequestOutputCollector
45 from vllm.v1.engine.parallel_sampling import ParentRequest
46 from vllm.v1.executor import Executor
47 +from vllm.v1.fault_tolerance.utils import FaultToleranceRequest, FaultToleranceResult
48 from vllm.v1.metrics.loggers import (
49 StatLoggerFactory,
50 StatLoggerManager,
@@ -1041,6 +1042,15 @@ class AsyncLLM(EngineClient):
1042 finally:
1043 set_scaling_elastic_ep(False)
1044
1045 + async def handle_fault(
1046 + self, fault_tolerance_request: FaultToleranceRequest
1047 + ) -> FaultToleranceResult:
1048 + """send fault tolerance instruction to the engine"""
1049 + return await self.engine_core.handle_fault(fault_tolerance_request)
1050 +
1051 + async def get_status(self):
1052 + return await self.engine_core.get_status()
1053 +
1054 @property
1055 def is_running(self) -> bool:
1056 # Is None before the loop is started.
vllm/v1/engine/core.py
+25
@@ -78,6 +78,11 @@ from vllm.v1.engine.utils import (
78 get_physical_gpu_ids_for_local_dp_rank,
79 )
80 from vllm.v1.executor import Executor
81 +from vllm.v1.fault_tolerance.engine_core_sentinel import (
82 + FT_UTILITY_METHOD,
83 + EngineCoreSentinel,
84 + fault_tolerant_wrapper,
85 +)
86 from vllm.v1.kv_cache_interface import KVCacheConfig, get_kv_cache_spec_kind
87 from vllm.v1.metrics.stats import SchedulerIterationDetails, SchedulerStats
88 from vllm.v1.outputs import ModelRunnerOutput
@@ -1070,6 +1075,16 @@ class EngineCoreProc(EngineCore):
1075 internal_dp_balancing,
1076 )
1077
1078 + # Initialize fault tolerance settings.
1079 + self.enable_fault_tolerance = (
1080 + vllm_config.parallel_config.enable_fault_tolerance
1081 + )
1082 + if self.enable_fault_tolerance:
1083 + self.ft_sentinel = EngineCoreSentinel(
1084 + engine=self,
1085 + parallel_config=vllm_config.parallel_config,
1086 + )
1087 +
1088 # Background Threads and Queues for IO. These enable us to
1089 # overlap ZMQ socket IO with GPU since they release the GIL,
1090 # and to overlap some serialization/deserialization with the
@@ -1355,6 +1370,7 @@ class EngineCoreProc(EngineCore):
1370 """Returns true if shutdown has not been requested."""
1371 return self.shutdown_state == EngineShutdownState.RUNNING
1372
1373 + @fault_tolerant_wrapper
1374 def run_busy_loop(self):
1375 """Core busy loop of the EngineCore."""
1376 while self._handle_shutdown():
@@ -1672,6 +1688,14 @@ class EngineCoreProc(EngineCore):
1688 except Exception:
1689 self._handle_request_preproc_error(req)
1690 continue
1691 + elif request_type == EngineCoreRequestType.UTILITY:
1692 + request = generic_decoder.decode(data_frames)
1693 + client_idx, call_id, method, args = request
1694 + if method == FT_UTILITY_METHOD:
1695 + self.ft_sentinel.handle_command(
1696 + client_idx, call_id, args[0]
1697 + )
1698 + continue
1699 else:
1700 request = generic_decoder.decode(data_frames)
1701
@@ -2021,6 +2045,7 @@ class DPEngineCoreProc(EngineCoreProc):
2045 and self.step_counter % self.prefill_schedule_interval != 0
2046 )
2047
2048 + @fault_tolerant_wrapper
2049 def run_busy_loop(self):
2050 """Core busy loop of the EngineCore for data parallel case."""
2051
vllm/v1/engine/core_client.py
+51 -1
@@ -16,6 +16,7 @@ from multiprocessing.queues import Queue
16 from threading import Thread
17 from typing import Any, TypeAlias, TypeVar
18
19 +import msgspec
20 import msgspec.msgpack
21 import zmq
22 import zmq.asyncio
@@ -35,6 +36,7 @@ from vllm.utils.network_utils import (
36 )
37 from vllm.v1.engine import (
38 EEP_NOTIFICATION_CALL_ID,
39 + FT_STATUS_CALL_ID,
40 EEPNotificationType,
41 EngineCoreOutputs,
42 EngineCoreReadyResponse,
@@ -56,6 +58,11 @@ from vllm.v1.engine.utils import (
58 launch_core_engines,
59 )
60 from vllm.v1.executor import Executor
61 +from vllm.v1.fault_tolerance.engine_core_sentinel import FT_UTILITY_METHOD
62 +from vllm.v1.fault_tolerance.utils import (
63 + FaultToleranceRequest,
64 + FaultToleranceResult,
65 +)
66 from vllm.v1.pool.late_interaction import get_late_interaction_engine_index
67 from vllm.v1.serial_utils import MsgpackDecoder, MsgpackEncoder, bytestr
68
@@ -272,6 +279,14 @@ class EngineCoreClient(ABC):
279 ) -> list[_R]:
280 raise NotImplementedError
281
282 + async def handle_fault(
283 + self, fault_tolerance_request: FaultToleranceRequest
284 + ) -> FaultToleranceResult:
285 + raise NotImplementedError
286 +
287 + async def get_status(self):
288 + raise NotImplementedError
289 +
290
291 class InprocClient(EngineCoreClient):
292 """
@@ -971,6 +986,14 @@ class AsyncMPClient(MPClient):
986 self.client_count = client_count
987 self.client_index = client_index
988 self.outputs_queue = asyncio.Queue[EngineCoreOutputs | Exception]()
989 +
990 + # locally-cached engine status
991 + self._engine_status: dict[int, dict] = {}
992 + if self.vllm_config.parallel_config.enable_fault_tolerance:
993 + self._engine_status = {
994 + rank: {"id": rank, "status": "healthy"}
995 + for rank in self.engine_ranks_managed
996 + }
997 try:
998 # If we are running in an asyncio event loop, start the queue task.
999 # Otherwise, it will be started lazily. If it is not started here,
@@ -994,7 +1017,7 @@ class AsyncMPClient(MPClient):
1017 output_handler: (
1018 Callable[[AsyncMPClient, EngineCoreOutputs], Awaitable[None]] | None
1019 ) = getattr(self.__class__, "process_engine_outputs", None)
997 - _self_ref = weakref.ref(self) if output_handler else None
1020 + _self_ref = weakref.ref(self)
1021 output_socket = resources.output_socket
1022 assert output_socket is not None
1023
@@ -1025,6 +1048,14 @@ class AsyncMPClient(MPClient):
1048 asyncio.create_task(
1049 notification_callback_handler(_self, notification_data)
1050 )
1051 + elif outputs.utility_output.call_id == FT_STATUS_CALL_ID:
1052 + _self = _self_ref()
1053 + if not _self:
1054 + return
1055 + if outputs.utility_output.result is not None:
1056 + _self._engine_status[outputs.engine_index] = (
1057 + outputs.utility_output.result.result
1058 + )
1059 else:
1060 _process_utility_output(
1061 outputs.utility_output, utility_results
@@ -1196,6 +1227,25 @@ class AsyncMPClient(MPClient):
1227 "collective_rpc", method, timeout, args, kwargs
1228 )
1229
1230 + async def handle_fault(
1231 + self, ft_request: FaultToleranceRequest
1232 + ) -> FaultToleranceResult:
1233 + res = await self.call_utility_async(FT_UTILITY_METHOD, ft_request)
1234 + result = msgspec.convert(res, FaultToleranceResult)
1235 + if not result.success:
1236 + status = self._engine_status.get(self.engine_ranks_managed[0])
1237 + if status is not None:
1238 + status["last_ft_request_id"] = result.request_id
1239 + status["ft_error"] = result.reason
1240 + return result
1241 +
1242 + async def get_status(self):
1243 + return {
1244 + "schema_version": 1,
1245 + "total_engines": len(self.engine_ranks_managed),
1246 + "engines": list(self._engine_status.values()),
1247 + }
1248 +
1249
1250 class DPAsyncMPClient(AsyncMPClient):
1251 """Asyncio-compatible client for multi-proc, multi-engine (data parallel)
vllm/v1/engine/utils.py
-3
@@ -234,9 +234,6 @@ class CoreEngineProcManager:
234 if exitcode != 0 and not self.manager_stopped.is_set():
235 self.failed_proc_name = proc.name
236 if died_sentinels:
237 - # Any engine exit currently triggers a shutdown. Future
238 - # work (e.g., Elastic and fault-tolerant EP) will add finer-grained
239 - # handling for different exit scenarios.
237 break
238
239 self.shutdown()
vllm/v1/fault_tolerance/__init__.py new
+8
@@ -0,0 +1,8 @@
1 +# SPDX-License-Identifier: Apache-2.0
2 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 +from .engine_core_sentinel import EngineCoreSentinel, fault_tolerant_wrapper
4 +
5 +__all__ = [
6 + "EngineCoreSentinel",
7 + "fault_tolerant_wrapper",
8 +]
vllm/v1/fault_tolerance/engine_core_sentinel.py new
+197
@@ -0,0 +1,197 @@
1 +# SPDX-License-Identifier: Apache-2.0
2 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 +"""EngineCoreSentinel and fault_tolerant_wrapper for the engine core."""
4 +
5 +import json
6 +import threading
7 +from collections.abc import Callable
8 +from typing import TYPE_CHECKING
9 +
10 +import msgspec
11 +
12 +from vllm.config import set_current_vllm_config
13 +from vllm.distributed import stateless_destroy_torch_distributed_process_group
14 +from vllm.distributed.utils import stateless_init_torch_distributed_process_group
15 +from vllm.logger import init_logger
16 +from vllm.utils.network_utils import get_open_port
17 +from vllm.v1.engine import (
18 + FT_STATUS_CALL_ID,
19 + EngineCoreOutputs,
20 + EngineStatusType,
21 + UtilityOutput,
22 +)
23 +from vllm.v1.fault_tolerance.utils import FaultToleranceRequest, FaultToleranceResult
24 +from vllm.v1.request import RequestStatus
25 +from vllm.v1.serial_utils import UtilityResult, run_method
26 +
27 +if TYPE_CHECKING:
28 + from vllm.v1.engine.core import EngineCoreProc
29 +
30 +logger = init_logger(__name__)
31 +
32 +FT_UTILITY_METHOD = "handle_fault_tolerance"
33 +
34 +
35 +class EngineCoreSentinel:
36 + """Manages fault tolerance state for a single engine core."""
37 +
38 + def __init__(self, engine: "EngineCoreProc", parallel_config):
39 + self.engine = engine
40 + self.engine_index = engine.engine_index
41 + self.parallel_config = parallel_config
42 + ft_config = parallel_config.fault_tolerance_config
43 + self.engine_recovery_timeout_sec = ft_config.engine_recovery_timeout_sec
44 +
45 + self.resumed = threading.Event()
46 + self.resumed.set()
47 + self.status_type = EngineStatusType.HEALTHY
48 + self.fault_info: str | None = None
49 + self._dp_reinit_epoch = 0
50 +
51 + def handle_command(self, client_idx: int, call_id: int, ft_args: dict):
52 + """Dispatch an FT command by instruction name."""
53 + ft_request = FaultToleranceRequest(**ft_args)
54 + if self.status_type != EngineStatusType.UNHEALTHY:
55 + reason = (
56 + f"[FT] Rejecting {ft_request.instruction} on engine "
57 + f"{self.engine_index}: status is {self.status_type.name}"
58 + )
59 + logger.warning(reason)
60 + result = FaultToleranceResult(
61 + request_id=ft_request.request_id,
62 + success=False,
63 + reason=reason,
64 + )
65 + else:
66 + try:
67 + result = run_method(self, ft_request.instruction, (ft_request,), {})
68 + except Exception as e:
69 + logger.exception("[FT] Instruction '%s' failed", ft_request.instruction)
70 + result = FaultToleranceResult(
71 + request_id=ft_request.request_id, success=False, reason=str(e)
72 + )
73 +
74 + uo = UtilityOutput(call_id)
75 + uo.result = UtilityResult(msgspec.structs.asdict(result))
76 + self.engine.output_queue.put_nowait(
77 + (client_idx, EngineCoreOutputs(utility_output=uo))
78 + )
79 +
80 + def on_fault(self, exc: Exception):
81 + """Called by the wrapper when the busy loop raises an exception."""
82 + self.resumed.clear()
83 + logger.warning(
84 + "[FT] Busy loop raised %s. Waiting for recovery.", type(exc).__name__
85 + )
86 +
87 + engine = self.engine
88 + aborted = engine.scheduler.finish_requests(None, RequestStatus.FINISHED_ABORTED)
89 + engine._send_abort_outputs(aborted)
90 + if engine.batch_queue is not None:
91 + engine.batch_queue.clear()
92 + if (
93 + hasattr(engine.model_executor, "is_failed")
94 + and engine.model_executor.is_failed
95 + ):
96 + self.status_type = EngineStatusType.DEAD
97 + else:
98 + self.status_type = EngineStatusType.UNHEALTHY
99 + self.fault_info = f"{type(exc).__name__}"
100 + logger.info(
101 + "[FT] Engine %d status -> %s:",
102 + self.engine_index,
103 + self.status_type.name,
104 + exc_info=exc,
105 + )
106 + self._push_status()
107 +
108 + def _push_status(self):
109 + """Push current health to the client so it can refresh its cache."""
110 + payload = {"id": self.engine_index, "status": self.status_type.name.lower()}
111 + if self.status_type == EngineStatusType.UNHEALTHY:
112 + payload["fault_info"] = self.fault_info
113 + outputs = EngineCoreOutputs(
114 + utility_output=UtilityOutput(
115 + call_id=FT_STATUS_CALL_ID,
116 + result=UtilityResult(payload),
117 + )
118 + )
119 + outputs.engine_index = self.engine_index
120 + self.engine.output_queue.put_nowait((0, outputs))
121 +
122 + def retry(self, ft_request: FaultToleranceRequest) -> FaultToleranceResult:
123 + engine = self.engine
124 + executor = engine.model_executor
125 +
126 + with set_current_vllm_config(engine.vllm_config):
127 + ft_request.params.update(self._reinit_dp_group())
128 + if hasattr(engine, "step_counter"):
129 + engine.step_counter = 0
130 +
131 + executor.collective_rpc("handle_ft_command", args=(ft_request,))
132 +
133 + self.status_type = EngineStatusType.HEALTHY
134 + logger.info("[FT] Engine %d status -> HEALTHY", self.engine_index)
135 + self.resumed.set()
136 + self._push_status()
137 + return FaultToleranceResult(request_id=ft_request.request_id, success=True)
138 +
139 + def _reinit_dp_group(self) -> dict:
140 + """Reinit DP process group if in DP mode. Returns worker params."""
141 + engine = self.engine
142 + if not hasattr(engine, "dp_group") or not hasattr(engine, "dp_store"):
143 + return {}
144 +
145 + parallel_config = engine.vllm_config.parallel_config
146 + worker_key = f"ft_worker_dp_ports_{self._dp_reinit_epoch}"
147 + engine_key = f"ft_engine_dp_port_{self._dp_reinit_epoch}"
148 + self._dp_reinit_epoch += 1
149 +
150 + if parallel_config.data_parallel_rank == 0:
151 + worker_ports = [get_open_port() for _ in range(parallel_config.world_size)]
152 + engine_port = get_open_port()
153 + engine.dp_store.set(worker_key, json.dumps(worker_ports).encode())
154 + engine.dp_store.set(engine_key, str(engine_port).encode())
155 + else:
156 + worker_ports = json.loads(engine.dp_store.get(worker_key).decode())
157 + engine_port = int(engine.dp_store.get(engine_key).decode())
158 +
159 + stateless_destroy_torch_distributed_process_group(engine.dp_group)
160 + engine.dp_group, engine.dp_store = (
161 + stateless_init_torch_distributed_process_group(
162 + parallel_config.data_parallel_master_ip,
163 + engine_port,
164 + parallel_config.data_parallel_rank,
165 + parallel_config.data_parallel_size,
166 + backend="gloo",
167 + return_store=True,
168 + )
169 + )
170 + return {"new_stateless_dp_group_ports": worker_ports}
171 +
172 +
173 +def fault_tolerant_wrapper(busy_loop_func: Callable):
174 + """Wrap the busy loop to catch faults and delegate recovery."""
175 +
176 + def run_with_fault_tolerance(self: "EngineCoreProc"):
177 + while True:
178 + try:
179 + busy_loop_func(self)
180 + except SystemExit:
181 + raise
182 + except Exception as exc:
183 + if not self.enable_fault_tolerance:
184 + raise
185 + self.ft_sentinel.on_fault(exc)
186 + recovered = self.ft_sentinel.resumed.wait(
187 + timeout=self.ft_sentinel.engine_recovery_timeout_sec
188 + )
189 + if recovered:
190 + continue
191 + logger.error(
192 + "[FT] No recovery within %ds timeout.",
193 + self.ft_sentinel.engine_recovery_timeout_sec,
194 + )
195 + raise
196 +
197 + return run_with_fault_tolerance
vllm/v1/fault_tolerance/utils.py new
+17
@@ -0,0 +1,17 @@
1 +# SPDX-License-Identifier: Apache-2.0
2 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 +from typing import Any
4 +
5 +import msgspec
6 +
7 +
8 +class FaultToleranceResult(msgspec.Struct):
9 + request_id: str
10 + success: bool
11 + reason: str | None = None
12 +
13 +
14 +class FaultToleranceRequest(msgspec.Struct):
15 + instruction: str
16 + params: dict[str, Any]
17 + request_id: str = ""
vllm/v1/worker/gpu/async_utils.py
+15
@@ -5,6 +5,7 @@ import contextlib
5 import numpy as np
6 import torch
7
8 +from vllm.model_executor.layers.fused_moe.all2all_utils import get_ep_all2all_manager
9 from vllm.v1.outputs import AsyncModelRunnerOutput, LogprobsTensors, ModelRunnerOutput
10 from vllm.v1.worker.gpu.sample.output import SamplerOutput
11
@@ -17,6 +18,7 @@ class AsyncOutput(AsyncModelRunnerOutput):
18 num_sampled_tokens: torch.Tensor,
19 main_stream: torch.cuda.Stream,
20 copy_stream: torch.cuda.Stream,
21 + check_ep_fault: bool = False,
22 ):
23 # NOTE(woosuk): We must retain references to the GPU tensors,
24 # as the copy operations are performed on a different CUDA stream than
@@ -26,6 +28,7 @@ class AsyncOutput(AsyncModelRunnerOutput):
28 self.num_sampled_tokens = num_sampled_tokens
29 # Blocking (sleep) event to avoid busy-polling the CUDA driver lock.
30 self.copy_event = torch.cuda.Event(blocking=True)
31 + self._has_fault: torch.Tensor | None = None
32
33 with stream(copy_stream, main_stream):
34 copy_stream.wait_stream(main_stream)
@@ -44,6 +47,9 @@ class AsyncOutput(AsyncModelRunnerOutput):
47 k: v.to_cpu_nonblocking() if v is not None else None
48 for k, v in self.model_runner_output.prompt_logprobs_dict.items()
49 }
50 + if check_ep_fault:
51 + has_fault = get_ep_all2all_manager().query_fault()
52 + self._has_fault = has_fault.to("cpu", non_blocking=True)
53 self.copy_event.record(copy_stream)
54
55 def get_output(self) -> ModelRunnerOutput:
@@ -67,6 +73,15 @@ class AsyncOutput(AsyncModelRunnerOutput):
73 if self.logprobs_tensors is not None:
74 self.model_runner_output.logprobs = self.logprobs_tensors.tolists()
75 self.model_runner_output.prompt_logprobs_dict = self.prompt_logprobs_dict
76 +
77 + if self._has_fault is not None and self._has_fault.item():
78 + mask = get_ep_all2all_manager().query_active_mask()
79 + raise RuntimeError(
80 + "Fault detected in EP all2all communication: "
81 + "one or more ranks timed out during dispatch/combine. "
82 + f"Mask: {mask.cpu().tolist()}"
83 + )
84 +
85 return self.model_runner_output
86
87
vllm/v1/worker/gpu/model_runner.py
+8
@@ -38,6 +38,7 @@ from vllm.distributed.parallel_state import (
38 )
39 from vllm.forward_context import BatchDescriptor, set_forward_context
40 from vllm.logger import init_logger
41 +from vllm.model_executor.layers.fused_moe.all2all_utils import get_ep_all2all_manager
42 from vllm.model_executor.layers.mamba.ops.ssu_dispatch import (
43 initialize_mamba_ssu_backend,
44 )
@@ -175,6 +176,12 @@ class GPUModelRunner(LoRAModelRunnerMixin):
176 self.dp_size = self.parallel_config.data_parallel_size
177 self.dp_rank = self.parallel_config.data_parallel_rank
178
179 + # Detect EP all2all peer faults to prevent emitting corrupted output.
180 + # Only meaningful for MoE + DP with an FT-capable all2all backend.
181 + self.check_ep_fault = False
182 + if self.dp_size > 1 and self.model_config.is_moe:
183 + self.check_ep_fault = get_ep_all2all_manager().support_fault_tolerance
184 +
185 # Decode context parallelism.
186 self.dcp_size = self.parallel_config.decode_context_parallel_size
187 self.use_dcp = self.dcp_size > 1
@@ -1488,6 +1495,7 @@ class GPUModelRunner(LoRAModelRunnerMixin):
1495 num_sampled_tokens=num_sampled,
1496 main_stream=self.main_stream,
1497 copy_stream=self.output_copy_stream,
1498 + check_ep_fault=self.check_ep_fault,
1499 )
1500
1501 mm_inputs: tuple[list[torch.Tensor], torch.Tensor] | None = None
vllm/v1/worker/gpu_worker.py
+8 -1
@@ -70,6 +70,7 @@ from vllm.v1.outputs import (
70 ModelRunnerOutput,
71 )
72 from vllm.v1.utils import compute_iteration_details, report_usage_stats
73 +from vllm.v1.worker.sentinel.gpu_worker_sentinel import WorkerSentinel
74 from vllm.v1.worker.startup_plan import (
75 maybe_apply_startup_plan,
76 maybe_save_startup_plan,
@@ -146,7 +147,9 @@ class Worker(WorkerBase):
147 from vllm.distributed.elastic_ep.elastic_execute import ElasticEPScalingExecutor
148
149 self.elastic_ep_executor = ElasticEPScalingExecutor(self)
149 -
150 + self.worker_sentinel: WorkerSentinel | None = None
151 + if self.parallel_config.enable_fault_tolerance:
152 + self.worker_sentinel = WorkerSentinel(worker=self)
153 # Buffers saved before sleep
154 self._sleep_saved_buffers: dict[str, torch.Tensor] = {}
155 self._sleep_rebuild_draft_metadata_buffers = False
@@ -414,6 +417,10 @@ class Worker(WorkerBase):
417 # If usage stat is enabled, collect relevant info.
418 report_usage_stats(self.vllm_config)
419
420 + def handle_ft_command(self, ft_request):
421 + assert self.worker_sentinel is not None
422 + return self.worker_sentinel.handle_command(ft_request)
423 +
424 # FIXME(youkaichao & ywang96): Use TorchDispatchMode instead of memory pool
425 # to hijack tensor allocation.
426 def load_model(self, *, load_dummy_weights: bool = False) -> None:
vllm/v1/worker/sentinel/__init__.py
vllm/v1/worker/sentinel/gpu_worker_sentinel.py new
+89
@@ -0,0 +1,89 @@
1 +# SPDX-License-Identifier: Apache-2.0
2 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 +from typing import TYPE_CHECKING, cast
4 +
5 +import torch
6 +
7 +from vllm.config import set_current_vllm_config
8 +from vllm.distributed import (
9 + get_dp_group,
10 + stateless_destroy_torch_distributed_process_group,
11 + stateless_init_torch_distributed_process_group,
12 +)
13 +from vllm.logger import init_logger
14 +from vllm.model_executor.layers.fused_moe.all2all_utils import get_ep_all2all_manager
15 +from vllm.v1.fault_tolerance.utils import FaultToleranceRequest
16 +from vllm.v1.serial_utils import run_method
17 +
18 +if TYPE_CHECKING:
19 + from vllm.v1.worker.gpu.model_runner import GPUModelRunner as GPUModelRunnerV2
20 + from vllm.v1.worker.gpu_worker import Worker
21 +
22 +logger = init_logger(__name__)
23 +
24 +# All2all backends that support fault-tolerant timeout + rank masking,
25 +# required for FT under DP+EP MoE deployments.
26 +FT_BACKEND_SET = frozenset({"deepep_low_latency", "nixl_ep"})
27 +
28 +
29 +class WorkerSentinel:
30 + """Holds FT state for a single worker (mask tensors, DP config).
31 +
32 + Methods are called via collective_rpc from EngineCoreSentinel.
33 + """
34 +
35 + def __init__(self, worker: "Worker"):
36 + self.worker = worker
37 + self.dp_rank = worker.parallel_config.data_parallel_rank
38 + self.dp_size = worker.parallel_config.data_parallel_size
39 + self.data_parallel_master_ip = worker.parallel_config.data_parallel_master_ip
40 + all2all_backend = worker.parallel_config.all2all_backend
41 + if all2all_backend not in FT_BACKEND_SET:
42 + raise ValueError(
43 + f"Fault tolerance requires an FT-capable all2all backend "
44 + f"(one of {sorted(FT_BACKEND_SET)}), but got '{all2all_backend}'."
45 + )
46 +
47 + def handle_command(self, ft_request: FaultToleranceRequest):
48 + """Dispatch an FT command by instruction name."""
49 + with set_current_vllm_config(self.worker.vllm_config):
50 + return run_method(self, ft_request.instruction, (ft_request,), {})
51 +
52 + def retry(self, ft_request: FaultToleranceRequest):
53 + torch.accelerator.synchronize()
54 + params = ft_request.params
55 + self._clean_worker_state()
56 + if self.dp_size > 1:
57 + get_ep_all2all_manager().clean_buffers()
58 + old_cpu_group = get_dp_group().cpu_group
59 + stateless_destroy_torch_distributed_process_group(old_cpu_group)
60 + world_size = self.worker.parallel_config.world_size
61 + port = params["new_stateless_dp_group_ports"][self.worker.rank % world_size]
62 + get_dp_group().cpu_group = stateless_init_torch_distributed_process_group(
63 + self.data_parallel_master_ip,
64 + port,
65 + self.dp_rank,
66 + self.dp_size,
67 + backend="gloo",
68 + )
69 +
70 + def _clean_worker_state(self):
71 + model_runner = self.worker.model_runner
72 + model_runner.execute_model_state = None
73 + if self.worker.use_v2_model_runner:
74 + runner = cast("GPUModelRunnerV2", model_runner)
75 + for req_id in list(runner.req_states.req_id_to_index):
76 + runner._remove_request(req_id)
77 + else:
78 + model_runner.kv_connector_output = None
79 +
80 + input_batch = model_runner.input_batch
81 + cached_req_ids = list(input_batch.req_id_to_index)
82 + for req_id in cached_req_ids:
83 + model_runner.requests.pop(req_id, None)
84 + model_runner.num_prompt_logprobs.pop(req_id, None)
85 + input_batch.remove_request(req_id)
86 +
87 + input_batch.condense()
88 + input_batch.refresh_metadata()
89 + input_batch.req_prompt_embeds.clear()