PyTorch Distributed Elastic のその他の使用方法

PyTorch Distributed Elastic の torch.distributed.elastic.agent.server.SimpleElasticAgent._restart_workers() 関数解説

torch.distributed.elastic.agent.server.SimpleElasticAgent._restart_workers() 関数は、PyTorch Distributed Elastic ランタイムにおいて、ワーカープロセスを再起動するために使用されます。これは、スケーラブルな分散学習の実行を可能にする重要な機能です。

詳細解説

この関数は、以下の引数を受け取ります。

• worker_spec_list: 再起動するワーカープロセスのリスト。
• max_restarts: ワーカープロセスが再起動される最大回数。
• timeout: ワーカープロセスが再起動するまでのタイムアウト時間。

この関数は、以下の処理を実行します。

1. 各ワーカープロセスについて、以下の処理を実行します。
   • ワーカープロセスが実行中の場合は、終了させます。
   • ワーカープロセスの状態が "READY" または "RUNNING" でない場合は、再起動しません。
2. ワーカープロセスを再起動します。
3. ワーカープロセスが再起動し、状態が "READY" になるまで待機します。

使用例

以下のコードは、torch.distributed.elastic.agent.server.SimpleElasticAgent._restart_workers() 関数の使用方法を示しています。

def restart_workers(worker_spec_list, max_restarts=5, timeout=60):
    """
    ワーカープロセスを再起動します。

    Args:
        worker_spec_list: 再起動するワーカープロセスのリスト。
        max_restarts: ワーカープロセスが再起動される最大回数。
        timeout: ワーカープロセスが再起動するまでのタイムアウト時間。

    Returns:
        None
    """

    agent = SimpleElasticAgent()
    agent._restart_workers(worker_spec_list, max_restarts, timeout)

# 使用例

worker_spec_list = [
    {"role": "worker", "host": "localhost", "port": 1234},
    {"role": "worker", "host": "localhost", "port": 1235},
]

restart_workers(worker_spec_list)

注意事項

• この関数は、ワーカープロセスが実行中の場合のみ、ワーカープロセスを再起動します。
• ワーカープロセスが異常終了した場合、この関数はワーカープロセスを再起動します。
• ワーカープロセスが再起動しても、問題が解決しない場合は、デバッグが必要となります。

PyTorch Distributed Elastic のサンプルコード

単純な分散学習

import torch
import torch.distributed as dist
import torch.distributed.elastic as elastic

# ランタイムを初期化します。
elastic.init()

# ワーカーグループを作成します。
worker_group = elastic.create_worker_group(role="worker", num_workers=2)

# モデルを定義します。
model = torch.nn.Linear(10, 1)

# データを定義します。
data = torch.randn(100, 10)

# オプティマイザを定義します。
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)

# 学習ループを実行します。
for epoch in range(10):
    # データを各ワーカーに送信します。
    data = elastic.broadcast(data, src=0)

    # モデルを更新します。
    outputs = model(data)
    loss = torch.nn.MSELoss()(outputs, torch.ones(100, 1))
    loss.backward()
    optimizer.step()

    # モデルの状態を各ワーカーに同期します。
    elastic.sync_state(model)

# ランタイムを終了します。
elastic.shutdown()

スケーラブルな分散学習

import torch
import torch.distributed as dist
import torch.distributed.elastic as elastic

# ランタイムを初期化します。
elastic.init()

# ワーカーグループを作成します。
worker_group = elastic.create_worker_group(role="worker", num_workers=2)

# モデルを定義します。
model = torch.nn.Linear(10, 1)

# データを定義します。
data = torch.randn(100, 10)

# オプティマイザを定義します。
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)

# スケジューラを定義します。
scheduler = torch.optim.lr_scheduler.ExponentialLR(optimizer, gamma=0.9)

# 学習ループを実行します。
for epoch in range(10):
    # データを各ワーカーに送信します。
    data = elastic.broadcast(data, src=0)

    # モデルを更新します。
    outputs = model(data)
    loss = torch.nn.MSELoss()(outputs, torch.ones(100, 1))
    loss.backward()
    optimizer.step()
    scheduler.step()

    # モデルの状態を各ワーカーに同期します。
    elastic.sync_state(model)

# ランタイムを終了します。
elastic.shutdown()

チェックポイントの保存と復元

import torch
import torch.distributed as dist
import torch.distributed.elastic as elastic

# ランタイムを初期化します。
elastic.init()

# ワーカーグループを作成します。
worker_group = elastic.create_worker_group(role="worker", num_workers=2)

# モデルを定義します。
model = torch.nn.Linear(10, 1)

# データを定義します。
data = torch.randn(100, 10)

# オプティマイザを定義します。
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)

# 学習ループを実行します。
for epoch in range(10):
    # データを各ワーカーに送信します。
    data = elastic.broadcast(data, src=0)

    # モデルを更新します。
    outputs = model(data)
    loss = torch.nn.MSELoss()(outputs, torch.ones(100, 1))
    loss.backward()
    optimizer.step()

    # チェックポイントを保存します。
    if epoch % 5 == 0:
        elastic.save_checkpoint(model, optimizer, epoch)

# ランタイムを終了します。
elastic.shutdown()

# チェックポイントを復元します。
model, optimizer = elastic.load_checkpoint(epoch=5)

カスタムデータローダーの使用

import torch
import torch.distributed as dist
import torch.distributed.elastic as elastic

# カスタムデータローダーを定義します。
class CustomDataLoader(torch.utils.data.DataLoader):
    def __init__(self, dataset, batch_size,

PyTorch Distributed Elastic のその他の使用方法

パラメータサーバーの使用方法

パラメータサーバーは、モデルのパラメータを保存し、ワーカープロセスに送信するために使用できます。

import torch
import torch.distributed as dist
import torch.distributed.elastic as elastic

# ランタイムを初期化します。
elastic.init()

# パラメータサーバーを作成します。
ps = elastic.create_parameter_server(role="ps", num_parameters=10)

# ワーカーグループを作成します。
worker_group = elastic.create_worker_group(role="worker", num_workers=2)

# モデルを定義します。
model = torch.nn.Linear(10, 1)

# データを定義します。
data = torch.randn(100, 10)

# オプティマイザを定義します。
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)

# 学習ループを実行します。
for epoch in range(10):
    # データを各ワーカーに送信します。
    data = elastic.broadcast(data, src=0)

    # モデルを更新します。
    outputs = model(data)
    loss = torch.nn.MSELoss()(outputs, torch.ones(100, 1))
    loss.backward()
    optimizer.step()

    # モデルのパラメータをパラメータサーバーに送信します。
    elastic.sync_state(model, ps)

# ランタイムを終了します。
elastic.shutdown()

スケジューリングを使用して、ジョブの実行を管理できます。

import torch
import torch.distributed as dist
import torch.distributed.elastic as elastic

# ランタイムを初期化します。
elastic.init()

# スケジューラを作成します。
scheduler = elastic.create_scheduler(role="scheduler")

# ジョブを登録します。
job = elastic.register_job(scheduler,