A high-throughput and memory-efficient inference and serving engine for LLMs
[Core][AMD] Propagate shutdown timeout to MultiprocExecutor (#43154)
Signed-off-by: Ryan Rock <ryan.rock@amd.com> Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
Ryan Rock committed
Jun 12, 2026 at 15:13 UTC
aab639c705dd5df1ca52f77e281ac23413a1993c
5 files changed
+71
-2
tests/v1/engine/test_core_engine_actor_manager.py
+13
@@ -8,6 +8,7 @@ import uuid
8
from pathlib import Path
9
from types import SimpleNamespace
10
from typing import Any
11
+from unittest.mock import Mock
12
13
import pytest
14
import ray
@@ -15,6 +16,7 @@ import zmq
16
17
from vllm.utils.network_utils import make_zmq_socket, split_zmq_path
18
from vllm.v1.engine.core import EngineCoreActorMixin
19
+from vllm.v1.engine.core_client import BackgroundResources
20
from vllm.v1.engine.utils import (
21
CoreEngineActorManager,
22
EngineZmqAddresses,
@@ -99,6 +101,17 @@ class _DummyExecutor:
101
pass
102
103
104
+def test_background_resources_passes_worker_shutdown_timeout(
105
+ monkeypatch: pytest.MonkeyPatch,
106
+) -> None:
107
+ timeout = 7
108
+ monkeypatch.setenv("VLLM_WORKER_SHUTDOWN_TIMEOUT_SECONDS", str(timeout))
109
+ engine_manager = Mock()
110
+ resources = BackgroundResources(ctx=None, engine_manager=engine_manager)
111
+ resources()
112
+ engine_manager.shutdown.assert_called_once_with(timeout=timeout)
113
+
114
+
115
def _make_vllm_config() -> SimpleNamespace:
116
return SimpleNamespace(
117
parallel_config=SimpleNamespace(
tests/v1/executor/test_executor.py
+45
@@ -14,6 +14,7 @@ from vllm.engine.arg_utils import AsyncEngineArgs, EngineArgs
14
from vllm.sampling_params import SamplingParams
15
from vllm.v1.engine.async_llm import AsyncLLM
16
from vllm.v1.engine.llm_engine import LLMEngine
17
+from vllm.v1.executor import multiproc_executor as multiproc_executor_module
18
from vllm.v1.executor.abstract import Executor
19
from vllm.v1.executor.multiproc_executor import MultiprocExecutor
20
from vllm.v1.executor.uniproc_executor import (
@@ -43,6 +44,50 @@ def test_supports_async_scheduling_multiproc_executor():
44
assert MultiprocExecutor.supports_async_scheduling() is True
45
46
47
+class _FakeClock:
48
+ def __init__(self) -> None:
49
+ self.now = 0.0
50
+
51
+ def time(self) -> float:
52
+ return self.now
53
+
54
+ def sleep(self, seconds: float) -> None:
55
+ self.now += seconds
56
+
57
+
58
+class _FakeProcess:
59
+ def __init__(self, clock: _FakeClock, exits_at: float) -> None:
60
+ self.clock = clock
61
+ self.exits_at = exits_at
62
+ self.terminate_called = False
63
+
64
+ def is_alive(self) -> bool:
65
+ return self.clock.time() < self.exits_at
66
+
67
+ def terminate(self) -> None:
68
+ self.terminate_called = True
69
+
70
+
71
+@pytest.mark.parametrize(
72
+ ("timeout", "exits_at", "expected_terminate"),
73
+ [
74
+ pytest.param(6, 5, False, id="worker-exits-before-timeout"),
75
+ pytest.param(6, 7, True, id="worker-exceeds-timeout"),
76
+ ],
77
+)
78
+def test_multiproc_executor_worker_termination_timeout(
79
+ monkeypatch, timeout, exits_at, expected_terminate
80
+):
81
+ monkeypatch.setenv("VLLM_WORKER_SHUTDOWN_TIMEOUT_SECONDS", str(timeout))
82
+ clock = _FakeClock()
83
+ monkeypatch.setattr(multiproc_executor_module.time, "time", clock.time)
84
+ monkeypatch.setattr(multiproc_executor_module.time, "sleep", clock.sleep)
85
+ executor = MultiprocExecutor.__new__(MultiprocExecutor)
86
+ proc = _FakeProcess(clock, exits_at=exits_at)
87
+ executor._ensure_worker_termination([proc])
88
+ assert proc.terminate_called is expected_terminate
89
+
90
+
91
class CustomMultiprocExecutor(MultiprocExecutor):
92
def collective_rpc(
93
self,
vllm/envs.py
+6
@@ -203,6 +203,7 @@ if TYPE_CHECKING:
203
VLLM_ENFORCE_STRICT_TOOL_CALLING: bool = True
204
VLLM_MQ_MAX_CHUNK_BYTES_MB: int = 16
205
VLLM_EXECUTE_MODEL_TIMEOUT_SECONDS: int = 300
206
+ VLLM_WORKER_SHUTDOWN_TIMEOUT_SECONDS: int = 5
207
VLLM_KV_CACHE_LAYOUT: Literal["NHD", "HND"] | None = None
208
VLLM_SSM_CONV_STATE_LAYOUT: Literal["SD", "DS"] | None = None
209
VLLM_COMPUTE_NANS_IN_LOGITS: bool = False
@@ -1552,6 +1553,10 @@ environment_variables: dict[str, Callable[[], Any]] = {
1553
"VLLM_EXECUTE_MODEL_TIMEOUT_SECONDS": lambda: int(
1554
os.getenv("VLLM_EXECUTE_MODEL_TIMEOUT_SECONDS", "300")
1555
),
1556
+ # Timeout in seconds for engine and worker process shutdown
1557
+ "VLLM_WORKER_SHUTDOWN_TIMEOUT_SECONDS": lambda: int(
1558
+ os.getenv("VLLM_WORKER_SHUTDOWN_TIMEOUT_SECONDS", "5")
1559
+ ),
1560
# KV Cache layout used throughout vllm.
1561
# Some common values are:
1562
# - NHD
@@ -1994,6 +1999,7 @@ def compile_factors() -> dict[str, object]:
1999
"VLLM_ENGINE_ITERATION_TIMEOUT_S",
2000
"VLLM_HTTP_TIMEOUT_KEEP_ALIVE",
2001
"VLLM_EXECUTE_MODEL_TIMEOUT_SECONDS",
2002
+ "VLLM_WORKER_SHUTDOWN_TIMEOUT_SECONDS",
2003
"VLLM_KEEP_ALIVE_ON_ENGINE_DEATH",
2004
"VLLM_IMAGE_FETCH_TIMEOUT",
2005
"VLLM_VIDEO_FETCH_TIMEOUT",
vllm/v1/engine/core_client.py
+4
-1
@@ -20,6 +20,7 @@ import msgspec.msgpack
20
import zmq
21
import zmq.asyncio
22
23
+from vllm import envs
24
from vllm.config import VllmConfig
25
from vllm.envs import VLLM_ENGINE_READY_TIMEOUT_S
26
from vllm.logger import init_logger
@@ -394,7 +395,9 @@ class BackgroundResources:
395
logger.debug_once("[shutdown] MPClient: background resource cleanup start")
396
self.engine_dead = True
397
if self.engine_manager is not None:
397
- self.engine_manager.shutdown()
398
+ self.engine_manager.shutdown(
399
+ timeout=envs.VLLM_WORKER_SHUTDOWN_TIMEOUT_SECONDS
400
+ )
401
if self.coordinator is not None:
402
self.coordinator.shutdown()
403
vllm/v1/executor/multiproc_executor.py
+3
-1
@@ -429,7 +429,9 @@ class MultiprocExecutor(Executor):
429
"[shutdown] Executor: waiting for worker exit count=%d",
430
initial_count,
431
)
432
- if wait_for_termination(active_procs(), 4):
432
+ if wait_for_termination(
433
+ active_procs(), timeout=envs.VLLM_WORKER_SHUTDOWN_TIMEOUT_SECONDS
434
+ ):
435
logger.info_once("[shutdown] Executor: all workers exited gracefully")
436
return
437