invalid argument Current pipeline object is no longer valid.

[zjliu-cs]

Describe the question.

Hello,
I am encountering an intermittent error when using NVIDIA DALI to load data. The same code sometimes runs successfully, but sometimes fails with a critical pipeline error. I would like to understand the possible reasons for this behavior.
Additionally, the failure location is not consistent: across different runs, the error may be reported in different operators or different parts of the code.

Critical error in pipeline:
Error in CPU operator 'nvidia.dali.fn.decoders.image',
which was used in the pipeline definition with the following traceback:

  File ".../dali_loader.py", line 36, in create_lmdb_dali_train_pipeline
    images = fn.decoders.image(images, device='cpu', output_type=types.RGB)

encountered:

Error in thread 0: CUDA runtime API error cudaErrorInvalidValue (1):
invalid argument
Current pipeline object is no longer valid.

Error in CPU operator 'nvidia.dali.fn.ones',

  File ".../dali_loader.py", line 64, in create_lmdb_dali_train_pipeline
    mask = fn.ones(shape=(crop, crop)).gpu()

CUDA runtime API error cudaErrorInvalidValue (1):
invalid argument
Current pipeline object is no longer valid.

Environment
OS: Ubuntu
GPU: A100
CUDA version: 12.2
NVIDIA DALI version: 1.51.2
PyTorch version: 2.4.1
Python version: 3.11.13
Data format: LMDB (ImageNet-style)

Check for duplicates

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

[JanuszL]

Hi @zjliu-cs,

Thank you for reaching out.

This behavior is unexpected, especially since you're seeing a CUDA error from a CPU operator. Could you please share a complete reproduction script that we can run on our end to help identify the issue?

[zjliu-cs]

Below is my DALI data loading and preprocessing script. thank you！

@pipeline_def(enable_conditionals=True)
def create_lmdb_dali_train_pipeline(crop, shard_id, num_shards):
    lmdb_path = [os.path.join(train_root_path, file) for file in os.listdir(path=train_root_path)]

    images, labels = fn.readers.caffe(
        path=lmdb_path,
        shard_id=shard_id,
        num_shards=num_shards,
        random_shuffle=True,
        name='Reader',
        prefetch_queue_depth=2
    )
    images = fn.decoders.image(images, device='cpu', output_type=types.RGB)
    images = images.gpu()
    images = fn.random_resized_crop(
        images,
        size=crop,
        random_area=[0.08, 1.0],
        random_aspect_ratio=[3 / 4, 4 / 3],
        device='gpu',
        interp_type=types.INTERP_TRIANGULAR
    )
    images = rand_augment.rand_augment(images, n=2, m=9)

    labels = labels.gpu()
    shuffled_indices = fn.batch_permutation(allow_repetitions=True, no_fixed_points=True)
    shuffled_labels = fn.permute_batch(labels, indices=shuffled_indices, device='gpu')

    images = fn.cast(images, dtype=types.FLOAT, device='gpu')
    shuffled_images = fn.permute_batch(images, device='gpu', indices=shuffled_indices)

    use_mixup = fn.random.coin_flip(probability=0.5, device='cpu')
    if use_mixup:
        lam = fn.random.beta(alpha=0.8, beta=0.8, device='cpu').gpu()
        images = images * lam + shuffled_images * (1. - lam)
    else:
        hw = fn.random.uniform(range=[0.3, 0.8], shape=(2), device='cpu')
        lam = (1. - hw[0] * hw[1]).gpu()
        mask = fn.ones(shape=(crop, crop)).gpu()
        mask = fn.erase(mask,
                        fill_value=0,
                        anchor=fn.random.uniform(range=[0., 1.], shape=(2)) * (1. - hw),
                        shape=hw, normalized=True, axes=(0, 1), device='gpu')
        mask = fn.expand_dims(mask, axes=[2], device='gpu')
        images = mask * images + (1 - mask) * shuffled_images

    use_erase = fn.random.coin_flip(probability=0.25)
    if use_erase:
        erase_shape = fn.random.uniform(range=[0.2, 0.5], shape=(2,), device='cpu')  # 擦除的宽高比例
        images = fn.erase(
            images,
            anchor=fn.random.uniform(range=[0.0, 1.], shape=(2,), device='cpu') * (1.0 - erase_shape),
            shape=erase_shape,
            fill_value=fn.random.uniform(range=[0, 255], shape=(3,), device='cpu'),
            axes=(1, 2),
            normalized_anchor=True,
            normalized_shape=True,
            device='gpu')

    images = fn.crop_mirror_normalize(
        images,
        dtype=types.FLOAT,
        mean=[0.485 * 255, 0.456 * 255, 0.406 * 255],
        std=[0.229 * 255, 0.224 * 255, 0.225 * 255],
        device='gpu',
        output_layout="CHW",
        mirror=fn.random.coin_flip(probability=0.5),
    )

    return images, labels, shuffled_labels, lam


