<jupyter_start><jupyter_text>「PyTorchで実装する分散アプリケーション」【原題】Writing Distributed Applications with PyTorch 【原著】[<NAME>](https://seba1511.com/)【元URL】https://pytorch.org/tutorials/intermediate/dist_tuto.html【翻訳】電通国際情報サービスISID HCM事業部　櫻井 亮佑【日付】2020年11月21日【チュトーリアル概要】前提知識- [PyTorch Distributedについて](https://pytorch.org/tutorials/beginner/dist_overview.html)(日本語チュートリアル6_1)本チュートリアルでは、PyTorchの分散パッケージについて紹介します。具体的には、分散設定のセットアップ方法を確認し、非同期処理における通信方法（のうちの3つ）を実際に実装してみます。  セットアップPyTorchに含まれている分散パッケージ（例えば、`torch.distributed`）を利用することで、研究者やエンジニアは、プロセスやマシンのクラスタ間での計算を簡単に並列化できます。並列化するには、メッセージパッシングセマンティクスを活用し、各プロセスが他のプロセスとデータ通信できるようにします。マルチプロセスパッケージ（`torch.multiprocessing`）とは違い、プロセスは異なる通信バックエンドを使用することが可能であり、同一マシン内という条件に制限されることはありません。 分散アプリケーションを始めるには、同時に複数のプロセスを実行するスキルが必要になります。コンピューティングマシン-クラスタに対するアクセス権を保有している場合は、ローカルのシステム管理者に確認するか、お好みの調整ツール（例えば、[pdsh](https://linux.die.net/man/1/pdsh)、[clustershell](https://cea-hpc.github.io/clustershell/)、または[その他](https://slurm.schedmd.com/)）を使ってください。 
本チュートリアルの目的は、これから実装するテンプレートを用いて、単一のマシン上で処理をマルチプロセスに分岐させることです。<jupyter_code>"""run.py:"""
#!/usr/bin/env python
import os
import torch
import torch.distributed as dist
from torch.multiprocessing import Process

def run(rank, size):
    """ 後ほど、分散する関数を実装します。"""
    pass

def init_process(rank, size, fn, backend='gloo'):
    """ 分散環境を初期化します。 """
    os.environ['MASTER_ADDR'] = '127.0.0.1'
    os.environ['MASTER_PORT'] = '29500'
    dist.init_process_group(backend, rank=rank, world_size=size)
    fn(rank, size)


if __name__ == "__main__":
    size = 2
    processes = []
    for rank in range(size):
        p = Process(target=init_process, args=(rank, size, run))
        p.start()
        processes.append(p)

    for p in processes:
        p.join()<jupyter_output><empty_output><jupyter_text>上記のスクリプトは、それぞれの分散環境をセットアップする2つのプロセスを生成します。これによりプロセスグループ（`dist.init_process_group`）を初期化し、最終的には与えられた `run` 関数を実行します。 `init_process` 関数を確認しましょう。同一のipアドレスとポートを使うことで、マスターを通して各プロセスが同期した動作ができるようにしています。なお、上記のコードでは `gloo` バックエンドを使用しましたが、その他のバックエンドも利用可能である点に留意してください（セクション「発展的なトピック」を参照）。本チュートリアルの最後には、`dist.init_process_group` で起こるマジックを紹介しますが、本質的には、各マシンの情報を共有することでプロセスが相互に通信を行えるようにしているだけです。  ポイントツーポイント通信 送受信 あるプロセスから別のプロセスへの移動は、ポイントツーポイント通信と呼ばれています。ポイントツーポイント通信は、`send`関数と `recv`関数、またはそれらと同様の処理を即時に行う `isend`関数と `irecv`関数によって実現されます。<jupyter_code>""" ポイントツーポイント通信 """

def run(rank, size):
    tensor = torch.zeros(1)
    if rank == 0:
        tensor += 1
        # プロセス1にテンソルを送信
        dist.send(tensor=tensor, dst=1)
    else:
        # プロセス0からテンソルを受信
        dist.recv(tensor=tensor, src=0)
    print('Rank ', rank, ' has data ', tensor[0])<jupyter_output><empty_output><jupyter_text>上記の実装例では、2つのプロセスが値ゼロのテンソルから始まっており、プロセス0がテンソルをインクリメントしてプロセス1に送信しているため、最終的には両方のプロセスに存在するテンソルが1.0の状態で終了します。なお、受信するデータを格納するために、プロセス1にはメモリを配分する必要がある点に留意してください。 また、`send` / `recv` は処理をブロックするため、通信が完了するまで両方のプロセスが止まる点にも留意してください。一方で、即時で通信を行う `isend`関数と `irecv`関数は、処理をブロックしません。この場合、スクリプトは処理の実行を継続し、メソッドは `wait()` を実行できる `Work` オブジェクトを返します。<jupyter_code>""" ブロックを行わないポイントツーポイント通信 """

def run(rank, size):
    tensor = torch.zeros(1)
    req = None
    if rank == 0:
        tensor += 1
        # テンソルをプロセス1に送信
        req = dist.isend(tensor=tensor, dst=1)
        print('Rank 0 started sending')
    else:
        # プロセス0からテンソルを受信
        req = dist.irecv(tensor=tensor, src=0)
        print('Rank 1 started receiving')
    req.wait()
    print('Rank ', rank, ' has data ', tensor[0])<jupyter_output><empty_output><jupyter_text>即時に処理を行う関数を使用する場合、送受信されたテンソルの使用方法に気を付ける必要があります。具体的には、データがいつ他のプロセスへ通信されるかについて私たちはわからないため、`req.wait()` の処理が完了する前に送信されたテンソルに手を加えたり、受信したテンソルにアクセスするべきではありません。 言い換えれば以下のようになります。

- `dist.isend()` の後に `tensor` に書き込みを行うと、未定義のエラーが発生します。
- `dist.irecv()` の後に `tensor` から読み込みを行うと、未定義のエラーが発生します。

しかし、`req.wait()` の実行が完了した後であれば、通信が発生・完了し、`tensor[0]` に格納されている値が1.0であることが保証されます。 ポイントツーポイント通信は、プロセスの通信を細部まで制御したい場合に便利です。また、[BaiduのDeepSpeech](https://github.com/baidu-research/baidu-allreduce) や [Facebookの大規模実験](https://research.fb.com/publications/imagenet1kin1h/) で使用されているような、大規模なアルゴリズムを実装する際にも使用することができます（セクション「独自のRing-Allreduceの実装」を参照）。  集合通信 ScatterGatherReduceAll-ReduceBroadcastAll-Gather ポイントツーポイント通信とは異なり、集合通信は**グループ内**の全てのプロセス間でのコミュニケーションパターンを可能にします。なお、ここでのグループとは、すべてのプロセスのサブセットを指しています。グループを作成するには、`dist.new_group(group)` にランクの配列を与えます。デフォルトでは、集合通信は**ワールド**とも表現される全てのプロセス上で実行されます。例えば、すべてのプロセスにおけるテンソルの合計を得るには、`dist.all_reduce(tensor, op, group)` の集合通信を使います。<jupyter_code>""" All-Reduce の例 """
def run(rank, size):
    """ 単純なポイントツーポイント通信 """
    group = dist.new_group([0, 1])
    tensor = torch.ones(1)
    dist.all_reduce(tensor, op=dist.reduce_op.SUM, group=group)
    print('Rank ', rank, ' has data ', tensor[0])<jupyter_output><empty_output><jupyter_text>グループ内のすべてのテンソルの合計を得たいため、reduce演算子として `dist.reduce_op.SUM` を使います。一般的に、数学的に可換性のある演算は、単一の演算子として使用することが可能です。なお、PyTorchは4つの演算子を用意しており、すべて要素単位のレベルで動作します。- `dist.reduce_op.SUM`,- `dist.reduce_op.PRODUCT`,- `dist.reduce_op.MAX`,- `dist.reduce_op.MIN`. `dist.all_reduce(tensor, op, group)` に加えて、現在PyTorchには、合計6つの集合通信の関数が実装されています（先ほどの図の通りです）。- `dist.scatter(tensor, src, scatter_list, group)`: i番目のテンソル（`scatter_list[i]`）を、i番目のプロセスにコピーします。- `dist.gather(tensor, dst, gather_list, group)`: `dst` 内のすべてのプロセスから `tensor` をコピーします。- `dist.reduce(tensor, dst, op, group)`: opをすべてのテンソルに適用し、結果を `dst` に格納します。- `dist.all_reduce(tensor, op, group)`: reduceと同様の処理を行いますが、すべてのプロセスに結果を格納します。- `dist.broadcast(tensor, src, group)`: srcから他のすべてのプロセスに対して `tensor` をコピーします。- `dist.all_gather(tensor_list, tensor, group)`: すべてのプロセス上で、すべてのプロセスから  `tensor_list` に `tensor` をコピーします。  そして、- `dist.barrier(group)`: 各プロセスがこの関数に到達するまで、グループ内のすべてのプロセスをブロックします。  分散訓練 **注意:** 本セクションのサンプルスクリプトは、[こちらのGitHubのリポジトリ](https://github.com/seba-1511/dist_tuto.pth/)で確認できます。 分散モジュールの仕組みを理解したところで、実際に分散モジュールを使った実装をしてみましょう。目標は、[DistributedDataParallel](https://pytorch.org/docs/stable/nn.htmltorch.nn.parallel.DistributedDataParallel) の機能を再現することです。もちろん、これは教育目的のサンプル例です。実際の状況では十分にテストされ、最適化された公式の[DistributedDataParallel](https://pytorch.org/docs/stable/nn.htmltorch.nn.parallel.DistributedDataParallel)を使用するべきです。 簡単に表せば、これから確率的勾配降下法の分散版を実装したいと考えています。スクリプトでは、データのバッチを使って、すべてのプロセスにモデルの勾配を演算させ、その後各プロセスで算出された勾配を平均化します。なお、プロセスの数を変えた際に似たような収束結果を得られるようにするためには、初めにデータセットを分割する必要があります。（下記に実装を記載するクラスの代わりに、[tnt.dataset.SplitDataset](https://github.com/pytorch/tnt/blob/master/torchnet/dataset/splitdataset.pyL4) を使用することも可能です。）<jupyter_code>""" データセットを分割するためのクラス """
class Partition(object):

    def __init__(self, data, index):
        self.data = data
        self.index = index

    def __len__(self):
        return len(self.index)

    def __getitem__(self, index):
        data_idx = self.index[index]
        return self.data[data_idx]


class DataPartitioner(object):

    def __init__(self, data, sizes=[0.7, 0.2, 0.1], seed=1234):
        self.data = data
        self.partitions = []
        rng = Random()
        rng.seed(seed)
        data_len = len(data)
        indexes = [x for x in range(0, data_len)]
        rng.shuffle(indexes)

        for frac in sizes:
            part_len = int(frac * data_len)
            self.partitions.append(indexes[0:part_len])
            indexes = indexes[part_len:]

    def use(self, partition):
        return Partition(self.data, self.partitions[partition])<jupyter_output><empty_output><jupyter_text>上記のクラスを利用することで、あとは数行実装すれば、どんなデータセットでもシンプルに分割できるようになりました。<jupyter_code>""" MNISTの分割 """
def partition_dataset():
    dataset = datasets.MNIST('./data', train=True, download=True,
                             transform=transforms.Compose([
                                 transforms.ToTensor(),
                                 transforms.Normalize((0.1307,), (0.3081,))
                             ]))
    size = dist.get_world_size()
    bsz = 128 / float(size)
    partition_sizes = [1.0 / size for _ in range(size)]
    partition = DataPartitioner(dataset, partition_sizes)
    partition = partition.use(dist.get_rank())
    train_set = torch.utils.data.DataLoader(partition,
                                         batch_size=bsz,
                                         shuffle=True)
    return train_set, bsz<jupyter_output><empty_output><jupyter_text>2つの複製されたプロセスがあると仮定したとき、各プロセスは 60,000 / 2 = 30,000 個のサンプルを含む `train_set` を有することになります。また、全体としてのバッチサイズを 128 に維持するために、各バッチサイズも複製されたプロセスの数で割ります。 これで、通常の（フォワード→バックワード→最適化）の訓練コードを実装し、モデルの勾配を平均化する関数の呼び出し部分を追加することができます。（次のコードの大部分は、公式の [PyTorchのMNISTの例](https://github.com/pytorch/examples/blob/master/mnist/main.py) を基にしています。）<jupyter_code>""" 分散同期SGDの例 """
def run(rank, size):
    torch.manual_seed(1234)
    train_set, bsz = partition_dataset()
    model = Net()
    optimizer = optim.SGD(model.parameters(),
                          lr=0.01, momentum=0.5)

    num_batches = ceil(len(train_set.dataset) / float(bsz))
    for epoch in range(10):
        epoch_loss = 0.0
        for data, target in train_set:
            optimizer.zero_grad()
            output = model(data)
            loss = F.nll_loss(output, target)
            epoch_loss += loss.item()
            loss.backward()
            average_gradients(model)
            optimizer.step()
        print('Rank ', dist.get_rank(), ', epoch ',
              epoch, ': ', epoch_loss / num_batches)<jupyter_output><empty_output><jupyter_text>そして、単にモデルを受け取り、ワールド全体の勾配を平均化する `average_gradients(model)` 関数を実装します。<jupyter_code>""" 勾配の平均化 """
def average_gradients(model):
    size = float(dist.get_world_size())
    for param in model.parameters():
        dist.all_reduce(param.grad.data, op=dist.reduce_op.SUM)  # all_reduceの働きは、チュートリアル前半の絵をご覧ください
        param.grad.data /= size<jupyter_output><empty_output><jupyter_text>いかがでしょうか。分散同期SGDの実装に成功し、大規模なコンピューティングマシン-クラスタ上で任意のモデルを訓練できるようになりました。 **注意:**後半の実装は、技術的には正しい内容ですが、同期させたSGDを製品化レベルで実装するには、[より沢山のトリック](https://seba-1511.github.io/dist_blog) が必要になります。繰り返しになりますが、製品レベルでは、テストが行われ、最適化されている[DistributedDataParallel](https://pytorch.org/docs/stable/nn.htmltorch.nn.parallel.DistributedDataParallel) を使用しましょう。  独自のRing-Allreduceの実装さらなるチャレンジとして、DeepSpeech([BaiduのDeepSpeech](https://github.com/baidu-research/baidu-allreduce) )で使用されている、効率的なRing-Allreduceを実装してみましょう。ポイントツーポイント通信（issend / isrecv）を使用することで、とても簡単に実装できます。<jupyter_code>""" 手を加えたring-reduceの実装 """
def allreduce(send, recv):
    rank = dist.get_rank()
    size = dist.get_world_size()
    send_buff = send.clone()
    recv_buff = send.clone()
    accum = send.clone()

    left = ((rank - 1) + size) % size
    right = (rank + 1) % size

    for i in range(size - 1):
        if i % 2 == 0:
            # send_buff の送信
            send_req = dist.isend(send_buff, right)
            dist.recv(recv_buff, left)
            accum[:] += recv_buff[:]
        else:
            # recv_buff の送信
            send_req = dist.isend(recv_buff, right)
            dist.recv(send_buff, left)
            accum[:] += send_buff[:]
        send_req.wait()
    recv[:] = accum[:]<jupyter_output><empty_output><jupyter_text>上記の実装における `allreduce(send, recv)` 関数は、PyTorchのものとは少々異なるインターフェースになっています。具体的には、`recv` テンソルを引数に取り、すべての `send` テンソルの合計を `recv` テンソルに格納しています。ちなみに、上記の実装とDeepSpeech（[BaiduのDeepSpeech](https://github.com/baidu-research/baidu-allreduce) ）の実装ではもう1つの違いがあります。DeepSpeechの実装では、勾配のテンソルを数個の塊に分割し、通信帯域幅を最適に利用できるようにしています。（ヒント:[torch.chunk](https://pytorch.org/docs/stable/torch.htmltorch.chunk)）  発展的なトピックこれで `torch.distributed` のより発展的な機能を学ぶ準備ができました。網羅する機能が多いため、本セクションは2つのサブセクションに分けます。1. 通信バックエンド: このサブセクションでは、GPU-GPU通信のためのMPIとGlooの使い方を学びます。2. 初期化メソッド: このサブセクションでは、`dist.init_process_group()` の初期調整フェーズのベストな設定方法を学びます。  通信バックエンド`torch.distributed` の最もエレガントな側面の一つは、異なるバックエンド上に抽象化して構築できることです。始めに紹介しましたが、現在PyTorchには3つのバックエンドが実装されています。Gloo、NCCL、そしてMPIです。各バックエンドはユースケースに応じた、異なる仕様を有し、トレードオフの関係にあります。それぞれのバックエンドでサポートされている関数を比較した表は、[こちら](https://pytorch.org/docs/stable/distributed.htmlmodule-torch.distributed)から確認できます。 **Glooバックエンド**これまでのところでは、[Glooバックエンド](https://github.com/facebookincubator/gloo)の拡張的な利用方法をしてきました。Glooバックエンドはコンパイル済みの PyTorch バイナリに含まれており、Linux (0.2 以降) と macOS (1.3 以降) の両方で動作するので、開発プラットフォームとしては非常に便利です。また、CPU上でのすべてのポイントツーポイント、集合演算、及びGPU上でのすべての集合演算をサポートしています。ただし、CUDAのテンソルの集合演算の実装は、NCCLバックエンドが提供するものほど最適化されていません。 お気づきのように、GPU 上にモデルを配置した場合、分散SGDのさきほどの実装例は動作しません。複数のGPUを使用するには、追加で次のような修正を行います。1. `device = torch.device("cuda:{}".format(rank))` の使用2. `model = Net()` → `model = Net().to(device)`3. `data, target = data.to(device), target.to(device)` の使用上記の修正により、2つのGPU上で訓練を行い、`watch nvidia-smi` によりGPUの使用率を監視することができます。 **MPIバックエンド**メッセージパッシングインターフェース（MPI）は、ハイパフォーマンスコンピューティングの領域で活用されてきた標準的なツールです。MPIバックエンドは、ポイントツーポイント通信と集合通信を可能にするものであり、`torch.distributed` APIの土台でもありました。MPIにはいくつかの実装(例: [Open-MPI](https://www.open-mpi.org/)、 [MVAPICH2](http://mvapich.cse.ohio-state.edu/)、 [Intel MPI](https://software.intel.com/en-us/intel-mpi-library))が存在しており、それぞれが異なる目的に対して最適化されています。 MPIバックエンドを使用する利点は、大規模なコンピュータークラスター上での広い可用性と高水準の最適化にあります。

例えば、[最近の](https://developer.nvidia.com/ibm-spectrum-mpi)　[一部の](https://developer.nvidia.com/mvapich)　[実装](https://www.open-mpi.org/)では、CUDA IPCとGPU Directの技術も利用し、CPUを介したメモリのコピーを行わないようにしています。 残念ながら、PyTorchのバイナリにMPIの実装は含まれていないため、手作業でリコンパイルする必要があります。ただ幸いなのは、この処理がコンパイル時にPyTorch自身がMPIの実装を探してくれる、というシンプルに完了する点です。次の手順で、PyTorchを [ソースコード](https://github.com/pytorch/pytorchfrom-source) からインストールすることで、MPIバックエンドをインストールできます。1. Anacondaの環境を作成、及び起動し、ガイドに倣って前提となるパッケージ等をすべてインストールします。なお、`python setup.py install` の実行はまだ**行わないでください。**2. お好みのMPIの実装を選択し、インストールします。なお、CUDAの使用を想定しているMPIを有効化するには、いくつかの追加手順が必要な場合がある点に注意してください。今回のケースでは、GPUサポートのないOpen-MPIを選びます。`conda install -c conda-forge openmpi`3. 最後に、クローンしたPyTorchのリポジトリで`python setup.py install`を実行します。 新規にインストールされたバックエンドのテストを行うには、少し手を加える必要があります。1. `if __name__ == '__main__':` 以下の内容を `init_process(0, 0, run, backend='mpi')` に置き換えます。2. `mpirun -n 4 python myscript.py` を実行します。上記の変更を行う理由は、プロセスを生成する前にMPIが独自の環境を作成する必要があるためです。MPIは独自のプロセスを生成し、次のサブセクション「初期化メソッド」内で解説されているハンドシェイク通信を行うことで、`init_process_group` における `rank` と `size` の引数を不要にします。これは、`mpirun` に追加の引数を渡すことで各プロセスに合わせた計算リソースの調整ができるため、実際には非常に強力です（プロセスごとのコア数、手作業で割り当てる特定ランクへのマシン、[その他](https://www.open-mpi.org/faq/?category=runningmpirun-hostfile)など）。これらを行うことで、他の通信バックエンドと同じように馴染みのある出力が得られるはずです。 **NCCLバックエンド**NCCLバックエンドは、CUDAのテンソルを対象にした集合演算に最適化された実装を提供しています。集合演算にCUDAのテンソルのみを使う場合は、ベストパフォーマンスを得るためにこのバックエンドの使用を検討してください。なお、NCCLバックエンドは、CUDAのサポートを伴っているビルド済みのバイナリに含まれています。  初期化メソッド本チュートリアルの最後に、一番最初に呼び出した `dist.init_process_group(backend, init_method)` 関数について説明します。特に、各プロセス間において初期調整のステップを担当する様々な初期化メソッドについて説明します。これらの初期化メソッドは、初期調整をどのように行うかを定義します。ハードウェアのセットアップ状況によりますが、様々な初期化メソッドのいずれかが最適な手法として使用できます。以下のセクションに加えて、[公式のドキュメント](https://pytorch.org/docs/stable/distributed.htmlinitialization)にも、目を通していただくことをおすすめします。 **環境変数**本チュートリアルでは、環境変数の初期化メソッドを使用してきました。下記の4つの環境変数をすべてのマシン上で設定することで、すべてのプロセスが適切にマスターに接続できるようになり、他のプロセスの情報を得ることができます。そして、最終的にはそれらのプロセスとハンドシェイク通信を行えるようになります。- `MASTER_PORT`: ランク0のプロセスをホストするマシン上の空きポート- `MASTER_ADDR`: ランク0のプロセスをホストするマシンのIPアドレス- `WORLD_SIZE`: マスターが、待機しているワーカーの数を知るために設定するプロセスの総数- `RANK`: 各プロセスがワーカーのマスターかどうかを判別するために設定する各プロセスのランク **共有ファイルシステム**共有ファイルシステムは、すべてのプロセスが共有ファイルシステムにアクセスできる必要があり、共有ファイルを通してプロセスの調整を行います。これは各プロセスがファイルを開き、情報を書き込み、すべてのプロセスが同様の処理を行うまで待機するということを意味します。なお、必要なすべての情報がすべてのプロセスで利用できるようになったあとは、競合を避けるために、ファイルシステムは [fcntl](http://man7.org/linux/man-pages/man2/fcntl.2.html) を通したロック機構を実装しなければなりません。<jupyter_code>dist.init_process_group(
    init_method='file:///mnt/nfs/sharedfile',
    rank=args.rank,
    world_size=4)<jupyter_output><empty_output><jupyter_text>**TCP**TCP経由での初期化は、ランク0のプロセスのIPアドレスと到達可能なポート番号を提供することで行われます。ここでは、すべてのワーカーがランク0のプロセスに繋がり、相互に到達方法の情報交換ができるようになります。<jupyter_code>dist.init_process_group(
    init_method='tcp://10.1.1.20:23456',
    rank=args.rank,
    world_size=4)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12484076433121019, 'fraction_numerical': 0.011210191082802547, 'mean_word_length': 6.1563354603464, 'pattern_counts': {'":': 1, '<': 46, '<?xml version=': 0, '>': 46, 'https://': 28, 'lorem ipsum': 0, 'www.': 3, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '157591', 'license': 'zlib-acknowledgement', 'n_tokens_mistral': 9866, 'n_tokens_neox': 9331, 'n_words': 599, 'path': 'notebook/6_Parallel_Distributed/6_4_writing_distributed_applications_with_pytorch_jp.ipynb', 'repo_name': 'seigot/pytorch_tutorials_jp', 'size': '15704'}