[Multimodal] Use GPU NVDEC for EPD encoder-only instance video media IO

tests/config/test_multimodal_config.py

@@ -375,3 +375,85 @@ def test_vllm_config_runs_the_mm_processor_device_check():
         pytest.raises(ValueError, match="also runs the language model"),
     ):
         VllmConfig._validate_mm_processor_device(vllm_config)
+
+
+def _resolve_mm_video_decode_device(
+    *,
+    ec_role: ECRole | None,
+    mm_tensor_ipc: str = "torch_shm",
+    device: str | None = None,
+    video_kwargs: dict | None = None,
+    torchcodec_available: bool = True,
+) -> dict:
+    """Run processor-device then video-decode-device resolution and report
+    the resulting video media IO kwargs."""
+    mm_config = MultiModalConfig(
+        mm_processor_kwargs={} if device is None else {"device": device},
+        mm_tensor_ipc=mm_tensor_ipc,  # type: ignore[arg-type]
+        media_io_kwargs={} if video_kwargs is None else {"video": dict(video_kwargs)},
+    )
+    model_config = MagicMock(spec=ModelConfig)
+    model_config.multimodal_config = mm_config
+    vllm_config = MagicMock(spec=VllmConfig)
+    vllm_config.model_config = model_config
+    vllm_config.ec_transfer_config = (
+        None
+        if ec_role is None
+        else ECTransferConfig(ec_connector="ECExampleConnector", ec_role=ec_role)
+    )
+
+    with (
+        patch("vllm.platforms.current_platform.device_type", "cuda"),
+        patch(
+            "vllm.utils.import_utils.check_torchcodec_available",
+            side_effect=None if torchcodec_available else ImportError("torchcodec"),
+        ),
+    ):
+        VllmConfig._resolve_mm_processor_device(vllm_config)
+        VllmConfig._resolve_mm_video_decode_device(vllm_config)
+    return mm_config.media_io_kwargs.get("video", {})
+
+
+def test_auto_video_decode_uses_nvdec_on_encoder_instance():
+    """The processor runs on the accelerator there, so decoded frames should
+    stay on-device too."""
+    assert _resolve_mm_video_decode_device(ec_role="ec_producer") == {
+        "backend": "torchcodec",
+        "device": "cuda",
+    }
+
+
+@pytest.mark.parametrize("ec_role", [None, "ec_consumer", "ec_both"])
+def test_auto_video_decode_stays_on_cpu_off_encoder_instance(
+    ec_role: ECRole | None,
+):
+    """The processor stays on CPU off encode-only instances, so video
+    decoding should too."""
+    assert _resolve_mm_video_decode_device(ec_role=ec_role) == {}
+
+
+def test_auto_video_decode_follows_explicit_processor_device():
+    assert _resolve_mm_video_decode_device(ec_role="ec_producer", device="cpu") == {}
+    assert _resolve_mm_video_decode_device(ec_role="ec_producer", device="cuda") == {
+        "backend": "torchcodec",
+        "device": "cuda",
+    }
+
+
+def test_auto_video_decode_respects_explicit_media_io_kwargs():
+    assert _resolve_mm_video_decode_device(
+        ec_role="ec_producer", video_kwargs={"backend": "opencv"}
+    ) == {"backend": "opencv"}
+    assert _resolve_mm_video_decode_device(
+        ec_role="ec_producer", video_kwargs={"device": "cpu"}
+    ) == {"device": "cpu"}
+
+
+def test_auto_video_decode_skipped_without_torchcodec():
+    assert (
+        _resolve_mm_video_decode_device(
+            ec_role="ec_producer",
+            torchcodec_available=False,
+        )
+        == {}
+    )

tests/multimodal/media/test_video.py

@@ -532,6 +532,50 @@ def test_strips_pool_size_from_runtime(self):
     def test_unknown_backend_not_treated_as_gpu(self):
         assert not VIDEO_LOADER_REGISTRY.backend_requires_gpu("totally_unknown")
 
