[Bugfix] Bound omni benchmark per-request timeout to 15 min by default (#7130)

Signed-off-by: tly <2200895168@qq.com>

tests/benchmarks/patch/test_patch.py

@@ -27,6 +27,8 @@ from vllm_omni.benchmarks.patch.patch import (
     _apply_stage0_token_timings,
     _apply_video_metrics_from_payload,
     _attach_seed_tts_to_request_func_input,
+    _build_benchmark_session,
+    _omni_request_timeout_s,
     async_request_openai_chat_omni_completions,
     async_request_openai_image_edits_omni,
     async_request_openai_image_generations_omni,
@@ -1201,6 +1203,71 @@ async def test_prompt_len_assigned_from_usage(mocker: MockerFixture):
     )
 
 
+class TestOmniRequestTimeout:
+    """``--omni-request-timeout-s`` precedence: explicit value > 900 s default."""
+
+    _OVERRIDE = "vllm_omni.benchmarks.patch.patch._REQUEST_TIMEOUT_OVERRIDE_S"
+
+    def test_default_timeout_is_900s_when_not_set(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(self._OVERRIDE, None)
+        assert _omni_request_timeout_s() == 900.0
+
+    def test_explicit_value_wins_over_default(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(self._OVERRIDE, 123.5)
+        assert _omni_request_timeout_s() == 123.5
+
+    def test_non_positive_restores_legacy_6h_cap(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(self._OVERRIDE, 0)
+        assert _omni_request_timeout_s() == 6 * 60 * 60.0
+
+    async def test_benchmark_session_uses_configured_timeout(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(self._OVERRIDE, 42.0)
+        session = _build_benchmark_session(max_concurrency=8, ssl_setting=False)
+        try:
+            assert session.timeout.total == 42.0
+            assert session.connector.limit == 8
+            assert session.connector.limit_per_host == 8
+        finally:
+            await session.close()
+
+    async def test_hung_server_request_times_out_as_failed(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """A server that accepts the request but never responds must surface as ``failed``."""
+
+        async def handler(reader, writer):
+            # Drain the request but never respond — the original hang failure mode.
+            try:
+                while await reader.read(4096):
+                    pass
+            except ConnectionResetError:
+                pass
+            finally:
+                # Close our side of the socket: the client aborts after the
+                # timeout and only sends FIN, so without writer.close() the
+                # half-closed connection keeps Server.wait_closed() (which on
+                # Python 3.12+ waits for every accepted connection to drop)
+                # blocked forever.
+                writer.close()
+                try:
+                    await writer.wait_closed()
+                except (ConnectionResetError, BrokenPipeError):
+                    pass
+
+        monkeypatch.setattr(self._OVERRIDE, 1.0)
+        server = await asyncio.start_server(handler, "127.0.0.1", 0)
+        port = server.sockets[0].getsockname()[1]
+        session = _build_benchmark_session(max_concurrency=1, ssl_setting=False)
+        try:
+            request_input = _seed_tts_request_func_input()
+            request_input.api_url = f"http://127.0.0.1:{port}/v1/chat/completions"
+            output = await async_request_openai_chat_omni_completions(request_input, session)
+            assert output.success is False
+            assert output.error
+        finally:
+            await session.close()
+            server.close()
+            await server.wait_closed()
+
+
 def test_video_rtf_prefers_generation_time_over_poll_latency():
     """RTF must not grow with client poll_interval overshoot in output.latency."""
     payload = {

tests/benchmarks/test_serve_cli.py

@@ -379,3 +379,17 @@ def test_bench_serve_cli_mocks_http_request(tmp_path: Path):
     )
     assert bench_requests
     assert all(url == expected_url for url in bench_requests), f"Unexpected target URLs: {bench_requests}"
+
+
+def test_omni_request_timeout_s_flag_defaults_and_parses() -> None:
+    parser = TrackingArgumentParser()
+    add_omni_args(parser)
+
+    args = parser.parse_args([])
+    assert args.omni_request_timeout_s is None
+
+    args = parser.parse_args(["--omni-request-timeout-s", "0"])
+    assert args.omni_request_timeout_s == 0.0
+
+    args = parser.parse_args(["--omni-request-timeout-s", "3600"])
+    assert args.omni_request_timeout_s == 3600.0

vllm_omni/benchmarks/patch/patch.py

@@ -211,6 +211,61 @@ def _seed_tts_capture_pcm_for_wer() -> bool:
     )
 
 
+_DEFAULT_REQUEST_TIMEOUT_S = 900.0
+_LEGACY_REQUEST_TIMEOUT_S = 6 * 60 * 60.0
+
+# Set from the ``--omni-request-timeout-s`` CLI flag by ``vllm bench serve``
+# before the benchmark session is built (``None`` = use the default above).
+_REQUEST_TIMEOUT_OVERRIDE_S: float | None = None
+
+
+def set_request_timeout_s(value: float) -> None:
+    """Record the explicitly requested per-request timeout (from the CLI)."""
+    global _REQUEST_TIMEOUT_OVERRIDE_S
+    _REQUEST_TIMEOUT_OVERRIDE_S = float(value)
+
+
+def _omni_request_timeout_s() -> float:
+    """Per-request total timeout for the shared benchmark ``aiohttp`` session.
+
+    An explicit ``--omni-request-timeout-s`` value wins over the 900 s default;
+    ``<= 0`` restores the legacy 6 h cap. A bounded per-request timeout makes a
+    hung server surface as ``failed`` requests once the deadline fires instead
+    of pinning the benchmark slot indefinitely.
+    """
+    value = _REQUEST_TIMEOUT_OVERRIDE_S
+    if value is None:
+        return _DEFAULT_REQUEST_TIMEOUT_S
+    if value <= 0:
+        return _LEGACY_REQUEST_TIMEOUT_S
+    return value
+
+
+def _build_benchmark_session(
+    max_concurrency: int | None,
+    ssl_setting: ssl.SSLContext | bool,
+) -> aiohttp.ClientSession:
+    """Build the session shared by every benchmark request.
+
+    Connections are reused across requests to reduce TLS handshake overhead;
+    the per-request total timeout comes from ``_omni_request_timeout_s()``.
+    """
+    connector = aiohttp.TCPConnector(
+        limit=max_concurrency or 0,
+        limit_per_host=max_concurrency or 0,
+        ttl_dns_cache=300,
+        use_dns_cache=True,
+        enable_cleanup_closed=True,
+        force_close=True,
+        ssl=ssl_setting,
+    )
+    return aiohttp.ClientSession(
+        connector=connector,
+        trust_env=True,
+        timeout=aiohttp.ClientTimeout(total=_omni_request_timeout_s()),
+    )
+
+
 def _merge_extra_body_mm_kwargs(base: dict | None, overlay: dict | None) -> dict | None:
     """Shallow-merge ``extra_body`` dicts; deep-merge ``mm_processor_kwargs`` if both set."""
     if not base and not overlay:
@@ -2616,21 +2671,8 @@ async def benchmark(
 
     # Reuses connections across requests to reduce TLS handshake overhead.
     ssl_setting = ssl_context if ssl_context is not None else ("https://" in api_url)
-    connector = aiohttp.TCPConnector(
-        limit=max_concurrency or 0,
-        limit_per_host=max_concurrency or 0,
-        ttl_dns_cache=300,
-        use_dns_cache=True,
-        enable_cleanup_closed=True,
-        force_close=True,
-        ssl=ssl_setting,
-    )
-
-    session = aiohttp.ClientSession(
-        connector=connector,
-        trust_env=True,
-        timeout=aiohttp.ClientTimeout(total=6 * 60 * 60),
-    )
+    session = _build_benchmark_session(max_concurrency, ssl_setting)
+    print(f"Per-request timeout: {_omni_request_timeout_s():g}s")
 
     print("Starting initial single prompt test run...")
     test_prompt, test_prompt_len, test_output_len, test_mm_content = (

vllm_omni/benchmarks/serve.py

@@ -20,6 +20,7 @@ from vllm_omni.benchmarks.omniinteract import omniinteract_output_lock
 from vllm_omni.benchmarks.patch.patch import (
     maybe_enable_stage_metrics,
     set_print_stage,
+    set_request_timeout_s,
     should_request_stage_metrics,
 )
 
@@ -62,6 +63,9 @@ def main(args: argparse.Namespace) -> dict[str, Any]:
         os.environ["SEED_TTS_WER_SAVE_ITEMS"] = "1"
     if getattr(args, "daily_omni_save_eval_items", False):
         os.environ["DAILY_OMNI_SAVE_EVAL_ITEMS"] = "1"
+    request_timeout_s = getattr(args, "omni_request_timeout_s", None)
+    if request_timeout_s is not None:
+        set_request_timeout_s(request_timeout_s)
     _use_endpoint_backend_when_implicit(args)
     set_print_stage(getattr(args, "print_stage", False))
     args.extra_body = maybe_enable_stage_metrics(

vllm_omni/entrypoints/cli/benchmark/cli_args.py

@@ -92,6 +92,16 @@ def add_multi_stage_cli_args(parser: argparse.ArgumentParser) -> None:
             "and tpop controls both text TPOP and internal stream TPOP."
         ),
     )
+    group.add_argument(
+        "--omni-request-timeout-s",
+        type=float,
+        default=None,
+        help=(
+            "Total per-request timeout in seconds for benchmark HTTP requests. Timed-out requests are "
+            "recorded as failed instead of stalling the run on a hung server. Defaults to 900 when "
+            "omitted; set <= 0 to restore the legacy 6 h cap."
+        ),
+    )
 
 
 def add_diffusion_cli_args(parser: argparse.ArgumentParser) -> None: