[Bugfix] Fix video benchmark timeout and failure accounting (#7259)

Signed-off-by: Alicia <115451386+congw729@users.noreply.github.com>

benchmarks/diffusion/README.md

@@ -17,18 +17,18 @@ The main entrypoint is:
 vllm serve Qwen/Qwen-Image --omni --port 8099
 ```
 
-2. Run a minimal benchmark:
+1. Run a minimal benchmark:
 
 ```bash
 python3 benchmarks/diffusion/diffusion_benchmark_serving.py \
-	--base-url http://localhost:8099 \
-	--model Qwen/Qwen-Image \
-	--task t2i \
-	--dataset vbench \
-	--num-prompts 5
+ --base-url http://localhost:8099 \
+ --model Qwen/Qwen-Image \
+ --task t2i \
+ --dataset vbench \
+ --num-prompts 5
 ```
 
-**Notes**
+### Notes
 
 - By default, image tasks talk to `http://<host>:<port>/v1/chat/completions`; video tasks talk to `/v1/videos`.
 - If you run the server on another host or port, pass `--base-url` accordingly.
@@ -52,14 +52,14 @@ Example (`t2v`):
 
 ```bash
 python3 benchmarks/diffusion/diffusion_benchmark_serving.py \
-	--base-url http://localhost:8099 \
-	--model Wan-AI/Wan2.2-T2V-A14B-Diffusers \
-	--task t2v \
-	--dataset vbench \
-	--num-prompts 50 \
-	--width 640 --height 480 \
-	--num-frames 81 --fps 16 \
-	--num-inference-steps 40
+ --base-url http://localhost:8099 \
+ --model Wan-AI/Wan2.2-T2V-A14B-Diffusers \
+ --task t2v \
+ --dataset vbench \
+ --num-prompts 50 \
+ --width 640 --height 480 \
+ --num-frames 81 --fps 16 \
+ --num-inference-steps 40
 ```
 
 Note: `vbench` can also be used for other tasks such as `t2i` / `i2v` (and `i2i`). For `t2i`, the loader reuses VBench t2v text prompts; for `i2v` / `i2i`, it loads the VBench i2v dataset (with image paths).
@@ -140,6 +140,21 @@ Traffic / concurrency flags:
 - `--request-rate`: Target request rate (requests/second). If set to `inf`, the script sends all requests immediately.
 - `--max-concurrency`: Max number of in-flight requests (default: `1`). This can hard-cap the achieved QPS: if it is too small, requests will queue behind the semaphore, and both achieved throughput and observed SLO attainment can be skewed.
 
+### Video job timeout and failures
+
+`--video-job-timeout` controls how long each `/v1/videos` job may be polled,
+including time queued on the server (default: `900` seconds). For long video
+jobs at high concurrency, use a larger budget, for example
+`--video-job-timeout 1800`. This also applies to warmup jobs. The polling budget
+starts after job creation; it excludes time waiting for the client concurrency
+semaphore. Individual HTTP calls still use the aiohttp session timeout.
+
+An expired job is counted as failed and deleted, which cancels queued or running
+server generation. The benchmark continues processing the remaining requests.
+The progress bar counts both successful and failed requests. The final report
+shows the failure count and the first 10 errors; `--output-file` saves all errors
+with their request IDs in `request_errors`.
+
 ### Batched warmup note
 
 For batched serving runs, warm the same in-flight shape you plan to measure.

benchmarks/diffusion/backends.py

@@ -1,3 +1,6 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
 import asyncio
 import base64
 import json
@@ -43,6 +46,7 @@ class RequestFuncInput:
     video_paths: list[str] | None = None
     request_id: str = field(default_factory=lambda: str(uuid.uuid4()))
     default_bot_task: str | None = DEFAULT_EDITS_BOT_TASK
+    video_job_timeout: float = 900.0
 
 
 @dataclass
@@ -331,64 +335,64 @@ async def async_request_v1_videos(
     output = RequestFuncOutput()
     output.start_time = time.perf_counter()
 
-    files = dict(input.extra_body)
-    if input.prompt:
-        files.setdefault("prompt", input.prompt)
-    if input.width and input.height:
-        files.setdefault("height", input.height)
-        files.setdefault("width", input.width)
-    if input.num_frames:
-        files.setdefault("num_frames", input.num_frames)
-    if input.num_inference_steps:
-        files.setdefault("num_inference_steps", input.num_inference_steps)
-    if input.seed is not None:
-        files.setdefault("seed", input.seed)
-    if input.fps:
-        files.setdefault("fps", input.fps)
-
-    form = aiohttp.FormData()
-    for k, v in files.items():
-        form.add_field(k, str(v))
-
     image_file = None
-    if input.image_paths and input.video_paths:
-        output.error = "Only one of image_paths or video_paths can be provided"
-        output.success = False
-        return output
+    job_id = None
+    try:
+        files = dict(input.extra_body)
+        if input.prompt:
+            files.setdefault("prompt", input.prompt)
+        if input.width and input.height:
+            files.setdefault("height", input.height)
+            files.setdefault("width", input.width)
+        if input.num_frames:
+            files.setdefault("num_frames", input.num_frames)
+        if input.num_inference_steps:
+            files.setdefault("num_inference_steps", input.num_inference_steps)
+        if input.seed is not None:
+            files.setdefault("seed", input.seed)
+        if input.fps:
+            files.setdefault("fps", input.fps)
 
-    if input.image_paths and len(input.image_paths) > 0:
-        image_path = input.image_paths[0]
-        image_file = open(image_path, "rb")
-        form.add_field(
-            "input_reference",
-            image_file,
-            filename=os.path.basename(image_path),
-            content_type=_guess_mime_type(image_path),
-        )
-    elif input.video_paths and len(input.video_paths) > 0:
-        video_path = input.video_paths[0]
-        if _is_minimax_h3_mixed_reference(input.model):
-            with open(video_path, "rb") as reference_file:
-                video_b64 = base64.b64encode(reference_file.read()).decode("ascii")
-            form.add_field(
-                "video_reference",
-                json.dumps([{"video_url": f"data:video/mp4;base64,{video_b64}"}]),
-            )
-        else:
-            image_file = open(video_path, "rb")
+        form = aiohttp.FormData()
+        for k, v in files.items():
+            form.add_field(k, str(v))
+
+        if input.image_paths and input.video_paths:
+            output.error = "Only one of image_paths or video_paths can be provided"
+            output.success = False
+            return output
+
+        if input.image_paths and len(input.image_paths) > 0:
+            image_path = input.image_paths[0]
+            image_file = open(image_path, "rb")
             form.add_field(
                 "input_reference",
                 image_file,
-                filename=os.path.basename(video_path),
-                content_type=_guess_mime_type(video_path),
+                filename=os.path.basename(image_path),
+                content_type=_guess_mime_type(image_path),
             )
+        elif input.video_paths and len(input.video_paths) > 0:
+            video_path = input.video_paths[0]
+            if _is_minimax_h3_mixed_reference(input.model):
+                with open(video_path, "rb") as reference_file:
+                    video_b64 = base64.b64encode(reference_file.read()).decode("ascii")
+                form.add_field(
+                    "video_reference",
+                    json.dumps([{"video_url": f"data:video/mp4;base64,{video_b64}"}]),
+                )
+            else:
+                image_file = open(video_path, "rb")
+                form.add_field(
+                    "input_reference",
+                    image_file,
+                    filename=os.path.basename(video_path),
+                    content_type=_guess_mime_type(video_path),
+                )
 
-    job_id = None
-    job_status = None
-    poll_json = {}
-    resp_json = {}
+        job_status = None
+        poll_json = {}
+        resp_json = {}
 
-    try:
         # invoke a post request (POST /v1/videos)
         async with session.post(input.api_url, data=form) as response:
             if response.status == 200:
@@ -406,7 +410,7 @@ async def async_request_v1_videos(
 
         # invoke a poll request (GET /v1/videos/{video_id})
         poll_interval = 2.0  # Unit(s)
-        timeout_seconds = 600.0
+        timeout_seconds = input.video_job_timeout
         deadline = time.perf_counter() + timeout_seconds
         job_url = f"{input.api_url}/{job_id}"
 
@@ -422,8 +426,11 @@ async def async_request_v1_videos(
                 poll_json = await poll_response.json()
                 job_status = poll_json.get("status")
 
-                if time.perf_counter() >= deadline:
-                    output.error = f"Timed out waiting for video job {job_id} to complete."
+                if job_status not in {"completed", "failed"} and time.perf_counter() >= deadline:
+                    output.error = (
+                        f"Timed out after {timeout_seconds:g}s waiting for video job {job_id} to complete. "
+                        "Increase --video-job-timeout for long-running or queued video jobs."
+                    )
                     output.success = False
                     return output
 
@@ -452,7 +459,7 @@ async def async_request_v1_videos(
             elif "peak_memory_mb" in resp_json:
                 output.peak_memory_mb = resp_json["peak_memory_mb"]
     except Exception as e:
-        output.error = str(e)
+        output.error = f"{type(e).__name__}: {e}"
         output.success = False
     finally:
         if image_file is not None:
@@ -465,13 +472,13 @@ async def async_request_v1_videos(
             except Exception as e:
                 print(f"Failed to clean up video job {job_id}: {e}")
 
-    output.latency = time.perf_counter() - output.start_time
+        output.latency = time.perf_counter() - output.start_time
+        if pbar is not None:
+            pbar.update(1)
 
     if output.success and input.slo_ms is not None:
         output.slo_achieved = (output.latency * 1000.0) <= float(input.slo_ms)
 
-    if pbar:
-        pbar.update(1)
     return output
 
 

benchmarks/diffusion/diffusion_benchmark_serving.py

@@ -1,6 +1,6 @@
 # adapted from fastvideo
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 
 """
 Benchmark online serving for diffusion models (Image/Video Generation).
@@ -87,6 +87,7 @@ import base64
 import glob
 import json
 import logging
+import math
 import os
 import random
 import tempfile
@@ -1116,6 +1117,11 @@ def calculate_metrics(
         "duration": total_duration,
         "completed_requests": num_success,
         "failed_requests": len(error_outputs),
+        "request_errors": [
+            {"request_id": req.request_id, "error": out.error}
+            for req, out in zip(requests_list, outputs)
+            if not out.success
+        ],
         "throughput_qps": num_success / total_duration if total_duration > 0 else 0,
         "latency_mean": np.mean(latencies) if latencies else 0,
         "latency_median": np.median(latencies) if latencies else 0,
@@ -1261,6 +1267,9 @@ def _default_endpoint_for_task(task: str) -> str:
 
 
 async def benchmark(args):
+    if not math.isfinite(args.video_job_timeout) or args.video_job_timeout <= 0:
+        raise ValueError("--video-job-timeout must be a positive finite number of seconds.")
+
     # Construct base_url if not provided
     if args.base_url is None:
         args.base_url = f"http://{args.host}:{args.port}"
@@ -1312,6 +1321,8 @@ async def benchmark(args):
     print("Loading requests...")
     requests_list = dataset.get_requests()
     print(f"Prepared {len(requests_list)} requests from {args.dataset} dataset.")
+    for req in requests_list:
+        req.video_job_timeout = args.video_job_timeout
 
     if args.return_stage_metrics and args.endpoint in _STAGE_METRICS_ENDPOINTS:
         for req in requests_list:
@@ -1376,6 +1387,8 @@ async def benchmark(args):
     metrics["model"] = args.model
     metrics["dataset"] = args.dataset
     metrics["task"] = args.task
+    if args.endpoint == "/v1/videos":
+        metrics["video_job_timeout"] = args.video_job_timeout
     if args.endpoint == "/v1/images/edits":
         metrics["bot_task"] = args.bot_task
 
@@ -1398,6 +1411,11 @@ async def benchmark(args):
         )
     )
     print("{:<40} {}/{:<15}".format("Successful requests:", metrics["completed_requests"], len(requests_list)))
+    print("{:<40} {:<15}".format("Failed requests:", metrics["failed_requests"]))
+    for failure in metrics["request_errors"][:10]:
+        print(f"Request {failure['request_id']} failed: {failure['error']}")
+    if metrics["failed_requests"] > 10:
+        print("Showing first 10 failures; use --output-file to save all request errors.")
 
     # Section 3: Performance Metrics
     print(f"{'-' * 50}")
@@ -1481,6 +1499,13 @@ if __name__ == "__main__":
         help="Path to local dataset file (optional).",
     )
     parser.add_argument("--num-prompts", type=int, default=10, help="Number of prompts to benchmark.")
+    parser.add_argument(
+        "--video-job-timeout",
+        type=float,
+        default=900.0,
+        help="Maximum polling time in seconds per video job, including server queue time (default: 900). "
+        "Increase this for long-running video generation or high concurrency.",
+    )
     parser.add_argument(
         "--max-concurrency",
         type=int,