dali-dynamic-mode

DALI Dynamic Mode

用途

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

说明

• 将动态模式导入为 nvidia.dali.experimental.dynamic as ndd，并在普通 Python 中直接调用 ndd 编写代码；不要使用管道模式 API，例如 Pipeline、@pipeline_def、pipe.build() 或 pipe.run()。
• 将读取器视为有状态对象：创建一次，在各个周期中重复使用，并向 next_epoch(...) 传入 batch_size。
• 为随机算子显式传入 batch_size；没有可继承的管道级批大小。
• 使用动态模式 API 约定：使用 device="gpu"，而不是管道模式的 "mixed"；使用 Batch.tensors[...] 选择样本，使用 Batch.slice[...] 对每个样本进行切片。
• 使用 .torch() 将张量或批次转换为 PyTorch 张量。对于形状可变的批次，使用 pad=True。

前置条件

• 若要运行或验证代码，必须安装 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.tensors[i]	Tensor
获取样本子集	batch.tensors[slice_or_list]	Batch
对每个样本内部切片	batch.slice[...]	Batch（batch_size 不变）
按样本切片	batch.slice[batch_of_indices]	Batch（batch_size 不变）

.tensors[] 用于选择哪些样本。.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.tensors[0]   # Tensor, the entire first sample [x, y]

高级切片

.slice[] API 接受索引批次，允许用户组合批次和标量值，例如：

imgs = ndd.imread(filenames)  # a batch of images, if `filenames` is a list
sliced = imgs.slice[
    42 :  # the range start is broadcast to all samples
    ndd.batch(imgs.shape).slice[0] // 2  # per-sample range stop (half of each image)
]

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_cpu:
    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.tensors[i]	Batch 没有 __getitem__
用 batch.tensors[0] 进行逐样本切片	batch.slice[0]	.tensors 选取样本；.slice 在每个样本内切片
每个算子之后调用 .evaluate()	让数据消费触发求值	.torch()、.shape 等会自动触发求值
在 GPU 模型之前调用 .cpu()	直接调用 .torch()	避免浪费性的 D2H + H2D 往返传输
每个 epoch 都重新创建读取器	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（或在首次迭代前注册）	否则后端不会保留快照
明确写出默认参数值	省略默认参数值	Python 端开销非常高，尤其是在参数接受 Tensor/Batch 时。省略参数会使用快速路径，实际上传入一个哨兵值。

流水线模式迁移

流水线模式	动态模式
@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(...)
流水线级别的 batch_size=64	reader.next_epoch(batch_size=64) + 随机算子的 batch_size=64
流水线级别的 seed=42	ndd.random.set_seed(42) 或 ndd.random.RNG(seed=42)
流水线级别的 num_threads=4	在启动时调用 ndd.set_num_threads(4)
output.at(i)	batch.tensors[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；Reader 需通过 enable_checkpointing=True 启用此功能

限制

动态模式比流水线模式更加灵活，但性能可能略差。若要获得最高吞吐量，请优先使用流水线模式。

故障排除

• 如果错误在出错调用之后才出现，请在 EvalMode.sync_cpu 或 EvalMode.sync_full 下重新运行该代码块。
• 如果 Reader 在不同 epoch 中表现异常，请检查它是否只创建了一次，以及每个 next_epoch() 迭代器是否都已完全遍历。