PyTorch Distributed Elastic のタイムアウト処理に関するトラブルシューティング

PyTorch Distributed Elastic における RendezvousTimeout.close の解説

RendezvousTimeout.close は、PyTorch Distributed Elastic で使用される RendezvousTimeout クラスのメソッドです。これは、分散訓練ジョブにおけるタイムアウト処理を制御するために使用されます。

詳細

RendezvousTimeout クラスは、以下の属性とメソッドを持ちます。

• 属性

  • timeout: タイムアウト時間 (秒)
  • closed: タイムアウトが発生したかどうか

• メソッド

  • close: タイムアウト処理を開始
  • is_timed_out: タイムアウトが発生したかどうかを確認

RendezvousTimeout.close メソッドは、以下の状況で使用されます。

• ランク0プロセスが、すべてのランクが待ち合わせ場所に集合するのを待機している間にタイムアウトが発生した場合
• ランク0プロセスが、すべてのランクから訓練データを受け取るのを待機している間にタイムアウトが発生した場合

コード例

from torch.distributed.elastic.rendezvous import RendezvousTimeout

# タイムアウト時間を設定
timeout = RendezvousTimeout(timeout=10)

# タイムアウト処理を開始
timeout.close()

# タイムアウトが発生したかどうかを確認
if timeout.is_timed_out():
    # タイムアウトが発生した場合の処理
    ...

注意事項

• RendezvousTimeout.close メソッドは、ランク0プロセスでのみ呼び出す必要があります。
• タイムアウトが発生すると、訓練ジョブは失敗します。

補足

• RendezvousTimeout クラスは、PyTorch Distributed Elastic の内部で使用されるクラスです。直接使用する必要はありませんが、タイムアウト処理の仕組みを理解するために役立ちます。
• タイムアウト時間は、訓練ジョブの規模やネットワーク環境によって調整する必要があります。

PyTorch Distributed Elastic における RendezvousTimeout.close のサンプルコード

ランク0プロセスにおけるタイムアウト処理

from torch.distributed.elastic.rendezvous import RendezvousTimeout

# タイムアウト時間を設定
timeout = RendezvousTimeout(timeout=10)

# ランク0プロセスが、すべてのランクが待ち合わせ場所に集合するのを待機
world_size = get_world_size()
gathered = torch.distributed.barrier()

# タイムアウトが発生したかどうかを確認
if not gathered and timeout.is_timed_out():
    # タイムアウトが発生した場合の処理
    print("Rendezvous timed out!")
    sys.exit(1)

# 訓練ジョブの続きを実行
...

ランク0プロセスにおける訓練データの受信タイムアウト処理

from torch.distributed.elastic.rendezvous import RendezvousTimeout

# タイムアウト時間を設定
timeout = RendezvousTimeout(timeout=10)

# ランク0プロセスが、すべてのランクから訓練データを受け取るのを待機
for rank in range(1, world_size):
    data = receive_data_from_rank(rank)

# タイムアウトが発生したかどうかを確認
if not all_data_received and timeout.is_timed_out():
    # タイムアウトが発生した場合の処理
    print("Data receive timed out!")
    sys.exit(1)

# 訓練ジョブの続きを実行
...

PyTorch Distributed Elastic におけるタイムアウト処理の代替方法

torch.distributed.barrier のタイムアウトオプション

torch.distributed.barrier 関数は、すべてのランクが完了するまで待機する関数です。この関数には、タイムアウトオプションを設定することができます。

# タイムアウト時間を設定
timeout = 10

# ランク0プロセスが、すべてのランクが待ち合わせ場所に集合するのを待機
gathered = torch.distributed.barrier(timeout=timeout)

# タイムアウトが発生したかどうかを確認
if not gathered:
    # タイムアウトが発生した場合の処理
    print("Rendezvous timed out!")
    sys.exit(1)

# 訓練ジョブの続きを実行
...

torch.distributed.all_gather 関数は、すべてのランクからデータを収集する関数です。この関数にも、タイムアウトオプションを設定することができます。

# タイムアウト時間を設定
timeout = 10

# ランク0プロセスが、すべてのランクから訓練データを受け取るのを待機
data = torch.distributed.all_gather(data, timeout=timeout)

# タイムアウトが発生したかどうかを確認
if not all_data_received:
    # タイムアウトが発生した場合の処理
    print("Data receive timed out!")
    sys.exit(1)

# 訓練ジョブの続きを実行
...

自身のタイムアウト処理の実装

上記の方法以外にも、独自のタイムアウト処理を実装することができます。

def my_timeout_handler(timeout):
    # タイムアウトが発生した場合の処理
    print("My timeout handler called!")
    sys.exit(1)

# タイムアウトハンドラを設定
torch.distributed.elastic.rendezvous.set_timeout_handler(my_timeout_handler)

# ランク0プロセスが、すべてのランクが待ち合わせ場所に集合するのを待機
gathered = torch.distributed.barrier()

# タイムアウトが発生したかどうかを確認
if not gathered:
    # タイムアウトが発生した場合の処理は、my_timeout_handler で行われる

# 訓練ジョブの続きを実行
...

• すべてのランクが同じタイミングで処理を行う必要がある場合は、torch.distributed.barrier のタイムアウトオプションを使用するのが簡単です。
• ランク0プロセスが、すべてのランクからデータを受け取る必要がある場合は、torch.distributed.all_gather のタイムアウトオプションを使用するのが簡単です。
• より複雑なタイムアウト処理が必要な場合は、独自のタイムアウト処理を実装する必要があります。