dali-dynamic-mode

DALI ダイナミックモード

目的

DALI の命令型ダイナミックモード API nvidia.dali.experimental.dynamic（ndd）を使用するコードの作成、レビュー、移行を AI エージェント向けにガイドします。

手順

• ダイナミックモードを nvidia.dali.experimental.dynamic as ndd としてインポートし、通常の Python で直接 ndd を呼び出すコードを書いてください。Pipeline、@pipeline_def、pipe.build()、pipe.run() などのパイプラインモード API は使用しないでください。
• リーダーは状態を持つものとして扱ってください。リーダーを一度作成したら、エポック間で再利用し、next_epoch(...) に batch_size を渡してください。
• ランダム演算には明示的に batch_size を渡してください。引き継ぐパイプラインレベルのバッチサイズはありません。
• ダイナミックモード API の規則に従ってください。パイプラインモードの "mixed" ではなく device="gpu" を使用し、サンプルの選択には Batch.tensors[...]、サンプルごとのスライスには Batch.slice[...] を使用します。
• テンソルまたはバッチを PyTorch テンソルに変換するには .torch() を使用してください。形状が可変のバッチには pad=True を使用してください。

前提条件

• コードの実行または検証には、ダイナミックモードを nvidia.dali.experimental.dynamic としてインポートできる状態で NVIDIA DALI をインストールする必要があります。
• 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 テンソルを直接取得できます。.cpu() が必要なのは、ホストメモリを必要とするコンシューマー（numpy、__array__）に渡す場合のみです。
• 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

.evaluate() の呼び出しをあちこちに追加する代わりに、デバッグには EvalMode.sync_full を使用してください。コードがすっきりし、すべての問題を一度に検出できます。多くの場合、sync_cpu で十分であり、sync_full よりも軽量です。

スレッド構成

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

CPU オペレーター用の DALI 内部ワーカースレッド数を制御します。デフォルトでは、CPU アフィニティ数または DALI_NUM_THREADS 環境変数が使用されます。Python レベルのスレッド処理とは無関係です。

RNG

次の 2 つの方法があります（どちらか一方を使用し、併用しないでください）。

# 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 は異なるチェックポイントフォーマットバージョンのペイロードを拒否します -- 自動アップグレードはありません。
• スレッドセーフではありません。 1 スレッドにつき 1 つの Checkpoint です。

手動の get_state / set_state は各 Reader と RNG で直接利用することもできます -- 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 の往復を避ける
エポックごとにリーダーを再作成	reader.next_epoch()	リーダーはステートフル -- 一度作成して再利用する
ndd.readers.file(...)	ndd.readers.File(...)	リーダークラスはパスカルケース
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(...)（パスカルケース、状態を保持）
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。リーダーでは enable_checkpointing=True を指定して有効にします

制限事項

ダイナミックモードはパイプラインモードより柔軟ですが、パフォーマンスがやや低下する場合があります。最大スループットを得るには、パイプラインモードをお使いください。

トラブルシューティング

• エラーが失敗した呼び出しより後で発生する場合は、EvalMode.sync_cpu または EvalMode.sync_full の下でブロックを再実行してください。
• エポック間でリーダーの動作が予期せず変わる場合は、リーダーが一度だけ作成され、各 next_epoch() イテレーターが最後まで消費されていることを確認してください。