[Core] Bound execute_dummy_batch RPC waits by the execute-model timeout (#60398)

Signed-off-by: NickLucche <nicolo.lucchesi@mistral.ai>

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

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)