From 48aa8d8d7529d2314858d8487cc0a21789fc7ec1 Mon Sep 17 00:00:00 2001 From: bugkeep <140229885+bugkeep@users.noreply.github.com> Date: Thu, 30 Jul 2026 07:41:34 +0800 Subject: [PATCH] [Bugfix] Prevent stale multiproc RPC deadlines from becoming unbounded waits (#41357) Signed-off-by: bugkeep <1921817430@qq.com> Co-authored-by: Claude --- .../test_multiproc_executor_timeout.py | 172 ++++++++++++++++++ .../device_communicators/shm_broadcast.py | 3 +- vllm/v1/executor/multiproc_executor.py | 2 +- 3 files changed, 175 insertions(+), 2 deletions(-) create mode 100644 tests/v1/executor/test_multiproc_executor_timeout.py diff --git a/tests/v1/executor/test_multiproc_executor_timeout.py b/tests/v1/executor/test_multiproc_executor_timeout.py new file mode 100644 index 0000000000..8d7998920c --- /dev/null +++ b/tests/v1/executor/test_multiproc_executor_timeout.py @@ -0,0 +1,172 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Regression tests for stale multiproc RPC deadlines.""" + +import time +from collections import deque +from concurrent.futures import Future, InvalidStateError +from contextlib import suppress +from unittest.mock import patch + + +def _dequeue_timeout(deadline: float | None) -> float | None: + return None if deadline is None else max(0.0, deadline - time.monotonic()) + + +class FutureWrapper(Future): + def __init__( + self, + futures_queue: deque["FutureWrapper"], + get_response, + aggregate=lambda x: x, + ): + self.futures_queue = futures_queue + self.get_response = get_response + self.aggregate = aggregate + super().__init__() + self.futures_queue.appendleft(self) + + def result(self, timeout=None): + if timeout is not None: + raise RuntimeError("timeout not implemented") + + while not self.done(): + future = self.futures_queue.pop() + future._wait_for_response() + return super().result() + + def _wait_for_response(self): + try: + response = self.aggregate(self.get_response()) + with suppress(InvalidStateError): + self.set_result(response) + except Exception as e: + with suppress(InvalidStateError): + self.set_exception(e) + + +class FakeClock: + def __init__(self, start: float = 0.0) -> None: + self._now = start + + def monotonic(self) -> float: + return self._now + + def advance(self, seconds: float) -> None: + self._now += seconds + + +class FakeResponseMQ: + def __init__(self, response: object = ("SUCCESS", "dummy_response")) -> None: + self.response = response + self.timeouts: list[float | None] = [] + + def dequeue(self, timeout: float | None = None) -> object: + assert timeout is None or timeout >= 0.0, ( + f"dequeue received negative timeout: {timeout}" + ) + self.timeouts.append(timeout) + return self.response + + +def test_future_wrapper_stale_deadline_never_passes_negative_timeout() -> None: + clock = FakeClock(100.0) + futures_queue: deque[FutureWrapper] = deque() + response_mq = FakeResponseMQ() + + deadline = clock.monotonic() + 1.0 + + def get_response() -> object: + return response_mq.dequeue(timeout=_dequeue_timeout(deadline)) + + future = FutureWrapper( + futures_queue, + get_response=get_response, + aggregate=lambda x: x, + ) + + clock.advance(5.0) + + with patch("time.monotonic", clock.monotonic): + future.result() + + assert response_mq.timeouts == [0.0] + + +def test_future_wrapper_non_expired_deadline_passes_positive_timeout() -> None: + clock = FakeClock(100.0) + futures_queue: deque[FutureWrapper] = deque() + response_mq = FakeResponseMQ() + + deadline = clock.monotonic() + 5.0 + + def get_response() -> object: + return response_mq.dequeue(timeout=_dequeue_timeout(deadline)) + + future = FutureWrapper( + futures_queue, + get_response=get_response, + aggregate=lambda x: x, + ) + + clock.advance(2.0) + + with patch("time.monotonic", clock.monotonic): + future.result() + + assert len(response_mq.timeouts) == 1 + timeout = response_mq.timeouts[0] + assert timeout is not None + assert timeout > 0.0 + assert timeout <= 5.0 + + +def test_future_wrapper_deadline_none_passes_none() -> None: + futures_queue: deque[FutureWrapper] = deque() + response_mq = FakeResponseMQ() + + def get_response() -> object: + return response_mq.dequeue(timeout=_dequeue_timeout(None)) + + future = FutureWrapper( + futures_queue, + get_response=get_response, + aggregate=lambda x: x, + ) + + future.result() + + assert response_mq.timeouts == [None] + + +def test_future_wrapper_drains_pending_before_own_get_response() -> None: + futures_queue: deque[FutureWrapper] = deque() + call_order: list[str] = [] + + def make_get_response(label: str): + def _get() -> object: + call_order.append(label) + return FakeResponseMQ().dequeue(timeout=None) + + return _get + + first = FutureWrapper(futures_queue, make_get_response("first")) + second = FutureWrapper(futures_queue, make_get_response("second")) + + second.result() + + assert first.done() + assert second.done() + assert call_order == ["first", "second"] + + +def test_recv_timeout_ms_clamps_negative_timeout() -> None: + def recv_timeout_ms(timeout: float | None) -> int | None: + return None if timeout is None else max(0, int(timeout * 1000)) + + assert recv_timeout_ms(None) is None + assert recv_timeout_ms(-1.0) == 0 + assert recv_timeout_ms(-0.001) == 0 + assert recv_timeout_ms(0.0) == 0 + assert recv_timeout_ms(0.001) == 1 + assert recv_timeout_ms(2.5) == 2500 diff --git a/vllm/distributed/device_communicators/shm_broadcast.py b/vllm/distributed/device_communicators/shm_broadcast.py index e59b14f7a6..f6d2a24755 100644 --- a/vllm/distributed/device_communicators/shm_broadcast.py +++ b/vllm/distributed/device_communicators/shm_broadcast.py @@ -909,7 +909,8 @@ class MessageQueue: @staticmethod def recv(socket: zmq.Socket, timeout: float | None) -> Any: - timeout_ms = None if timeout is None else int(timeout * 1000) + # Ensure non-negative timeout passed to zmq poll. + timeout_ms = None if timeout is None else max(0, int(timeout * 1000)) if not socket.poll(timeout=timeout_ms): raise TimeoutError recv, *recv_oob = socket.recv_multipart(copy=False) diff --git a/vllm/v1/executor/multiproc_executor.py b/vllm/v1/executor/multiproc_executor.py index 0a3a6d1ec3..b464f2dcf6 100644 --- a/vllm/v1/executor/multiproc_executor.py +++ b/vllm/v1/executor/multiproc_executor.py @@ -395,7 +395,7 @@ class MultiprocExecutor(Executor): responses = [] for mq in response_mqs: dequeue_timeout = ( - None if deadline is None else (deadline - time.monotonic()) + None if deadline is None else max(0.0, deadline - time.monotonic()) ) try: status, result = mq.dequeue(timeout=dequeue_timeout)