@pipeline_def()
def create_lmdb_dali_val_pipeline(crop, size, shard_id, num_shards):
    images, labels = fn.readers.caffe(
        path=val_root_path,
        shard_id=shard_id,
        num_shards=num_shards,
        name='Reader',
        prefetch_queue_depth=2
    )
    images = fn.decoders.image(images, device='cpu', output_type=types.RGB)
    images = images.gpu()
    images = fn.resize(
        images,
        device='gpu',
        resize_shorter=size,
        interp_type=types.INTERP_TRIANGULAR
    )
    images = fn.crop_mirror_normalize(
        images,
        dtype=types.FLOAT,
        crop=(crop, crop),
        mean=[0.485 * 255, 0.456 * 255, 0.406 * 255],
        std=[0.229 * 255, 0.224 * 255, 0.225 * 255],
        device='gpu',
        output_layout="CHW"
    )
    return images, labels


class TrainDALIWrapper:
    def __init__(self, dali_iter):
        self.dali_iter = dali_iter

    def __iter__(self):
        return self

    def __next__(self):
        data = next(self.dali_iter)
        images = data[0]["images"]
        labels_a = data[0]["labels_a"].squeeze().long()
        labels_b = data[0]["labels_b"].squeeze().long()
        lam = data[0]['lam'].squeeze().float()
        return images, labels_a, labels_b, lam

    def __len__(self):
        return len(self.dali_iter)

    def reset(self):
        self.dali_iter.reset()


class ValDALIWrapper:

    def __init__(self, dali_iter):
        self.dali_iter = dali_iter

    def __iter__(self):
        return self

    def __next__(self):
        data = next(self.dali_iter)
        images = data[0]["data"]
        labels = data[0]["label"].squeeze().long()
        return images, labels

    def __len__(self):
        return len(self.dali_iter)

    def reset(self):
        self.dali_iter.reset()


def create_dali_lmdb_loaders(
        batch_size, num_workers, shard_id, device_id, world_size, crop_size=224, val_size=256):
    if not DALI_AVAILABLE:
        raise ImportError("NVIDIA DALI is required but not installed")

    train_pipe = create_lmdb_dali_train_pipeline(
        crop=crop_size,
        shard_id=shard_id,
        num_shards=world_size,
        batch_size=batch_size,
        num_threads=num_workers,
        device_id=device_id,
    )

    val_pipe = create_lmdb_dali_val_pipeline(
        crop=crop_size,
        size=val_size,
        shard_id=shard_id,
        num_shards=world_size,
        batch_size=batch_size,
        num_threads=num_workers,
        device_id=device_id,
    )

    train_pipe.build()
    val_pipe.build()

    train_loader = DALIGenericIterator(
        train_pipe,
        reader_name="Reader",
        output_map=["images", "labels_a", "labels_b", "lam"],
        last_batch_policy=LastBatchPolicy.DROP,
        auto_reset=True
    )

    val_loader = DALIClassificationIterator(
        val_pipe,
        reader_name="Reader",
        last_batch_policy=LastBatchPolicy.DROP,
        auto_reset=True
    )

    return TrainDALIWrapper(train_loader), ValDALIWrapper(val_loader)

