dali-dynamic-mode

DALI 动态模式

用途

指导 AI Agent 编写、审查和迁移使用 DALI 命令式动态模式 API nvidia.dali.experimental.dynamic（ndd）的代码。

说明

• 将动态模式导入为 nvidia.dali.experimental.dynamic as ndd，并在普通 Python 代码中直接调用 ndd；不要使用 Pipeline、@pipeline_def、pipe.build() 或 pipe.run() 等流水线模式 API。
• 将读取器视为有状态对象：只创建一次，在多个轮次间重复使用，并将 batch_size 传给 next_epoch(...)。
• 为随机算子显式传入 batch_size；没有可供继承的流水线级批次大小。
• 遵循动态模式 API 约定：使用 device="gpu"，而不是流水线模式中的 "mixed"；使用 Batch.tensors[...] 选择样本，使用 Batch.slice[...] 对每个样本进行切片。

前提条件

• 要运行或验证代码，必须安装 NVIDIA DALI，并确保可以导入动态模式 nvidia.dali.experimental.dynamic。
• GPU 解码或 GPU 算子需要支持 CUDA 的 DALI 构建版本，以及可用的 NVIDIA GPU／驱动程序。
• 框架转换示例需要安装目标框架，例如，使用 .torch() 时需要安装 PyTorch。

简介

动态模式是 DALI 的命令式 Python API。它支持在普通 Python 控制流中直接调用 DALI 算子，而不是构建并运行流水线图。

核心数据类型

Tensor——单个样本

t = ndd.tensor(data)           # copy
t = ndd.as_tensor(data)        # wrap, no copy if possible
t.cpu()                        # move to CPU
t.gpu()                        # move to GPU
t.torch(copy=False)            # conversion to PyTorch tensor with no copy (default)
t[1:3]                         # slicing supported
np.asarray(t)                  # NumPy via __array__ (CPU only)

支持 __dlpack__、__cuda_array_interface__、__array__ 和算术运算符。

Batch——样本集合（支持形状不一的样本）

b = ndd.batch([arr1, arr2])    # copy
b = ndd.as_batch(data)         # wrap, no copy if possible

Batch 没有 __getitem__——batch[i] 会引发 TypeError，因为索引的含义不明确（选择样本还是对每个样本切片）。请改用以下显式 API：

用途	方法	返回值
获取样本 i	batch.select(i)	Tensor
获取样本子集	batch.select(slice_or_list)	Batch
对每个样本内部切片	batch.slice[...]	Batch（batch_size 不变）

.select() 用于选择哪些样本。.slice 用于索引每个样本内部。

xy = ndd.random.uniform(batch_size=16, range=[0, 1], shape=2)
crop_x = xy.slice[0]       # Batch of 16 scalars, first element from each sample
crop_y = xy.slice[1]       # Batch of 16 scalars, second element from each sample
sample_0 = xy.select(0)    # Tensor, the entire first sample [x, y]

转换为 PyTorch：

• batch.torch()——适用于形状一致的批次；形状不一时会引发错误
• batch.torch(pad=True)——将形状不一的批次补零至最大形状（适用于变长音频、检测框等）
• batch.torch(copy=None) 是默认设置（尽可能避免复制）
• Batch 没有 __dlpack__——供 DLPack 消费者使用时，请先调用 ndd.as_tensor(batch)。ndd.as_tensor 同样支持 pad。
• Tensor.torch(copy=False) 是默认设置（不复制）

迭代：for sample in batch: 会生成 Tensor。

读取器

读取器是有状态对象——只创建一次，并在多个轮次间重复使用。这很重要，因为读取器会跟踪打乱顺序、分片位置等内部状态。

reader = ndd.readers.File(file_root=image_dir, random_shuffle=True)

for epoch in range(num_epochs):
    for jpegs, labels in reader.next_epoch(batch_size=64):
        # jpegs, labels are Batch objects
        ...

要点：

• 读取器的输出（JPEG 图像、标签等）是 CPU 张量／批次。标签通常保留在 CPU 上，直到将其转换为框架使用的格式（例如，labels.torch().to(device)）。
• 读取器类采用 PascalCase 命名：ndd.readers.File(...)、ndd.readers.COCO(...)、ndd.readers.TFRecord(...)
• batch_size 应传给 next_epoch()，而不是读取器构造函数
• next_epoch(batch_size=N) 会生成由 Batch 组成的元组；不带 batch_size 调用 next_epoch() 时，会生成由 Tensor 组成的元组
• 再次调用 next_epoch() 前，必须完整耗尽 next_epoch() 返回的迭代器
• 读取器一旦使用某个 batch_size，就不能更改。同样，使用批次模式的读取器不能切换到样本模式，反之亦然。

