[Bugfix][Multimodal] Recover PyNvVideoCodec decoder slots after failures (#59597)

Signed-off-by: Daniel Afrimi <dafrimi@nvidia.com> Signed-off-by: danielafrimi <45691845+danielafrimi@users.noreply.github.com> Signed-off-by: Daniel Afrimi <dafrimi@oci-hsg-cs-001-login-02.cm.cluster> Co-authored-by: OpenAI Codex <noreply@openai.com> Co-authored-by: Daniel Afrimi <dafrimi@oci-hsg-cs-001-login-02.cm.cluster> Co-authored-by: tomeras91 <57313761+tomeras91@users.noreply.github.com>
2026-10-15 09:17:19 +03:00
parent c5f8adc0b1
commit a9fe30e9f6
8 changed files with 912 additions and 88 deletions
@@ -387,9 +387,14 @@ def test_pynvvideocodec_unrelated_error_propagates(
    monkeypatch.setitem(sys.modules, "PyNvVideoCodec", fake_nvc)
    original_error = RuntimeError("GPU decoder unavailable")

-    def raise_unrelated_error(cls, file_path, nvc):
+    def raise_unrelated_error(cls, file_path, nvc, max_width, max_height):
        raise original_error

+    monkeypatch.setattr(
+        PyNvVideoCodecVideoBackendMixin,
+        "_validate_decoder_caps",
+        classmethod(lambda cls, file_path, nvc, **kwargs: (4096, 4096)),
+    )
    monkeypatch.setattr(
        PyNvVideoCodecVideoBackendMixin,
        "_read_source_metadata",
@@ -464,6 +469,18 @@ class TestMergeKwargsGpuBackendPolicy:
        )
        assert result["hw_decoders"] == 2

+    def test_prevents_request_level_pynv_dimension_override(self):
+        result = VideoMediaIO.merge_kwargs(
+            default_kwargs={
+                "backend": "pynvvideocodec",
+                "max_width": 3840,
+                "max_height": 2160,
+            },
+            runtime_kwargs={"max_width": 1920, "max_height": 1080},
+        )
+        assert result["max_width"] == 3840
+        assert result["max_height"] == 2160
+
    @pytest.mark.parametrize("backend", ["opencv", "torchcodec"])
    def test_software_video_backend_passes_through(self, backend: str):
        result = VideoMediaIO.merge_kwargs(
@@ -18,6 +18,7 @@ from vllm.multimodal.video_decoders import PYNVVIDEOCODEC_VIDEO_BACKEND
 from vllm.multimodal.video_decoders.pynvvideocodec import (
    PYNVVIDEOCODEC_CUDA_CONTEXT_BYTES,
    PYNVVIDEOCODEC_DECODER_GPU_MEMORY_BYTES,
+    get_pynvvideocodec_decoder_gpu_memory_bytes,
 )
 from vllm.utils.mem_constants import GiB_bytes

@@ -27,12 +28,18 @@ def _mm_config(
    mm_ipc_gpu_memory_gb: float = 0,
    video_backend: str | None = None,
    hw_decoders: int | None = None,
+    max_width: int | None = None,
+    max_height: int | None = None,
 ) -> MultiModalConfig:
    video_kwargs: dict[str, object] = (
        {} if video_backend is None else {"video_backend": video_backend}
    )
    if hw_decoders is not None:
        video_kwargs["hw_decoders"] = hw_decoders
+    if max_width is not None:
+        video_kwargs["max_width"] = max_width
+    if max_height is not None:
+        video_kwargs["max_height"] = max_height

    return MultiModalConfig(
        mm_ipc_gpu_memory_gb=mm_ipc_gpu_memory_gb,
@@ -43,9 +50,11 @@ def _mm_config(
 def _pynvvideocodec_decoder_budget(
    api_process_count: int = 1,
    hw_decoders: int = 2,
+    max_width: int | None = None,
+    max_height: int | None = None,
 ) -> int:
    return api_process_count * (
-        PYNVVIDEOCODEC_DECODER_GPU_MEMORY_BYTES * hw_decoders
+        get_pynvvideocodec_decoder_gpu_memory_bytes(max_width, max_height) * hw_decoders
        + PYNVVIDEOCODEC_CUDA_CONTEXT_BYTES
    )

@@ -246,3 +255,29 @@ def test_reserve_mm_ipc_gpu_memory_uses_configured_hw_decoders(
    assert reserve_mm_ipc_gpu_memory(available_bytes, mm_config) == (
        available_bytes - _pynvvideocodec_decoder_budget(hw_decoders=3)
    )
+
+
+def test_reserve_mm_ipc_gpu_memory_scales_for_configured_decoder_dimensions(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    monkeypatch.setenv("VLLM_VIDEO_LOADER_BACKEND", "opencv")
+    available_bytes = 8 * GiB_bytes
+    max_width = 7680
+    max_height = 4320
+    mm_config = _mm_config(
+        video_backend=PYNVVIDEOCODEC_VIDEO_BACKEND,
+        max_width=max_width,
+        max_height=max_height,
+    )
+
+    assert reserve_mm_ipc_gpu_memory(available_bytes, mm_config) == (
+        available_bytes
+        - _pynvvideocodec_decoder_budget(
+            max_width=max_width,
+            max_height=max_height,
+        )
+    )
+    assert (
+        get_pynvvideocodec_decoder_gpu_memory_bytes(max_width, max_height)
+        > PYNVVIDEOCODEC_DECODER_GPU_MEMORY_BYTES
+    )
@@ -38,6 +38,7 @@ from vllm.multimodal.video import (
 from vllm.multimodal.video_decoders import decode_video, resolve_video_backend_kwargs
 from vllm.multimodal.video_decoders.opencv import OpenCVVideoBackendMixin
 from vllm.multimodal.video_decoders.pynvvideocodec import (
+    PYNVVIDEOCODEC_DECODE_BATCH_SIZE,
    PYNVVIDEOCODEC_DECODER_CACHE_SIZE,
    PyNvVideoCodecDecoderSlot,
    PyNvVideoCodecVideoBackendMixin,
@@ -196,6 +197,17 @@ def test_decode_video_imports_only_selected_backend(
            {"max_frames": 16},
            {"pool_size": 3, "timeout_sec": 10.0},
        ),
+        (
+            PYNVVIDEOCODEC_VIDEO_BACKEND,
+            {
+                "min_frames": 4,
+                "hw_decoders": 3,
+                "max_width": 3840,
+                "max_height": 2160,
+            },
+            {"min_frames": 4},
+            {"hw_decoders": 3, "max_width": 3840, "max_height": 2160},
+        ),
    ],
 )
 def test_video_backend_kwargs_are_separated_from_sampling_kwargs(
@@ -283,7 +295,14 @@ def test_pynvvideocodec_backend_accounts_raw_decoded_frames(
            self.acquired.append(size)
            yield

-    def fake_decode(cls, file_path: str, frame_idx: list[int], nvc):
+    def fake_decode(
+        cls,
+        file_path: str,
+        frame_idx: list[int],
+        nvc,
+        max_width: int,
+        max_height: int,
+    ):
        return np.zeros((len(frame_idx), 20, 10, 3), dtype=np.uint8)

    pool = RecordingPool()
@@ -296,6 +315,11 @@ def test_pynvvideocodec_backend_accounts_raw_decoded_frames(
        "_decode_to_pinned_host",
        classmethod(fake_decode),
    )
+    monkeypatch.setattr(
+        PyNvVideoCodecVideoBackendMixin,
+        "_validate_decoder_caps",
+        classmethod(lambda cls, file_path, nvc, **kwargs: (8192, 8192)),
+    )

    loader = VIDEO_LOADER_REGISTRY.load(PYNVVIDEOCODEC_VIDEO_BACKEND)
    frames, metadata = loader.load_bytes(b"fake video", num_frames=4)
@@ -344,7 +368,14 @@ def test_pynvvideocodec_codec_uses_dynamic_sampling_strategy(
            self.acquired.append(size)
            yield

-    def fake_decode(cls, file_path: str, frame_idx: list[int], nvc):
+    def fake_decode(
+        cls,
+        file_path: str,
+        frame_idx: list[int],
+        nvc,
+        max_width: int,
+        max_height: int,
+    ):
        decoded_indices.append(frame_idx)
        return np.zeros((len(frame_idx), 20, 10, 3), dtype=np.uint8)

@@ -358,6 +389,11 @@ def test_pynvvideocodec_codec_uses_dynamic_sampling_strategy(
        "_decode_to_pinned_host",
        classmethod(fake_decode),
    )
+    monkeypatch.setattr(
+        PyNvVideoCodecVideoBackendMixin,
+        "_validate_decoder_caps",
+        classmethod(lambda cls, file_path, nvc, **kwargs: (8192, 8192)),
+    )

    loader = VIDEO_LOADER_REGISTRY.load("opencv_dynamic")
    frames, metadata = loader.load_bytes(
@@ -392,7 +428,8 @@ def test_pynvvideocodec_corrupted_videos_raise_value_error():
                hw_decoders=1,
            )

-        assert malformed_exc.value.__cause__ is not None
+        assert malformed_exc.value.__cause__ is None
+        assert malformed_exc.value.__context__ is None

        with pytest.raises(
            ValueError,
@@ -404,7 +441,8 @@ def test_pynvvideocodec_corrupted_videos_raise_value_error():
                hw_decoders=1,
            )

-        assert exc_info.value.__cause__ is not None
+        assert exc_info.value.__cause__ is None
+        assert exc_info.value.__context__ is None

        frames, _ = loader.load_bytes(
            valid_video,
@@ -479,12 +517,17 @@ def test_pynvvideocodec_decoder_slots_are_configured_once(
        PyNvVideoCodecVideoBackendMixin._configure_decoder_slots(3)


-def test_pynvvideocodec_failed_rebuild_invalidates_decoder_slot():
+def test_pynvvideocodec_failed_rebuild_retires_decoder_slot(
+    monkeypatch: pytest.MonkeyPatch,
+):
    events: list[tuple[str, str]] = []

    class FakeStream:
        cuda_stream = "cuda-stream"

+        def synchronize(self):
+            events.append(("synchronize", "cuda-stream"))
+
    class FakeDecoder:
        poisoned = False

@@ -493,10 +536,15 @@ def test_pynvvideocodec_failed_rebuild_invalidates_decoder_slot():
            events.append(("reconfigure", file_path))
            raise RuntimeError("reconfigure failed")

+        def stop(self):
+            events.append(("stop", "valid.mp4"))
+
    old_decoder = FakeDecoder()
    slot = PyNvVideoCodecDecoderSlot(FakeStream())
    slot.decoder = old_decoder
    slot.source_path = "valid.mp4"
+    slot.max_width = 4096
+    slot.max_height = 4096

    class FakeNvc:
        class OutputColorType:
@@ -529,16 +577,34 @@ def test_pynvvideocodec_failed_rebuild_invalidates_decoder_slot():
                "unsupported-8k.mp4",
                FakeNvc,
                device_index=0,
+                max_width=4096,
+                max_height=4096,
            )

        assert events == [
            ("reconfigure", "unsupported-8k.mp4"),
+            ("synchronize", "cuda-stream"),
+            ("stop", "valid.mp4"),
            ("construct", "unsupported-8k.mp4"),
+            ("synchronize", "cuda-stream"),
        ]
        assert old_decoder.poisoned
        assert slot.decoder is None
        assert slot.source_path is None
-        assert pool.slots == [slot]
+        assert pool.slots == []
+        assert pool.active == 0
+
+        replacement_slot = object()
+        monkeypatch.setattr(
+            PyNvVideoCodecVideoBackendMixin,
+            "_create_decoder_slot",
+            classmethod(lambda cls: replacement_slot),
+        )
+        with PyNvVideoCodecVideoBackendMixin._borrow_decoder_slot() as borrowed:
+            assert borrowed is replacement_slot
+
+        assert pool.slots == [replacement_slot]
+        assert pool.active == 1
    finally:
        pool.slots = old_slots
        pool.active = old_active
@@ -546,13 +612,291 @@ def test_pynvvideocodec_failed_rebuild_invalidates_decoder_slot():
        pool.max_slots = old_max


+def test_pynvvideocodec_native_metadata_error_is_not_chained(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    class FakeUnsupported(Exception):
+        pass
+
+    class FakeNvc:
+        PyNvVCExceptionUnsupported = FakeUnsupported
+
+    def raise_unsupported(cls, file_path, nvc, max_width, max_height):
+        raise FakeUnsupported("metadata failed")
+
+    monkeypatch.setitem(sys.modules, "PyNvVideoCodec", FakeNvc)
+    monkeypatch.setattr(
+        PyNvVideoCodecVideoBackendMixin,
+        "_read_source_metadata",
+        classmethod(raise_unsupported),
+    )
+    monkeypatch.setattr(
+        PyNvVideoCodecVideoBackendMixin,
+        "_validate_decoder_caps",
+        classmethod(lambda cls, file_path, nvc, **kwargs: (4096, 4096)),
+    )
+
+    with pytest.raises(
+        ValueError,
+        match=r"^Invalid or unsupported video file\.$",
+    ) as exc_info:
+        PyNvVideoCodecVideoBackendMixin.decode_frames_pynvvideocodec(
+            loader_cls=None,
+            data=b"unsupported",
+            target=None,
+        )
+
+    assert exc_info.value.__cause__ is None
+    assert exc_info.value.__context__ is None
+
+
+def test_pynvvideocodec_rejects_unsupported_decoder_caps():
+    class FakeDemuxer:
+        def Width(self):
+            return 7680
+
+        def Height(self):
+            return 4320
+
+        def GetNvCodecId(self):
+            return "h264"
+
+        def ChromaFormat(self):
+            return "420"
+
+        def BitDepth(self):
+            return 8
+
+    class FakeNvc:
+        @staticmethod
+        def CreateDemuxer(file_path):
+            return FakeDemuxer()
+
+        @staticmethod
+        def GetDecoderCaps(device_index, codec, chroma_format, bit_depth):
+            assert (device_index, codec, chroma_format, bit_depth) == (
+                0,
+                "h264",
+                "420",
+                8,
+            )
+            return {
+                "supported": 1,
+                "width_min": 48,
+                "width_max": 4096,
+                "height_min": 16,
+                "height_max": 4096,
+                "mb_num_max": 65536,
+            }
+
+    with pytest.raises(
+        ValueError,
+        match=r"^Invalid or unsupported video file\.$",
+    ):
+        PyNvVideoCodecVideoBackendMixin._validate_decoder_caps(
+            "unsupported-8k.mp4", FakeNvc
+        )
+
+
+def test_pynvvideocodec_uses_bounded_orientation_envelope_by_default():
+    class FakeDemuxer:
+        def Width(self):
+            return 1920
+
+        def Height(self):
+            return 1080
+
+        def GetNvCodecId(self):
+            return "h264"
+
+        def ChromaFormat(self):
+            return "420"
+
+        def BitDepth(self):
+            return 8
+
+    class FakeNvc:
+        @staticmethod
+        def CreateDemuxer(file_path):
+            return FakeDemuxer()
+
+        @staticmethod
+        def GetDecoderCaps(device_index, codec, chroma_format, bit_depth):
+            return {
+                "supported": 1,
+                "width_min": 48,
+                "width_max": 8192,
+                "height_min": 16,
+                "height_max": 8192,
+                "mb_num_max": 262144,
+            }
+
+    assert PyNvVideoCodecVideoBackendMixin._validate_decoder_caps(
+        "valid.mp4", FakeNvc
+    ) == (1920, 1920)
+
+
+def test_pynvvideocodec_default_orientation_envelope_respects_decoder_caps():
+    class FakeDemuxer:
+        def Width(self):
+            return 4096
+
+        def Height(self):
+            return 2160
+
+        def GetNvCodecId(self):
+            return "h264"
+
+        def ChromaFormat(self):
+            return "420"
+
+        def BitDepth(self):
+            return 8
+
+    class FakeNvc:
+        @staticmethod
+        def CreateDemuxer(file_path):
+            return FakeDemuxer()
+
+        @staticmethod
+        def GetDecoderCaps(device_index, codec, chroma_format, bit_depth):
+            return {
+                "supported": 1,
+                "width_min": 48,
+                "width_max": 4096,
+                "height_min": 16,
+                "height_max": 4096,
+                "mb_num_max": 40000,
+            }
+
+    assert PyNvVideoCodecVideoBackendMixin._validate_decoder_caps(
+        "valid.mp4", FakeNvc
+    ) == (4096, 2160)
+
+
+def test_pynvvideocodec_validates_configured_decoder_bounds():
+    class FakeDemuxer:
+        def Width(self):
+            return 1920
+
+        def Height(self):
+            return 1080
+
+        def GetNvCodecId(self):
+            return "h264"
+
+        def ChromaFormat(self):
+            return "420"
+
+        def BitDepth(self):
+            return 8
+
+    class FakeNvc:
+        @staticmethod
+        def CreateDemuxer(file_path):
+            return FakeDemuxer()
+
+        @staticmethod
+        def GetDecoderCaps(device_index, codec, chroma_format, bit_depth):
+            return {
+                "supported": 1,
+                "width_min": 48,
+                "width_max": 8192,
+                "height_min": 16,
+                "height_max": 8192,
+                "mb_num_max": 262144,
+            }
+
+    assert PyNvVideoCodecVideoBackendMixin._validate_decoder_caps(
+        "valid.mp4",
+        FakeNvc,
+        max_width=3840,
+        max_height=2160,
+    ) == (3840, 2160)
+
+    with pytest.raises(
+        ValueError,
+        match=r"^Invalid or unsupported video file\.$",
+    ):
+        PyNvVideoCodecVideoBackendMixin._validate_decoder_caps(
+            "valid.mp4",
+            FakeNvc,
+            max_width=1280,
+            max_height=720,
+        )
+
+    assert PyNvVideoCodecVideoBackendMixin._validate_decoder_caps(
+        "valid.mp4",
+        FakeNvc,
+        max_width=16384,
+        max_height=16384,
+    ) == (8192, 8192)
+
+
+def test_pynvvideocodec_clamps_configured_bounds_per_codec():
+    class FakeDemuxer:
+        def Width(self):
+            return 1920
+
+        def Height(self):
+            return 1080
+
+        def GetNvCodecId(self):
+            return "h264"
+
+        def ChromaFormat(self):
+            return "420"
+
+        def BitDepth(self):
+            return 8
+
+    class FakeNvc:
+        @staticmethod
+        def CreateDemuxer(file_path):
+            return FakeDemuxer()
+
+        @staticmethod
+        def GetDecoderCaps(device_index, codec, chroma_format, bit_depth):
+            return {
+                "supported": 1,
+                "width_min": 48,
+                "width_max": 4096,
+                "height_min": 16,
+                "height_max": 4096,
+                "mb_num_max": 65536,
+            }
+
+    assert PyNvVideoCodecVideoBackendMixin._validate_decoder_caps(
+        "h264.mp4",
+        FakeNvc,
+        max_width=7680,
+        max_height=4320,
+    ) == (4096, 4096)
+
+
+@pytest.mark.parametrize(
+    ("max_width", "max_height"),
+    [(3840, None), (None, 2160), (0, 2160), (3840, -1), (True, 2160)],
+)
+def test_pynvvideocodec_rejects_invalid_max_dimensions(
+    max_width: object,
+    max_height: object,
+):
+    with pytest.raises(ValueError, match="max_width|max_height"):
+        VideoBackend.load_bytes(
+            b"fake video",
+            backend=PYNVVIDEOCODEC_VIDEO_BACKEND,
+            max_width=max_width,
+            max_height=max_height,
+        )
+
+
@pytest.mark.skipif(not current_platform.is_cuda(), reason="Requires CUDA")
-def test_pynvvideocodec_h200_recovers_after_unsupported_8k():
-    import PyNvVideoCodec as nvc
+def test_pynvvideocodec_h100_h200_recovers_after_repeated_unsupported_8k():
    import torch

-    if "H200" not in torch.cuda.get_device_name(0):
-        pytest.skip("Requires H200 NVDEC resolution limits")
+    if not any(gpu in torch.cuda.get_device_name(0) for gpu in ("H100", "H200")):
+        pytest.skip("Requires H100 or H200 NVDEC resolution limits")

    valid_video = create_long_gop_video(num_frames=2, width=64, height=64)
    unsupported_video = (ASSETS_DIR / "unsupported_8k_h264.mp4").read_bytes()
@@ -574,26 +918,27 @@ def test_pynvvideocodec_h200_recovers_after_unsupported_8k():
            hw_decoders=1,
        )

-        with pytest.raises(Exception) as exc_info:
-            loader.load_bytes(
-                unsupported_video,
+        for _ in range(2):
+            with pytest.raises(
+                ValueError,
+                match=r"^Invalid or unsupported video file\.$",
+            ) as exc_info:
+                loader.load_bytes(
+                    unsupported_video,
+                    num_frames=1,
+                    hw_decoders=1,
+                )
+
+            assert exc_info.value.__cause__ is None
+            assert exc_info.value.__context__ is None
+
+            frames_after, _ = loader.load_bytes(
+                valid_video,
                num_frames=1,
                hw_decoders=1,
            )

-        root_cause = exc_info.value
-        while root_cause.__cause__ is not None:
-            root_cause = root_cause.__cause__
-        assert isinstance(root_cause, nvc.PyNvVCExceptionUnsupported)
-        assert "MBCount not supported" in str(root_cause)
-
-        frames_after, _ = loader.load_bytes(
-            valid_video,
-            num_frames=1,
-            hw_decoders=1,
-        )
-
-        assert frames_after.shape == frames_before.shape
+            assert frames_after.shape == frames_before.shape
    finally:
        for slot in _pynv_decoder_pool.slots:
            slot.invalidate()
@@ -689,6 +1034,8 @@ def test_pynvvideocodec_decoder_slot_retains_simple_decoder():
                    kwargs["gpu_id"],
                    kwargs["cuda_stream"],
                    kwargs["decoder_cache_size"],
+                    kwargs["max_width"],
+                    kwargs["max_height"],
                )
            )

@@ -703,9 +1050,33 @@ def test_pynvvideocodec_decoder_slot_retains_simple_decoder():

    slot = PyNvVideoCodecDecoderSlot(FakeStream())

-    decoder = slot.get_decoder("first.mp4", FakeNvc, device_index=7)
-    assert slot.get_decoder("first.mp4", FakeNvc, device_index=7) is decoder
-    assert slot.get_decoder("second.mp4", FakeNvc, device_index=7) is decoder
+    decoder = slot.get_decoder(
+        "first.mp4",
+        FakeNvc,
+        device_index=7,
+        max_width=4096,
+        max_height=4096,
+    )
+    assert (
+        slot.get_decoder(
+            "first.mp4",
+            FakeNvc,
+            device_index=7,
+            max_width=4096,
+            max_height=4096,
+        )
+        is decoder
+    )
+    assert (
+        slot.get_decoder(
+            "second.mp4",
+            FakeNvc,
+            device_index=7,
+            max_width=4096,
+            max_height=4096,
+        )
+        is decoder
+    )

    assert events == [
        (
@@ -714,12 +1085,163 @@ def test_pynvvideocodec_decoder_slot_retains_simple_decoder():
            7,
            "cuda-stream",
            PYNVVIDEOCODEC_DECODER_CACHE_SIZE,
+            4096,
+            4096,
        ),
        ("reconfigure", "second.mp4"),
    ]
    assert slot.source_path == "second.mp4"


+def test_pynvvideocodec_decoder_slot_rebuilds_for_larger_bounds():
+    events: list[tuple[object, ...]] = []
+
+    class FakeStream:
+        cuda_stream = "cuda-stream"
+
+        def synchronize(self):
+            events.append(("synchronize",))
+
+    class FakeDecoder:
+        def __init__(self, file_path: str, **kwargs):
+            events.append(
+                (
+                    "create",
+                    file_path,
+                    kwargs["max_width"],
+                    kwargs["max_height"],
+                )
+            )
+
+        def reconfigure_decoder(self, file_path: str):
+            events.append(("reconfigure", file_path))
+
+        def stop(self):
+            events.append(("stop",))
+
+    class FakeNvc:
+        class OutputColorType:
+            RGB = "rgb"
+
+        SimpleDecoder = FakeDecoder
+
+    slot = PyNvVideoCodecDecoderSlot(FakeStream())
+    first = slot.get_decoder(
+        "720p.mp4",
+        FakeNvc,
+        device_index=0,
+        max_width=1920,
+        max_height=1080,
+    )
+    second = slot.get_decoder(
+        "4k.mp4",
+        FakeNvc,
+        device_index=0,
+        max_width=4096,
+        max_height=4096,
+    )
+
+    assert second is not first
+    assert events == [
+        ("create", "720p.mp4", 1920, 1080),
+        ("synchronize",),
+        ("stop",),
+        ("create", "4k.mp4", 4096, 4096),
+    ]
+
+
+@pytest.mark.skipif(not current_platform.is_cuda(), reason="Requires CUDA")
+def test_pynvvideocodec_decodes_frames_in_bounded_batches(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    requested_indices: list[list[int]] = []
+
+    class FakeDecoder:
+        def get_batch_frames_by_index(self, indices: list[int]):
+            requested_indices.append(indices)
+            return [
+                torch.full(
+                    (3, 2, 4),
+                    index,
+                    dtype=torch.uint8,
+                    device="cuda",
+                )
+                for index in indices
+            ]
+
+    class FakeStream:
+        def __init__(self):
+            self.sync_count = 0
+
+        def synchronize(self):
+            self.sync_count += 1
+            torch.accelerator.synchronize()
+
+    class FakeSlot:
+        def __init__(self):
+            self.stream = FakeStream()
+            self.decoder = FakeDecoder()
+
+        def get_decoder(self, *args, **kwargs):
+            return self.decoder
+
+    class FakeNvc:
+        pass
+
+    slot = FakeSlot()
+
+    @contextmanager
+    def fake_borrow_decoder_slot(cls):
+        yield slot
+
+    @contextmanager
+    def fake_torch_stream_context(stream):
+        yield
+
+    monkeypatch.setattr(
+        PyNvVideoCodecVideoBackendMixin,
+        "_borrow_decoder_slot",
+        classmethod(fake_borrow_decoder_slot),
+    )
+    monkeypatch.setattr(
+        PyNvVideoCodecVideoBackendMixin,
+        "_torch_stream_context",
+        staticmethod(fake_torch_stream_context),
+    )
+
+    frame_idx = list(range(PYNVVIDEOCODEC_DECODE_BATCH_SIZE * 2 + 1))
+    frames = PyNvVideoCodecVideoBackendMixin._decode_to_pinned_host(
+        "video.mp4",
+        frame_idx,
+        FakeNvc,
+        max_width=4096,
+        max_height=4096,
+    )
+
+    assert [len(indices) for indices in requested_indices] == [32, 32, 1]
+    assert slot.stream.sync_count == 3
+    assert frames.shape == (65, 2, 4, 3)
+    np.testing.assert_array_equal(frames[:, 0, 0, 0], np.arange(65))
+
+
+@pytest.mark.skipif(not current_platform.is_cuda(), reason="Requires CUDA")
+def test_pynvvideocodec_chunked_decode_preserves_frame_order():
+    video = create_long_gop_video(num_frames=65, width=64, height=64)
+
+    with _fresh_decoder_pool():
+        loader = VIDEO_LOADER_REGISTRY.load(PYNVVIDEOCODEC_VIDEO_BACKEND)
+        frames, metadata = loader.load_bytes(
+            video,
+            num_frames=-1,
+            hw_decoders=1,
+        )
+
+    assert frames.shape == (65, 64, 64, 3)
+    assert metadata["frames_indices"] == list(range(65))
+    mean_green = frames[..., 1].reshape(65, -1).mean(axis=1)
+    np.testing.assert_allclose(mean_green, np.arange(65), atol=5)
+
+
 # ============================================================================
 # Video Processor → Video Loader Tests (via model repo)
 # ============================================================================
@@ -196,6 +196,7 @@ def reserve_mm_ipc_gpu_memory(
    from vllm.multimodal.video_decoders.pynvvideocodec import (
        PYNVVIDEOCODEC_CUDA_CONTEXT_BYTES,
        PYNVVIDEOCODEC_DECODER_GPU_MEMORY_BYTES,
+        get_pynvvideocodec_decoder_gpu_memory_bytes,
        validate_pynvvideocodec_hw_decoders,
    )

@@ -220,9 +221,15 @@ def reserve_mm_ipc_gpu_memory(
        if uses_pynvvideocodec
        else 1
    )
+    decoder_bytes = (
+        get_pynvvideocodec_decoder_gpu_memory_bytes(
+            video_kwargs.get("max_width"), video_kwargs.get("max_height")
+        )
+        if uses_pynvvideocodec
+        else PYNVVIDEOCODEC_DECODER_GPU_MEMORY_BYTES
+    )
    per_server_decoder_bytes = (
-        PYNVVIDEOCODEC_DECODER_GPU_MEMORY_BYTES * hw_decoders
-        + PYNVVIDEOCODEC_CUDA_CONTEXT_BYTES
+        decoder_bytes * hw_decoders + PYNVVIDEOCODEC_CUDA_CONTEXT_BYTES
    )
    decoder_reserved_bytes = (
        num_api_servers * per_server_decoder_bytes
@@ -48,6 +48,8 @@ class VideoMediaIO(MediaIO[MediaWithBytes[tuple[DecodedFrames, dict[str, Any]]]]
            # Decoder GPU memory is reserved from the startup value.
            runtime_kwargs = dict(runtime_kwargs)
            runtime_kwargs.pop("hw_decoders", None)
+            runtime_kwargs.pop("max_width", None)
+            runtime_kwargs.pop("max_height", None)
            runtime_kwargs.pop("pool_size", None)
            # The decode device determines which device the decoded frames
            # (and thus the processor pipeline) live on; startup-only.
@@ -282,6 +282,12 @@ class VideoBackend(VideoLoader):
                - ``hw_decoders`` (PyNvVideoCodec): maximum number of
                  concurrent decoder slots. Defaults to 2 and must be a
                  positive integer.
+                - ``max_width`` / ``max_height`` (PyNvVideoCodec): optional
+                  maximum input dimensions configured for each decoder. Both
+                  must be positive integers and set together. Values are
+                  clamped to each codec's hardware limits. By default, decoder
+                  dimensions also cover the input's opposite orientation when
+                  that fits the reserved memory and hardware limits.
                - ``pool_size`` / ``timeout_sec`` (DeepStream): decoder pool
                  size and pool acquisition timeout in seconds.

@@ -23,6 +23,8 @@ _BACKEND_OPTION_DEFAULTS: dict[str, dict[str, Any]] = {
    },
    PYNVVIDEOCODEC_VIDEO_BACKEND: {
        "hw_decoders": PYNVVIDEOCODEC_DEFAULT_HW_DECODERS,
+        "max_width": None,
+        "max_height": None,
    },
    "deepstream": {
        "pool_size": None,
@@ -30,12 +30,19 @@ def decode_pynvvideocodec(
    sampling_kwargs: dict,
    *,
    hw_decoders: int = PYNVVIDEOCODEC_DEFAULT_HW_DECODERS,
+    max_width: int | None = None,
+    max_height: int | None = None,
 ) -> tuple[npt.NDArray, VideoSourceMetadata, list[int], list[int]]:
+    max_width, max_height = validate_pynvvideocodec_max_dimensions(
+        max_width, max_height
+    )
    PyNvVideoCodecVideoBackendMixin._configure_decoder_slots(hw_decoders)
    return PyNvVideoCodecVideoBackendMixin.decode_frames_pynvvideocodec(
        loader_cls,
        data,
        target,
+        max_width=max_width,
+        max_height=max_height,
        **sampling_kwargs,
    )

@@ -48,9 +55,11 @@ class PyNvVideoCodecSourceMetadata(NamedTuple):
    height: int


-# Per-decoder upper bound reserved for persistent PyNvVideoCodec surfaces.
+# Per-decoder baseline reserved for persistent PyNvVideoCodec surfaces.
 PYNVVIDEOCODEC_DECODER_GPU_MEMORY_BYTES = 128 * MiB_bytes
 PYNVVIDEOCODEC_DECODER_CACHE_SIZE = 2
+# Bound native frame surfaces retained by each batch decode call.
+PYNVVIDEOCODEC_DECODE_BATCH_SIZE = 32
 # Per-API-server CUDA context and driver allocation, measured with
 # PyNvVideoCodec 2.0.4 on H100.
 PYNVVIDEOCODEC_CUDA_CONTEXT_BYTES = int(1.8 * 1024 * MiB_bytes)
@@ -66,6 +75,43 @@ def validate_pynvvideocodec_hw_decoders(hw_decoders: object) -> int:
    return hw_decoders


+def validate_pynvvideocodec_max_dimensions(
+    max_width: object,
+    max_height: object,
+) -> tuple[int | None, int | None]:
+    if max_width is None and max_height is None:
+        return None, None
+    if max_width is None or max_height is None:
+        raise ValueError("max_width and max_height must be set together")
+    if isinstance(max_width, bool) or not isinstance(max_width, int) or max_width < 1:
+        raise ValueError("max_width must be a positive integer")
+    if (
+        isinstance(max_height, bool)
+        or not isinstance(max_height, int)
+        or max_height < 1
+    ):
+        raise ValueError("max_height must be a positive integer")
+    return max_width, max_height
+
+
+def get_pynvvideocodec_decoder_gpu_memory_bytes(
+    max_width: object,
+    max_height: object,
+) -> int:
+    max_width, max_height = validate_pynvvideocodec_max_dimensions(
+        max_width, max_height
+    )
+    if max_width is None or max_height is None:
+        return PYNVVIDEOCODEC_DECODER_GPU_MEMORY_BYTES
+
+    configured_surface_bytes = _decoder_surface_bytes(max_width, max_height)
+    return max(PYNVVIDEOCODEC_DECODER_GPU_MEMORY_BYTES, configured_surface_bytes)
+
+
+def _decoder_surface_bytes(width: int, height: int) -> int:
+    return width * height * 3 * PYNVVIDEOCODEC_DECODER_CACHE_SIZE
+
+
 def _pynvvideocodec_exception_types(nvc) -> tuple[type[Exception], ...]:
    return tuple(
        exception_type
@@ -90,22 +136,54 @@ class PyNvVideoCodecDecoderSlot:
    existing decoder at each new source instead of paying a fresh
    ``SimpleDecoder`` construction per request. Construction (CUVID parser +
    decoder + surface-pool allocation) is the dominant per-request cost, so
-    reconfiguring is far cheaper. A single decoder serves both metadata
-    (``len``/``get_stream_metadata``) and frame decode -- no separate
-    metadata decoder.
+    reconfiguring is far cheaper. Decoder dimensions are bounded at construction
+    so resolution changes do not replace its native frame pool. A single decoder
+    serves both metadata (``len``/``get_stream_metadata``) and frame decode -- no
+    separate metadata decoder.
    """

    def __init__(self, stream) -> None:
        self.stream = stream
        self.decoder = None
        self.source_path: str | None = None
+        self.max_width = 0
+        self.max_height = 0

    def invalidate(self) -> None:
+        decoder = self.decoder
        self.decoder = None
        self.source_path = None
+        self.max_width = 0
+        self.max_height = 0
+        if decoder is not None:
+            with suppress(Exception):
+                decoder.stop()

-    def _construct(self, file_path: str, nvc, device_index: int) -> None:
+    def reset(self) -> None:
+        """Rebuild this slot after draining only its CUDA stream."""
+        with suppress(Exception):
+            self.stream.synchronize()
        self.invalidate()
+
+    def recover(self) -> None:
+        """Release decoder resources and drain work after a failed request."""
+        self.reset()
+
+        with suppress(Exception):
+            import torch
+
+            with torch.accelerator.device_index(self.stream.device.index):
+                torch.accelerator.synchronize()
+                torch.accelerator.empty_cache()
+
+    def _construct(
+        self,
+        file_path: str,
+        nvc,
+        device_index: int,
+        max_width: int,
+        max_height: int,
+    ) -> None:
        decoder = nvc.SimpleDecoder(
            file_path,
            output_color_type=nvc.OutputColorType.RGB,
@@ -114,20 +192,36 @@ class PyNvVideoCodecDecoderSlot:
            gpu_id=device_index,
            cuda_stream=self.stream.cuda_stream,
            decoder_cache_size=PYNVVIDEOCODEC_DECODER_CACHE_SIZE,
+            max_width=max_width,
+            max_height=max_height,
        )
        self.decoder = decoder
        self.source_path = file_path
+        self.max_width = max_width
+        self.max_height = max_height

-    def get_decoder(self, file_path: str, nvc, device_index: int):
+    def get_decoder(
+        self,
+        file_path: str,
+        nvc,
+        device_index: int,
+        max_width: int,
+        max_height: int,
+    ):
        if self.decoder is None:
-            self._construct(file_path, nvc, device_index)
+            self._construct(file_path, nvc, device_index, max_width, max_height)
        elif self.source_path != file_path:
-            try:
-                self.decoder.reconfigure_decoder(file_path)
-                self.source_path = file_path
-            except Exception:
-                # reconfigure unsupported/unsafe for this source -> rebuild.
-                self._construct(file_path, nvc, device_index)
+            if max_width > self.max_width or max_height > self.max_height:
+                self.reset()
+                self._construct(file_path, nvc, device_index, max_width, max_height)
+            else:
+                try:
+                    self.decoder.reconfigure_decoder(file_path)
+                    self.source_path = file_path
+                except Exception:
+                    # reconfigure unsupported/unsafe for this source -> rebuild.
+                    self.reset()
+                    self._construct(file_path, nvc, device_index, max_width, max_height)
        return self.decoder


@@ -222,9 +316,12 @@ class PyNvVideoCodecVideoBackendMixin:
            borrow_succeeded = True
        finally:
            if not borrow_succeeded:
-                slot.invalidate()
+                slot.recover()
            with pool.cond:
-                pool.slots.append(slot)
+                if borrow_succeeded:
+                    pool.slots.append(slot)
+                else:
+                    pool.active -= 1
                pool.cond.notify()

    @staticmethod
@@ -235,19 +332,86 @@ class PyNvVideoCodecVideoBackendMixin:
                return value
        return default

+    @classmethod
+    def _validate_decoder_caps(
+        cls,
+        file_path: str,
+        nvc,
+        max_width: int | None = None,
+        max_height: int | None = None,
+    ) -> tuple[int, int]:
+        demuxer = nvc.CreateDemuxer(file_path)
+        width = demuxer.Width()
+        height = demuxer.Height()
+        caps = nvc.GetDecoderCaps(
+            cls._DEVICE_INDEX,
+            demuxer.GetNvCodecId(),
+            demuxer.ChromaFormat(),
+            demuxer.BitDepth(),
+        )
+        width_min = int(caps["width_min"])
+        width_max = int(caps["width_max"])
+        height_min = int(caps["height_min"])
+        height_max = int(caps["height_max"])
+        mb_num_max = int(caps["mb_num_max"])
+        macroblock_count = ((width + 15) // 16) * ((height + 15) // 16)
+        if (
+            not caps["supported"]
+            or width < width_min
+            or width > width_max
+            or height < height_min
+            or height > height_max
+            or macroblock_count > mb_num_max
+        ):
+            raise ValueError("Invalid or unsupported video file.")
+
+        if max_width is None or max_height is None:
+            max_dimension = max(width, height)
+            orientation_mb_count = ((max_dimension + 15) // 16) ** 2
+            if (
+                max_dimension <= width_max
+                and max_dimension <= height_max
+                and orientation_mb_count <= mb_num_max
+                and _decoder_surface_bytes(max_dimension, max_dimension)
+                <= PYNVVIDEOCODEC_DECODER_GPU_MEMORY_BYTES
+            ):
+                return max_dimension, max_dimension
+            return width, height
+
+        decoder_max_width = min(max_width, width_max)
+        decoder_max_height = min(max_height, height_max)
+        decoder_mb_count = ((decoder_max_width + 15) // 16) * (
+            (decoder_max_height + 15) // 16
+        )
+        if decoder_mb_count > mb_num_max:
+            decoder_max_width = width
+            decoder_max_height = height
+        if width > decoder_max_width or height > decoder_max_height:
+            raise ValueError("Invalid or unsupported video file.")
+        return decoder_max_width, decoder_max_height
+
    @classmethod
    def _read_source_metadata(
        cls,
        file_path: str,
        nvc,
+        max_width: int,
+        max_height: int,
    ) -> PyNvVideoCodecSourceMetadata:
        with cls._borrow_decoder_slot() as decoder_slot:
            with cls._torch_stream_context(decoder_slot.stream):
                decoder = decoder_slot.get_decoder(
-                    file_path, nvc, device_index=cls._DEVICE_INDEX
+                    file_path,
+                    nvc,
+                    device_index=cls._DEVICE_INDEX,
+                    max_width=max_width,
+                    max_height=max_height,
                )
-                metadata = decoder.get_stream_metadata()
-                total_frames_num = len(decoder)
+                try:
+                    metadata = decoder.get_stream_metadata()
+                    total_frames_num = len(decoder)
+                finally:
+                    del decoder
            width = int(cls._metadata_value(metadata, "width", default=0))
            height = int(cls._metadata_value(metadata, "height", default=0))
            original_fps = float(
@@ -280,6 +444,8 @@ class PyNvVideoCodecVideoBackendMixin:
        file_path: str,
        frame_idx: list[int],
        nvc,
+        max_width: int,
+        max_height: int,
    ) -> npt.NDArray:
        import torch

@@ -289,45 +455,79 @@ class PyNvVideoCodecVideoBackendMixin:
        with cls._borrow_decoder_slot() as decoder_slot:
            stream = decoder_slot.stream
            with cls._torch_stream_context(stream):
-                try:
-                    decoder = decoder_slot.get_decoder(
-                        file_path, nvc, device_index=cls._DEVICE_INDEX
+                decoder = decoder_slot.get_decoder(
+                    file_path,
+                    nvc,
+                    device_index=cls._DEVICE_INDEX,
+                    max_width=max_width,
+                    max_height=max_height,
+                )
+                host_frames = None
+                decoded_count = 0
+                for start in range(0, len(frame_idx), PYNVVIDEOCODEC_DECODE_BATCH_SIZE):
+                    chunk_indices = frame_idx[
+                        start : start + PYNVVIDEOCODEC_DECODE_BATCH_SIZE
+                    ]
+                    invalid_video = False
+                    try:
+                        decoded_frames = decoder.get_batch_frames_by_index(
+                            chunk_indices
+                        )
+                    except Exception as exc:
+                        decoder = None
+                        if not isinstance(
+                            exc,
+                            _pynvvideocodec_exception_types(nvc) + (IndexError,),
+                        ):
+                            raise
+                        invalid_video = True
+                    if invalid_video:
+                        raise ValueError("Invalid or unsupported video file.")
+
+                    torch_frames = [
+                        torch.from_dlpack(frame) for frame in decoded_frames
+                    ]
+                    if not torch_frames:
+                        stream.synchronize()
+                        break
+                    device_frames = _pynvvc_frames_to_nhwc(torch.stack(torch_frames))
+                    if device_frames.ndim != 4:
+                        raise ValueError(
+                            "PyNvVideoCodec returned frames with unexpected shape "
+                            f"{tuple(device_frames.shape)}"
+                        )
+                    if host_frames is None:
+                        host_frames = torch.empty(
+                            (len(frame_idx), *device_frames.shape[1:]),
+                            dtype=device_frames.dtype,
+                            device="cpu",
+                            pin_memory=True,
+                        )
+                    elif tuple(device_frames.shape[1:]) != tuple(host_frames.shape[1:]):
+                        raise ValueError(
+                            "PyNvVideoCodec returned frames with inconsistent shapes"
+                        )
+
+                    next_count = decoded_count + len(device_frames)
+                    host_frames[decoded_count:next_count].copy_(
+                        device_frames, non_blocking=True
                    )
-                    decoded_frames = decoder.get_batch_frames_by_index(frame_idx)
-                except Exception as exc:
-                    if not isinstance(
-                        exc,
-                        _pynvvideocodec_exception_types(nvc) + (IndexError,),
-                    ):
-                        raise
-                    raise ValueError("Invalid or unsupported video file.") from exc
-                if len(decoded_frames) < len(frame_idx):
+                    stream.synchronize()
+                    decoded_count = next_count
+                    del decoded_frames, torch_frames, device_frames
+
+                    if decoded_count < start + len(chunk_indices):
+                        break
+
+                if decoded_count < len(frame_idx):
                    logger.warning(
                        "pynvvideocodec video loading: expected %d frames but got %d.",
                        len(frame_idx),
-                        len(decoded_frames),
+                        decoded_count,
                    )
-                torch_frames = [torch.from_dlpack(frame) for frame in decoded_frames]
-                if not torch_frames:
+                if host_frames is None:
                    return np.empty((0,), dtype=np.uint8)
-                device_frames = torch.stack(torch_frames)
-                if device_frames.ndim != 4:
-                    raise ValueError(
-                        "PyNvVideoCodec returned frames with unexpected shape "
-                        f"{tuple(device_frames.shape)}"
-                    )
-                device_frames = _pynvvc_frames_to_nhwc(device_frames)
-                host_frames = torch.empty(
-                    device_frames.shape,
-                    dtype=device_frames.dtype,
-                    device="cpu",
-                    pin_memory=True,
-                )
-                host_frames.copy_(device_frames, non_blocking=True)
-                stream.synchronize()
-                host_array = host_frames.numpy()
-                del decoded_frames, torch_frames, device_frames
-                return host_array
+                return host_frames[:decoded_count].numpy()

    @classmethod
    def decode_frames_pynvvideocodec(
@@ -335,6 +535,8 @@ class PyNvVideoCodecVideoBackendMixin:
        loader_cls,
        data: bytes,
        target: VideoTargetMetadata,
+        max_width: int | None = None,
+        max_height: int | None = None,
        **kwargs,
    ) -> tuple[npt.NDArray, VideoSourceMetadata, list[int], list[int]]:
        import PyNvVideoCodec as nvc
@@ -346,12 +548,31 @@ class PyNvVideoCodecVideoBackendMixin:
            with os.fdopen(temp_fd, "wb") as temp_file:
                temp_file.write(data)

+            invalid_video = False
            try:
-                gpu_source = cls._read_source_metadata(temp_path, nvc)
+                decoder_max_width, decoder_max_height = cls._validate_decoder_caps(
+                    temp_path,
+                    nvc,
+                    max_width=max_width,
+                    max_height=max_height,
+                )
            except Exception as exc:
                if not isinstance(exc, _pynvvideocodec_exception_types(nvc)):
                    raise
-                raise ValueError("Invalid or unsupported video file.") from exc
+                invalid_video = True
+            if invalid_video:
+                raise ValueError("Invalid or unsupported video file.")
+
+            try:
+                gpu_source = cls._read_source_metadata(
+                    temp_path, nvc, decoder_max_width, decoder_max_height
+                )
+            except Exception as exc:
+                if not isinstance(exc, _pynvvideocodec_exception_types(nvc)):
+                    raise
+                invalid_video = True
+            if invalid_video:
+                raise ValueError("Invalid or unsupported video file.")
            check_frame_pixel_limit(gpu_source.width, gpu_source.height)
            source = loader_cls._prepare_source(gpu_source.source)
            frame_idx = loader_cls.compute_frames_index_to_sample(
@@ -360,10 +581,22 @@ class PyNvVideoCodecVideoBackendMixin:
            raw_frame_bytes = len(frame_idx) * gpu_source.height * gpu_source.width * 3
            pool = get_mm_gpu_ipc_pool()
            if pool is None or raw_frame_bytes == 0:
-                frames = cls._decode_to_pinned_host(temp_path, frame_idx, nvc)
+                frames = cls._decode_to_pinned_host(
+                    temp_path,
+                    frame_idx,
+                    nvc,
+                    decoder_max_width,
+                    decoder_max_height,
+                )
            else:
                with pool.acquire(raw_frame_bytes):
-                    frames = cls._decode_to_pinned_host(temp_path, frame_idx, nvc)
+                    frames = cls._decode_to_pinned_host(
+                        temp_path,
+                        frame_idx,
+                        nvc,
+                        decoder_max_width,
+                        decoder_max_height,
+                    )
        finally:
            with suppress(FileNotFoundError):
                os.unlink(temp_path)