分散トレーニング入門: PyTorch を使用してマルチ GPU 分散トレーニングを実装する方法

具体的には、この記事ではまず、分散コンピューティングの基本概念と、分散コンピューティングがディープラーニングにどのように使用されるかを紹介します。次に、分散アプリケーションを処理する環境を構成するための標準要件 (ハードウェアとソフトウェア) を示します。 ***、実践的な経験を提供するために、この記事では、理論と実装の両方の観点から、ディープラーニングモデルをトレーニングするための分散アルゴリズム (同期確率的勾配降下法、同期 SGD) について説明します。

[[264239]]

分散コンピューティングとは何ですか?

分散コンピューティングとは、ネットワーク上の複数の接続されたコンポーネントを活用するプログラムの作成方法を指します。通常、大規模なコンピューティングは、高密度の数値演算を並列処理できるようにコンピューターを配置することによって実現されます。分散コンピューティングの用語では、これらのコンピューターはノードと呼ばれることが多く、これらのノードの集合はクラスターと呼ばれます。これらのノードは通常イーサネット経由で接続されますが、他の高帯域幅ネットワークでも分散アーキテクチャを活用できます。

ディープラーニングは分散コンピューティングからどのような恩恵を受けるのでしょうか?

ディープラーニングの主力であるニューラルネットワークは、以前から文献に登場していますが、その潜在能力が最大限に発揮されるようになったのはごく最近のことです。ニューラルネットワークが普及した主な理由の 1 つは、膨大な計算能力です。この記事では、この点について説明します。ディープラーニングでは、多数のパラメータを含む大量のデータに基づいてディープニューラルネットワークをトレーニングする必要があります。分散コンピューティングは、最新のハードウェアを最大限に活用するための最適なツールです。その中心となる考え方は次のとおりです。

適切に設計された分散アルゴリズムでは、次のことが可能になります。

一貫した処理のために、計算 (ディープラーニングモデルでの順方向および逆方向伝播) とデータを複数のノードに「分散」します。
一貫性を実現するために、複数のノード間で効果的な「同期」を確立できます。

MPI: 分散コンピューティングの標準

慣れておく必要があるもう一つの用語があります。それは、Message Passing Interface (MPI) です。 MPI は、ほぼすべての分散コンピューティングの主力製品です。 MPI は、ノードが相互に通信するための一連のルールを定義するオープンスタンダードです。MPI はプログラミングモデル/API でもあります。 MPI はソフトウェアでもツールでもなく、仕様です。

1991 年の夏、学界と産業界の組織と個人のグループが集まり、最終的に MPI フォーラムが設立されました。フォーラムは合意に達し、さまざまなハードウェアベンダーが移植可能で柔軟性があり最適化された実装を実現するためのガイダンスを提供するライブラリの構文とセマンティクスの仕様を起草しました。いくつかのハードウェアベンダーは、OpenMPI、MPICH、MVAPICH、Intel MPI など、独自の MPI 実装を持っています。

このチュートリアルでは、非常に効率的で Intel プラットフォームに最適化されている Intel MPI を使用します。オリジナルの Intel MPI は C ライブラリであり、非常に低レベルです。

構成

分散システムでは、適切な構成が非常に重要です。プログラミングモデルの概念を理解していたとしても、適切なハードウェアとネットワークレイアウトがなければほとんど役に立ちません。行う必要がある主な手配は次のとおりです。

通常、クラスターは共通のネットワークによって接続された一連のノードによって形成されます。ノードとしてハイエンドサーバーを使用し、InfiniBand などの高帯域幅ネットワークを使用することをお勧めします。
クラスター内のすべてのノードには、まったく同じユーザー名を持つ Linux システムが必要です。
シームレスな接続には、ノード間でパスワードなしの SSH 接続を確立することが不可欠です。
MPI 実装をインストールする必要があります。この記事では、Intel MPI のみに焦点を当てます。
すべてのノードから参照でき、分散アプリケーションが常駐する必要がある共通のファイルシステムが必要です。これを実現する方法の 1 つは、ネットワークファイルシステム (NFS) です。

並列戦略の種類

ディープラーニングモデルを並列化する一般的な方法は 2 つあります。

モデルの並列処理
データ並列処理

1. モデルの並列性

モデルの並列処理とは、モデルが論理的に複数の部分に分割され (たとえば、一部のレイヤーが 1 つの部分に、他のレイヤーが別の部分に)、異なるハードウェア/デバイスに展開されることを意味します。

