From 8733a21d5517d56fb7ef573b379eaeb7c08e1d03 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=B2=20Lucchesi?= Date: Thu, 8 Oct 2026 03:22:04 +0200 Subject: [PATCH] [Core] Bound execute_dummy_batch RPC waits by the execute-model timeout (#60398) Signed-off-by: NickLucche --- tests/v1/executor/test_multiproc_executor.py | 22 +++++++++++++------- vllm/v1/executor/multiproc_executor.py | 6 +++++- 2 files changed, 20 insertions(+), 8 deletions(-) diff --git a/tests/v1/executor/test_multiproc_executor.py b/tests/v1/executor/test_multiproc_executor.py index 706bfae381..6a3b3d516e 100644 --- a/tests/v1/executor/test_multiproc_executor.py +++ b/tests/v1/executor/test_multiproc_executor.py @@ -70,16 +70,24 @@ def test_execute_worker_rpc_returns_worker_exception(): @pytest.mark.parametrize("stalled", [False, True]) -def test_take_draft_token_ids_uses_execute_model_timeout(monkeypatch, stalled): - """A wedged draft-token readback must not block EngineCore forever.""" +@pytest.mark.parametrize( + "method,result", + [ + ("take_draft_token_ids", DraftTokenIds(["req"], [[1, 2, 3]])), + ("execute_dummy_batch", None), + ], +) +def test_model_rpc_uses_execute_model_timeout(monkeypatch, method, result, stalled): + """A wedged model RPC must not block EngineCore forever. For + execute_dummy_batch this is an idle DP rank stuck in a collective with a + hung peer.""" monkeypatch.setenv("VLLM_EXECUTE_MODEL_TIMEOUT_SECONDS", "7") - draft = DraftTokenIds(["req"], [[1, 2, 3]]) def dequeue(*, timeout=None): assert timeout is not None and 0 < timeout <= 7 if stalled: raise TimeoutError - return WorkerProc.ResponseStatus.SUCCESS, draft + return WorkerProc.ResponseStatus.SUCCESS, result executor: Any = MultiprocExecutor.__new__(MultiprocExecutor) executor.is_failed = False @@ -89,7 +97,7 @@ def test_take_draft_token_ids_uses_execute_model_timeout(monkeypatch, stalled): executor.response_mqs = [None, SimpleNamespace(dequeue=dequeue)] if stalled: - with pytest.raises(TimeoutError, match="take_draft_token_ids timed out"): - executor.take_draft_token_ids() + with pytest.raises(TimeoutError, match=f"{method} timed out"): + getattr(executor, method)() else: - assert executor.take_draft_token_ids() is draft + assert getattr(executor, method)() is result diff --git a/vllm/v1/executor/multiproc_executor.py b/vllm/v1/executor/multiproc_executor.py index 78290d25ba..cad3436854 100644 --- a/vllm/v1/executor/multiproc_executor.py +++ b/vllm/v1/executor/multiproc_executor.py @@ -364,7 +364,11 @@ class MultiprocExecutor(Executor): ) def execute_dummy_batch(self) -> None: - self.collective_rpc("execute_dummy_batch", unique_reply_rank=self.output_rank) + self.collective_rpc( + "execute_dummy_batch", + unique_reply_rank=self.output_rank, + timeout=envs.VLLM_EXECUTE_MODEL_TIMEOUT_SECONDS, + ) def take_draft_token_ids(self) -> DraftTokenIds | None: # OPTIMIZATION: Get output only from a single worker (output_rank)