+    def test_strips_request_level_device(self):
+        """The decode device is a startup-only knob: a request must not move
+        decoding onto the GPU when the startup config did not opt in, nor off
+        it when it did."""
+        result = VideoMediaIO.merge_kwargs(
+            default_kwargs={"backend": "torchcodec"},
+            runtime_kwargs={"device": "cuda"},
+        )
+        assert "device" not in result
+
+        result = VideoMediaIO.merge_kwargs(
+            default_kwargs={"backend": "torchcodec", "device": "cuda"},
+            runtime_kwargs={"device": "cpu", "num_frames": 8},
+        )
+        assert result["device"] == "cuda"
+        assert result["num_frames"] == 8
+
+
+@pytest.mark.parametrize(
+    "default_kwargs",
+    [
+        {"backend": "torchcodec", "device": "cuda", "seek_mode": "approximate"},
+        {"backend": "pynvvideocodec", "hw_decoders": 2},
+    ],
+)
+def test_switching_backend_drops_stale_codec_options(default_kwargs):
+    """Codec-specific options from the static config must not leak into a
+    different codec backend selected per-request, where they would fail the
+    new backend's option validation."""
+    result = VideoMediaIO.merge_kwargs(
+        default_kwargs={**default_kwargs, "num_frames": 8},
+        runtime_kwargs={"backend": "opencv"},
+    )
+    assert result == {"backend": "opencv", "num_frames": 8}
+
+
+def test_same_backend_keeps_codec_options():
+    result = VideoMediaIO.merge_kwargs(
+        default_kwargs={"backend": "torchcodec", "device": "cuda"},
+        runtime_kwargs={"backend": "torchcodec", "num_frames": 8},
+    )
+    assert result["device"] == "cuda"
+    assert result["num_frames"] == 8
+
 
 @pytest.mark.parametrize("layout", ["nhwc", "nchw"])
 def test_pynvvc_frames_normalized_to_nhwc(layout: str):

tests/multimodal/test_hasher.py

@@ -137,6 +137,33 @@ def item_for_hash(num_frames: int):
     )
 
 
+def test_hash_video_tensor_frames():
+    """Videos holding tensor frames (e.g. NVDEC-decoded) hash like
+    array-framed ones, from the original bytes without a D2H copy."""
+    source = b"x" * 100
+
+    def item_for_hash(frames):
+        metadata = {
+            "total_num_frames": 2,
+            "fps": 2.0,
+            "duration": 1.0,
+            "video_backend": "torchcodec",
+            "frames_indices": [0, 1],
+            "do_sample_frames": False,
+        }
+        video = MediaWithBytes((frames, metadata), source)
+        items = MultiModalDataParser()._parse_video_data([video])
+        return items.get_all_items_for_hash()[0]
+
+    np_frames = np.zeros((2, 8, 8, 3), dtype=np.uint8)
+    torch_frames = torch.zeros((2, 8, 8, 3), dtype=torch.uint8)
+
+    hasher = MultiModalHasher
+    assert hasher.hash_kwargs("blake3", video=item_for_hash(np_frames)) == (
+        hasher.hash_kwargs("blake3", video=item_for_hash(torch_frames))
+    )
+
+
 def test_hash_non_contiguous_array():
     arr = np.arange(24).reshape(4, 6).T
     assert not arr.flags.c_contiguous

tests/multimodal/test_parse.py

@@ -56,6 +56,28 @@ def test_frame_size_hwc_chw(frame):
     assert items.get_frame_size(0) == (W, H)
 
 