実行時間の点で、モデルのさまざまな部分をさまざまなデバイスに展開することには利点がありますが、これは通常、メモリの制約を回避するために行われます。特に多くのパラメータを持つモデルは、大きなメモリフットプリントを必要とし、単一のシステムに収めるのが難しいため、この並列戦略の恩恵を受けるでしょう。

2. データの並列処理

一方、データの並列処理とは、異なるハードウェア/デバイス上にある同じネットワークの複数のコピーを通じて、異なるデータバッチを処理することを指します。モデルの並列処理とは異なり、各レプリカはネットワークの一部ではなく、ネットワーク全体である場合があります。

ご想像のとおり、この戦略はデータの増加に合わせて拡張できます。ただし、ネットワーク全体を単一のデバイスに展開する必要があるため、メモリフットプリントが大きいモデルには役立たない可能性があります。次の図を見ればこれが明らかになるはずです。

モデル並列処理とデータ並列処理

実際、大規模な組織では、データ並列処理が普及し、本番品質のディープラーニングトレーニングアルゴリズムを実行するために一般的に使用されるようになっています。したがって、このチュートリアルではデータの並列処理に焦点を当てます。

torch.distributed API

PyTorch は、C で記述された基盤となる MPI ライブラリへのインターフェースとして、非常にエレガントで使いやすい API を提供します。 PyTorch はソースからコンパイルする必要があり、システムにインストールされている Intel MPI とリンクする必要があります。ここでは、torch.distributed の基本的な使い方と実行方法について見ていきます。

 # ファイル名 'ptdist.py'
輸入トーチ
torch.distributed を dist としてインポートします。 
 
 def main(ランク、ワールド):
    ランク== 0 の場合:
 x = torch.tensor ([1., -1.]) # 対象のテンソル
        送信(x, dst = 1 )
        print('Rank-0 は次のテンソルを Rank-1 に送信しました')
        印刷(x)
    それ以外：
 z = torch.tensor ([0., 0.]) # テンソルを受け取るホルダー
        dist.recv(z, src = 0 ) は、
        print('Rank-1 は Rank-0 から次のテンソルを受け取りました')
        印刷(z) 
 
 __name__ == '__main__' の場合:
    dist.init_process_group(バックエンド= 'mpi' )
    メイン(dist.get_rank()、dist.get_world_size())

ポイントツーポイント通信

上記のコードを mpiexec で実行すると、標準 MPI 実装に基づいた分散プロセススケジューラが生成されます。結果は次のとおりです。

クラスター@miriad2a:~/nfs$ mpiexec -n 2 -ppn 1 -hosts miriad2a,miriad2b python ptdist.py
ランク0は次のテンソルをランク1に送信しました
テンソル([ 1., -1.])
ランク1はランク0から次のテンソルを受け取りました
テンソル([ 1., -1.])

実行される最初の行は dist.init_process_group(backend) であり、これは基本的に参加ノード間の内部通信チャネルを設定します。使用するバックエンドを指定するためのパラメータが必要です。この例では MPI のみを使用しているため、backend='mpi' です。他にもバックエンドがあります (例: TCP、Gloo、NCCL)。
取得する必要があるパラメーターは、ワールドサイズとランクの 2 つです。ワールドは、特定の mpiexec 呼び出しのコンテキスト内のすべてのノードのセットを指します (mpiexec の -hosts フラグを参照)。ランクは、MPI ランタイムによって各プロセスに割り当てられる一意の整数です。 0から始まります。 -hosts で指定された順序を使用して値が割り当てられます。したがって、この例では、ノード「miriad2a」のプロセスにランク 0 が割り当てられ、ノード「miriad2b」のプロセスにランク 1 が割り当てられます。
x は、Rank 0 が Rank 1 に送信するテンソルであり、dist.send(x, dst=1) を介して実行されます。
z は、Rank 1 がテンソルを受け取る前に作成するものです。渡されたテンソルを受け取るには、プレースホルダーと同じ次元の事前に作成されたテンソルが必要です。 z の値は最終的に x に置き換えられます。
dist.send(..) と同様に、受信を担当する対応する関数は dist.recv(z, src=0) であり、テンソルを z に受信します。

コミュニケーション集団

前のセクションで見たのは、「ポイントツーポイント」通信の例で、特定の環境内の特定のランクにランクがデータを送信するというものでした。このタイプの通信は、通信をきめ細かく制御できるため便利ですが、集団と呼ばれる、よく使用される他の標準的な通信パターンもあります。以下では、同期 SGD アルゴリズムで注目する集団、つまり all-reduce 集団について説明します。