用于分布式训练的分片读取：

reader = ndd.readers.File(
    file_root=image_dir,
    shard_id=rank, num_shards=world_size,
    stick_to_shard=True,
    pad_last_batch=True,
)

设备处理

• 设备会根据输入推断——只要任一输入位于 GPU 上，就使用 GPU
• 对于混合解码：使用 device="gpu"（不要使用 "mixed"）。"mixed" 关键字是流水线模式的概念，用于隐式地将数据从 CPU 传输到 GPU；在动态模式下，传入 device="gpu" 会触发相同的硬件加速解码路径。
• 将数据传给 GPU 模型前，不要调用 .cpu()——.torch() 会直接返回 GPU 张量。只有需要主机内存的使用方（如 numpy、__array__）才需要调用 .cpu()。
• DALI 与 PyTorch 之间的 CUDA 流同步会通过 DLPack 自动完成——无需手动管理流。

执行模式

默认模式为 eager——在后台线程中异步执行，并立即返回。

大多数情况下无需调用 .evaluate()。 任何数据消费操作（.torch()、__dlpack__、__array__、.shape、属性访问、迭代）都会自动触发求值。

调试时，切换到同步模式，让错误在确切的调用位置显现，而不是稍后才在异步队列中出现：

with ndd.EvalMode.sync_full:
    images = ndd.decoders.image(jpegs, device="gpu")
    images = ndd.resize(images, size=[224, 224])
    # Any error surfaces here, at the exact op that failed

模式（同步程度递增）：deferred < eager < sync_cpu < sync_full

调试时使用 EvalMode.sync_full，而不是到处调用 .evaluate()——这样更简洁，也能一次性捕获所有问题。sync_cpu 通常已足够，而且比 sync_full 更轻量。

线程配置

ndd.set_num_threads(4)  # Call once at startup, only if necessary to override the defaults

用于控制 DALI 内部 CPU 运算符的工作线程。默认值为 CPU 亲和性计数，或 DALI_NUM_THREADS 环境变量的值。与 Python 层级的线程处理无关。

随机数生成器

两种方式（二选一，不要同时使用）：

# Approach 1: set the thread-local default seed (simple, good enough for most cases)
ndd.random.set_seed(42)
angles = ndd.random.uniform(batch_size=64, range=(-30, 30))

# Approach 2: explicit RNG object (finer control, pass rng= to each op)
rng = ndd.random.RNG(seed=42)
values = ndd.random.uniform(batch_size=64, range=[0, 1], shape=2, rng=rng)

向随机运算传入 rng= 时，显式 RNG 会覆盖默认种子。此设置是线程局部的：每个线程都有独立的随机状态。

处理批次时，随机运算需要显式指定 batch_size——不存在可供继承的流水线级批次大小。

检查点

动态模式没有流水线级检查点。检查点汇总各个有状态对象的状态：读取器和 RNG 实例。无状态运算（解码器、调整大小、旋转、归一化等）不属于检查点。

ckpt = ndd.checkpoint.Checkpoint()
ckpt.register(reader, "my_reader")
ckpt.register(rng, "rng")

# ... iterate for a while ...

ckpt.collect()                       # snapshot the registered objects
ckpt.save("ckpt_{seq:04d}.json")     # writes ckpt_0000.json, ckpt_0001.json, ...

恢复是对称操作——构建一个全新的读取器和 RNG，然后执行 load + register。加载的状态会在 register 时应用到每个对象：

reader = ndd.readers.File(file_root=..., enable_checkpointing=True, name="my_reader")
rng = ndd.random.RNG()

ckpt = ndd.checkpoint.Checkpoint()
ckpt.load("ckpt_{seq:04d}.json")     # picks the highest sequence number
ckpt.register(reader, "my_reader")   # state applied here
ckpt.register(rng, "rng")            # ditto

for batch in reader.next_epoch(batch_size=N):
    ...  # produces the next batch after the checkpointed iteration

关键规则：