[zjliu-cs]

At the beginning, we usedimages = fn.decoders.image(images, device='mixed', output_type=types.RGB),
but it still resulted in the same error, so we later replaced it with the code images = fn.decoders.image(images, device='cpu', output_type=types.RGB) images = images.gpu().

[JanuszL]

Hi @zjliu-cs,

I run this script multiple times in a clean docker env:

import os
import nvidia.dali.fn as fn
import nvidia.dali.types as types
from nvidia.dali.pipeline import pipeline_def
from nvidia.dali.auto_aug import rand_augment, auto_augment
from nvidia.dali.plugin.pytorch import DALIClassificationIterator, LastBatchPolicy, DALIGenericIterator

train_root_path = [os.path.join(os.environ.get("DALI_EXTRA_PATH"), "db", "lmdb")]
val_root_path = [os.path.join(os.environ.get("DALI_EXTRA_PATH"), "db", "lmdb")]

@pipeline_def(enable_conditionals=True)
def create_lmdb_dali_train_pipeline(crop, shard_id, num_shards):
    lmdb_path = train_root_path

    images, labels = fn.readers.caffe(
        path=lmdb_path,
        shard_id=shard_id,
        num_shards=num_shards,
        random_shuffle=True,
        name='Reader',
        prefetch_queue_depth=2
    )
    images = fn.decoders.image(images, device='cpu', output_type=types.RGB)
    images = images.gpu()
    images = fn.random_resized_crop(
        images,
        size=crop,
        random_area=[0.08, 1.0],
        random_aspect_ratio=[3 / 4, 4 / 3],
        device='gpu',
        interp_type=types.INTERP_TRIANGULAR
    )
    images = rand_augment.rand_augment(images, n=2, m=9)

    labels = labels.gpu()
    shuffled_indices = fn.batch_permutation(allow_repetitions=True, no_fixed_points=True)
    shuffled_labels = fn.permute_batch(labels, indices=shuffled_indices, device='gpu')

    images = fn.cast(images, dtype=types.FLOAT, device='gpu')
    shuffled_images = fn.permute_batch(images, device='gpu', indices=shuffled_indices)

    use_mixup = fn.random.coin_flip(probability=0.5, device='cpu')
    if use_mixup:
        lam = fn.random.beta(alpha=0.8, beta=0.8, device='cpu').gpu()
        images = images * lam + shuffled_images * (1. - lam)
    else:
        hw = fn.random.uniform(range=[0.3, 0.8], shape=(2), device='cpu')
        lam = (1. - hw[0] * hw[1]).gpu()
        mask = fn.ones(shape=(crop, crop)).gpu()
        mask = fn.erase(mask,
                        fill_value=0,
                        anchor=fn.random.uniform(range=[0., 1.], shape=(2)) * (1. - hw),
                        shape=hw, normalized=True, axes=(0, 1), device='gpu')
        mask = fn.expand_dims(mask, axes=[2], device='gpu')
        images = mask * images + (1 - mask) * shuffled_images

    use_erase = fn.random.coin_flip(probability=0.25)
    if use_erase:
        erase_shape = fn.random.uniform(range=[0.2, 0.5], shape=(2,), device='cpu')  # 擦除的宽高比例
        images = fn.erase(
            images,
            anchor=fn.random.uniform(range=[0.0, 1.], shape=(2,), device='cpu') * (1.0 - erase_shape),
            shape=erase_shape,
            fill_value=fn.random.uniform(range=[0, 255], shape=(3,), device='cpu'),
            axes=(1, 2),
            normalized_anchor=True,
            normalized_shape=True,
            device='gpu')

    images = fn.crop_mirror_normalize(
        images,
        dtype=types.FLOAT,
        mean=[0.485 * 255, 0.456 * 255, 0.406 * 255],
        std=[0.229 * 255, 0.224 * 255, 0.225 * 255],
        device='gpu',
        output_layout="CHW",
        mirror=fn.random.coin_flip(probability=0.5),
    )

    return images, labels, shuffled_labels, lam


