Deepstream video backend (#42424)

Signed-off-by: Viranjan Pagar <vpagar@nvidia.com>
Signed-off-by: Isotr0py <Isotr0py@outlook.com>
Signed-off-by: Isotr0py <mozf@mail2.sysu.edu.cn>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: Isotr0py <mozf@mail2.sysu.edu.cn>
Co-authored-by: Isotr0py <Isotr0py@outlook.com>
Co-authored-by: Roger Wang <hey@rogerw.io>

Author: 4 people

docs/features/multimodal_inputs.md

@@ -879,6 +879,55 @@ vllm serve Qwen/Qwen3-VL-30B-A3B-Instruct \
 
 Works with common video formats like MP4 when using OpenCV backends.
 
+#### GPU Video Decoding with DeepStream (NVDEC)
+
+By default vLLM decodes video on the CPU. On NVIDIA GPUs you can instead decode
+directly on the hardware video engine (NVDEC) with the DeepStream backend, which
+keeps decoding off the CPU and can significantly increase video throughput.
+
+Install the backend (Linux x86-64 only):
+
+```bash
+pip install vllm[deepstream]
+```
+
+The pip wheel bundles the DeepStream libraries but still relies on a few system
+packages that pip cannot install. On Ubuntu:
+
+```bash
+apt-get install -y \
+  gstreamer1.0-tools gstreamer1.0-plugins-base gstreamer1.0-plugins-good \
+  gstreamer1.0-plugins-bad gstreamer1.0-libav \
+  python3-gi python3-gst-1.0 libv4l-0 cuda-libraries-13-0
+```
+
+Select the backend either with an environment variable:
+
+```bash
+export VLLM_VIDEO_LOADER_BACKEND=deepstream
+vllm serve Qwen/Qwen3-VL-30B-A3B-Instruct
+```
+
+or per request via `--media-io-kwargs`:
+
+```bash
+vllm serve Qwen/Qwen3-VL-30B-A3B-Instruct \
+  --media-io-kwargs '{"video": {"backend": "deepstream"}}'
+```
+
+**Parameters:**
+
+- `pool_size`: Number of GPU decode workers in the process-wide decode pool
+  (clamped to `[1, 16]`). When unset it defaults to
+  `VLLM_MEDIA_LOADING_THREAD_COUNT` (default `8`). The pool is a singleton, so
+  the first request's value wins.
+
+```bash
+# Example: 12 decode workers
+vllm serve Qwen/Qwen3-VL-30B-A3B-Instruct \
+  --media-io-kwargs '{"video": {"backend": "deepstream", "pool_size": 12}}'
+```
+
 #### Pre-extracted Frame Sequences with `media_io_kwargs`
 
 When you extract video frames on the client side and send them as `video/jpeg` (base64-concatenated JPEG frames), you can preserve the original video metadata by using `media_io_kwargs` in your request. This enables more accurate video understanding by preserving temporal information that would otherwise be lost during client-side frame extraction.

setup.py

@@ -1257,6 +1257,9 @@ def add_vllm_package_data(filename: str) -> None:
             "mistral_common[audio]",
         ],  # Required for audio processing
         "video": [],  # Kept for backwards compatibility
+        # NVIDIA DeepStream (NVDEC) GPU video-decode backend. Linux x86-64
+        # only; also needs system GStreamer + libv4l (see docs).
+        "deepstream": ["nvidia-deepstream-videodecode-cu13>=9.0.2"],
         "flashinfer": [],  # Kept for backwards compatibility
         # Optional deps for Helion kernel development
         # NOTE: When updating helion version, also update CI files:

vllm/multimodal/video.py