1. すべて削減

All-reduce は同期通信方式です。すべてのランクに対して削減操作が実行され、その結果はすべてのランクで表示されます。次の図は、この考え方 (減算演算としての合計) を紹介しています。

すべて削減集合

def main(ランク、ワールド):
    ランク== 0 の場合:
 x =トーチテンソル([1.])
    ランク== 1 の場合:
 x =トーチテンソル([2.])
    ランク== 2 の場合:
 x =トーチテンソル([-3.]) 
 
    dist.all_reduce(x, op = dist.reduce_op.SUM )は、すべての要素を合計して計算します。
    print('ランク {} には {} があります'.format(rank, x)) 
 
 __name__ == '__main__' の場合:
    dist.init_process_group(バックエンド= 'mpi' )
    メイン(dist.get_rank()、dist.get_world_size())

PyTorch における all-reduce 集合の基本的な使用法

3 環境の世界で開始した場合、結果は次のようになります。

クラスター@miriad2a:~/nfs$ mpiexec -n 3 -ppn 1 -hosts miriad2a,miriad2b,miriad2c python ptdist.py
ランク1にはテンソル([0.])がある。
ランク0にはテンソル([0.])がある。
ランク2にはテンソル([0.])がある。

ランク == … elif は分散コンピューティングで頻繁に遭遇するパターンです。この例では、異なるランクのテンソルを作成するために使用されます。
これらは、合計 (dist.reduce_op.SUM) をリダクション操作として使用して、一緒に all-reduce を実行します (ご覧のとおり、dist.all_reduce(..) は if … elif ブロックの外側にあります)。
各ランクの x を合計し、その合計を各ランクの x 内に配置します。

ディープラーニングへの転換

読者は、ディープラーニングモデルのトレーニングによく使用される標準的な確率的勾配降下法 (SGD) アルゴリズムに精通していることを前提としています。私たちが今見ているのは、スケーリングのために all-reduce 集合を使用する SGD のバリエーションである同期 SGD です。標準的な SGD 計算から始めましょう。

ここで、D はサンプルセット (ミニバッチ)、θ はすべてのパラメーターのセット、λ は学習率、Loss(X, y) は D 内のすべてのサンプルの損失関数の平均です。

同期 SGD が依存する中心的なトリックは、更新ルールの合計を小さな (ミニ) バッチサブセットに分割することです。 D は R 個のサブセット D₁、D₂、… に分割されます (各サブセットのサンプル数が同じであることが推奨されます)。そのため、標準の SGD 更新式の合計は次のように分割されて得られます。

ここで、勾配演算子は合計演算子に分散されるため、次のようになります。

私たちはこれから何を得るのでしょうか?

上記の式の個々の勾配項 (角括弧内) を見てみましょう。これらは個別に計算され、その後加算されて、損失や近似なしに元の勾配が得られます。これはデータの並列処理です。全体のプロセスは次のとおりです。

データセット全体を R 個の等サイズのデータブロック (サブセット) に分割します。ここでの文字 R はレプリカを表します。
MPI を使用して R プロセス/ランクを起動し、各プロセスをデータブロックにバインドします。
各ランクでは、サイズ B のミニバッチ (dᵣ) (dᵣ はランクに割り当てられたデータブロック D_r から取得) を使用して勾配を計算します (つまり、ランク r を計算します)。
すべてのランクの勾配が合計され、結果の勾配は、さらに処理する前に各ランクで表示されます。

最後のポイントは、all-reduce アルゴリズムです。したがって、サイズ B のミニバッチ (dᵣ) を使用してすべてのランクの勾配を計算した後は、毎回 all-reduce を実行する必要があります。注目すべき点は、すべての R ランクの勾配を合計すると (サイズ B のミニバッチを使用して計算)、有効なバッチサイズが得られることです。

実装の重要な部分は次のとおりです (定型コードは示していません)。

モデル= LeNet ()
 # 初期重みの最初の同期
sync_initial_weights(モデル、ランク、ワールドサイズ) 
 
最適化オプティマイザー= optim.SGD(model.parameters(), lr = 1e -3,モーメンタム= 0.85 ) 
 
モデル.train()
範囲(1, エポック + 1)内のエポックの場合:
    データの場合、train_loader のターゲット:
        オプティマイザ.zero_grad()