+def test_video_with_metadata_tensor_passthrough():
+    """Tensor frames pass through unchanged regardless of device: HF video
+    processors accept tensors, and device-resident frames (e.g. NVDEC-decoded)
+    must not be copied back to host."""
+    frames = torch.zeros((4, H, W, 3), dtype=torch.uint8)
+    video, metadata = MultiModalDataParser()._get_video_with_metadata(frames)
+
+    assert video is frames
+    assert metadata is None
+
+
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="Requires CUDA")
+def test_video_with_metadata_keeps_device_tensor():
+    """Device-resident frames (e.g. NVDEC-decoded) pass through as tensors,
+    so a device-side HF processor can consume them without a D2H copy."""
+    frames = torch.zeros((4, H, W, 3), dtype=torch.uint8, device="cuda")
+    video, metadata = MultiModalDataParser()._get_video_with_metadata(frames)
+
+    assert video is frames
+    assert metadata is None
+
+
 @pytest.mark.parametrize(
     "modality,processor_cls",
     [

tests/multimodal/test_video.py

@@ -11,6 +11,7 @@
 import numpy as np
 import numpy.typing as npt
 import pytest
+import torch
 from transformers import AutoVideoProcessor
 from transformers.video_utils import VideoMetadata
 
@@ -173,9 +174,14 @@ def fake_import_module(name: str, package: str):
     [
         (
             "torchcodec",
-            {"min_frames": 4, "num_ffmpeg_threads": 2, "seek_mode": "approximate"},
+            {
+                "min_frames": 4,
+                "num_ffmpeg_threads": 2,
+                "seek_mode": "approximate",
+                "device": "cuda",
+            },
             {"min_frames": 4},
-            {"num_ffmpeg_threads": 2, "seek_mode": "approximate"},
+            {"num_ffmpeg_threads": 2, "seek_mode": "approximate", "device": "cuda"},
         ),
         (
             "deepstream",
@@ -205,6 +211,11 @@ def test_video_backend_rejects_options_for_another_decoder():
     ):
         resolve_video_backend_kwargs("opencv", {"num_ffmpeg_threads": 2})
 
+    with pytest.raises(
+        ValueError, match="device is not supported by the 'opencv' backend"
+    ):
+        resolve_video_backend_kwargs("opencv", {"device": "cuda"})
+
 
 @pytest.mark.parametrize(
     ("backend", "error"),
@@ -1205,6 +1216,30 @@ def test_torchcodec_backend_rejects_frame_recovery(dummy_video_path):
         )
 
 
+@pytest.mark.skipif(not current_platform.is_cuda(), reason="Requires CUDA")
+def test_torchcodec_backend_cuda_decodes_on_gpu(dummy_video_path):
+    """With device="cuda", torchcodec decodes via NVDEC and keeps the frames
+    on the GPU instead of returning a host-side numpy array."""
+    pytest.importorskip("torchcodec")
+
+    with open(dummy_video_path, "rb") as f:
+        video_data = f.read()
+
+    loader = VIDEO_LOADER_REGISTRY.load("opencv")
+    frames, metadata = loader.load_bytes(
+        video_data, num_frames=8, backend="torchcodec", device="cuda"
+    )
+
+    assert isinstance(frames, torch.Tensor)
+    assert frames.device.type == "cuda"
+    assert frames.dtype == torch.uint8
+    assert frames.ndim == 4
+    assert frames.shape[3] == 3  # RGB
+    assert frames.shape[0] == 8
+    assert frames.shape[0] == len(metadata["frames_indices"])
+    assert metadata["video_backend"] == "torchcodec"
+
+
 def test_torchcodec_backend_returns_target_frames_not_keyframes():
     """Regression test: torchcodec must return the requested frames, not the
     GOP keyframe they seek back to.

vllm/config/vllm.py

@@ -1718,6 +1718,7 @@ def has_blocked_weights():
 
         self._resolve_allow_missing_mm_embeddings()
         self._resolve_mm_processor_device()
+        self._resolve_mm_video_decode_device()
         self._validate_mm_processor_device()
 
         if self.use_v2_model_runner:
@@ -2595,6 +2596,61 @@ def _resolve_mm_processor_device(self) -> None:
             device_type,
         )
 
+    def _resolve_mm_video_decode_device(self) -> None:
+        """Default video decoding to torchcodec GPU backend for EPD encoder-only
+        instance if the mm processor runs on CUDA.
+
+        The processor consumes the decoded frames on-device in that case, so
+        keeping the frames on the GPU skips the host round-trip through the
+        CPU media path. An explicit codec/backend choice in
+        `--media-io-kwargs` is left alone, and the default is skipped where
+        torchcodec (or its FFmpeg runtime) is unavailable.
+        """
+        if self.model_config is None or self.model_config.multimodal_config is None:
+            return
+        mm_config = self.model_config.multimodal_config
+
+        ec_config = self.ec_transfer_config
+        # An EC producer that is not also a consumer runs no forward pass and
+        # allocates no KV cache, so frontend accelerator work has the device to
+        # itself.
+        if ec_config is None or not ec_config.is_encode_only:
+            return
+
+        from vllm.platforms import current_platform
+
+        device_type = current_platform.device_type
+        if (
+            device_type != "cuda"
+            or mm_config.get_mm_processor_device_type() != device_type
+        ):
+            return
+
+        # User set video backend or device explicitly
+        video_kwargs = mm_config.media_io_kwargs.setdefault("video", {})
+        if "backend" in video_kwargs or "device" in video_kwargs:
+            return
+
+        from vllm.utils.import_utils import check_torchcodec_available
+
+        try:
+            check_torchcodec_available()
+        except (ImportError, RuntimeError):
+            # torchcodec is not installed, or is installed without a usable
+            # FFmpeg runtime (it raises rather than returning False).
+            logger.info_once(
+                "EPD encoder instance: keeping CPU video decoding because "
+                "torchcodec is not available (needs a CUDA build with FFmpeg)."
+            )
+            return
+
+        video_kwargs["backend"] = "torchcodec"
+        video_kwargs["device"] = device_type
+        logger.info_once(
+            "EPD encoder instance: decoding video with NVDEC (torchcodec device=%s).",
+            device_type,
+        )
+
     def _validate_mm_processor_device(self) -> None:
         """Hand the EC config to `MultiModalConfig`, which owns the rule."""
         model_config = self.model_config

vllm/multimodal/hasher.py

@@ -89,8 +89,11 @@ def serialize_item(cls, obj: object) -> Iterable[bytes | memoryview]:
                 )
             return cls.iter_item_to_bytes("image", obj.original_bytes)
 
-        if isinstance(obj, MediaWithBytes) and isinstance(obj.media, np.ndarray):
+        if isinstance(obj, MediaWithBytes) and isinstance(
+            obj.media, (np.ndarray, torch.Tensor)
+        ):
             frames = obj.media
+            # Both np.ndarray and torch.Tensor expose .nbytes.
             if frames.nbytes < len(obj.original_bytes):
                 return cls.iter_item_to_bytes("video", frames)
             return cls.iter_item_to_bytes("video", obj.original_bytes)