@@ -826,21 +826,130 @@ def decode_frames_pynvvideocodec(
         return frames, source, frame_idx, valid_frame_indices
 
 
+class DeepStreamVideoBackendMixin:
+    """NVIDIA DeepStream (NVDEC) GPU-decode codec utilities.
+
+    Decoding runs on a shared pool of daemon threads inside one CUDA
+    context (see the ``nvidia-deepstream-videodecode-cu13`` package). The
+    container bytes are pushed into an ``appsrc`` GStreamer pipeline, so no
+    local file path is required — HTTP and base64 sources decode identically
+    to local files.
+
+    Like the OpenCV/PyAV mixins, this provides only the codec layer.
+    Frame *selection* lives in the loader's
+    ``compute_frames_index_to_sample`` and arrives here as an explicit
+    list of frame indices.
+    """
+
+    # Process-wide lazy decode pool, shared across all DeepStream backends.
+    _pool: ClassVar[Any] = None
+    _pool_lock: ClassVar[Any] = None
+
+    @classmethod
+    def _get_pool(cls, pool_size: int | None = None):
+        """Lazy-initialize the shared decode pool on first use.
+
+        ``pool_size`` (number of decode worker threads) comes from
+        ``--media-io-kwargs`` (``{"video": {"pool_size": N}}``); when unset it
+        defaults to the existing ``VLLM_MEDIA_LOADING_THREAD_COUNT`` so no
+        DeepStream-specific env var is needed. The pool is a process-wide
+        singleton, so the first decode's value wins.
+        """
+        if cls._pool is not None:
+            return cls._pool
+        if cls._pool_lock is None:
+            cls._pool_lock = threading.Lock()
+        with cls._pool_lock:
+            if cls._pool is not None:
+                return cls._pool
+            import os
+
+            from nvidia.deepstream_videodecode import DecodePool
+
+            if pool_size is None:
+                pool_size = int(os.environ.get("VLLM_MEDIA_LOADING_THREAD_COUNT", 8))
+            pool_size = max(1, min(int(pool_size), 16))
+            logger.info(
+                "[DeepStream] initializing decode pool with %d workers",
+                pool_size,
+            )
+            cls._pool = DecodePool(num_workers=pool_size)
+            return cls._pool
+
+    @classmethod
+    def decode_indices(
+        cls,
+        data: bytes,
+        frame_indices: list[int],
+        source: VideoSourceMetadata,
+        codec: str = "",
+        pool_size: int | None = None,
+        timeout_sec: float = 120.0,
+    ) -> tuple[npt.NDArray, list[int]]:
+        """Decode the requested frame indices from raw container bytes.
+
+        The whole stream is decoded; the pool keeps exactly the frames whose
+        decode-order index is in ``frame_indices`` (1:1, frame-exact) and
+        sends EOS once the last one is matched.
+
+        ``codec`` (e.g. ``"h264"``/``"hevc"``) lets the pool keep its NVDEC
+        session warm across same-codec streams and rebuild only on a codec
+        change. Frames are returned as a CPU NHWC uint8 array so the
+        upstream multimodal parser sees the same shape as the other
+        backends.
+        """
+        if not frame_indices:
+            raise ValueError("DeepStream backend received no frame indices")
+
+        result = cls._get_pool(pool_size).decode(
+            data,
+            target_indices=frame_indices,
+            codec=codec,
+            max_frames=len(frame_indices),
+            timeout_sec=timeout_sec,
+        )
+        if result.error:
+            raise ValueError(f"DeepStream decode failed: {result.error}")
+        if result.frames is None or result.n_kept == 0:
+            raise ValueError("DeepStream decode produced no frames")
+
+        valid = frame_indices[: result.n_kept]
+        # GPU -> CPU NHWC uint8 at the codec boundary (one PCIe copy); keeps
+        # the array shape identical to the OpenCV/PyAV backends. Copy into
+        # PINNED host memory (reused across calls by PyTorch's pinned caching
+        # allocator) so the D2H runs at full PCIe bandwidth (~13 GB/s) rather
+        # than the ~1 GB/s pageable path that plain ``.cpu()`` takes — ~12x
+        # faster for a 1080p x8 frame batch (~46ms -> ~4ms). ``numpy()`` keeps
+        # the pinned tensor alive via the array's base.
+        import torch
+
+        gpu = result.frames
+        if gpu.is_cuda:
+            host = torch.empty(gpu.shape, dtype=gpu.dtype, pin_memory=True)
+            host.copy_(gpu, non_blocking=True)
+            torch.cuda.current_stream().synchronize()
+            arr = host.numpy()
+        else:
+            arr = gpu.numpy()
+        return arr, valid
+
+
 @VIDEO_LOADER_REGISTRY.register("opencv")
 class VideoBackend(
     VideoLoader,
     OpenCVVideoBackendMixin,
     PyAVVideoBackendMixin,
     TorchCodecVideoBackendMixin,
     PyNvVideoCodecVideoBackendMixin,
+    DeepStreamVideoBackendMixin,
 ):
     """Uniform-sampling video backend.
 
     Samples ``num_frames`` uniformly across the video (or one frame every
     ``1/fps`` seconds, whichever produces fewer frames). The decoding codec
     is selected via the ``backend`` kwarg (``"opencv"``, ``"pyav"``,
-    ``"torchcodec"`` or ``"pynvvideocodec"``), which can be passed through
-    ``--media-io-kwargs``. Defaults to ``"opencv"``.
+    ``"torchcodec"``, ``"pynvvideocodec"``, or ``"deepstream"``),
+    which can be passed through ``--media-io-kwargs``. Defaults to ``"opencv"``.
     """
 
     _sampling_suffix: ClassVar[str] = ""
@@ -885,7 +994,9 @@ def load_bytes(
         max_duration: int = 300,
         frame_recovery: bool = False,
         *,
-        backend: Literal["opencv", "pyav", "torchcodec", "pynvvideocodec"] = "opencv",
+        backend: Literal[
+            "opencv", "pyav", "torchcodec", "pynvvideocodec", "deepstream"
+        ] = "opencv",
         num_ffmpeg_threads: int = 0,
         seek_mode: Literal["exact", "approximate"] = "exact",
         **kwargs,
@@ -901,7 +1012,7 @@ def load_bytes(
             frame_recovery: Enable forward-scan recovery for failed frames.
                 Only honored by the OpenCV codec.
             backend: Decoding codec — ``"opencv"``, ``"pyav"``,
-                ``"torchcodec"`` or ``"pynvvideocodec"``.
+                ``"torchcodec"``, ``"pynvvideocodec"`` or ``"deepstream"``.
             num_ffmpeg_threads: Number of FFmpeg decoding threads, only used by
                 TorchCodec: ``0`` (default) relies on the FFmpeg default value
                 which is ``min(cpu_count + 1, 16)``.
@@ -982,11 +1093,37 @@ def load_bytes(
                 target,
                 **kwargs,
             )
+        elif backend == "deepstream":
+            assert not frame_recovery, (
+                "frame_recovery is only available for `opencv` backend"
+            )
+            # Decode-pool size comes from media-io-kwargs (no env var); the
+            # pool is a process-wide singleton so the first decode's value
+            # wins. Pop it so it isn't forwarded to the frame sampler.
+            pool_size = kwargs.pop("pool_size", None)
+            # Probe container metadata from the bytes via GStreamer (in
+            # the deepstream video-decode wheel) — no PyAV/pymediainfo, no path.
+            from nvidia.deepstream_videodecode import probe_metadata
+
+            total_frames, original_fps, duration, _w, _h, codec = probe_metadata(data)
+            source = cls._prepare_source(
+                VideoSourceMetadata(
+                    total_frames_num=total_frames,
+                    original_fps=original_fps,
+                    duration=duration,
+                )
+            )
+            frame_idx = cls.compute_frames_index_to_sample(
+                source=source, target=target, **kwargs
+            )
+            frames, valid = cls.decode_indices(
+                data, frame_idx, source, codec=codec, pool_size=pool_size
+            )
         else:
             raise ValueError(
                 f"Unknown video codec backend {backend!r}; "
                 "valid options: 'opencv', 'pyav', 'torchcodec', "
-                "'pynvvideocodec'."
+                "'pynvvideocodec' and 'deepstream'."
             )
 
         if len(valid) < len(frame_idx):
@@ -1073,7 +1210,9 @@ def load_bytes(
         max_duration: int = 300,
         frame_recovery: bool = False,
         *,
-        backend: Literal["opencv", "pyav", "torchcodec", "pynvvideocodec"] = "opencv",
+        backend: Literal[
+            "opencv", "pyav", "torchcodec", "pynvvideocodec", "deepstream"
+        ] = "opencv",
         **kwargs,
     ) -> tuple[npt.NDArray, dict[str, Any]]:
         return super().load_bytes(
@@ -1152,7 +1291,9 @@ def load_bytes(
         max_duration: int = 300,
         frame_recovery: bool = False,
         *,
-        backend: Literal["opencv", "pyav", "torchcodec", "pynvvideocodec"] = "opencv",
+        backend: Literal[
+            "opencv", "pyav", "torchcodec", "pynvvideocodec", "deepstream"
+        ] = "opencv",
         **kwargs,
     ) -> tuple[npt.NDArray, dict[str, Any]]:
         return super().load_bytes(
@@ -1244,7 +1385,9 @@ def load_bytes(
         max_duration: int = 300,
         frame_recovery: bool = False,
         *,
-        backend: Literal["opencv", "pyav", "torchcodec", "pynvvideocodec"] = "opencv",
+        backend: Literal[
+            "opencv", "pyav", "torchcodec", "pynvvideocodec", "deepstream"
+        ] = "opencv",
         **kwargs,
     ) -> tuple[npt.NDArray, dict[str, Any]]:
         return super().load_bytes(
@@ -1369,7 +1512,9 @@ def load_bytes(
         max_duration: int = 300,
         frame_recovery: bool = False,
         *,
-        backend: Literal["opencv", "pyav", "torchcodec", "pynvvideocodec"] = "opencv",
+        backend: Literal[
+            "opencv", "pyav", "torchcodec", "pynvvideocodec", "deepstream"
+        ] = "opencv",
         **kwargs,
     ) -> tuple[npt.NDArray, dict[str, Any]]:
         return super().load_bytes(
@@ -1467,7 +1612,9 @@ def load_bytes(
         max_duration: int = 300,
         frame_recovery: bool = False,
         *,
-        backend: Literal["opencv", "pyav", "torchcodec", "pynvvideocodec"] = "opencv",
+        backend: Literal[
+            "opencv", "pyav", "torchcodec", "pynvvideocodec", "deepstream"
+        ] = "opencv",
         **kwargs,
     ) -> tuple[npt.NDArray, dict[str, Any]]:
         frames, metadata = super().load_bytes(
@@ -1790,7 +1937,9 @@ def load_bytes(
         max_duration: int = 300,
         frame_recovery: bool = False,
         *,
-        backend: Literal["opencv", "pyav", "torchcodec", "pynvvideocodec"] = "opencv",
+        backend: Literal[
+            "opencv", "pyav", "torchcodec", "pynvvideocodec", "deepstream"
+        ] = "opencv",
         **kwargs,
     ) -> tuple[npt.NDArray, dict[str, Any]]:
         frames, metadata = super().load_bytes(