出力=モデル（データ）
損失= F .nll_loss(出力、ターゲット)
        損失.後方() 
 
        # 勾配の全縮約
        sync_gradients(モデル、ランク、ワールドサイズ) 
 
        オプティマイザ.ステップ() 
 
 def sync_initial_weights(モデル、ランク、ワールドサイズ):
    model.parameters() の param の場合:
        ランク== 0 の場合:
            # ランク0は自身の重みを送信しています
            # すべての兄弟に (1 から world_size)
            範囲(1, world_size)内の兄弟の場合:
                dist.send(param.data, dst =兄弟)
        それ以外：
            # 兄弟はパラメータを受け取る必要がある
            dist.recv(param.data, src = 0 ) です。 
 
 
 def sync_gradients(モデル、ランク、ワールドサイズ):
    model.parameters() の param の場合:
        dist.all_reduce(param.grad.data, op = dist.reduce_op.SUM )は、

すべての R ランクは、ランダムな重みを使用してモデルの独自のコピーを作成します。
ランダムな重みを持つ単一のレプリカは、最初は同期されていない可能性があります。すべてのレプリカの初期重みを同期することをお勧めします。sync_initial_weights(..) がこれを実行します。任意のランクが重みをその兄弟ランクに送信するようにします。兄弟ランクはこれらの重みを受信し、それを使用して自身を初期化する必要があります。
各ランクに対応するデータ部分からミニバッチ（サイズ B）を取得し、順方向パスと逆方向パス（勾配）を計算します。構成の一部として、すべてのプロセス/ランクが独自のデータを表示できるようにする必要があることに注意することが重要です (通常は独自のハードディスク上または共有ファイルシステム内)。
合計を縮小演算として扱い、各レプリカの勾配に対して all-reduce を実行します。 sync_gradients(..) はグラデーションを同期します。
勾配が同期された後、各レプリカは独自の重みに対して標準の SGD 更新を独立して実行できます。 optimizer.step() は正常に動作します。

ここで問題となるのは、独立した更新が同期された状態を維持するにはどうすればよいかということです。

更新方程式の***更新を見てみましょう。

上記のポイント 2 と 4 により、それぞれの初期重みと勾配が同期していることが保証されます。明らかに、それらの線形結合も同期しています (λ は定数)。後続のすべての更新も同様のロジックに従うため、同期されます。

パフォーマンス比較

すべての分散アルゴリズムの最大のボトルネックは同期です。分散アルゴリズムは、同期時間が計算時間よりも大幅に短い場合にのみ有益です。標準 SGD と同期 SGD を簡単に比較して、後者の方が優れている場合を見てみましょう。

定義: データセット全体のサイズが N であると仮定します。ネットワークがサイズ B のミニバッチを処理するには、時間 Tcomp がかかります。分散の場合、all-reduce 同期には Tsync が使用されます。

非分散型 (標準) SGD の場合、エポックごとにかかる時間は次のとおりです。

同期 SGD の場合、エポックごとにかかる時間は次のとおりです。

したがって、分散環境が非分散環境よりも大きな利点を持つためには、次の条件を満たす必要があります。

上記の不等式に影響する 3 つの要素を調整することで、分散アルゴリズムからより多くのメリットを得ることができます。

ノードを高帯域幅の高速ネットワークに接続することで、Tsync を削減できます。
バッチサイズ B を増やすと、Tcomp が増加します。
より多くのノードを接続し、レプリカを増やすことで R を増やします。

この記事では、ディープラーニングの文脈における分散コンピューティングの中心的な考え方をわかりやすく紹介します。同期 SGD は人気がありますが、頻繁に使用される他の分散アルゴリズムもあります (非同期 SGD やそのバリエーションなど)。しかし、ディープラーニングの手法を並行して考えることができることの方が重要です。すべてのアルゴリズムがすぐに並列化できるわけではなく、一部のアルゴリズムでは元のアルゴリズムによって与えられた理論的な保証を破る近似値が必要になることに注意してください。これらの近似を効率的に処理できるかどうかは、アルゴリズムの設計者と実装者に依存します。

元の住所:

https://medium.com/intel-student-ambassadors/distributed-training-of-deep-learning-models-with-pytorch-1123fa538848

[この記事は51CTOコラム「Machine Heart」、WeChatパブリックアカウント「Machine Heart（id:almosthuman2014）」によるオリジナル翻訳です]

この著者の他の記事を読むにはここをクリックしてください

<<: データ分析を使用して協調フィルタリングアルゴリズムの2つの一般的な問題を定量化する

>>: 中国でドローン配送用の商用「操縦免許」が発行されるまでにどれくらいの時間がかかるのでしょうか？