Implementation:NVIDIA DALI Pipeline Run Validation

Knowledge Sources	NVIDIA DALI
Domains	Custom_Operators, Data_Pipeline, Testing
Last Updated	2026-02-08 00:00 GMT

Overview

Concrete pipeline execution and validation code provided by the NVIDIA DALI NaiveHistogram example for running a pipeline that includes a custom operator and converting the GPU output to a NumPy array for inspection.

Description

The test_naive_histogram.py script demonstrates the complete end-to-end validation pattern for a custom DALI operator:

1. Plugin Loading: plugin_manager.load_library("./build/libnaivehistogram.so") loads the custom operator into the DALI runtime.

1. Pipeline Definition: The @pipeline_def decorated function naive_hist_pipe() constructs the processing graph:
   • fn.readers.file(files=test_file_list) reads JPEG images from disk.
   • fn.decoders.image(img, device="mixed", output_type=DALIImageType.GRAY) decodes images to single-channel grayscale on the GPU via hybrid CPU/GPU decoding.
   • img.gpu() ensures the data resides on the GPU.
   • fn.naive_histogram(img, n_bins=24) applies the custom histogram operator.

1. Pipeline Instantiation: naive_hist_pipe(batch_size=2, num_threads=1, device_id=0) creates the pipeline with a batch size of 2, one CPU worker thread, and GPU device 0.

1. Execution and Output: pipe.run() executes one iteration, returning a list of TensorList outputs. out[0].as_cpu().as_array() transfers the GPU histogram result to CPU memory and converts it to a NumPy array of shape (2, 24) with dtype int32.

Usage

Run the test script after building the plugin:

export DALI_EXTRA_PATH=/path/to/dali_extra
python test_naive_histogram.py

Or use it as a pytest test function: pytest test_naive_histogram.py::test_naive_histogram.

Code Reference

Source Location

• Repository: NVIDIA DALI
• File: docs/examples/custom_operations/custom_operator/naive_histogram/test_naive_histogram.py (lines 49-52)

Signature

@pipeline_def
def naive_hist_pipe():
    # ... pipeline graph definition ...
    return img

def test_naive_histogram():
    pipe = naive_hist_pipe(batch_size=2, num_threads=1, device_id=0)
    out = pipe.run()
    result = out[0].as_cpu().as_array()  # numpy.ndarray, shape (2, 24), dtype int32

Import

from nvidia.dali import pipeline_def
from nvidia.dali.types import DALIImageType
import nvidia.dali.fn as fn
import nvidia.dali.plugin_manager as plugin_manager
import numpy

I/O Contract

Inputs

Name	Type	Required	Description
batch_size	int	Yes	Number of samples processed per pipeline iteration. Determines the first dimension of the output array.
num_threads	int	Yes	Number of CPU worker threads for data loading and CPU operators.
device_id	int	Yes	CUDA device index (0-based) on which GPU operators execute.
test_file_list	list[str]	Yes	List of file paths to input images. Must contain at least batch_size entries.
n_bins	int	No (default: 24)	Number of histogram bins, passed to fn.naive_histogram().

Outputs

Name	Type	Description
out	list[TensorListGPU]	List of pipeline outputs from pipe.run(). Each element corresponds to one return value of the pipeline function.
out[0].as_cpu().as_array()	numpy.ndarray	NumPy array of shape (batch_size, n_bins) with dtype int32. Each row contains the histogram for one sample.

Usage Examples

Example: Complete test function

import os
from nvidia.dali import pipeline_def
from nvidia.dali.types import DALIImageType
import nvidia.dali.fn as fn
import nvidia.dali.plugin_manager as plugin_manager

# Load the custom operator
plugin_manager.load_library("./build/libnaivehistogram.so")

# Prepare test files
dali_extra_path = os.environ["DALI_EXTRA_PATH"]
test_file_list = [
    dali_extra_path + "/db/single/jpeg/100/swan-3584559_640.jpg",
    dali_extra_path + "/db/single/jpeg/113/snail-4368154_1280.jpg",
]

@pipeline_def
def naive_hist_pipe():
    img, _ = fn.readers.file(files=test_file_list)
    img = fn.decoders.image(img, device="mixed", output_type=DALIImageType.GRAY)
    img = img.gpu()
    img = fn.naive_histogram(img, n_bins=24)
    return img

def test_naive_histogram():
    pipe = naive_hist_pipe(batch_size=2, num_threads=1, device_id=0)
    out = pipe.run()
    result = out[0].as_cpu().as_array()
    print(result)          # shape: (2, 24), dtype: int32
    assert result.shape == (2, 24)
    assert result.dtype == numpy.int32

Example: Minimal pipeline run

pipe = naive_hist_pipe(batch_size=2, num_threads=1, device_id=0)
out = pipe.run()
print(out[0].as_cpu().as_array())

Related Pages

Implements Principle

• Principle:NVIDIA_DALI_Pipeline_Validation

Requires Environment

• Environment:NVIDIA_DALI_CUDA_GPU_Environment