From 611c083587ee6506d6648b294917ec2b98d7abdd Mon Sep 17 00:00:00 2001 From: Yueqian Lin <70319226+linyueqian@users.noreply.github.com> Date: Sun, 16 Aug 2026 19:51:17 -0700 Subject: [PATCH] [BugFix][TTS] R1.1-R1.4: turn silent async-chunk failures into visible ones (#6033) Signed-off-by: Yueqian Lin Co-authored-by: Hongsheng Liu --- .../sched/test_omni_scheduler_mixin_shared.py | 21 +- .../test_omni_scheduler_mixin_timeouts.py | 253 +++++++++++++++++- .../test_chunk_transfer_adapter.py | 109 ++++++++ .../test_orchestrator_error_handling.py | 45 ++++ .../test_orchestrator_stage_input_bridge.py | 28 +- vllm_omni/core/sched/omni_scheduler_mixin.py | 140 +++++++++- .../omni_connectors/transfer_adapter/base.py | 33 ++- .../chunk_transfer_adapter.py | 71 +++++ vllm_omni/engine/orchestrator.py | 70 ++++- .../omni_connector_model_runner_mixin.py | 12 +- 10 files changed, 750 insertions(+), 32 deletions(-) diff --git a/tests/core/sched/test_omni_scheduler_mixin_shared.py b/tests/core/sched/test_omni_scheduler_mixin_shared.py index 9aaa4c656..4f98fd092 100644 --- a/tests/core/sched/test_omni_scheduler_mixin_shared.py +++ b/tests/core/sched/test_omni_scheduler_mixin_shared.py @@ -4,7 +4,10 @@ from types import SimpleNamespace import pytest from vllm.v1.engine import FinishReason -from vllm_omni.core.sched.omni_scheduler_mixin import OmniSchedulerMixin +from vllm_omni.core.sched.omni_scheduler_mixin import ( + DEFAULT_INPUT_WAIT_TIMEOUT_S, + OmniSchedulerMixin, +) from vllm_omni.core.sched.output import OmniChunkRecvHandle pytestmark = [pytest.mark.core_model, pytest.mark.cpu] @@ -22,13 +25,25 @@ def test_schedule_lifecycle_helpers_process_and_restore_both_input_paths(): scheduler.requests = {"request": object()} scheduler._consume_pending_connector_output = lambda mode: calls.append(("consume", mode)) scheduler._process_pending_input_timeouts = lambda: calls.append(("timeouts",)) + + def _collect_timed_out(timeout_s): + calls.append(("chunk-timeouts", timeout_s)) + return set() + + def _collect_failed_sends(): + calls.append(("failed-sends",)) + return {} + scheduler.chunk_transfer_adapter = SimpleNamespace( + receives_chunks=True, process_pending_chunks=lambda waiting, running, scheduler_requests: calls.append( ("process", waiting, running, scheduler_requests) ), restore_queues=lambda waiting, running, scheduler_requests: calls.append( ("restore-chunks", waiting, running, scheduler_requests) ), + collect_timed_out_request_ids=_collect_timed_out, + collect_failed_send_request_ids=_collect_failed_sends, ) scheduler.input_coordinator = SimpleNamespace( restore_queues=lambda waiting: calls.append(("restore-full", waiting)) @@ -41,6 +56,10 @@ def test_schedule_lifecycle_helpers_process_and_restore_both_input_paths(): ("consume", "ar"), ("timeouts",), ("process", scheduler.waiting, scheduler.running, scheduler.requests), + # The chunk deadline runs after chunks are applied, so a chunk that + # arrived this cycle resets the clock before it is measured (R1.1). + ("chunk-timeouts", DEFAULT_INPUT_WAIT_TIMEOUT_S), + ("failed-sends",), ("restore-chunks", scheduler.waiting, scheduler.running, scheduler.requests), ("restore-full", scheduler.waiting), ] diff --git a/tests/core/sched/test_omni_scheduler_mixin_timeouts.py b/tests/core/sched/test_omni_scheduler_mixin_timeouts.py index 5fad4340c..1824d2809 100644 --- a/tests/core/sched/test_omni_scheduler_mixin_timeouts.py +++ b/tests/core/sched/test_omni_scheduler_mixin_timeouts.py @@ -1,6 +1,10 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project -"""Unit coverage for _process_pending_input_timeouts. +"""Unit coverage for the two stage-input timeout hooks. + +``_process_pending_input_timeouts`` covers the full-payload transport; +``_process_pending_chunk_timeouts`` covers async-chunk. Both share +``VLLM_OMNI_INPUT_WAIT_TIMEOUT_S``. Verifies that the mixin correctly *delegates* timed-out requests to the base scheduler's ``finish_requests`` API with ``RequestStatus.FINISHED_ERROR``. @@ -12,6 +16,10 @@ this file only asserts the wiring from the mixin to that API. from __future__ import annotations +import importlib +import logging +import math +import os from types import SimpleNamespace import pytest @@ -102,5 +110,248 @@ def test_process_pending_input_timeouts_disabled_when_timeout_zero(monkeypatch): assert scheduler.finish_calls == [] +class _FakeChunkAdapter: + """Stands in for OmniChunkTransferAdapter's timeout-collection surface.""" + + def __init__(self, timed_out_ids, *, receives_chunks=True): + self._timed_out_ids = set(timed_out_ids) + self.receives_chunks = receives_chunks + self.calls = [] + + def collect_timed_out_request_ids(self, timeout_s): + self.calls.append(timeout_s) + return set(self._timed_out_ids) + + +class _FakeChunkScheduler(OmniSchedulerMixin): + def __init__(self, requests, adapter): + self.requests = requests + self.chunk_transfer_adapter = adapter + self.finish_calls = [] + + def finish_requests(self, req_ids, status): + self.finish_calls.append((set(req_ids), status)) + + +def test_process_pending_chunk_timeouts_delegates_to_finish_requests(): + """A request stalled in WAITING_FOR_CHUNK is failed, not left parked. + + This is the async-chunk half of the safety net (RFC #4855 R1.1); before it + existed, the chunk path had no deadline at all and issue #3833 could park a + request forever. + """ + req_id = "stalled-stream" + adapter = _FakeChunkAdapter(timed_out_ids={req_id}) + scheduler = _FakeChunkScheduler({req_id: SimpleNamespace(request_id=req_id)}, adapter) + + scheduler._process_pending_chunk_timeouts() + + assert len(adapter.calls) == 1 + assert adapter.calls[0] > 0 + assert len(scheduler.finish_calls) == 1 + finished_ids, status = scheduler.finish_calls[0] + assert finished_ids == {req_id} + assert getattr(status, "name", str(status)).endswith("FINISHED_ERROR") + + +def test_process_pending_chunk_timeouts_skips_already_freed_request(): + adapter = _FakeChunkAdapter(timed_out_ids={"already-freed"}) + scheduler = _FakeChunkScheduler(requests={}, adapter=adapter) + + scheduler._process_pending_chunk_timeouts() + + assert scheduler.finish_calls == [] + + +def test_process_pending_chunk_timeouts_noop_on_a_producer_stage(): + """Stage 0 does not receive chunks, so it has nothing to expire.""" + adapter = _FakeChunkAdapter(timed_out_ids={"r1"}, receives_chunks=False) + scheduler = _FakeChunkScheduler({"r1": SimpleNamespace(request_id="r1")}, adapter) + + scheduler._process_pending_chunk_timeouts() + + assert adapter.calls == [] + assert scheduler.finish_calls == [] + + +def test_process_pending_chunk_timeouts_noop_without_adapter(): + class _NoAdapter(OmniSchedulerMixin): + def __init__(self): + self.requests = {} + self.chunk_transfer_adapter = None + self.finish_calls = [] + + def finish_requests(self, req_ids, status): + self.finish_calls.append((set(req_ids), status)) + + scheduler = _NoAdapter() + scheduler._process_pending_chunk_timeouts() + assert scheduler.finish_calls == [] + + +def test_process_pending_chunk_timeouts_disabled_when_timeout_zero(monkeypatch): + """The shared knob disables both transports' nets together.""" + from vllm_omni.core.sched import omni_scheduler_mixin + + monkeypatch.setattr(omni_scheduler_mixin, "DEFAULT_INPUT_WAIT_TIMEOUT_S", 0.0) + + adapter = _FakeChunkAdapter(timed_out_ids={"r1"}) + scheduler = _FakeChunkScheduler({"r1": SimpleNamespace(request_id="r1")}, adapter) + scheduler._process_pending_chunk_timeouts() + + assert adapter.calls == [] + assert scheduler.finish_calls == [] + + +class _FakeSendFailureAdapter: + """Stands in for the adapter's drained send-failure map (R1.2).""" + + def __init__(self, failures): + self._failures = dict(failures) + self.collect_calls = 0 + + def collect_failed_send_request_ids(self): + self.collect_calls += 1 + failures, self._failures = self._failures, {} + return failures + + +class _FakeSendFailureScheduler(OmniSchedulerMixin): + def __init__(self, requests, adapter): + self.requests = requests + self.chunk_transfer_adapter = adapter + + +def test_log_failed_chunk_sends_reports_a_live_request(caplog): + adapter = _FakeSendFailureAdapter({"req-live": "OSError: No space left on device"}) + scheduler = _FakeSendFailureScheduler({"req-live": SimpleNamespace(request_id="req-live")}, adapter) + + with caplog.at_level(logging.ERROR): + scheduler._log_failed_chunk_sends() + + assert "req-live" in caplog.text + assert "No space left on device" in caplog.text + + +def test_log_failed_chunk_sends_still_reports_a_departed_request(caplog): + """``collect_*`` drains the map, so a failure for a request that already left + this scheduler is the one place a drop could still go unrecorded.""" + adapter = _FakeSendFailureAdapter({"req-gone": "OSError: No space left on device"}) + scheduler = _FakeSendFailureScheduler({}, adapter) + + with caplog.at_level(logging.WARNING): + scheduler._log_failed_chunk_sends() + + assert adapter.collect_calls == 1 + assert "req-gone" in caplog.text + assert "No space left on device" in caplog.text + assert "finished or aborted" in caplog.text + + +def test_log_failed_chunk_sends_noop_without_adapter(): + scheduler = _FakeSendFailureScheduler({}, None) + scheduler._log_failed_chunk_sends() + + +# --- R1.4: VLLM_OMNI_INPUT_WAIT_TIMEOUT_S is parsed at import time, so each of +# these reloads the module under a different environment. One knob now gates the +# deadline on both transports, which is why a silent 0 matters more than it did +# when only the full-payload path used it. + + +@pytest.fixture +def reload_with(monkeypatch): + """Reload the mixin module under a given env value, then put it back. + + ``importlib.reload`` re-executes into the live module dict, so whatever the + last test parsed is what every later test in the session sees -- ``-k + zero_warns`` alone would leave ``DEFAULT_INPUT_WAIT_TIMEOUT_S = 0.0`` and + silently disable both deadlines. The rejection cases are worse: the + ValueError fires *after* the rebind, so ``nan`` lingers in the module. + ``monkeypatch`` restores the environment but not the module, so reload once + more on the way out. + """ + import vllm_omni.core.sched.omni_scheduler_mixin as mod + + def _reload(raw): + if raw is None: + monkeypatch.delenv("VLLM_OMNI_INPUT_WAIT_TIMEOUT_S", raising=False) + else: + monkeypatch.setenv("VLLM_OMNI_INPUT_WAIT_TIMEOUT_S", raw) + return importlib.reload(mod) + + yield _reload + + monkeypatch.undo() + importlib.reload(mod) + + +@pytest.mark.parametrize( + "raw,expected", + [ + (None, 600.0), + ("30", 30.0), + ("0.5", 0.5), + ("0", 0.0), # explicit opt-out is still honoured + ], +) +def test_timeout_accepts_valid_values(reload_with, raw, expected): + mod = reload_with(raw) + assert mod.DEFAULT_INPUT_WAIT_TIMEOUT_S == expected + + +@pytest.mark.parametrize("raw", ["-1", "-600", "-0.5"]) +def test_negative_timeout_is_rejected(reload_with, raw): + """`-1` is a common "no limit" idiom; substituting 600 would give an operator + who meant "never time out" mysterious failures ten minutes in. Fail loudly.""" + with pytest.raises(ValueError, match="negative"): + reload_with(raw) + + +@pytest.mark.parametrize("raw", ["nan", "inf", "-inf", "1e999"]) +def test_non_finite_timeout_is_rejected(reload_with, raw): + """nan compares false against every deadline check and inf is unreachable; + both disable the net while looking like a number.""" + with pytest.raises(ValueError, match="not finite"): + reload_with(raw) + + +def test_zero_warns_and_names_both_transports(reload_with, caplog): + with caplog.at_level(logging.WARNING): + mod = reload_with("0") + assert mod.DEFAULT_INPUT_WAIT_TIMEOUT_S == 0.0 + text = caplog.text + assert "DISABLED" in text + assert "WAITING_FOR_INPUT" in text and "WAITING_FOR_CHUNK" in text + + +def test_unparsable_timeout_falls_back(reload_with): + mod = reload_with("not-a-number") + assert mod.DEFAULT_INPUT_WAIT_TIMEOUT_S == 600.0 + + +def test_a_rejected_value_does_not_outlive_its_test(monkeypatch): + """The reason ``reload_with`` reloads on the way out, done here by hand. + + ``nan`` is bound before the ValueError fires, so the reload that rejects it + still leaves it in the module -- and nan compares false against every + deadline check, which is exactly the silent-disable this item exists to + stop. Restoring the environment is not enough; the module has to be reloaded + too. + """ + import vllm_omni.core.sched.omni_scheduler_mixin as mod + + ambient = float(os.environ.get("VLLM_OMNI_INPUT_WAIT_TIMEOUT_S", "600")) + + monkeypatch.setenv("VLLM_OMNI_INPUT_WAIT_TIMEOUT_S", "nan") + with pytest.raises(ValueError, match="not finite"): + importlib.reload(mod) + assert math.isnan(mod.DEFAULT_INPUT_WAIT_TIMEOUT_S) + + monkeypatch.undo() + importlib.reload(mod) + assert mod.DEFAULT_INPUT_WAIT_TIMEOUT_S == ambient + + if __name__ == "__main__": pytest.main([__file__, "-v"]) diff --git a/tests/distributed/omni_connectors/test_chunk_transfer_adapter.py b/tests/distributed/omni_connectors/test_chunk_transfer_adapter.py index da4e81995..063751b01 100644 --- a/tests/distributed/omni_connectors/test_chunk_transfer_adapter.py +++ b/tests/distributed/omni_connectors/test_chunk_transfer_adapter.py @@ -1728,3 +1728,112 @@ def test_purge_is_noop_on_empty_deques(build_adapter): adapter.restore_queues(waiting_queue, running_queue, scheduler_requests={}) assert running_queue == [] assert waiting_queue == [] + + +# --------------------------------------------------------------------------- # +# Chunk-wait deadline (RFC #4855 R1.1, issue #3833) +# +# Before this, the async-chunk path had no deadline of any kind, so a dropped +# terminal chunk or an upstream stage that died mid-stream parked the request +# in WAITING_FOR_CHUNK forever. None of these scenarios were covered. +# --------------------------------------------------------------------------- # + + +def _park_in_chunk_wait(adapter, request, *, waiting=True): + """Drive a request through one process_pending_chunks round into the wait.""" + queue = DummyWaitingQueue([request]) if waiting else [request] + if waiting: + adapter.process_pending_chunks(queue, []) + else: + adapter.process_pending_chunks(DummyWaitingQueue(), queue) + return queue + + +def test_chunk_wait_clock_starts_when_a_request_parks(build_adapter): + adapter, _ = build_adapter(stage_id=1, model_mode="ar") + request = _req("r1", RequestStatus.WAITING) + + _park_in_chunk_wait(adapter, request) + + assert request.status == RequestStatus.WAITING_FOR_CHUNK + assert "r1" in adapter._waiting_since + + +def test_chunk_wait_does_not_expire_before_the_deadline(build_adapter): + adapter, _ = build_adapter(stage_id=1, model_mode="ar") + _park_in_chunk_wait(adapter, _req("r1", RequestStatus.WAITING)) + + assert adapter.collect_timed_out_request_ids(timeout_s=600.0) == set() + assert "r1" in adapter._waiting_since + + +def test_dropped_terminal_chunk_expires_the_request(build_adapter): + """The #3833 scenario: upstream stops sending and never marks the stream done.""" + adapter, _ = build_adapter(stage_id=1, model_mode="ar") + _park_in_chunk_wait(adapter, _req("r1", RequestStatus.WAITING)) + + adapter._waiting_since["r1"] -= 601.0 + + assert adapter.collect_timed_out_request_ids(timeout_s=600.0) == {"r1"} + # Cleared, so a second sweep does not re-report the same request. + assert adapter.collect_timed_out_request_ids(timeout_s=600.0) == set() + + +def test_arriving_chunk_resets_the_clock(build_adapter): + """A slow but healthy stream must never expire: the deadline measures stall + time between chunks, not the lifetime of the stream.""" + adapter, _ = build_adapter(stage_id=1, model_mode="ar") + request = _req("r1", RequestStatus.WAITING) + queue = _park_in_chunk_wait(adapter, request) + adapter.restore_queues(queue, [], scheduler_requests={"r1": request}) + + # Age the wait to just under the deadline, then deliver a chunk. + adapter._waiting_since["r1"] -= 599.0 + adapter._finished_load_reqs.add("r1") + adapter.process_pending_chunks(queue, []) + + assert "r1" not in adapter._waiting_since + assert adapter.collect_timed_out_request_ids(timeout_s=600.0) == set() + + +def test_a_disabled_deadline_never_expires(build_adapter): + adapter, _ = build_adapter(stage_id=1, model_mode="ar") + _park_in_chunk_wait(adapter, _req("r1", RequestStatus.WAITING)) + adapter._waiting_since["r1"] -= 10_000.0 + + assert adapter.collect_timed_out_request_ids(timeout_s=0.0) == set() + assert adapter.collect_timed_out_request_ids(timeout_s=-1.0) == set() + + +def test_finished_request_leaves_no_stale_timestamp(build_adapter): + """Otherwise a completed request would be reported as timed out later.""" + adapter, _ = build_adapter(stage_id=1, model_mode="ar") + request = _req("r1", RequestStatus.WAITING) + _park_in_chunk_wait(adapter, request) + + adapter.finish_requests(["r1"], RequestStatus.FINISHED_STOPPED, {"r1": request}) + + assert "r1" not in adapter._waiting_since + assert adapter.collect_timed_out_request_ids(timeout_s=0.001) == set() + + +def test_aborted_request_leaves_no_stale_timestamp(build_adapter): + adapter, _ = build_adapter(stage_id=1, model_mode="ar") + _park_in_chunk_wait(adapter, _req("r1", RequestStatus.WAITING)) + + adapter.cleanup_receiver("r1") + + assert "r1" not in adapter._waiting_since + + +def test_expiry_is_per_request(build_adapter): + """One stalled stream must not take down its healthy neighbours.""" + adapter, _ = build_adapter(stage_id=1, model_mode="ar", max_num_seqs=4) + stalled = _req("stalled", RequestStatus.WAITING) + healthy = _req("healthy", RequestStatus.WAITING) + adapter.process_pending_chunks(DummyWaitingQueue([stalled, healthy]), []) + + adapter._waiting_since["stalled"] -= 601.0 + + assert adapter.collect_timed_out_request_ids(timeout_s=600.0) == {"stalled"} + assert "healthy" in adapter._waiting_since diff --git a/tests/engine/test_orchestrator_error_handling.py b/tests/engine/test_orchestrator_error_handling.py index 780c5a17b..4176764bc 100644 --- a/tests/engine/test_orchestrator_error_handling.py +++ b/tests/engine/test_orchestrator_error_handling.py @@ -411,6 +411,51 @@ async def test_async_chunk_prewarm_failure_reports_failing_downstream_stage(orch orchestrator_fixture.thread.join(timeout=5) +@pytest.mark.asyncio +async def test_async_chunk_prewarm_without_prompt_token_ids_fails_only_that_request( + orchestrator_factory, +) -> None: + """R1.3 of #4855: an async-chunk request whose stage-0 prompt carries no + token ids cannot prewarm anything, so it must fail with a client error + instead of stalling. + + Non-fatal is the point: ``fatal=True`` reads as "the engine died" and the + entrypoints act on it (``OmniLLM.generate`` raises out of its batch loop, + the serving layer calls ``terminate_if_errored``). One client's embeds-only + prompt must not reach other requests. + """ + stage0 = FakeStageClient(stage_type="llm", final_output=False) + stage1 = FakeStageClient(stage_type="llm", final_output=True) + orchestrator_fixture = orchestrator_factory([stage0, stage1], async_chunk=True) + + try: + await _enqueue_add_request( + orchestrator_fixture, + request_id="req-embeds", + # No prompt_token_ids: what an embeds-only prompt looks like here. + prompt=SimpleNamespace(request_id="req-embeds"), + original_prompt={"prompt": "embeds only"}, + sampling_params_list=[_sampling_params(), _sampling_params()], + final_stage_id=1, + ) + + error_msg = await _wait_for_error_message(orchestrator_fixture, request_id="req-embeds") + assert error_msg.fatal is False + assert error_msg.status_code == 400 + assert error_msg.error_type == "BadRequestError" + assert error_msg.stage_id == 0 + assert "prompt_token_ids" in error_msg.error + + # Nothing was prewarmed downstream, and the request is gone. + assert stage1.add_request_calls == [] + await _wait_for(lambda: "req-embeds" not in orchestrator_fixture.orchestrator.request_states) + assert orchestrator_fixture.thread.is_alive() + finally: + if orchestrator_fixture.thread.is_alive(): + orchestrator_fixture.request_sync_q.put_nowait(ShutdownRequestMessage()) + orchestrator_fixture.thread.join(timeout=5) + + # ───────── Direct unit tests for the fault-isolation helpers ───────── diff --git a/tests/engine/test_orchestrator_stage_input_bridge.py b/tests/engine/test_orchestrator_stage_input_bridge.py index cf347e49f..25637913b 100644 --- a/tests/engine/test_orchestrator_stage_input_bridge.py +++ b/tests/engine/test_orchestrator_stage_input_bridge.py @@ -256,12 +256,13 @@ async def test_async_prewarm_skips_outgoing_only_stage() -> None: duplex_identity=SimpleNamespace(), ) - await orchestrator._prewarm_async_chunk_stages( + prewarmed = await orchestrator._prewarm_async_chunk_stages( "req-prewarm", SimpleNamespace(prompt_token_ids=[1, 2], resumable=True), req_state, ) + assert prewarmed is True assert stage1.submitted == [] assert len(stage2.submitted) == 1 assert 1 not in req_state.stage_submit_ts @@ -277,6 +278,7 @@ async def test_async_prewarm_skips_outgoing_only_stage() -> None: @pytest.mark.asyncio async def test_duplex_prewarm_runs_after_first_stage0_submission() -> None: port, stage_pools, request_states, prewarm, submission = _duplex_stage_port_submission() + prewarm.return_value = True result = await port.submit(submission) @@ -285,6 +287,30 @@ async def test_duplex_prewarm_runs_after_first_stage0_submission() -> None: prewarm.assert_awaited_once_with("req-duplex", ANY, request_states["req-duplex"]) +@pytest.mark.asyncio +async def test_duplex_submit_bails_out_when_prewarm_failed_the_request() -> None: + """A failed prewarm has already aborted the request and popped its state. + + Returning a success result here would hand the control plane a replica for a + request that no longer exists, and the trailing bookkeeping would re-register + a running counter the cleanup just released -- a leak that never decrements. + """ + port, stage_pools, request_states, prewarm, submission = _duplex_stage_port_submission() + counter = MagicMock() + port._running_counter = counter + prewarm.return_value = False + request_state = request_states["req-duplex"] + + with pytest.raises(RuntimeError, match="prewarm failed"): + await port.submit(submission) + + prewarm.assert_awaited_once() + assert request_state.duplex_stage_fences == {} + assert request_state.stage_submit_ts == {} + assert request_state.running_counter_registered is False + counter.increment.assert_not_called() + + @pytest.mark.asyncio async def test_async_route_forwards_to_outgoing_only_stage() -> None: orchestrator = object.__new__(Orchestrator) diff --git a/vllm_omni/core/sched/omni_scheduler_mixin.py b/vllm_omni/core/sched/omni_scheduler_mixin.py index e0df774c0..7ce5f4470 100644 --- a/vllm_omni/core/sched/omni_scheduler_mixin.py +++ b/vllm_omni/core/sched/omni_scheduler_mixin.py @@ -1,5 +1,6 @@ from __future__ import annotations +import math import os import time from collections.abc import Iterable @@ -42,25 +43,56 @@ logger = init_logger(__name__) _STATS_INTERVAL_S = 1.0 -# Upper bound on how long a request may sit in full-payload-input wait -# (the state ``OmniSchedulingCoordinator`` records via ``_waiting_since``) -# before the scheduler force-fails it. Defends against stuck consumer-side -# requests when the producer drops a full-payload, send fails, or recv -# never arrives. Override per-deployment via -# VLLM_OMNI_INPUT_WAIT_TIMEOUT_S; set <=0 to disable the safety net. +# Upper bound on how long a request may wait for stage input before the +# scheduler force-fails it. Defends against stuck consumer-side requests when +# the producer drops a payload, send fails, or recv never arrives. Override +# per-deployment via VLLM_OMNI_INPUT_WAIT_TIMEOUT_S; set exactly 0 to disable +# the safety net. Negative and non-finite values are rejected at startup. # -# Scope: this constant only covers the full-payload coordinator path -# (``input_coordinator``). The async-chunk path uses -# ``chunk_transfer_adapter`` and is not affected by this constant. +# Scope: both transports. The full-payload path measures time parked in +# WAITING_FOR_INPUT (``OmniSchedulingCoordinator._waiting_since``); the +# async-chunk path measures time stalled in WAITING_FOR_CHUNK +# (``OmniChunkTransferAdapter._waiting_since``, reset on each chunk arrival). +# One knob, because the question it answers -- "how long may a request wait for +# stage input?" -- does not depend on which transport carries it. +_INPUT_WAIT_TIMEOUT_DEFAULT = 600.0 _INPUT_WAIT_TIMEOUT_RAW = os.environ.get("VLLM_OMNI_INPUT_WAIT_TIMEOUT_S", "600") try: DEFAULT_INPUT_WAIT_TIMEOUT_S: float = float(_INPUT_WAIT_TIMEOUT_RAW) except ValueError: logger.warning( - "Invalid VLLM_OMNI_INPUT_WAIT_TIMEOUT_S=%r; falling back to 600 seconds.", + "Invalid VLLM_OMNI_INPUT_WAIT_TIMEOUT_S=%r; falling back to %s seconds.", _INPUT_WAIT_TIMEOUT_RAW, + _INPUT_WAIT_TIMEOUT_DEFAULT, + ) + DEFAULT_INPUT_WAIT_TIMEOUT_S = _INPUT_WAIT_TIMEOUT_DEFAULT + +if not math.isfinite(DEFAULT_INPUT_WAIT_TIMEOUT_S): + # nan compares false against everything, so every deadline check silently + # passes; inf makes the deadline unreachable. Both disable the net while + # looking like a number, which is the failure this item exists to stop. + raise ValueError( + f"VLLM_OMNI_INPUT_WAIT_TIMEOUT_S={_INPUT_WAIT_TIMEOUT_RAW!r} is not finite. " + "Use a positive number of seconds, or exactly 0 to disable the deadline." + ) +if DEFAULT_INPUT_WAIT_TIMEOUT_S < 0: + # Fail at startup rather than substituting a default. `-1` is a common idiom + # for "no limit", so silently turning it into 600 would give an operator who + # meant "never time out" mysterious failures ten minutes in. + raise ValueError( + f"VLLM_OMNI_INPUT_WAIT_TIMEOUT_S={_INPUT_WAIT_TIMEOUT_RAW!r} is negative. " + "Use a positive number of seconds, or exactly 0 to disable the deadline." + ) +elif DEFAULT_INPUT_WAIT_TIMEOUT_S == 0: + # Zero stays a supported opt-out, but it is no longer silent: it disables + # the deadline on BOTH transports, so a request whose stage input never + # arrives waits forever. Say which nets just went away. + logger.warning( + "VLLM_OMNI_INPUT_WAIT_TIMEOUT_S=0: stage-input deadlines are DISABLED for " + "both the full-payload path (WAITING_FOR_INPUT) and the async-chunk path " + "(WAITING_FOR_CHUNK). A request whose input never arrives will wait " + "indefinitely instead of failing." ) - DEFAULT_INPUT_WAIT_TIMEOUT_S = 600.0 class OmniSchedulerMixin: @@ -184,6 +216,8 @@ class OmniSchedulerMixin: self.running, scheduler_requests=self.requests, ) + self._process_pending_chunk_timeouts() + self._log_failed_chunk_sends() def _restore_omni_wait_queues(self) -> None: """Restore requests temporarily parked by Omni input gates.""" @@ -211,9 +245,9 @@ class OmniSchedulerMixin: Disabled when ``DEFAULT_INPUT_WAIT_TIMEOUT_S`` is <= 0. Scope: only covers ``input_coordinator`` (full-payload path). - Async-chunk requests park in ``chunk_transfer_adapter`` instead - and are not handled here -- if a similar safety net is needed - for the chunk path, it belongs in the chunk adapter. + Async-chunk requests park in ``chunk_transfer_adapter`` instead and + are handled by ``_process_pending_chunk_timeouts`` below, which shares + this timeout. """ if DEFAULT_INPUT_WAIT_TIMEOUT_S <= 0: return @@ -234,6 +268,84 @@ class OmniSchedulerMixin: ) self.finish_requests(present_ids, RequestStatus.FINISHED_ERROR) + def _log_failed_chunk_sends(self) -> None: + """Surface chunks the sender gave up on (R1.2 of #4855). + + ``save_loop`` pops a task with ``popleft()`` and never re-queues it, and + ``connector.put`` can report failure without raising, so a dropped chunk + is a give-up rather than a retry. Until this change the only trace was a + warning that printed ``None`` for the request id, because the task dict + has no ``request_id`` key. + + This only reports. Failing the request from here does not work: the + producer is not the ``final_output`` stage, so an abort synthesized by + ``finish_requests`` is not turned into a client-visible termination by + ``Orchestrator._route_output`` (see the ``final_output`` guards there). + Making a producer-side drop client-visible needs an orchestrator-side + path like ``_fail_request_dead_stage``; the consumer's R1.1 deadline is + what actually ends the request today. + """ + adapter = getattr(self, "chunk_transfer_adapter", None) + if adapter is None: + return + failures = adapter.collect_failed_send_request_ids() + for request_id, reason in failures.items(): + if request_id in self.requests: + logger.error( + "[OmniScheduler] req=%s: chunk send gave up (%s); the consumer will " + "fail on the stage-input deadline", + request_id, + reason, + ) + else: + # Drained but no longer scheduled here: the request finished or + # was aborted between the failed send and this sweep. Still say + # so -- collect_* empties the map, so staying quiet would make + # this the one drop path in a change whose point is visibility. + logger.warning( + "[OmniScheduler] req=%s: chunk send gave up (%s) after the request " + "left this scheduler (finished or aborted)", + request_id, + reason, + ) + + def _process_pending_chunk_timeouts(self) -> None: + """Force-fail requests stalled in ``WAITING_FOR_CHUNK`` too long. + + The async-chunk counterpart of ``_process_pending_input_timeouts``, + called right after ``chunk_transfer_adapter.process_pending_chunks``. + Until now the chunk path had no safety net at all: the scheduler-side + comment on the full-payload net said a similar guard "belongs in the + chunk adapter", while the worker mixin assumed the full-payload net + already covered it. Only the first was true, so a dropped terminal + chunk or a producer that exhausted its send retries parked the request + forever (vllm-project/vllm-omni#3833). Since ``async_chunk: true`` is + the default for the flagship TTS pipelines, that was the default + streaming path. + + Shares ``VLLM_OMNI_INPUT_WAIT_TIMEOUT_S`` with the full-payload net: + one knob for "how long may a request wait for stage input", regardless + of which transport carries it. Disabled when the value is <= 0. + """ + if DEFAULT_INPUT_WAIT_TIMEOUT_S <= 0: + return + adapter = getattr(self, "chunk_transfer_adapter", None) + if adapter is None or not getattr(adapter, "receives_chunks", False): + return + timed_out_ids = adapter.collect_timed_out_request_ids(timeout_s=DEFAULT_INPUT_WAIT_TIMEOUT_S) + if not timed_out_ids: + return + present_ids = {req_id for req_id in timed_out_ids if req_id in self.requests} + if not present_ids: + return + logger.warning( + "Marking %d request(s) as FINISHED_ERROR after stalling > %.0fs waiting for a chunk: %s", + len(present_ids), + DEFAULT_INPUT_WAIT_TIMEOUT_S, + sorted(present_ids), + ) + self.finish_requests(present_ids, RequestStatus.FINISHED_ERROR) + def _capture_omni_connector_output(self, model_runner_output: Any) -> None: """Stash the model runner's omni_connector_output for next schedule(). diff --git a/vllm_omni/distributed/omni_connectors/transfer_adapter/base.py b/vllm_omni/distributed/omni_connectors/transfer_adapter/base.py index bbaca48e0..578fbb233 100644 --- a/vllm_omni/distributed/omni_connectors/transfer_adapter/base.py +++ b/vllm_omni/distributed/omni_connectors/transfer_adapter/base.py @@ -35,6 +35,8 @@ class OmniTransferAdapterBase: self.stop_event = threading.Event() self._recv_cond = threading.Condition() self._save_cond = threading.Condition() + self._send_failures: dict[str, str] = {} + self._send_failure_lock = threading.Lock() self.recv_thread = threading.Thread(target=self.recv_loop, daemon=True) self.recv_thread.start() @@ -82,6 +84,24 @@ class OmniTransferAdapterBase: elif not any_success and not self.stop_event.is_set(): self._recv_cond.wait(timeout=0.001) + def record_send_failure(self, request_id: str | None, reason: str) -> None: + """Note that a chunk for *request_id* will never be delivered. + + Called from the sender thread; drained by the scheduler thread, hence + the lock. ``_waiting_since`` needs none because only the scheduler + touches it. + """ + if not request_id: + return + with self._send_failure_lock: + self._send_failures.setdefault(request_id, reason) + + def collect_failed_send_request_ids(self) -> dict[str, str]: + """Drain and return the requests whose send gave up.""" + with self._send_failure_lock: + failures, self._send_failures = self._send_failures, {} + return failures + def save_loop(self): """Loop to send outgoing data.""" while not self.stop_event.is_set(): @@ -90,7 +110,18 @@ class OmniTransferAdapterBase: try: self._send_single_request(task) except Exception as e: - logger.warning(f"Error saving data for {task.get('request_id')}: {e}") + # R1.2 of #4855. The task was popleft()-ed and is not re-queued, + # so this is a give-up, not a retry: the consumer will wait for a + # chunk that is never coming. Record it so the scheduler can fail + # the request now instead of leaving it to the wait deadline. + # The task dict carries the Request object, not a bare id -- + # `task.get("request_id")` is always None (it was that way in the + # original warning too, which logged "None" for every failure). + # Use the scheduler-side id: that is what `self.requests` is keyed + # by, and what `finish_requests` expects. + failed = getattr(task.get("request"), "request_id", None) + logger.error("Send gave up for %s: %s", failed, e) + self.record_send_failure(failed, f"{type(e).__name__}: {e}") with self._save_cond: if not self._pending_save_reqs and not self.stop_event.is_set(): diff --git a/vllm_omni/distributed/omni_connectors/transfer_adapter/chunk_transfer_adapter.py b/vllm_omni/distributed/omni_connectors/transfer_adapter/chunk_transfer_adapter.py index d722adad2..ae1cb74b9 100644 --- a/vllm_omni/distributed/omni_connectors/transfer_adapter/chunk_transfer_adapter.py +++ b/vllm_omni/distributed/omni_connectors/transfer_adapter/chunk_transfer_adapter.py @@ -2,6 +2,7 @@ # SPDX-FileCopyrightText: Copyright contributors to the vLLM project import importlib +import time from collections import defaultdict, deque from collections.abc import Callable, Mapping from typing import Any @@ -94,6 +95,12 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase): self._held_non_active: deque[Any] = deque() self.requests_num_chunks_sent: dict[str, int] = defaultdict(int) self._pending_streaming_prefills: dict[str, dict] = {} + # Monotonic timestamp of when each request last began waiting for a + # chunk, refreshed every time one arrives. Read by + # collect_timed_out_request_ids() so a stream that stops advancing + # becomes a client-visible error instead of parking forever. Mirrors + # OmniSchedulingCoordinator._waiting_since on the full-payload path. + self._waiting_since: dict[str, float] = {} @staticmethod def _is_truthy_scalar(value: Any) -> bool: @@ -383,6 +390,24 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase): # has been sent successfully. This avoids cleanup->save races. if is_payload_finished: self.cleanup(request.request_id, external_req_id) + else: + # R1.2 of #4855. connector.put returning False is a silent drop: no + # exception, no retry, and the caller's `if success:` block simply + # does not run. /dev/shm exhaustion in SharedMemoryConnector.put is + # one way to get here. Record it so the request fails now rather + # than parking in WAITING_FOR_CHUNK until the deadline. + logger.error( + "Chunk send failed for %s (stage %s -> %s); giving up on this chunk", + external_req_id, + stage_id, + next_stage_id, + ) + # Key on the scheduler-side id. `external_req_id` is the user-facing id + # (InputProcessor renames request_id to an internal UUID and keeps the + # original in external_req_id, see async_omni_engine.py), while + # `self.requests` -- and therefore `finish_requests` -- is keyed by the + # internal one. + self.record_send_failure(request.request_id, "connector.put reported failure") if is_segment_finished: self.code_prompt_token_ids.pop(external_req_id, None) @@ -438,6 +463,7 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase): self._cancelled_load_reqs.add(request_id) self._finished_load_reqs.discard(request_id) + self._waiting_since.pop(request_id, None) @staticmethod def _discard_from_chunk_deque(deque_list: deque[Any], request_id: str) -> None: @@ -579,6 +605,42 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase): self._active_streams[request_id] = request return True + def collect_timed_out_request_ids(self, timeout_s: float) -> set[str]: + """Return IDs whose chunk wait has exceeded *timeout_s*. + + The async-chunk path had no deadline of any kind: a request parks in + ``WAITING_FOR_CHUNK`` and the receiver re-queues it on every failed + poll (``transfer_adapter/base.py``) with no attempt counter, so a + producer that crashed, gave up after its send retries, or simply never + emitted a terminal chunk left the request waiting forever + (vllm-project/vllm-omni#3833). The full-payload path has had a net + since ``OmniSchedulingCoordinator.collect_timed_out_request_ids``; this + is its async-chunk counterpart, and both are driven by the same + ``VLLM_OMNI_INPUT_WAIT_TIMEOUT_S``. + + The clock measures *stall* time, not stream lifetime: it starts when a + request begins waiting for a chunk and resets each time one arrives, so + a long but healthy stream is never failed. + + Clears ``_waiting_since`` for the expired IDs. The caller marks them + ``FINISHED_ERROR`` via the scheduler's ``finish_requests``, which routes + back through this adapter's own ``finish_requests`` and releases the + rest of the per-request state. + """ + if timeout_s <= 0 or not self._waiting_since: + return set() + now = time.monotonic() + timed_out_ids = {req_id for req_id, started in self._waiting_since.items() if now - started > timeout_s} + for req_id in timed_out_ids: + self._waiting_since.pop(req_id, None) + logger.warning( + "[Stage-%s] Request %s timed out waiting for a chunk (stalled > %.0fs)", + self.connector.stage_id, + req_id, + timeout_s, + ) + return timed_out_ids + @property def num_running_waiting_for_chunk(self) -> int: """Count running requests temporarily removed while awaiting a chunk.""" @@ -624,11 +686,15 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase): # Requests that waiting for chunk self.load_async(request) request.status = RequestStatus.WAITING_FOR_CHUNK + self._waiting_since.setdefault(request.request_id, time.monotonic()) else: if request.request_id in finished_load_reqs: request.status = target_status finished_load_reqs.remove(request.request_id) self.requests_with_ready_chunks.add(request.request_id) + # A chunk landed: restart the clock for the next one, so the + # deadline measures stall time rather than total stream time. + self._waiting_since.pop(request.request_id, None) continue queue.remove(request) self.requests_origin_status[request.request_id] = target_status @@ -767,11 +833,15 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase): # Requests that waiting for chunk self.load_async(request) request.status = RequestStatus.WAITING_FOR_CHUNK + self._waiting_since.setdefault(request.request_id, time.monotonic()) else: if request.request_id in finished_load_reqs: request.status = target_status finished_load_reqs.remove(request.request_id) self.requests_with_ready_chunks.add(request.request_id) + # A chunk landed: restart the clock for the next one, so the + # deadline measures stall time rather than total stream time. + self._waiting_since.pop(request.request_id, None) continue queue.remove(request) self.requests_origin_status[request.request_id] = target_status @@ -826,5 +896,6 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase): self.upstream_exhausted_requests.discard(req_id) self._finished_load_reqs.discard(req_id) self._cancelled_load_reqs.add(req_id) + self._waiting_since.pop(req_id, None) return [] diff --git a/vllm_omni/engine/orchestrator.py b/vllm_omni/engine/orchestrator.py index 901bdb980..f112983e1 100644 --- a/vllm_omni/engine/orchestrator.py +++ b/vllm_omni/engine/orchestrator.py @@ -15,6 +15,7 @@ import asyncio import time as _time from collections.abc import Awaitable, Callable from dataclasses import dataclass, field +from http import HTTPStatus from typing import TYPE_CHECKING, Any import janus @@ -51,6 +52,7 @@ from vllm_omni.engine.messages import ( from vllm_omni.engine.orchestrator_monitor import create_orch_monitor, replica_key from vllm_omni.engine.serialization import serialize_additional_information from vllm_omni.engine.stage_pool import StagePool, StageUnavailableError +from vllm_omni.errors import DEFAULT_CLIENT_ERROR_TYPE from vllm_omni.metrics.prometheus import OmniRequestCounter from vllm_omni.metrics.stat_logger import OmniPrometheusStatLogger from vllm_omni.outputs import OmniRequestOutput @@ -221,7 +223,7 @@ class _OrchestratorDuplexStagePort: async_chunk: bool, prewarm_async_chunk_stages: Callable[ [str, Any, OrchestratorRequestState], - Awaitable[None], + Awaitable[bool], ], ) -> None: self._stage_pools = stage_pools @@ -307,11 +309,23 @@ class _OrchestratorDuplexStagePort: else: replica_id = await pool.submit_initial(context.request_id, request_state, request, prompt_text=None) if self._async_chunk and context.stage_id == 0: - await self._prewarm_async_chunk_stages( + prewarmed = await self._prewarm_async_chunk_stages( context.request_id, request, request_state, ) + if not prewarmed: + # The prewarm already failed the request, aborted it, and + # popped its state. Falling through would write fences and + # a submit timestamp onto that orphaned object and + # re-increment the running counter the cleanup just + # released, then hand the control plane a success result + # for a request that no longer exists. Raise instead: + # ``DuplexControlPlane.handle_append`` turns this into an + # error result for the append. + raise RuntimeError( + f"async-chunk prewarm failed for duplex request {context.request_id}; the request was aborted" + ) request_state.duplex_stage_fences[context.stage_id] = context.fence request_state.stage_submit_ts[context.stage_id] = _time.time() if not request_state.running_counter_registered and self._running_counter is not None: @@ -2217,18 +2231,54 @@ class Orchestrator: request_id: str, stage0_request: Any, req_state: OrchestratorRequestState, - ) -> None: - """Pre-submit downstream stages for async-chunk mode.""" + ) -> bool: + """Pre-submit downstream stages for async-chunk mode. + + Returns False when the request was failed and cleaned up in here, so a + caller still holding ``req_state`` stops instead of recording state on + an object the cleanup already popped from ``request_states``. + """ if req_state.final_stage_id <= 0: - return + return True prompt_token_ids = getattr(stage0_request, "prompt_token_ids", None) if prompt_token_ids is None: - logger.warning( - "[Orchestrator] async_chunk prewarm skipped for req=%s: stage0 prompt_token_ids missing", + # R1.3 of #4855. Skipping the prewarm leaves every downstream stage + # unsubmitted, so the request produces no output and no error -- it + # just stops. An embeds-only prompt in async-chunk mode is a config + # mistake, and the client should hear about it now rather than wait + # for the stage-input deadline to notice nothing ever arrived. + # + # Non-fatal, with a 4xx: `fatal=True` means "the engine is dead", and + # the entrypoints act on it -- `OmniBase._handle_output_message` + # raises `OmniEngineDeadError` out of `OmniLLM.generate`'s batch loop, + # failing every other request in that batch, and the serving layer + # calls `terminate_if_errored`. One client's bad prompt must not do + # either. This is the same shape `_handle_stage_error` uses for a + # request-scoped client error. + logger.error( + "[Orchestrator] req=%s: async_chunk prewarm needs stage0 prompt_token_ids " + "and none were provided; failing the request", request_id, ) - return + await self.output_async_queue.put( + ErrorMessage( + error=( + "async_chunk requires prompt_token_ids on the stage-0 request; " + "an embeds-only prompt cannot prewarm downstream stages" + ), + status_code=HTTPStatus.BAD_REQUEST.value, + error_type=DEFAULT_CLIENT_ERROR_TYPE, + request_id=request_id, + stage_id=0, + ) + ) + await self._cleanup_request_ids( + [request_id, *self._cfg_tracker.cleanup_parent(request_id)], + abort=True, + close_duplex_sessions=True, + ) + return False for next_stage_id in range(1, req_state.final_stage_id + 1): next_pool = self.stage_pools[next_stage_id] @@ -2303,7 +2353,7 @@ class Orchestrator: # Attribute the failure to the stage that failed, not stage 0; the # request is already cleaned up, so stop prewarming. if not submitted: - return + return False # The guard swallows submit_initial's return value; the pool binding # it recorded carries the same replica. @@ -2329,6 +2379,8 @@ class Orchestrator: tx_ms=_tx_ms, ) + return True + def _build_kv_sender_info( self, sender_stage_ids: list[int], diff --git a/vllm_omni/worker/omni_connector_model_runner_mixin.py b/vllm_omni/worker/omni_connector_model_runner_mixin.py index 07a0327a6..70e9dafd2 100644 --- a/vllm_omni/worker/omni_connector_model_runner_mixin.py +++ b/vllm_omni/worker/omni_connector_model_runner_mixin.py @@ -406,11 +406,13 @@ class OmniConnectorModelRunnerMixin: # thread by one execute_model() cycle, especially when the request was # added after the current scheduler_output snapshot. # - # Orphaned pending recv entries (e.g. from upstream stage crash) - # are handled by OmniSchedulingCoordinator.collect_timed_out_request_ids() - # which detects wait-time violations. The scheduler then removes the - # request from its queues, sets FINISHED_ERROR, and calls _free_request() - # which ultimately triggers cleanup_finished_request() here. + # Orphaned pending recv entries (e.g. from upstream stage crash) are + # handled by collect_timed_out_request_ids() -- on + # OmniSchedulingCoordinator for full-payload requests, and on + # OmniChunkTransferAdapter for async-chunk ones -- which detect + # wait-time violations. The scheduler then removes the request from + # its queues, sets FINISHED_ERROR, and calls _free_request() which + # ultimately triggers cleanup_finished_request() here. for attr_name in ( "_request_ids_mapping", "_get_req_chunk",