diff --git a/recipes/MiniMaxAI/MiniMax-H3.md b/recipes/MiniMaxAI/MiniMax-H3.md index d022cce51..9b306bc2f 100644 --- a/recipes/MiniMaxAI/MiniMax-H3.md +++ b/recipes/MiniMaxAI/MiniMax-H3.md @@ -69,11 +69,12 @@ reference-video preparation and MP4 output. For CUDA and ROCm deployments, non-streaming MP4 responses are encoded on the host CPU through PyAV/libx264 after generation. The response encoder selects the -path automatically at runtime: inputs with a supported frame shape, common dtype, -and RGB per-channel contiguous layout use direct planar frames; other inputs -fall back to the legacy path before the PyAV container is opened. No CLI flag, -model declaration, or user configuration is required. Streaming fMP4 output is -unchanged. +path automatically at runtime. The server-owned parallel converter accepts +supported frame shapes and dtypes with either per-channel-contiguous or strided +RGB planes, including interleaved arrays materialized by output transport. +Standalone callers without a parallel converter retain the legacy fallback for +strided planes. No CLI flag, model declaration, or user configuration is +required. Streaming fMP4 output is unchanged. A community benchmark on 2x Xeon 8480C reported the following comparison between the legacy and direct planar paths diff --git a/tests/benchmarks/benchmark_video_response_encoding.py b/tests/benchmarks/benchmark_video_response_encoding.py index 575d13664..ec589e734 100644 --- a/tests/benchmarks/benchmark_video_response_encoding.py +++ b/tests/benchmarks/benchmark_video_response_encoding.py @@ -1,5 +1,5 @@ # SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project """Benchmark legacy and automatic MP4 response encoding on synthetic input.""" from __future__ import annotations @@ -11,13 +11,16 @@ import statistics import time from collections.abc import Callable from dataclasses import dataclass +from functools import partial from pathlib import Path +from typing import cast import numpy as np from vllm_omni.entrypoints.openai.video_api_utils import ( _encode_video_bytes, _encode_video_bytes_legacy, + _PlanarFrameConverter, ) @@ -42,12 +45,19 @@ def _build_inputs( fps: int, audio_sample_rate: int, seed: int, + input_layout: str = "planar-backed", ) -> tuple[np.ndarray, np.ndarray]: rng = np.random.default_rng(seed) planar = rng.random((3, frames, height, width), dtype=np.float32) video = planar.transpose(1, 2, 3, 0) - if not all(video[..., channel].flags.c_contiguous for channel in range(3)): + if input_layout == "interleaved": + video = np.ascontiguousarray(video) + if not video.flags.c_contiguous: + raise RuntimeError("synthetic video is not C-contiguous") + elif input_layout == "planar-backed" and not all(video[..., channel].flags.c_contiguous for channel in range(3)): raise RuntimeError("synthetic video does not expose contiguous channel planes") + elif input_layout != "planar-backed": + raise ValueError(f"unsupported input layout: {input_layout}") sample_count = max(1, round(frames / fps * audio_sample_rate)) timeline = np.arange(sample_count, dtype=np.float32) / audio_sample_rate @@ -86,8 +96,8 @@ def _measure( def _summarize(records: list[dict[str, object]], label: str) -> dict[str, object]: selected = [record for record in records if record["label"] == label] - wall_values = [float(record["wall_ms"]) for record in selected] - cpu_values = [float(record["process_cpu_ms"]) for record in selected] + wall_values = [cast(float, record["wall_ms"]) for record in selected] + cpu_values = [cast(float, record["process_cpu_ms"]) for record in selected] return { "runs": len(selected), "wall_ms": { @@ -113,6 +123,16 @@ def _parse_args() -> argparse.Namespace: parser.add_argument("--warmup", type=int, default=1) parser.add_argument("--rounds", type=_positive_int, default=3) parser.add_argument("--seed", type=int, default=20260817) + parser.add_argument( + "--input-layout", + choices=("planar-backed", "interleaved"), + default="planar-backed", + ) + parser.add_argument( + "--frame-conversion-workers", + type=_positive_int, + default=8, + ) parser.add_argument("--output-json", type=Path) args = parser.parse_args() if args.warmup < 0: @@ -129,33 +149,41 @@ def main() -> None: fps=args.fps, audio_sample_rate=args.audio_sample_rate, seed=args.seed, + input_layout=args.input_layout, ) baseline = BenchmarkVariant("legacy", _encode_video_bytes_legacy) - candidate = BenchmarkVariant("automatic", _encode_video_bytes) + converter = _PlanarFrameConverter(max_workers=args.frame_conversion_workers) + candidate = BenchmarkVariant( + "automatic", + partial(_encode_video_bytes, frame_converter=converter), + ) - for _ in range(args.warmup): - for variant in (baseline, candidate): - _measure( - variant, - video=video, - audio=audio, - fps=args.fps, - audio_sample_rate=args.audio_sample_rate, - ) + try: + for _ in range(args.warmup): + for variant in (baseline, candidate): + _measure( + variant, + video=video, + audio=audio, + fps=args.fps, + audio_sample_rate=args.audio_sample_rate, + ) - records: list[dict[str, object]] = [] - for round_index in range(args.rounds): - order = (baseline, candidate) if round_index % 2 == 0 else (candidate, baseline) - for variant in order: - _, record = _measure( - variant, - video=video, - audio=audio, - fps=args.fps, - audio_sample_rate=args.audio_sample_rate, - ) - record["round"] = round_index + 1 - records.append(record) + records: list[dict[str, object]] = [] + for round_index in range(args.rounds): + order = (baseline, candidate) if round_index % 2 == 0 else (candidate, baseline) + for variant in order: + _, record = _measure( + variant, + video=video, + audio=audio, + fps=args.fps, + audio_sample_rate=args.audio_sample_rate, + ) + record["round"] = round_index + 1 + records.append(record) + finally: + converter.shutdown() output_hashes = {str(record["output_sha256"]) for record in records} if len(output_hashes) != 1: @@ -165,8 +193,8 @@ def main() -> None: paired_savings_percent = [] for round_index in range(args.rounds): pair = [record for record in records if record["round"] == round_index + 1] - baseline_ms = float(next(record["wall_ms"] for record in pair if record["label"] == baseline.label)) - candidate_ms = float(next(record["wall_ms"] for record in pair if record["label"] == candidate.label)) + baseline_ms = cast(float, next(record["wall_ms"] for record in pair if record["label"] == baseline.label)) + candidate_ms = cast(float, next(record["wall_ms"] for record in pair if record["label"] == candidate.label)) paired_savings_ms.append(baseline_ms - candidate_ms) paired_savings_percent.append((baseline_ms - candidate_ms) / baseline_ms * 100) @@ -180,6 +208,8 @@ def main() -> None: "warmup": args.warmup, "rounds": args.rounds, "seed": args.seed, + "input_layout": args.input_layout, + "frame_conversion_workers": args.frame_conversion_workers, "video_shape": list(video.shape), "video_strides": list(video.strides), "codec_options": {"preset": "ultrafast", "threads": "0"}, diff --git a/tests/entrypoints/openai_api/test_video_api_utils.py b/tests/entrypoints/openai_api/test_video_api_utils.py index 80deb445b..be27c4c97 100644 --- a/tests/entrypoints/openai_api/test_video_api_utils.py +++ b/tests/entrypoints/openai_api/test_video_api_utils.py @@ -1,10 +1,11 @@ # SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project """Unit tests for OpenAI-compatible video API encoding helpers.""" import base64 import threading from io import BytesIO +from typing import Any import av import httpx @@ -41,7 +42,7 @@ def _install_http_transport(monkeypatch, handler, client_kwargs): @pytest.mark.asyncio async def test_decode_image_url_follows_redirects_when_allowed(monkeypatch): requested_paths = [] - client_kwargs = [] + client_kwargs: list[dict[str, object]] = [] def _handler(request): requested_paths.append(request.url.path) @@ -63,7 +64,7 @@ async def test_decode_image_url_follows_redirects_when_allowed(monkeypatch): @pytest.mark.asyncio async def test_decode_image_url_rejects_redirects_when_disabled(monkeypatch): requested_paths = [] - client_kwargs = [] + client_kwargs: list[dict[str, object]] = [] def _handler(request): requested_paths.append(request.url.path) @@ -81,7 +82,7 @@ async def test_decode_image_url_rejects_redirects_when_disabled(monkeypatch): @pytest.mark.asyncio async def test_decode_image_url_reports_http_status(monkeypatch): - client_kwargs = [] + client_kwargs: list[dict[str, object]] = [] def _handler(request): return httpx.Response(404) @@ -95,7 +96,7 @@ async def test_decode_image_url_reports_http_status(monkeypatch): @pytest.mark.asyncio async def test_decode_image_url_reports_connection_failure(monkeypatch): - client_kwargs = [] + client_kwargs: list[dict[str, object]] = [] def _handler(request): raise httpx.ConnectError("connection refused", request=request) @@ -141,7 +142,7 @@ def _install_fake_video_mux(monkeypatch, mux_calls): def test_encode_video_bytes_exports_frames_without_interpolation(monkeypatch): - mux_calls = [] + mux_calls: list[dict[str, Any]] = [] _install_fake_video_mux(monkeypatch, mux_calls) frames = [np.full((2, 2, 3), fill_value=i / 5, dtype=np.float32) for i in range(5)] @@ -549,6 +550,44 @@ def test_prepared_automatic_fallback_preserves_legacy_quantization(monkeypatch, np.testing.assert_array_equal(mux_inputs[0], reference) +@pytest.mark.parametrize( + ("worker_count", "expected_path", "expected_reason"), + [ + (None, "legacy_fallback", "non_contiguous_rgb_planes"), + (1, "legacy_fallback", "non_contiguous_rgb_planes"), + (8, "direct_planar", None), + ], +) +def test_interleaved_video_routing(monkeypatch, worker_count, expected_path, expected_reason): + routes = [] + + def fake_planar_mux(*args, **kwargs): + return b"planar-video" + + def fake_compat_mux(frames, audio, **kwargs): + return b"legacy-video" + + monkeypatch.setattr(media_utils, "mux_av_video_audio_bytes", fake_planar_mux) + monkeypatch.setattr(media_utils, "mux_video_audio_bytes", fake_compat_mux) + monkeypatch.setattr(video_api_utils, "_log_video_encoding_path", lambda **kwargs: routes.append(kwargs)) + + converter = None if worker_count is None else video_api_utils._PlanarFrameConverter(max_workers=worker_count) + try: + encoded = video_api_utils._encode_video_bytes( + np.zeros((2, 4, 6, 3), dtype=np.float32), + fps=12, + frame_converter=converter, + ) + finally: + if converter is not None: + converter.shutdown() + + assert encoded == (b"planar-video" if expected_path == "direct_planar" else b"legacy-video") + assert len(routes) == 1 + assert routes[0]["selected_path"] == expected_path + assert routes[0].get("reason") == expected_reason + + @pytest.mark.parametrize( ("video", "reason"), [ @@ -690,7 +729,8 @@ def test_planar_bool_frames_match_bounded_compatible_output(): ) def test_planar_frame_rejects_undersized_plane(monkeypatch, plane_height, line_size): class FakePlane: - pass + height: int + line_size: int plane = FakePlane() plane.height = plane_height @@ -758,6 +798,32 @@ def test_direct_and_compatible_paths_produce_equivalent_mp4(with_audio): np.testing.assert_array_equal(direct_frames, compatible_frames) +@pytest.mark.parametrize("with_audio", [False, True]) +def test_parallel_interleaved_and_legacy_paths_produce_equivalent_mp4(with_audio): + rng = np.random.default_rng(7) + video = rng.random((3, 32, 32, 3), dtype=np.float32) + audio = np.zeros((2, 2400), dtype=np.float32) if with_audio else None + kwargs = { + "fps": 12, + "audio": audio, + "audio_sample_rate": 24000, + "video_codec_options": {"preset": "ultrafast", "threads": "1"}, + } + + converter = video_api_utils._PlanarFrameConverter(max_workers=8) + try: + parallel_bytes = video_api_utils._encode_video_bytes( + video, + **kwargs, + frame_converter=converter, + ) + finally: + converter.shutdown() + legacy_bytes = video_api_utils._encode_video_bytes_legacy(video, **kwargs) + + assert parallel_bytes == legacy_bytes + + def test_mux_closes_container_and_preserves_generator_error(monkeypatch): class GeneratorError(RuntimeError): pass diff --git a/vllm_omni/entrypoints/openai/video_api_utils.py b/vllm_omni/entrypoints/openai/video_api_utils.py index 81c7540ba..dcaf2451b 100644 --- a/vllm_omni/entrypoints/openai/video_api_utils.py +++ b/vllm_omni/entrypoints/openai/video_api_utils.py @@ -1,5 +1,5 @@ # SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project """ Shared helper utilities for OpenAI-compatible video generation API. """ @@ -15,7 +15,7 @@ from collections import deque from collections.abc import Generator from concurrent.futures import Future, ThreadPoolExecutor from io import BytesIO -from typing import TYPE_CHECKING, Any, Literal +from typing import TYPE_CHECKING, Any, Literal, TypeAlias import httpx import numpy as np @@ -50,9 +50,9 @@ logger = init_logger(__name__) DEFAULT_AUDIO_SAMPLE_RATE = 24_000 -VideoInput = torch.Tensor | np.ndarray | list[torch.Tensor | np.ndarray | Image.Image] +VideoInput: TypeAlias = torch.Tensor | np.ndarray | list[torch.Tensor | np.ndarray | Image.Image] AudioSample = int | float -AudioInput = torch.Tensor | np.ndarray | list[AudioSample] | list[list[AudioSample]] +AudioInput: TypeAlias = torch.Tensor | np.ndarray | list[AudioSample] | list[list[AudioSample]] class VideoFrames(list[Image.Image]): @@ -561,6 +561,8 @@ def _direct_planar_fallback_reason( frames: list[np.ndarray], frame_shape: tuple[int, ...], common_dtype: np.dtype, + *, + allow_strided_rgb_planes: bool = False, ) -> str | None: """Return a stable reason when direct planar muxing cannot consume frames.""" if len(frame_shape) != 3 or frame_shape[0] <= 0 or frame_shape[1] <= 0 or frame_shape[2] not in (3, 4): @@ -573,7 +575,9 @@ def _direct_planar_fallback_reason( ): return "unsupported_dtype" - if not all(frame[..., channel].flags.c_contiguous for frame in frames for channel in range(3)): + if not allow_strided_rgb_planes and not all( + frame[..., channel].flags.c_contiguous for frame in frames for channel in range(3) + ): return "non_contiguous_rgb_planes" return None @@ -627,6 +631,7 @@ class _PlanarFrameConverter: if frame.dtype == np.uint8: plane_view[:height, :width] = frame[..., channel] else: + assert scratch is not None np.copyto(scratch, frame[..., channel], casting="unsafe") np.clip(scratch, 0.0, 1.0, out=scratch) scratch *= 255.0 @@ -779,7 +784,12 @@ def _encode_video_bytes( # input is reported before any muxer is opened. frames, frame_shape, common_dtype = _prepare_video_frames(video) effective_audio_sample_rate = _resolve_audio_sample_rate(audio, audio_sample_rate) if audio is not None else None - fallback_reason = _direct_planar_fallback_reason(frames, frame_shape, common_dtype) + fallback_reason = _direct_planar_fallback_reason( + frames, + frame_shape, + common_dtype, + allow_strided_rgb_planes=frame_converter is not None and frame_converter.max_workers > 1, + ) if fallback_reason is not None: _log_video_encoding_path( selected_path="legacy_fallback",