PyTorch Distributed Elasticでスマートに終了処理を行う：RendezvousHandler.shutdown()の使い方

torch.distributed.elastic.rendezvous.RendezvousHandler.shutdown() は、PyTorch Distributed Elastic における分散ワークフローの終了処理を実行する関数です。この関数は、分散ワークフローの終了時に呼び出され、以下のタスクを実行します。

詳細

torch.distributed.elastic.rendezvous.RendezvousHandler.shutdown() 関数は、以下の引数を取ります。

graceful_shutdown (bool)
これが True の場合、関数は他のワーカーが完了するのを待ってから終了処理を実行します。デフォルトは False です。

使用方法

torch.distributed.elastic.rendezvous.RendezvousHandler.shutdown() 関数は、分散ワークフローの終了処理を実行するために使用されます。通常、これはすべてのワーカーが完了した後に呼び出されます。

以下の例は、torch.distributed.elastic.rendezvous.RendezvousHandler.shutdown() 関数の使用方法を示しています。

import torch.distributed.elastic as dist_elastic

# ... 分散ワークフローを実行 ...

dist_elastic.rendezvous.RendezvousHandler.shutdown(graceful_shutdown=True)

注意点

この関数は、一度だけ呼び出す必要があります。
torch.distributed.elastic.rendezvous.RendezvousHandler.shutdown() 関数は、すべてのワーカーが完了するのを待ってから呼び出す必要があります。

torch.distributed.elastic.rendezvous.RendezvousHandler.shutdown() 関数は、以下のエラーを発生させる可能性があります。
- RuntimeError: ワークフローがまだ進行中の場合に発生します。
- ValueError: 無効な引数が渡された場合に発生します。

torch.distributed.elastic.rendezvous.RendezvousHandler.shutdown() 関数は、PyTorch Distributed Elastic における分散ワークフローの終了処理を実行するために使用されます。この関数は、ワークフローのクリーンアップとリソースの解放を実行します。

import torch
import torch.distributed.elastic as dist_elastic


def worker_main(rank: int, world_size: int):
    # 分散ワークフローを実行する

    # ...

    # ワークフローが完了したら、終了処理を実行する
    dist_elastic.rendezvous.RendezvousHandler.shutdown(graceful_shutdown=True)


if __name__ == "__main__":
    # 分散ワークフローを初期化する
    dist_elastic.init_process_group(backend="gloo", world_size=4)

    # 各ワーカーで worker_main 関数を呼び出す
    for rank in range(dist_elastic.get_world_size()):
        dist_elastic.launch_process_group(rank=rank, world_size=4, function=worker_main)

このコードでは、以下の処理が行われます。

dist_elastic.init_process_group() 関数を使用して、分散ワークフローを初期化します。
dist_elastic.launch_process_group() 関数を使用して、各ワーカーで worker_main 関数を呼び出します。
worker_main 関数では、分散ワークフローを実行します。
ワークフローが完了したら、dist_elastic.rendezvous.RendezvousHandler.shutdown() 関数を使用して終了処理を実行します。

このコードは、torch.distributed.elastic.rendezvous.RendezvousHandler.shutdown() 関数の基本的な使用方法を示しています。実際の使用例では、ワークフローの具体的な要件に応じてコードを変更する必要があります。

graceful_shutdown 引数の使用

dist_elastic.rendezvous.RendezvousHandler.shutdown(graceful_shutdown=True)

このコードでは、graceful_shutdown 引数を True に設定することで、他のワーカーが完了するのを待ってから終了処理を実行します。

エラー処理

try:
    dist_elastic.rendezvous.RendezvousHandler.shutdown()
except RuntimeError as e:
    print(f"エラーが発生しました: {e}")

しかし、いくつかの状況では、torch.distributed.elastic.rendezvous.RendezvousHandler.shutdown() 関数の代替方法が必要になる場合があります。

代替方法

以下の代替方法を検討することができます。

- ワークフローに特化した終了処理を実行するカスタム関数を作成します。

各方法の詳細

手動でクリーンアップとリソース解放を実行する

この方法は、最も柔軟性がありますが、最も複雑でもあります。以下の手順を実行する必要があります。

共有された状態を解放する
- 共有された状態を保持するために使用しているすべてのデータ構造を解放します。
分散ストアとの接続を切断する
- 使用している分散ストアのドキュメントを参照して、接続を切断する方法を確認してください。

別の終了処理関数を使用する

この方法は、最も簡単ですが、最も柔軟性に欠けます。以下の手順を実行する必要があります。

終了処理関数を呼び出す
- ワークフローが完了したら、作成した終了処理関数を呼び出します。
ワークフローに特化した終了処理関数を作成する
- ワークフローの具体的な要件に応じて、終了処理を実行するカスタム関数を作成します。

どちらの方法を選択するべきか

どちらの方法を選択するべきかは、ワークフローの具体的な要件によって異なります。

ワークフローが複雑で、特別なクリーンアップ処理が必要な場合は、手動でクリーンアップとリソース解放を実行するか、別の終了処理関数を使用する必要があります。
ワークフローが単純で、特別なクリーンアップ処理が必要ない場合は、torch.distributed.elastic.rendezvous.RendezvousHandler.shutdown() 関数を使用するのが最善です。

例

以下の例は、手動でクリーンアップとリソース解放を実行する方法を示しています。

import torch
import torch.distributed as dist


def worker_main(rank: int, world_size: int):
    # 分散ワークフローを実行する

    # ...

    # ワークフローが完了したら、クリーンアップとリソース解放を実行する
    dist.destroy_process_group()
    torch.cuda.empty(0).zero_()


if __name__ == "__main__":
    # 分散ワークフローを初期化する
    dist.init_process_group(backend="gloo", world_size=4)

    # 各ワーカーで worker_main 関数を呼び出す
    for rank in range(dist.get_world_size()):
        dist.launch_process_group(rank=rank, world_size=4, function=worker_main)

dist.destroy_process_group() 関数を使用して、分散ワークフローを終了します。
torch.cuda.empty(0).zero_() 関数を使用して、GPU メモリを解放します。

このコードは、手動でクリーンアップとリソース解放を実行する方法を示す基本的な例です。実際の使用例では、ワークフローの具体的な要件に応じてコードを変更する必要があります。