• 读取器必须选择启用检查点。 使用 enable_checkpointing=True 构建。注册一个已经迭代过、但未启用该选项的读取器会引发 RuntimeError；如果读取器尚未迭代，register 会追溯启用该选项。
• 必须在首次调用 next_epoch 之前应用读取器状态。 首次迭代时预取线程会启动，此后快照队列会被锁定。对已经迭代过的读取器调用 set_state（或从已加载的检查点调用 register）会引发 RuntimeError。
• enable_checkpointing=True 与 compile=True 不兼容。 对已启用检查点的读取器调用 reader.next_epoch(..., compile=True) 会引发 NotImplementedError。
• 使用命名注册更安全。 匿名调用 register(op) 会使用顺序键（__op_0、__op_1……），因此保存和恢复时的注册顺序必须一致。类型标记可以检测不同类型之间的错配，但无法检测兼容类型之间的顺序变化。建议使用 register(op, name)。
• ndd.checkpoint.current() 会返回绑定到当前线程本地 EvalContext 的 Checkpoint。它会在多次调用之间共享——如果要将默认上下文用于彼此无关的运行，请调用 ckpt.clear()。
• 文件名模式： save/load 接受一个带有单个 {seq} 占位符的 Python 格式字符串（例如 "ckpt_{seq:04d}.json"）。save 会选择下一个可用序号；load 会选择磁盘上匹配的最大序号。
• 格式版本严格。 deserialize 会拒绝使用不同检查点格式版本的载荷——不会自动升级。
• 非线程安全。 每个线程使用一个 Checkpoint。

每个 Reader 和 RNG 本身也提供手动 get_state / set_state；Checkpoint 汇总器基于这些接口构建。仅在与外部检查点系统集成时使用手动 API。

示例

图像分类流水线

import nvidia.dali.experimental.dynamic as ndd

reader = ndd.readers.File(file_root="/data/imagenet/train", random_shuffle=True)

for epoch in range(num_epochs):
    for jpegs, labels in reader.next_epoch(batch_size=64):
        images = ndd.decoders.image(jpegs, device="gpu")
        images = ndd.resize(images, size=[224, 224])
        images = ndd.crop_mirror_normalize(
            images,
            mean=[0.485 * 255, 0.456 * 255, 0.406 * 255],
            std=[0.229 * 255, 0.224 * 255, 0.225 * 255],
        )
        train_step(images.torch(), labels.torch())

常见错误

错误	正确	原因
device="mixed"	device="gpu"	"mixed" 仅适用于流水线模式
batch[i]	batch.select(i)	Batch 没有 __getitem__
使用 batch.select(0) 进行逐样本切片	batch.slice[0]	.select() 选择样本；.slice 在每个样本内部切片
每个运算后都调用 .evaluate()	让数据消费触发计算	.torch()、.shape 等会自动触发计算
在 GPU 模型前调用 .cpu()	直接调用 .torch()	避免浪费 D2H + H2D 往返传输
每个周期都重新创建读取器	reader.next_epoch()	读取器是有状态的——创建一次并重复使用
ndd.readers.file(...)	ndd.readers.File(...)	读取器类使用 PascalCase 命名
从 next_epoch() 循环中 break	耗尽迭代器或创建新读取器	必须完全耗尽迭代器，才能再次调用 next_epoch()
对随机运算不传入 batch_size	ndd.random.uniform(batch_size=N, ...)	没有可供继承的流水线级批次大小
首次调用 next_epoch 后调用 register(reader) 进行恢复	在首次迭代前注册刚构建的读取器	只能在预取线程启动前应用读取器状态
迭代后恢复到构建时未设置 enable_checkpointing=True 的读取器	构建时传入 enable_checkpointing=True（或在首次迭代前注册）	否则后端不会保留快照

Pipeline 模式迁移

Pipeline 模式	动态模式
@pipeline_def / pipe.build() / pipe.run()	在循环中直接调用函数
fn.readers.file(...)	ndd.readers.File(...)（PascalCase，有状态）
fn.decoders.image(jpegs, device="mixed")	ndd.decoders.image(jpegs, device="gpu")
fn.op_name(...)	ndd.op_name(...)
Pipeline 级别的 batch_size=64	reader.next_epoch(batch_size=64) + 随机操作的 batch_size=64
Pipeline 级别的 seed=42	ndd.random.set_seed(42) 或 ndd.random.RNG(seed=42)
Pipeline 级别的 num_threads=4	启动时调用 ndd.set_num_threads(4)
output.at(i)	batch.select(i)
output.as_cpu()	batch.cpu()
pipe.run() 返回 TensorList 元组	reader.next_epoch(batch_size=N) 生成 Batch 元组
Pipeline(..., enable_checkpointing=True) + pipe.checkpoint() / pipeline(checkpoint=...)	ndd.checkpoint.Checkpoint + 每个对象的 register / collect / save / load；读取器通过 enable_checkpointing=True 启用

局限性

动态模式比 Pipeline 模式更灵活，但性能可能略差。如需最大吞吐量，请优先使用 Pipeline 模式。

故障排除

• 如果错误出现得晚于发生故障的调用，请在 with ndd.EvalMode.sync_full: 下重新运行该代码块。
• 如果读取器在不同 epoch 中表现异常，请检查它是否只创建了一次，以及是否完整遍历了每个 next_epoch() 迭代器。