@tduncan / Nvidia-Optimization-Stack-VLLM Mirror · read-only / commits / 0ddd7dd656
Fork 0

A high-throughput and memory-efficient inference and serving engine for LLMs

Mirrored from https://github.com/tduncan-lum/Nvidia-Optimization-Stack---VLLM.git

Synced Nov 3, 01:45 UTC

[Frontend] DP Supervisor (#40841)

Signed-off-by: yewentao256 <zhyanwentao@126.com> Signed-off-by: Robert Shaw <robertgshaw2@gmail.com> Signed-off-by: Robert Shaw <114415538+robertgshaw2-redhat@users.noreply.github.com> Co-authored-by: robertgshaw2-redhat <robertgshaw2@gmail.com> Co-authored-by: Robert Shaw <114415538+robertgshaw2-redhat@users.noreply.github.com> Co-authored-by: Nick Hill <nickhill123@gmail.com>

Wentao Ye committed May 21, 2026 at 23:33 UTC 0ddd7dd6564f5e403a15bd7c973c7d358ec82454
7 files changed +1168 -17
tests/entrypoints/openai/test_dp_supervisor.py new
+610
@@ -0,0 +1,610 @@
1 +# SPDX-License-Identifier: Apache-2.0
2 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 +
4 +"""
5 +Tests for DPSupervisor: unit tests and lifecycle integration tests.
6 +
7 +Lifecycle integration tests replace child vLLM servers with lightweight
8 +aiohttp "fake" servers controlled by the test, so the suite runs without GPUs.
9 +_start_children is monkeypatched to install FakeProcess objects (with
10 +controllable liveness/timing) alongside those fake HTTP servers.
11 +
12 +Port allocation (kept far from default vLLM ports to avoid conflicts):
13 + Supervisor : 19256
14 + Children : 18000, 18001
15 +"""
16 +
17 +from __future__ import annotations
18 +
19 +import argparse
20 +import asyncio
21 +import contextlib
22 +import os
23 +import signal
24 +import time
25 +from types import SimpleNamespace
26 +
27 +import aiohttp
28 +import pytest
29 +import uvicorn
30 +from fastapi import FastAPI, Response
31 +
32 +import vllm.entrypoints.openai.dp_supervisor as dp_sup
33 +from vllm.entrypoints.openai.dp_supervisor import (
34 + CHILD_EXIT_GRACE_S,
35 + DPSupervisor,
36 + _build_vllm_dp_server_args,
37 + infer_multi_port_external_lb_start_rank,
38 +)
39 +from vllm.logger import init_logger
40 +
41 +logger = init_logger(__name__)
42 +
43 +_SUPERVISOR_PORT = 19256
44 +_CHILD_PORT_BASE = 18000
45 +_N_CHILDREN = 2
46 +_PROBE_INTERVAL = 1.0
47 +_POLL_INTERVAL = 1.0
48 +
49 +
50 +# ---------------------------------------------------------------------------
51 +# Args factories
52 +# ---------------------------------------------------------------------------
53 +
54 +
55 +def _make_unit_args(**overrides) -> argparse.Namespace:
56 + """Minimal args for unit tests (no real network activity)."""
57 + base = {
58 + "host": None,
59 + "port": 8000,
60 + "data_parallel_multi_port_external_lb": True,
61 + "data_parallel_supervisor_port": 9256,
62 + "dp_supervisor_probe_interval_s": 5.0,
63 + "dp_supervisor_probe_timeout_s": 5.0,
64 + "dp_supervisor_probe_failure_threshold": 3,
65 + "data_parallel_size": 8,
66 + "data_parallel_size_local": 4,
67 + "data_parallel_start_rank": None,
68 + "data_parallel_rank": None,
69 + "data_parallel_external_lb": False,
70 + "data_parallel_hybrid_lb": False,
71 + "api_server_count": None,
72 + "headless": False,
73 + "grpc": False,
74 + "uds": None,
75 + "ssl_keyfile": None,
76 + "ssl_certfile": None,
77 + "ssl_ca_certs": None,
78 + "node_rank": 1,
79 + "tensor_parallel_size": 1,
80 + "pipeline_parallel_size": 1,
81 + "uvicorn_log_level": "info",
82 + "shutdown_timeout": 5.0,
83 + }
84 + base.update(overrides)
85 + return argparse.Namespace(**base)
86 +
87 +
88 +def _make_args(**overrides) -> argparse.Namespace:
89 + """Args for lifecycle integration tests (real loopback servers)."""
90 + base: dict = dict(
91 + host="127.0.0.1",
92 + port=_CHILD_PORT_BASE,
93 + data_parallel_multi_port_external_lb=True,
94 + data_parallel_supervisor_port=_SUPERVISOR_PORT,
95 + dp_supervisor_probe_interval_s=_PROBE_INTERVAL,
96 + dp_supervisor_probe_timeout_s=1.0,
97 + dp_supervisor_probe_failure_threshold=3,
98 + data_parallel_size=_N_CHILDREN,
99 + data_parallel_size_local=_N_CHILDREN,
100 + data_parallel_start_rank=0,
101 + data_parallel_rank=None,
102 + data_parallel_external_lb=False,
103 + data_parallel_hybrid_lb=False,
104 + api_server_count=None,
105 + headless=False,
106 + grpc=False,
107 + uds=None,
108 + ssl_keyfile=None,
109 + ssl_certfile=None,
110 + ssl_ca_certs=None,
111 + node_rank=0,
112 + tensor_parallel_size=1,
113 + pipeline_parallel_size=1,
114 + uvicorn_log_level="warning",
115 + shutdown_timeout=0.0,
116 + )
117 + base.update(overrides)
118 + return argparse.Namespace(**base)
119 +
120 +
121 +# ---------------------------------------------------------------------------
122 +# Unit tests
123 +# ---------------------------------------------------------------------------
124 +
125 +
126 +def test_infer_multi_port_external_lb_start_rank_uses_node_rank():
127 + args = _make_unit_args()
128 + assert infer_multi_port_external_lb_start_rank(args) == 4
129 +
130 +
131 +def test_build_multi_port_external_lb_child_args_sets_external_rank_server():
132 + args = _make_unit_args(data_parallel_start_rank=8, api_server_count=None)
133 + child_args = _build_vllm_dp_server_args(args, local_rank=2)
134 +
135 + assert child_args.port == 8002
136 + assert child_args.data_parallel_rank == 10
137 + assert child_args.data_parallel_size_local == 1
138 + assert child_args.data_parallel_external_lb is True
139 + assert child_args.data_parallel_hybrid_lb is False
140 + assert child_args.data_parallel_multi_port_external_lb is False
141 + assert child_args.api_server_count == 1
142 +
143 +
144 +def test_aggregates_health():
145 + supervisor = DPSupervisor(_make_unit_args())
146 + supervisor._is_ready = True
147 + assert supervisor.is_ready is True
148 +
149 +
150 +def test_handles_shutdown_event():
151 + supervisor = DPSupervisor(_make_unit_args())
152 + supervisor._is_ready = True
153 + supervisor._shutdown_event.set()
154 + assert supervisor.is_ready is False
155 +
156 +
157 +@pytest.mark.asyncio
158 +async def test_handles_child_exit(
159 + monkeypatch: pytest.MonkeyPatch,
160 +):
161 + supervisor = DPSupervisor(_make_unit_args())
162 + supervisor._processes = [
163 + SimpleNamespace(
164 + name="APIServer_DPRank_4", exitcode=None, is_alive=lambda: True
165 + ),
166 + SimpleNamespace(name="APIServer_DPRank_5", exitcode=17, is_alive=lambda: False),
167 + ]
168 +
169 + async def fake_probe(*_args, **_kwargs) -> bool:
170 + return True
171 +
172 + monkeypatch.setattr(dp_sup, "_probe_endpoint", fake_probe)
173 +
174 + await supervisor._monitor_children()
175 + assert supervisor._is_ready is False
176 +
177 +
178 +@pytest.mark.asyncio
179 +async def test_handles_probe_failure(
180 + monkeypatch: pytest.MonkeyPatch,
181 +):
182 + supervisor = DPSupervisor(_make_unit_args(dp_supervisor_probe_interval_s=0.0))
183 + supervisor.child_ports = [8000]
184 + probe_results = iter([True, False])
185 +
186 + async def fake_probe(*_args, **_kwargs) -> bool:
187 + return next(probe_results)
188 +
189 + monkeypatch.setattr(dp_sup, "_probe_endpoint", fake_probe)
190 +
191 + await supervisor._monitor_children()
192 + assert supervisor._is_ready is False
193 +
194 +
195 +@pytest.mark.asyncio
196 +async def test_shutdown_if_supervisor_server_error_on_startup(
197 + monkeypatch: pytest.MonkeyPatch,
198 +):
199 + class FakeLoop:
200 + def add_signal_handler(self, *_args, **_kwargs):
201 + pass
202 +
203 + def remove_signal_handler(self, *_args, **_kwargs):
204 + pass
205 +
206 + class FakeServer:
207 + def __init__(self, _config):
208 + self.started = False
209 + self.should_exit = False
210 +
211 + async def serve(self):
212 + raise ValueError("supervisor boom")
213 +
214 + async def fake_shutdown_children(self):
215 + return None
216 +
217 + monkeypatch.setattr(dp_sup.asyncio, "get_running_loop", lambda: FakeLoop())
218 + monkeypatch.setattr(dp_sup.uvicorn, "Server", FakeServer)
219 + monkeypatch.setattr(DPSupervisor, "_shutdown_children", fake_shutdown_children)
220 +
221 + supervisor = DPSupervisor(_make_unit_args())
222 +
223 + with pytest.raises(ValueError, match="supervisor boom"):
224 + await supervisor.run()
225 +
226 +
227 +# ---------------------------------------------------------------------------
228 +# Lifecycle integration tests – MockVLLMServer
229 +# ---------------------------------------------------------------------------
230 +
231 +
232 +class MockVLLMServer:
233 + """
234 + Minimal FastAPI server that mimics one vLLM replica.
235 + GET /health returns 200 when healthy, 503 otherwise.
236 + Health state is toggled by the test via set_healthy().
237 + """
238 +
239 + def __init__(self, port: int, drain_seconds: float = 0.0) -> None:
240 + self.port = port
241 + self._healthy = False
242 + self._drain_seconds = drain_seconds
243 + self._server: uvicorn.Server | None = None
244 + self._serve_task: asyncio.Task | None = None
245 +
246 + async def start(self) -> None:
247 + app = FastAPI()
248 +
249 + @app.get("/health")
250 + async def health() -> Response:
251 + print(f"MockServer {self.port}: /health: {self._healthy}")
252 + return Response(status_code=200 if self._healthy else 503)
253 +
254 + @app.get("/set_healthy")
255 + async def set_healthy() -> Response:
256 + print(f"MockServer {self.port}: /set_healthy")
257 + self._healthy = True
258 + return Response(status_code=200)
259 +
260 + @app.get("/set_unhealthy")
261 + async def set_unhealthy() -> Response:
262 + print(f"MockServer {self.port}: /set_unhealthy")
263 + self._healthy = False
264 + return Response(status_code=200)
265 +
266 + @app.get("/kill")
267 + async def kill() -> Response:
268 + print(f"MockServer {self.port}: /kill")
269 + os.kill(os.getpid(), signal.SIGKILL)
270 +
271 + config = uvicorn.Config(
272 + app,
273 + host="127.0.0.1",
274 + port=self.port,
275 + log_level="warning",
276 + lifespan="off",
277 + )
278 + self._server = uvicorn.Server(config)
279 +
280 + # Configure request draining if needed.
281 + # Uvicorn's capture_signals() installs signal.signal(SIGTERM, self.handle_exit),
282 + # which sets should_exit=True immediately. Override handle_exit on the instance
283 + # so capture_signals() picks up our version that drains first.
284 + if self._drain_seconds > 0:
285 + self._shutdown_event = asyncio.Event()
286 + loop = asyncio.get_running_loop()
287 +
288 + async def _drain_and_stop() -> None:
289 + await self._shutdown_event.wait()
290 + print(f"MockServer {self.port}: draining for {self._drain_seconds}s.")
291 + await asyncio.sleep(self._drain_seconds)
292 + print("Setting should_exit")
293 + if self._server is not None:
294 + self._server.should_exit = True
295 +
296 + self._drain_task = asyncio.create_task(_drain_and_stop())
297 +
298 + def _custom_handle_exit(sig: int, frame: object) -> None:
299 + print("Got SIGTERM, setting shutdown.")
300 + if not self._shutdown_event.is_set():
301 + loop.call_soon_threadsafe(self._shutdown_event.set)
302 +
303 + self._server.handle_exit = _custom_handle_exit
304 +
305 + self._serve_task = asyncio.create_task(self._server.serve())
306 + while not self._server.started:
307 + await asyncio.sleep(0.01)
308 + print(f"Mock DP Server on port {self.port} started")
309 +
310 + await self._serve_task
311 +
312 +
313 +def launch_mock_vllm(child_args: argparse.Namespace, env_updates: dict[str, str]):
314 + logger.info("Launching mock vLLM on port %s", child_args.port)
315 + mock_vllm = MockVLLMServer(port=child_args.port)
316 + asyncio.run(mock_vllm.start())
317 +
318 +
319 +def launch_mock_vllm_with_drain(
320 + child_args: argparse.Namespace, env_updates: dict[str, str]
321 +):
322 + logger.info("Launching mock vLLM with 15s drain on port %s", child_args.port)
323 + mock_vllm = MockVLLMServer(port=child_args.port, drain_seconds=10.0)
324 + asyncio.run(mock_vllm.start())
325 +
326 +
327 +# ---------------------------------------------------------------------------
328 +# Lifecycle test helpers
329 +# ---------------------------------------------------------------------------
330 +
331 +
332 +async def _poll_supervisor_health(expected_status: int) -> bool:
333 + """
334 + Poll GET /health on the supervisor until expected_status is seen.
335 + A connection error is treated as 503-equivalent when expected_status != 200.
336 + """
337 + url = f"http://127.0.0.1:{_SUPERVISOR_PORT}/health"
338 + async with aiohttp.ClientSession() as session:
339 + try:
340 + async with session.get(url) as resp:
341 + if resp.status != expected_status:
342 + print(f"expected: {expected_status=}, got: {resp.status=}")
343 + return False
344 + return True
345 + except aiohttp.ClientError:
346 + if expected_status != -1:
347 + print(f"expected: {expected_status=}, got: aiohttp.ClientError")
348 + return False
349 + return True
350 +
351 +
352 +async def _poll_until_api_server_running(port: int, retries: int = 10) -> None:
353 + url = f"http://127.0.0.1:{port}/health"
354 + async with aiohttp.ClientSession() as session:
355 + for _ in range(retries):
356 + try:
357 + async with session.get(url) as resp:
358 + if resp.status != 200:
359 + return
360 + await asyncio.sleep(1.0)
361 + except aiohttp.ClientError:
362 + print("Test detected not started yet, sleeping for 1s")
363 + await asyncio.sleep(1.0)
364 +
365 +
366 +async def _set_healthy(port: int) -> None:
367 + url = f"http://127.0.0.1:{port}/set_healthy"
368 + async with aiohttp.ClientSession() as session, session.get(url) as resp:
369 + assert resp.status == 200
370 +
371 +
372 +async def _set_unhealthy(port: int) -> None:
373 + url = f"http://127.0.0.1:{port}/set_unhealthy"
374 + async with aiohttp.ClientSession() as session, session.get(url) as resp:
375 + assert resp.status == 200
376 +
377 +
378 +async def _kill_server(port: int) -> None:
379 + url = f"http://127.0.0.1:{port}/kill"
380 + try:
381 + async with aiohttp.ClientSession() as session, session.get(url) as resp:
382 + assert resp.status != 200
383 + except Exception as e:
384 + assert isinstance(e, aiohttp.ClientConnectorError)
385 +
386 +
387 +@contextlib.asynccontextmanager
388 +async def _run_supervisor(
389 + args: argparse.Namespace,
390 + monkeypatch: pytest.MonkeyPatch,
391 + launch_fn=None,
392 +):
393 + if launch_fn is None:
394 + launch_fn = launch_mock_vllm
395 + monkeypatch.setattr(dp_sup, "_run_vllm_dp_server", launch_fn)
396 + supervisor = DPSupervisor(args)
397 + task = asyncio.create_task(supervisor.run())
398 + await asyncio.sleep(1.0)
399 + try:
400 + yield supervisor, task
401 + finally:
402 + task.cancel()
403 + with contextlib.suppress(asyncio.CancelledError):
404 + await task
405 +
406 +
407 +# ---------------------------------------------------------------------------
408 +# Lifecycle integration tests
409 +# ---------------------------------------------------------------------------
410 +
411 +
412 +@pytest.mark.asyncio
413 +async def test_basic_lifecycle(monkeypatch):
414 + """
415 + A) Supervisor /health returns 503 while children are unhealthy.
416 + B) /health returns 200 once every child reports healthy.
417 + C) SIGTERM and shutdown
418 + """
419 + args = _make_args()
420 +
421 + vllm_server_ports = [_CHILD_PORT_BASE + i for i in range(_N_CHILDREN)]
422 +
423 + async with _run_supervisor(args, monkeypatch) as (supervisor, _task):
424 + assert await _poll_supervisor_health(503)
425 + assert not supervisor.is_ready
426 +
427 + for port in vllm_server_ports:
428 + assert await _poll_supervisor_health(503)
429 + assert not supervisor.is_ready
430 + await _poll_until_api_server_running(port)
431 +
432 + await _set_healthy(vllm_server_ports[0])
433 + await asyncio.sleep(1.0)
434 + assert await _poll_supervisor_health(503)
435 + assert not supervisor.is_ready
436 + print("/health is 503 --- expected!")
437 +
438 + for port in vllm_server_ports:
439 + await _set_healthy(port)
440 + await asyncio.sleep(1.0)
441 + assert await _poll_supervisor_health(200)
442 + assert supervisor.is_ready
443 + print("/health is 200 --- expected!")
444 +
445 + await asyncio.sleep(1.0)
446 + assert await _poll_supervisor_health(200)
447 + assert supervisor.is_ready
448 + print("/health is 200 --- expected!")
449 +
450 + os.kill(os.getpid(), signal.SIGTERM)
451 +
452 + await asyncio.wait_for(_task, timeout=5.0)
453 + for p in supervisor._processes:
454 + assert not p.is_alive()
455 + print("everything was cleaned up!")
456 +
457 +
458 +@pytest.mark.asyncio
459 +async def test_failed_startup(monkeypatch):
460 + """
461 + A) One of the vLLM servers crashes during startup.
462 + B) DPSupervisor detects this, and cleans up resources.
463 + """
464 + args = _make_args()
465 +
466 + vllm_server_ports = [_CHILD_PORT_BASE + i for i in range(_N_CHILDREN)]
467 +
468 + async with _run_supervisor(args, monkeypatch) as (supervisor, _task):
469 + assert await _poll_supervisor_health(503)
470 + assert not supervisor.is_ready
471 +
472 + for port in vllm_server_ports:
473 + await _poll_until_api_server_running(port)
474 +
475 + await _kill_server(port)
476 +
477 + await asyncio.wait_for(_task, timeout=5.0)
478 + for p in supervisor._processes:
479 + assert not p.is_alive()
480 +
481 +
482 +@pytest.mark.asyncio
483 +async def test_becomes_unhealthy(monkeypatch):
484 + """
485 + A) Supervisor /health returns 503 while children are unhealthy.
486 + B) /health returns 200 once every child reports healthy.
487 + C) Child process becomes unhealthy.
488 + D) Detected and shutdown.
489 + """
490 + args = _make_args()
491 +
492 + vllm_server_ports = [_CHILD_PORT_BASE + i for i in range(_N_CHILDREN)]
493 +
494 + async with _run_supervisor(args, monkeypatch) as (supervisor, _task):
495 + assert await _poll_supervisor_health(503)
496 + assert not supervisor.is_ready
497 +
498 + for port in vllm_server_ports:
499 + assert await _poll_supervisor_health(503)
500 + assert not supervisor.is_ready
501 + await _poll_until_api_server_running(port)
502 +
503 + await _set_healthy(vllm_server_ports[0])
504 + await asyncio.sleep(1.0)
505 + assert await _poll_supervisor_health(503)
506 + assert not supervisor.is_ready
507 + print("/health is 503 --- expected!")
508 +
509 + for port in vllm_server_ports:
510 + await _set_healthy(port)
511 + await asyncio.sleep(1.0)
512 + assert await _poll_supervisor_health(200)
513 + assert supervisor.is_ready
514 + print("/health is 200 --- expected!")
515 +
516 + await _set_unhealthy(port)
517 +
518 + await asyncio.wait_for(_task, timeout=5.0)
519 + for p in supervisor._processes:
520 + assert not p.is_alive()
521 + print("everything was cleaned up!")
522 +
523 +
524 +@pytest.mark.asyncio
525 +async def test_dp_server_fails(monkeypatch):
526 + """
527 + A) Supervisor /health returns 503 while children are unhealthy.
528 + B) /health returns 200 once every child reports healthy.
529 + C) Child process fails.
530 + D) Detected and shutdown.
531 + """
532 + args = _make_args()
533 +
534 + vllm_server_ports = [_CHILD_PORT_BASE + i for i in range(_N_CHILDREN)]
535 +
536 + async with _run_supervisor(args, monkeypatch) as (supervisor, _task):
537 + assert await _poll_supervisor_health(503)
538 + assert not supervisor.is_ready
539 +
540 + for port in vllm_server_ports:
541 + assert await _poll_supervisor_health(503)
542 + assert not supervisor.is_ready
543 + await _poll_until_api_server_running(port)
544 +
545 + await _set_healthy(vllm_server_ports[0])
546 + await asyncio.sleep(1.0)
547 + assert await _poll_supervisor_health(503)
548 + assert not supervisor.is_ready
549 + print("/health is 503 --- expected!")
550 +
551 + for port in vllm_server_ports:
552 + await _set_healthy(port)
553 + await asyncio.sleep(1.0)
554 + assert await _poll_supervisor_health(200)
555 + assert supervisor.is_ready
556 + print("/health is 200 --- expected!")
557 +
558 + dp_mock_server_process = supervisor._processes[0]
559 + os.kill(dp_mock_server_process.pid, signal.SIGKILL)
560 + await asyncio.sleep(1.0)
561 + assert not dp_mock_server_process.is_alive()
562 +
563 + await asyncio.wait_for(_task, timeout=5.0)
564 + for p in supervisor._processes:
565 + assert not p.is_alive()
566 + print("everything was cleaned up!")
567 +
568 +
569 +@pytest.mark.asyncio
570 +async def test_shutdown_timeout(monkeypatch: pytest.MonkeyPatch):
571 + """
572 + Child mock servers delay shutdown by 10s on SIGTERM (simulating in-flight
573 + request drain). The supervisor is configured with shutdown_timeout=10,
574 + so its total wait budget is 10 + CHILD_EXIT_GRACE_S seconds. The
575 + children exit naturally within that window, so no force-kill should occur
576 + and the measured wall-clock time must be >= 10s.
577 + """
578 + _DRAIN_SECONDS = 10.0
579 + _SHUTDOWN_TIMEOUT = 10.0
580 +
581 + args = _make_args(shutdown_timeout=_SHUTDOWN_TIMEOUT)
582 + vllm_server_ports = [_CHILD_PORT_BASE + i for i in range(_N_CHILDREN)]
583 +
584 + async with _run_supervisor(
585 + args, monkeypatch, launch_fn=launch_mock_vllm_with_drain
586 + ) as (supervisor, _task):
587 + for port in vllm_server_ports:
588 + await _poll_until_api_server_running(port)
589 +
590 + for port in vllm_server_ports:
591 + await _set_healthy(port)
592 + await asyncio.sleep(1.0)
593 + assert await _poll_supervisor_health(200)
594 + assert supervisor.is_ready
595 +
596 + start_t = time.perf_counter()
597 + os.kill(os.getpid(), signal.SIGTERM)
598 +
599 + print(f"DRAINING FOR {_DRAIN_SECONDS}")
600 + await asyncio.wait_for(_task, timeout=_DRAIN_SECONDS + CHILD_EXIT_GRACE_S + 5.0)
601 + elapsed = time.perf_counter() - start_t
602 +
603 + assert elapsed >= _DRAIN_SECONDS, (
604 + f"Supervisor exited after only {elapsed:.1f}s; "
605 + f"expected >= {_DRAIN_SECONDS}s for request draining"
606 + )
607 +
608 + for p in supervisor._processes:
609 + assert not p.is_alive()
610 + print(f"Supervisor waited {elapsed:.1f}s for children to drain — expected!")
vllm/engine/arg_utils.py
+10
@@ -476,6 +476,7 @@ class EngineArgs:
476 data_parallel_rpc_port: int | None = None
477 data_parallel_hybrid_lb: bool = False
478 data_parallel_external_lb: bool = False
479 + data_parallel_multi_port_external_lb: bool = False
480 data_parallel_backend: DataParallelBackend = ParallelConfig.data_parallel_backend
481 enable_expert_parallel: bool = ParallelConfig.enable_expert_parallel
482 enable_ep_weight_filter: bool = ParallelConfig.enable_ep_weight_filter
@@ -1043,6 +1044,15 @@ class EngineArgs:
1044 "-dpe",
1045 **parallel_kwargs["data_parallel_external_lb"],
1046 )
1047 + parallel_group.add_argument(
1048 + "--data-parallel-multi-port-external-lb",
1049 + "-dpm",
1050 + action="store_true",
1051 + default=False,
1052 + help="Run a node-local supervisor that launches one external-LB API "
1053 + "server per local data parallel rank and exposes aggregated health on "
1054 + "a supervisor port.",
1055 + )
1056 parallel_group.add_argument(
1057 "--enable-expert-parallel",
1058 "-ep",
vllm/entrypoints/cli/serve.py
+30 -14
@@ -12,6 +12,9 @@ import vllm.envs as envs
12 from vllm.entrypoints.cli.types import CLISubcommand
13 from vllm.entrypoints.openai.api_server import run_server, setup_server
14 from vllm.entrypoints.openai.cli_args import make_arg_parser, validate_parsed_serve_args
15 +from vllm.entrypoints.openai.dp_supervisor import (
16 + run_dp_supervisor,
17 +)
18 from vllm.entrypoints.utils import VLLM_SUBCMD_PARSER_EPILOG
19 from vllm.logger import init_logger
20 from vllm.usage.usage_lib import UsageContext
@@ -66,31 +69,41 @@ class ServeSubcommand(CLISubcommand):
69 args.api_server_count = 0
70
71 # Detect LB mode for defaulting api_server_count.
72 + # Multi-port: --data-parallel-multi-port-external-lb
73 # External LB: --data-parallel-external-lb or --data-parallel-rank
74 # Hybrid LB: --data-parallel-hybrid-lb or --data-parallel-start-rank
75 is_external_lb = (
76 args.data_parallel_external_lb or args.data_parallel_rank is not None
77 )
74 - is_hybrid_lb = (
75 - args.data_parallel_hybrid_lb or args.data_parallel_start_rank is not None
76 - )
78
78 - if is_external_lb and is_hybrid_lb:
79 + # If --data_parallel_multi_port_external_lb and --data_parallel_hybrid_lb
80 + # are unset, default to hybrid if --data-parallel-start-rank is set
81 + is_hybrid_lb = is_multi_port = False
82 + if (
83 + not args.data_parallel_hybrid_lb
84 + and not args.data_parallel_multi_port_external_lb
85 + ):
86 + is_hybrid_lb = args.data_parallel_start_rank is not None
87 + else:
88 + is_hybrid_lb = args.data_parallel_hybrid_lb
89 + is_multi_port = args.data_parallel_multi_port_external_lb
90 +
91 + if sum([is_multi_port, is_external_lb, is_hybrid_lb]) > 1:
92 raise ValueError(
80 - "Cannot use both external and hybrid data parallel load "
81 - "balancing modes. External LB is enabled via "
82 - "--data-parallel-external-lb or --data-parallel-rank. "
83 - "Hybrid LB is enabled via --data-parallel-hybrid-lb or "
84 - "--data-parallel-start-rank. Use one mode or the other."
93 + "Cannot use more than one data parallel load balancing mode. "
94 + "Choose one of: --data-parallel-multi-port-external-lb, "
95 + "--data-parallel-external-lb (or --data-parallel-rank), "
96 + "--data-parallel-hybrid-lb (or --data-parallel-start-rank)."
97 )
98
99 # Default api_server_count if not explicitly set.
88 - # - Rust frontend: Use 1 (not applicable as it's multithreaded)
89 - # - External LB: Leave as 1 (external LB handles distribution)
100 + # - Multi-port: 1 (supervisor spawns one server per local DP rank)
101 + # - Rust frontend: 1 (not applicable as it's multithreaded)
102 + # - External LB: 1 (external LB handles distribution)
103 # - Hybrid LB: Use local DP size (internal LB for local ranks only)
104 # - Internal LB: Use full DP size
105 if args.api_server_count is None:
93 - if is_external_lb or envs.VLLM_RUST_FRONTEND_PATH:
106 + if is_multi_port or is_external_lb or envs.VLLM_RUST_FRONTEND_PATH:
107 args.api_server_count = 1
108 elif is_hybrid_lb:
109 args.api_server_count = args.data_parallel_size_local or 1
@@ -109,7 +122,8 @@ class ServeSubcommand(CLISubcommand):
122 )
123 elif envs.VLLM_RUST_FRONTEND_PATH and args.api_server_count > 1:
124 logger.warning(
112 - "Ignoring --api-server-count=%d when using rust front-end process"
125 + "Ignoring --api-server-count=%d when using rust front-end process",
126 + args.api_server_count,
127 )
128 args.api_server_count = 1
129
@@ -122,7 +136,9 @@ class ServeSubcommand(CLISubcommand):
136 )
137 args.api_server_count = 1
138
125 - if args.api_server_count < 1:
139 + if is_multi_port:
140 + run_dp_supervisor(args)
141 + elif args.api_server_count < 1:
142 run_headless(args)
143 elif args.api_server_count > 1 or envs.VLLM_RUST_FRONTEND_PATH:
144 run_multi_api_server(args)
vllm/entrypoints/openai/api_server.py
+1 -2
@@ -665,8 +665,7 @@ async def build_and_serve_renderer(
665 async def run_server(args, **uvicorn_kwargs) -> None:
666 """Run a single-worker API server."""
667
668 - # Add process-specific prefix to stdout and stderr.
669 - decorate_logs("APIServer")
668 + decorate_logs("APIServer", skip_if_decorated=True)
669
670 # Interrupt initialization if SIGTERM arrives before uvicorn installs its
671 # own signal handlers. Once uvicorn is running it replaces this.
vllm/entrypoints/openai/cli_args.py
+18
@@ -229,6 +229,17 @@ class FrontendArgs(BaseFrontendArgs):
229 """Host name."""
230 port: int = 8000
231 """Port number."""
232 + data_parallel_supervisor_port: int = 9256
233 + """HTTP port for aggregated health endpoints in multi-port external LB
234 + mode."""
235 + dp_supervisor_probe_interval_s: float = 5.0
236 + """Seconds between aggregated health probes in multi-port external LB mode."""
237 + dp_supervisor_probe_timeout_s: float = 5.0
238 + """Seconds to wait between retries when a child health probe fails with a
239 + connection error in multi-port external LB mode."""
240 + dp_supervisor_probe_failure_threshold: int = 3
241 + """Number of consecutive connection-error retries before a child health
242 + probe is declared failed in multi-port external LB mode."""
243 uds: str | None = None
244 """Unix domain socket path. If set, host and port arguments are ignored."""
245 uvicorn_log_level: Literal[
@@ -387,6 +398,13 @@ def validate_parsed_serve_args(args: argparse.Namespace):
398 if args.enable_log_outputs and not args.enable_log_requests:
399 raise TypeError("Error: --enable-log-outputs requires --enable-log-requests")
400
401 + if args.data_parallel_multi_port_external_lb:
402 + from vllm.entrypoints.openai.dp_supervisor import (
403 + validate_multi_port_external_lb_args,
404 + )
405 +
406 + validate_multi_port_external_lb_args(args)
407 +
408
409 def create_parser_for_docs() -> FlexibleArgumentParser:
410 parser_for_docs = FlexibleArgumentParser(
vllm/entrypoints/openai/dp_supervisor.py new
+493
@@ -0,0 +1,493 @@
1 +#!/usr/bin/env python3
2 +# SPDX-License-Identifier: Apache-2.0
3 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
4 +
5 +from __future__ import annotations
6 +
7 +import argparse
8 +import asyncio
9 +import contextlib
10 +import copy
11 +import multiprocessing
12 +import os
13 +import signal
14 +import time
15 +from functools import partial
16 +from http import HTTPStatus
17 +from multiprocessing.process import BaseProcess
18 +
19 +import aiohttp
20 +import psutil
21 +import uvicorn
22 +import uvloop
23 +from fastapi import FastAPI, Response
24 +
25 +from vllm.logger import init_logger
26 +from vllm.platforms import current_platform
27 +from vllm.utils.system_utils import (
28 + decorate_logs,
29 + kill_process_tree,
30 + set_process_title,
31 + update_environment_variables,
32 +)
33 +
34 +logger = init_logger(__name__)
35 +
36 +CHILD_EXIT_GRACE_S = 5.0
37 +
38 +
39 +def infer_multi_port_external_lb_start_rank(args: argparse.Namespace) -> int:
40 + start_rank = getattr(args, "data_parallel_start_rank", None)
41 + if start_rank is not None:
42 + return start_rank
43 +
44 + node_rank = getattr(args, "node_rank", 0) or 0
45 + local_size = getattr(args, "data_parallel_size_local", 0) or 0
46 + return node_rank * local_size
47 +
48 +
49 +def validate_multi_port_external_lb_args(args: argparse.Namespace) -> None:
50 + if getattr(args, "grpc", False):
51 + raise ValueError(
52 + "Error: --data-parallel-multi-port-external-lb does not support --grpc"
53 + )
54 + if args.uds is not None:
55 + raise ValueError(
56 + "Error: --data-parallel-multi-port-external-lb does not support --uds"
57 + )
58 + if any((args.ssl_keyfile, args.ssl_certfile, args.ssl_ca_certs)):
59 + raise ValueError(
60 + "Error: --data-parallel-multi-port-external-lb does not support HTTPS yet"
61 + )
62 + if args.api_server_count not in (None, 1):
63 + raise ValueError(
64 + "Error: --data-parallel-multi-port-external-lb currently requires "
65 + "--api-server-count=1"
66 + )
67 + if args.data_parallel_rank is not None:
68 + raise ValueError(
69 + "Error: --data-parallel-multi-port-external-lb manages child "
70 + "--data-parallel-rank values internally"
71 + )
72 + if args.data_parallel_external_lb or args.data_parallel_hybrid_lb:
73 + raise ValueError(
74 + "Error: --data-parallel-multi-port-external-lb cannot be combined with "
75 + "--data-parallel-external-lb or --data-parallel-hybrid-lb"
76 + )
77 + if args.data_parallel_size < 2:
78 + raise ValueError(
79 + "Error: --data-parallel-multi-port-external-lb requires "
80 + "--data-parallel-size > 1"
81 + )
82 +
83 + local_size = args.data_parallel_size_local
84 + if local_size is None or local_size < 2:
85 + raise ValueError(
86 + "Error: --data-parallel-multi-port-external-lb requires "
87 + "--data-parallel-size-local >= 2"
88 + )
89 + if local_size > args.data_parallel_size:
90 + raise ValueError(
91 + "Error: --data-parallel-size-local cannot exceed --data-parallel-size"
92 + )
93 + if args.data_parallel_size % local_size != 0:
94 + raise ValueError(
95 + "Error: --data-parallel-size must be divisible by "
96 + "--data-parallel-size-local"
97 + )
98 +
99 + start_rank = infer_multi_port_external_lb_start_rank(args)
100 + if start_rank + local_size > args.data_parallel_size:
101 + raise ValueError(
102 + "Error: multi-port supervised ranks would exceed --data-parallel-size"
103 + )
104 +
105 + supervisor_port = args.data_parallel_supervisor_port
106 + child_port_min = args.port
107 + child_port_max = args.port + local_size - 1
108 + if child_port_min <= supervisor_port <= child_port_max:
109 + raise ValueError(
110 + f"Error: --data-parallel-supervisor-port {supervisor_port} "
111 + f"overlaps with child rank ports {child_port_min}-{child_port_max}"
112 + )
113 +
114 +
115 +def _build_vllm_dp_server_args(
116 + args: argparse.Namespace, local_rank: int
117 +) -> argparse.Namespace:
118 + child_args = copy.copy(args)
119 + child_args.port = args.port + local_rank
120 + child_args.data_parallel_rank = (
121 + infer_multi_port_external_lb_start_rank(args) + local_rank
122 + )
123 + child_args.data_parallel_start_rank = None
124 + child_args.data_parallel_size_local = 1
125 + child_args.data_parallel_external_lb = True
126 + child_args.data_parallel_hybrid_lb = False
127 + child_args.data_parallel_multi_port_external_lb = False
128 + child_args.data_parallel_supervisor_port = None
129 + child_args.api_server_count = 1
130 + return child_args
131 +
132 +
133 +def _build_vllm_dp_server_env(
134 + args: argparse.Namespace, local_rank: int
135 +) -> dict[str, str]:
136 + # set visible devices for the child process
137 + devices_per_rank = args.tensor_parallel_size * args.pipeline_parallel_size
138 + start = local_rank * devices_per_rank
139 + stop = start + devices_per_rank
140 + device_env = current_platform.device_control_env_var
141 + visible_devices = ",".join(
142 + str(current_platform.device_id_to_physical_device_id(idx))
143 + for idx in range(start, stop)
144 + )
145 + return {device_env: visible_devices}
146 +
147 +
148 +def _child_base_url(args: argparse.Namespace, port: int) -> str:
149 + host = args.host or "127.0.0.1"
150 + if host == "0.0.0.0":
151 + host = "127.0.0.1"
152 + elif host == "::":
153 + host = "::1"
154 + return f"http://{host}:{port}"
155 +
156 +
157 +def _join_processes_with_timeout(processes: list[BaseProcess], timeout: float) -> None:
158 + deadline = time.monotonic() + timeout
159 + for process in processes:
160 + remaining = deadline - time.monotonic()
161 + if remaining <= 0:
162 + break
163 + process.join(timeout=remaining)
164 +
165 +
166 +async def _probe_endpoint(
167 + session: aiohttp.ClientSession,
168 + args: argparse.Namespace,
169 + port: int,
170 + path: str,
171 + conn_err_failure_threshold: int = 3,
172 + conn_err_retry_delay: float = 5.0,
173 +) -> bool:
174 + """
175 + Probe /health endpoint for 200 status.
176 +
177 + If there is a connection error, retry every N seconds.
178 + """
179 + for iteration in range(conn_err_failure_threshold):
180 + try:
181 + async with session.get(_child_base_url(args, port) + path) as response:
182 + # vLLM returns 503 on EngineDeadError, so we should return
183 + # immediately if vLLM responds with a non-200 status code.
184 + return response.status == HTTPStatus.OK
185 + except (aiohttp.ClientError, asyncio.TimeoutError) as e:
186 + # Allow retry of connection errors.
187 + logger.debug(
188 + "Probe attempt %d/%d failed on port %d: %r",
189 + iteration + 1,
190 + conn_err_failure_threshold,
191 + port,
192 + e,
193 + )
194 +
195 + if iteration < conn_err_failure_threshold - 1:
196 + await asyncio.sleep(conn_err_retry_delay)
197 +
198 + return False
199 +
200 +
201 +def _build_dp_supervisor_app(supervisor: DPSupervisor) -> FastAPI:
202 + app = FastAPI(openapi_url=None, docs_url=None, redoc_url=None)
203 + app.state.supervisor = supervisor
204 +
205 + def _status_response(ok: bool) -> Response:
206 + return Response(
207 + status_code=(HTTPStatus.OK if ok else HTTPStatus.SERVICE_UNAVAILABLE)
208 + )
209 +
210 + @app.get("/health", include_in_schema=False)
211 + async def health() -> Response:
212 + return _status_response(app.state.supervisor.is_ready)
213 +
214 + @app.get("/ready", include_in_schema=False)
215 + @app.get("/readyz", include_in_schema=False)
216 + async def ready() -> Response:
217 + return _status_response(app.state.supervisor.is_ready)
218 +
219 + return app
220 +
221 +
222 +def _run_vllm_dp_server(
223 + child_args: argparse.Namespace, env_updates: dict[str, str]
224 +) -> None:
225 + """
226 + Entrypoint function for the vLLM DP Server.
227 + """
228 + from vllm.entrypoints.openai.api_server import run_server
229 +
230 + # Create a fresh process group for the vLLM DP Server,
231 + # so that CTRL-C is propagated cleanly.
232 + os.setpgrp()
233 +
234 + name = f"APIServer_DP{child_args.data_parallel_rank}"
235 + update_environment_variables(env_updates)
236 + set_process_title(name)
237 + decorate_logs(name)
238 + uvloop.run(run_server(child_args))
239 +
240 +
241 +class DPSupervisor:
242 + def __init__(self, args: argparse.Namespace):
243 + validate_multi_port_external_lb_args(args)
244 + self.args = args
245 + self.supervisor_port = args.data_parallel_supervisor_port
246 + self.child_ports = [
247 + args.port + local_rank
248 + for local_rank in range(args.data_parallel_size_local)
249 + ]
250 + self._is_ready = False
251 + self._processes: list[BaseProcess] = []
252 + self._shutdown_event = asyncio.Event()
253 + self._shutdown_signal = signal.SIGTERM
254 +
255 + @property
256 + def is_ready(self) -> bool:
257 + return self._is_ready and not self._shutdown_event.is_set()
258 +
259 + async def run(self) -> None:
260 + loop = asyncio.get_running_loop()
261 +
262 + # K8s sends SIGTERM for shutdown - begin graceful termination.
263 + for sig in (signal.SIGTERM, signal.SIGINT):
264 + loop.add_signal_handler(sig, partial(self._handle_signal, sig))
265 +
266 + # Launch DPSupervisor Server.
267 + app = _build_dp_supervisor_app(self)
268 + decorate_logs("DPSupervisor")
269 + host = self.args.host or "0.0.0.0"
270 + config = uvicorn.Config(
271 + app,
272 + host=host,
273 + port=self.supervisor_port,
274 + log_level=self.args.uvicorn_log_level,
275 + )
276 + supervisor_server = uvicorn.Server(config)
277 + supervisor_server_task = asyncio.create_task(
278 + supervisor_server.serve(),
279 + name="dp-supervisor",
280 + )
281 + supervisor_server_task.add_done_callback(
282 + lambda _task: self._shutdown_event.set()
283 + )
284 +
285 + # Ensure DPSupervisor task starts on the event loop.
286 + while not supervisor_server.started:
287 + if supervisor_server_task.done():
288 + supervisor_server_task.result()
289 + raise RuntimeError("DPSupervisor exited before startup.")
290 + await asyncio.sleep(0.05)
291 + logger.info("Started DPSupervisor on %s:%d", host, self.supervisor_port)
292 +
293 + # Launch and Monitor vLLM Server Processes.
294 + try:
295 + self._start_children()
296 + await self._monitor_children()
297 + finally:
298 + self._is_ready = False
299 + await self._shutdown_children()
300 +
301 + # Shutdown the DP Supervisor server.
302 + supervisor_server.should_exit = True
303 + await supervisor_server_task
304 +
305 + def _handle_signal(self, signum: int) -> None:
306 + """
307 + Signal handler that is added to the event loop.
308 +
309 + This catches the SIGTERM from K8s and begins graceful shutdown,
310 + by setting the _shutdown_event(), which is watched by the main
311 + coroutine monitoring the vLLM DP Servers.
312 + """
313 +
314 + if self._shutdown_event.is_set():
315 + return
316 +
317 + self._shutdown_signal = signal.Signals(signum)
318 + logger.info(
319 + "DPSupervisor received %s, shutting down.",
320 + self._shutdown_signal.name,
321 + )
322 +
323 + self._shutdown_event.set()
324 + self._is_ready = False
325 +
326 + def _start_children(self) -> None:
327 + """
328 + Launch vLLM DP Servers on separate GPUs.
329 + """
330 + logger.info("Launching vLLM DP Servers")
331 + context = multiprocessing.get_context("spawn")
332 + for local_rank in range(self.args.data_parallel_size_local):
333 + child_args = _build_vllm_dp_server_args(self.args, local_rank)
334 + child_env = _build_vllm_dp_server_env(self.args, local_rank)
335 + process = context.Process(
336 + target=_run_vllm_dp_server,
337 + name=f"APIServer_DPRank_{child_args.data_parallel_rank}",
338 + args=(child_args, child_env),
339 + )
340 + process.start()
341 + self._processes.append(process)
342 +
343 + async def _probe_all_children(self) -> None:
344 + """
345 + Background coroutine: probes all child endpoints on each interval.
346 +
347 + Exits when any server becomes unhealthy after being ready, signalling
348 + _monitor_children to initiate shutdown.
349 + """
350 + timeout = aiohttp.ClientTimeout(total=self.args.dp_supervisor_probe_timeout_s)
351 + async with aiohttp.ClientSession(timeout=timeout) as session:
352 + while not self._shutdown_event.is_set():
353 + threshold = (
354 + self.args.dp_supervisor_probe_failure_threshold
355 + if self._is_ready
356 + else 1
357 + )
358 + results = await asyncio.gather(
359 + *(
360 + _probe_endpoint(
361 + session,
362 + self.args,
363 + port,
364 + "/health",
365 + conn_err_failure_threshold=threshold,
366 + conn_err_retry_delay=self.args.dp_supervisor_probe_interval_s,
367 + )
368 + for port in self.child_ports
369 + ),
370 + return_exceptions=True,
371 + )
372 + all_healthy = all(r is True for r in results)
373 +
374 + if all_healthy:
375 + # If all healthy, we are ready to receive requests.
376 + # This conditional avoids a potential race condition
377 + # where shutdown is set, THEN the probe returns true.
378 + if not self._shutdown_event.is_set():
379 + self._is_ready = True
380 + elif self._is_ready:
381 + # Once ready, any failure in the probe means vLLM is dead.
382 + num_unhealthy = sum(1 for r in results if r is not True)
383 + logger.info(
384 + "DPSupervisor probe found %s unhealthy DP Servers.",
385 + num_unhealthy,
386 + )
387 + self._is_ready = False
388 + self._shutdown_event.set()
389 + return
390 +
391 + with contextlib.suppress(asyncio.TimeoutError):
392 + logger.debug(
393 + "Waiting for %s seconds before next probe",
394 + self.args.dp_supervisor_probe_interval_s,
395 + )
396 + await asyncio.wait_for(
397 + self._shutdown_event.wait(),
398 + timeout=self.args.dp_supervisor_probe_interval_s,
399 + )
400 +
401 + async def _monitor_children(self) -> None:
402 + """
403 + Main coroutine task that monitors the children vLLM servers.
404 +
405 + Before the vLLM servers are /ready:
406 + - if the pid is dead, we will shut down
407 + - if the probe fails, we try again after dp_supervisor_probe_interval_s
408 +
409 + After the vLLM servers are /ready:
410 + - if the pid is dead, we will shut down
411 + - if the probe fails, we will shut down
412 + """
413 + probe_task = asyncio.create_task(
414 + self._probe_all_children(), name="dp-health-probe"
415 + )
416 +
417 + try:
418 + while not self._shutdown_event.is_set():
419 + # 1. Check for dead processes
420 + n_failed = len([p for p in self._processes if not p.is_alive()])
421 + if n_failed > 0:
422 + logger.info("DPSupervisor found %s exited DP Servers.", n_failed)
423 + break
424 +
425 + # 2. Check if the probe background task crashed or failed.
426 + if probe_task.done():
427 + # Extract exception if it crashed, or log failure
428 + exc = probe_task.exception() if not probe_task.cancelled() else None
429 + logger.info("DPSupervisor probe task stopped. Exception: %s", exc)
430 + break
431 +
432 + # Sleep for probe_interval seconds or until a shutdown.
433 + with contextlib.suppress(asyncio.TimeoutError):
434 + logger.debug(
435 + "Waiting for %s seconds before next monitor",
436 + self.args.dp_supervisor_probe_interval_s,
437 + )
438 + await asyncio.wait_for(
439 + self._shutdown_event.wait(),
440 + timeout=self.args.dp_supervisor_probe_interval_s,
441 + )
442 +
443 + finally:
444 + # Cleanup probe task if needed.
445 + if not probe_task.done():
446 + probe_task.cancel()
447 + with contextlib.suppress(asyncio.CancelledError):
448 + await probe_task
449 +
450 + async def _shutdown_children(self) -> None:
451 + """Terminate the vLLM DP servers."""
452 + timeout = self.args.shutdown_timeout + CHILD_EXIT_GRACE_S
453 +
454 + try:
455 + logger.info(
456 + "DPSupervisor forwarding %s to DP Servers.",
457 + self._shutdown_signal.name,
458 + )
459 + for process in self._processes:
460 + pid = process.pid
461 + if not process.is_alive() or pid is None:
462 + continue
463 + with contextlib.suppress(ProcessLookupError, OSError):
464 + os.kill(pid, self._shutdown_signal)
465 +
466 + try:
467 + await asyncio.to_thread(
468 + _join_processes_with_timeout, self._processes, timeout
469 + )
470 + except asyncio.CancelledError:
471 + logger.warning("Shutdown await cancelled")
472 + raise
473 + finally:
474 + for process in self._processes:
475 + pid = process.pid
476 + if not process.is_alive() or pid is None:
477 + continue
478 + logger.warning(
479 + "DP server %s did not exit within %.1fs; force killing.",
480 + process.name,
481 + timeout,
482 + )
483 + with contextlib.suppress(
484 + ProcessLookupError,
485 + OSError,
486 + psutil.NoSuchProcess,
487 + psutil.AccessDenied,
488 + ):
489 + kill_process_tree(pid)
490 +
491 +
492 +def run_dp_supervisor(args: argparse.Namespace) -> None:
493 + uvloop.run(DPSupervisor(args).run())
vllm/utils/system_utils.py
+6 -1
@@ -234,12 +234,17 @@ def _add_prefix(file: TextIO, worker_name: str, pid: int) -> None:
234 file.write = write_with_prefix # type: ignore[method-assign]
235
236
237 -def decorate_logs(process_name: str | None = None) -> None:
237 +def decorate_logs(
238 + process_name: str | None = None, *, skip_if_decorated: bool = False
239 +) -> None:
240 """Decorate stdout/stderr with process name and PID prefix."""
241 # Respect VLLM_CONFIGURE_LOGGING environment variable
242 if not envs.VLLM_CONFIGURE_LOGGING:
243 return
244
245 + if skip_if_decorated and hasattr(sys.stdout, "_original_write"):
246 + return
247 +
248 if process_name is None:
249 process_name = get_mp_context().current_process().name
250