@pipeline_def()
def create_lmdb_dali_val_pipeline(crop, size, shard_id, num_shards):
    images, labels = fn.readers.caffe(
        path=val_root_path,
        shard_id=shard_id,
        num_shards=num_shards,
        name='Reader',
        prefetch_queue_depth=2
    )
    images = fn.decoders.image(images, device='cpu', output_type=types.RGB)
    images = images.gpu()
    images = fn.resize(
        images,
        device='gpu',
        resize_shorter=size,
        interp_type=types.INTERP_TRIANGULAR
    )
    images = fn.crop_mirror_normalize(
        images,
        dtype=types.FLOAT,
        crop=(crop, crop),
        mean=[0.485 * 255, 0.456 * 255, 0.406 * 255],
        std=[0.229 * 255, 0.224 * 255, 0.225 * 255],
        device='gpu',
        output_layout="CHW"
    )
    return images, labels


class TrainDALIWrapper:
    def __init__(self, dali_iter):
        self.dali_iter = dali_iter

    def __iter__(self):
        return self

    def __next__(self):
        data = next(self.dali_iter)
        images = data[0]["images"]
        labels_a = data[0]["labels_a"].squeeze().long()
        labels_b = data[0]["labels_b"].squeeze().long()
        lam = data[0]['lam'].squeeze().float()
        return images, labels_a, labels_b, lam

    def __len__(self):
        return len(self.dali_iter)

    def reset(self):
        self.dali_iter.reset()


class ValDALIWrapper:

    def __init__(self, dali_iter):
        self.dali_iter = dali_iter

    def __iter__(self):
        return self

    def __next__(self):
        data = next(self.dali_iter)
        images = data[0]["data"]
        labels = data[0]["label"].squeeze().long()
        return images, labels

    def __len__(self):
        return len(self.dali_iter)

    def reset(self):
        self.dali_iter.reset()


def create_dali_lmdb_loaders(
        batch_size, num_workers, shard_id, device_id, world_size, crop_size=224, val_size=256):

    train_pipe = create_lmdb_dali_train_pipeline(
        crop=crop_size,
        shard_id=shard_id,
        num_shards=world_size,
        batch_size=batch_size,
        num_threads=num_workers,
        device_id=device_id,
    )

    val_pipe = create_lmdb_dali_val_pipeline(
        crop=crop_size,
        size=val_size,
        shard_id=shard_id,
        num_shards=world_size,
        batch_size=batch_size,
        num_threads=num_workers,
        device_id=device_id,
    )

    train_pipe.build()
    val_pipe.build()

    train_loader = DALIGenericIterator(
        train_pipe,
        reader_name="Reader",
        output_map=["images", "labels_a", "labels_b", "lam"],
        last_batch_policy=LastBatchPolicy.DROP,
        auto_reset=True
    )

    val_loader = DALIClassificationIterator(
        val_pipe,
        reader_name="Reader",
        last_batch_policy=LastBatchPolicy.DROP,
        auto_reset=True
    )

    return TrainDALIWrapper(train_loader), ValDALIWrapper(val_loader)

i1 ,i2 = create_dali_lmdb_loaders(batch_size=1, num_workers=1, shard_id=0, device_id=0, world_size=1)
for i in i1:
    pass
for i in i2:
    pass
print("OK")

using

docker run --gpus all --rm -ti ubuntu:24.04

apt update && apt install -y nano python3-pip python3.12-venv python3-full git-lfs && \
git clone https://github.com/NVIDIA/DALI_extra /opt/dali_extra && \
python3 -m venv /opt/python3.12 && \
source /opt/python3.12/bin/activate && \
pip install nvidia-dali-cuda130 torch

export DALI_EXTRA_PATH=/opt/dali_extra
for i in {1..10}
do
  echo "Running test $i" && \
  python test.py;
done

Driver Version: 580.56

and I found no issues.