Utilizing all NVDEC engines on a single GPU with multiple pipelines

[olympiquemarcel]

Describe the question.

When running some video decoding benchmarks by building a DALIGenericIterator with a single pipeline on my GH200 system and checking the NVDEC engine utilization via

nvidia-smi dmon -s u

I find my DEC utilization to be capped at ~14%, which would align with only one (out of seven) NVDEC engines being used. According to the documentation I found, the NVIDIA driver should take care of load balancing between the different decoding units.

However, when creating multiple DALI piplines (e.g., seven), I find my DEC utilization to be close to 100%, indicating that all NVDEC engines are used. In raw decoding performance, running multiple pipelines on the same GPU also give me a performance boost.

My questions are:

• Can a single DALI pipeline only use a single decoding unit?
• Is running multiple pipelines on the same GPU the default way to utilize all decoding units?

Check for duplicates

• I have searched the open bugs/issues and have found no duplicates for this bug report

[JanuszL]

Hi @olympiquemarcel,

Thank you for reaching out.

From what I recall, only the experimental video decoder supports parallel video decoding. The standard video reader processes videos sequentially, utilizing a single NVDEC instance at a time.

If you could provide a minimal example illustrating your use case, I’d be happy to test it on our end to confirm my understanding.

[olympiquemarcel]

Please find a minimal example of my benchmark attached. I am using it with a number of .mp4 videos (h264 codec) in resolution 1920x1080. In the code, you can see I adjust the batch size by the number of pipelines to keep GPU memory comparable between runs.

These are my benchmarking results:

Pipeline	Frames per second	NVDEC usage (approx.)
1	45	14%
2	87	28%
4	166	55%
7	275	89%

"""
Minimal DALI multi-pipeline video decoding benchmark.

Creating N pipelines, each owning a disjoint
sub-shard of the file list, keeps all N NVDEC engines busy concurrently.
DALIGenericIterator merges their outputs into one iterator transparently.

Usage
-----
    python dali_multi_pipeline_example.py --data-root /path/to/videos \\
        --num-pipelines 1 2 4 7 --max-batches 50

Requirements
------------
    pip install nvidia-dali-cuda120 torch
"""

import argparse
import glob
import os
import time

import torch
from nvidia.dali import fn, types
from nvidia.dali.pipeline import pipeline_def
from nvidia.dali.plugin.pytorch import DALIGenericIterator, LastBatchPolicy


# ---------------------------------------------------------------------------
# Pipeline definition
# ---------------------------------------------------------------------------

@pipeline_def(prefetch_queue_depth=4, set_affinity=True)
def video_pipeline(filenames, sequence_length, shard_id, num_shards):
    # device="gpu" -> NVDEC hardware decoding; one NVDEC context per pipeline instance
    frames = fn.readers.video(
        filenames=filenames,
        sequence_length=sequence_length,
        shard_id=shard_id,
        num_shards=num_shards,
        random_shuffle=False,
        pad_last_batch=True,
        skip_vfr_check=True,
        name="VideoReader",
        device="gpu",
        dtype=types.UINT8,
        lazy_init=False,
        stick_to_shard=True,
    )
    return frames


# ---------------------------------------------------------------------------
# Loader builder
# ---------------------------------------------------------------------------

def build_loader(filenames, num_pipelines, batch_size, sequence_length, num_threads, device_id, seed=42):
    pipes = []
    for i in range(num_pipelines):
        pipe = video_pipeline(
            filenames=filenames,
            sequence_length=sequence_length,
            shard_id=i,           # each pipeline owns 1/num_pipelines of the data
            num_shards=num_pipelines,
            batch_size=batch_size,
            num_threads=num_threads,
            device_id=device_id,
            seed=seed + i,
        )
        pipe.build()
        pipes.append(pipe)

    return DALIGenericIterator(
        pipes,
        output_map=["frames"],
        reader_name="VideoReader",
        last_batch_policy=LastBatchPolicy.PARTIAL,
        auto_reset=True,
    )


# ---------------------------------------------------------------------------
# Benchmark
# ---------------------------------------------------------------------------

def benchmark(filenames, num_pipelines, batch_size, sequence_length, num_threads, device_id, max_batches):
    loader = build_loader(filenames, num_pipelines, batch_size, sequence_length, num_threads, device_id)

    total_frames = 0
    torch.cuda.synchronize(device_id)
    t0 = time.perf_counter()

    for idx, batch in enumerate(loader):
        clips = torch.cat([b["frames"] for b in batch], dim=0)
        total_frames += clips.shape[0] * clips.shape[1]  # num_clips * frames_per_clip
        if max_batches > 0 and idx + 1 >= max_batches:
            break

    torch.cuda.synchronize(device_id)
    return total_frames / (time.perf_counter() - t0)


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description="Benchmark DALI multi-pipeline video loading throughput."
    )
    parser.add_argument("--data-root",       type=str, required=True,
                        help="Directory containing .mp4 files.")
    parser.add_argument("--num-pipelines",   type=int, nargs="+", default=[1, 2, 4],
                        help="Pipeline counts to benchmark (space-separated).")
    parser.add_argument("--total-batch-size", type=int, default=300,
                        help="Total clips per step, divided evenly across pipelines.")
    parser.add_argument("--sequence-length", type=int, default=8,
                        help="Frames per clip.")
    parser.add_argument("--num-threads",     type=int, default=4,
                        help="CPU threads per pipeline.")
    parser.add_argument("--device-id",       type=int, default=0)
    parser.add_argument("--max-batches",     type=int, default=50,
                        help="Batches per trial (0 = full epoch).")
    args = parser.parse_args()

    # Print environment info
    print(f"PyTorch {torch.__version__}  |  CUDA {torch.version.cuda}")
    try:
        import nvidia.dali as dali
        print(f"DALI {dali.__version__}")
    except Exception:
        pass
    print()

    pattern = os.path.join(args.data_root, "*.mp4")
    filenames = sorted(glob.glob(pattern))
    if not filenames:
        raise FileNotFoundError(f"No .mp4 files found in {args.data_root}")
    print(f"Found {len(filenames)} video files.\n")

    results = {}
    for n in args.num_pipelines:
        print(f"--- num_pipelines={n} ---")
        fps = benchmark(
            filenames=filenames,
            num_pipelines=n,
            batch_size=args.total_batch_size // n, # divide batch size evenly across pipelines to keep GPU memory usage constant
            sequence_length=args.sequence_length,
            num_threads=args.num_threads,
            device_id=args.device_id,
            max_batches=args.max_batches,
        )
        results[n] = fps
        print(f"  Throughput: {fps:,.0f} frames/s\n")

    # Print summary table
    baseline = results[args.num_pipelines[0]]
    print("=" * 45)
    print(f"{'Pipelines':>10}  {'frames/s':>12}  {'speedup':>8}")
    print("-" * 45)
    for n, fps in results.items():
        print(f"{n:>10}  {fps:>12,.0f}  {fps / baseline:>7.2f}x")
    print("=" * 45)


if __name__ == "__main__":
    main()

[JanuszL]

I can confirm that only fn.decoders.video supports parallel decoding of samples. Both fn.readers.video and fn.experimental.readers.video decode sequences one at a time.

I’ll add this to our ToDo list.