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[...] を使用してください。

前提条件

• コードの実行または検証には、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.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 で使用したリーダーの 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_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

.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} プレースホルダーを1つ含む Python のフォーマット文字列を受け取ります（例："ckpt_{seq:04d}.json"）。save は次に空いているシーケンス番号を選び、load はディスク上で一致するもののうち最大の番号を選びます。
• フォーマットのバージョンは厳密です。 deserialize は、異なるチェックポイント形式のバージョンのペイロードを拒否し、自動アップグレードは行いません。
• スレッドセーフではありません。 スレッドごとに Checkpoint を1つ使用してください。

各 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(...)	リーダークラスはパスカルケースです
next_epoch() のループから break する	イテレーターを最後まで処理するか、新しいリーダーを作成する	次の next_epoch() を呼び出す前に、イテレーターを最後まで処理する必要があります
ランダム演算に batch_size を指定しない	ndd.random.uniform(batch_size=N, ...)	継承するパイプラインレベルのバッチサイズはありません
最初の next_epoch の後に register(reader) を呼び出して復元する	新しく作成したリーダーを最初のイテレーションより前に登録する	リーダーの状態を適用できるのは、プリフェッチスレッドの開始前だけです
enable_checkpointing=True を指定せずに作成したリーダーで、イテレーション後に復元する	作成時に enable_checkpointing=True を渡す（または最初のイテレーションより前に登録する）	それ以外の場合、バックエンドはスナップショットを保持しません

パイプラインモードからの移行

パイプラインモード	ダイナミックモード
@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.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 でオプトインする

制限事項

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

トラブルシューティング

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