OpenMPIを用いて並列化する方法を10分で話す

今回は「棋譜からの学習を並列化(クラスター化)したい」という人のためにOpenMPIを用いて並列化する方法を10分で話します。

@yaneuraou Windows で MPI クラスタ並列学習の記事も期待してます！

— かず@なのは (@kazu_nanoha) November 11, 2015

OpenMPIに関しては私は時間がなくてきちんと調べてなくて、とりあえずMS-MPIを用いて棋譜からの学習の並列化(クラスター化)自体は達成出来たので、注意点などをざっと箇条書きにしたいと思います。間違っている点や、「それ、こうやれば出来るよ」みたいなのは、どなたかコメント欄にてフォローしていただけると助かります。

MS-MPIを使え

今回、対象とするのはボナメソ型のバッチ学習です。つまりPC間の同期はそれほどの頻度で行わないことが前提です。やはりネットワークごしに大きなファイルを短い頻度で送り合うとそこがボトルネックになります。それを回避するために、学習メソッドを改良したりする話になるのですが、SGDなどの並列化(クラスター化)はそれ自体に難しい話題をいくつも含んでいるので今回の解説の対象とはしません。

次に、今回クラスター内の他PCとの通信にはMPIを用います。MPIは仕様の名前で、その実装がたくさんあるというのが実状のようです。(OpenMPIが一番有名) Windowsで動かすならMicrosoft製のMPIの実装がお勧めです。MS-MPI(略称)と呼ばれます。最新版は、「Microsoft MPI v6」です。Microsoftのサイトはpermanent linkの重要性がわからない~~馬鹿ども~~人たちが運営しているのか、URLがしょっちゅう変わるので、ググってたどり着いてください。

一応、現時点ではここにあります。
https://msdn.microsoft.com/en-us/library/bb524831(v=vs.85).aspx

MS-MPIの使い方

さて、MS-MPIをセットアップして、Visual Studioからinclude pathやlibrary pathを指定してやると、
#include <mpi.h>
とするとMPIの命令が使えるようになります。

セットアップ手順などはBonanzaのMPI並列化をした人(マイボナの作者)の記事がとても参考になります。
http://www.geocities.jp/shogi_depot/doc/bonanza_mpi.htm

2台でクラスター並列にすると仮定して話を進めます。
片方のPCをmaster、もう片方のPCをslaveと呼ぶことにします。

1. 2台のPCでMPIの通信用のデーモンを走らせます。
例)
> smpd.exe -port 8677

2. master側で
> mpiexec.exe /debug 0 /lines -hosts 2 localhost 1 【slave側のPCの名前】 1 C:\mpi\YaneuraOu2016.exe

のように実行します。

※ /linesをつけておくと標準出力上に[0](masterの意味)とか[1](slaveの意味)とか、先頭について、どのPCからの出力かがわかるのでお勧め。
※ 「-hosts 2」の「2」は2台のPCの意味。「localhost 1」の1は、1プロセスの意味。各PC、1プロセスずつ割り当てて、自前でスレッドを生成するのがお勧め。

設定上の注意点

・【slave側のPCの名前】のところには、PC-5124 のような、Windowsのネットワークのところを開いたときに同じLAN内の他のPC名が表示されているかと思いますが、それのことです。IPアドレス直打ちはどうもNGのようです。

・フォルダpathはmasterとslave側とで統一しておかないといけないようです。何かオプションでpathを指定すれば変更できるのかも知れませんが結構面倒くさそうだったので、私はWindowsのjunctionの機能を使ってmasterとslave側で無理やり同じpathにしました。

・MS-MPIは標準出力をhookしてバッファリングしているようで、std::cout << “Hello MPI World!!” << std::flush; のようにしてflushをしないと表示されない(ことがある)。

・MS-MPIはslave側の標準出力もmaster側の標準出力にリダイレクトされて表示される。(この機能、余計なお世話のような気がしなくもないが、master側で一元管理したいときに便利なのだろう)

・MPIはもともと科学計算などを並列化するために設計されたものなので、要らない機能がいっぱいあるが、今回の用途では全く使えない。今回のような用途においてはMPIは、1対N(1つのmaster 対複数のslave)の通信の手段でしかない。

・Visual Studioから実行ファイルを直接実行する方法がよくわからない。(用意されていない？) 仕方ないのでmpiexec.exe経由で実行して、そのあとプロセスにアタッチしてデバッグした。

設計上の注意点

・複数のスレッドからMPI_Send()を呼び出すならマルチスレッド用にコンパイルされたlibファイルが必要のようです。MS-MPIに付属しているmsmpi.libがどういうオプションでコンパイルされているかは調べてません。master/slaveともに通信用のスレッドを一つ生成して、そいつにのみ通信を担当させるような設計にするほうが無難だと思います。

・シリアライズに関してはboost::serializationを用いるのが一般的。しかし、どうせstd::vectorぐらいしか送り合わないのならシリアライザぐらい自分で書いたほうがよっぽど手っ取り早いと思います。boost::serializationの使い方を覚える時間と設定をする時間とコンパイルが遅くなるデメリットとソース管理が面倒になるデメリットetc…。

・MPIではmasterはrank == 0、slaveはrank == 1,2,3…と順番にrankが割り振られます。1台のPCは1プロセスのみを指定して、自前でスレッドを生成したほうがよほどすっきりするので、このとき、PCごとにrankが異なることになります。

・MPI、設計が古くて、int型(32bit)で表せる範囲のデータしか一度に送れない。それを超えるなら分割して送る必要がある。

・MPI、設計がださくて、ノンブロッキング(非同期)で読みだすメソッドを使ったときに、そのデータのサイズがわからない。std::vectorのような可変長のデータを送るときに受け手側は事前に送られてきたデータのサイズがわからないので、メモリの確保のしようがない。仕方ないので、まずメッセージヘッダーのようなものを送って、そこにbodyのサイズを書いておく、みたいな設計になる。→ 補足＆訂正あり。

実装の概要

1. KKP/KPPの配列をmasterからslaveに丸投げ
2. masterからslaveに1局分の棋譜を送る
3. slave側はPV(読み筋)をmasterに返す
4. 2.と3.を全局分、繰り返す
以下略

おまけ — 簡易serializer/deserializer

私が10分ぐらいで書いたserializer/deserializer。自由に使ってもらって構わない。

#ifndef EASY_SERIALIZER_H_INCLUDED
#define EASY_SERIALIZER_H_INCLUDED

#include "learn.h"

#ifdef EVAL_LEARN_MPI

#include <vector>

namespace EasySerializer
{
  // 超簡易なシリアライザ実装。Boost::Serializerの代わり。
  class ReadSerializer
  {
  public:

    // ptr = バッファの先頭
    ReadSerializer(void* ptr_)
    {
      ptr = (char*)ptr_;
    }

    template <typename T>
    ReadSerializer& operator >> (T& n)
    {
      n = *(T*)ptr;
      ptr += sizeof(T);
      return *this;
    }

    // vectorのdeserialize
    template <typename T>
    ReadSerializer& operator >> (std::vector<T>& v)
    {
      size_t vec_size;
      *this >> vec_size;
      v.resize(vec_size);
      size_t size = vec_size * sizeof(T);
      memcpy(&v[0], ptr, size);
      ptr += size;

      return *this;
    }

  protected:
    char* ptr; // 次にreadすべき場所
  };

  class WriteSerializer
  {
  public:

    template <typename T>
    WriteSerializer& operator << (const T n)
    {
      size_t last = buffer.size();
      buffer.resize(last + sizeof(T));

      *(T*)&buffer[last] = n;
      return *this;
    }

    // vectorのserialize
    template <typename T>
    WriteSerializer& operator << (const std::vector<T>& v)
    {
      *this << v.size();

      size_t size = buffer.size();
      buffer.resize(size + sizeof(T)*v.size());
      memcpy(&buffer[size], &v[0], v.size()*sizeof(T));

      return *this;
    }

    // 内部バッファをコピーして返す。受け取り側でこのポインターをdelete[]すること。
    void* get_ptr()
    {
      char* ptr = new char[buffer.size()];
      memcpy(ptr, &buffer[0], buffer.size());
      return ptr;
    }

    // 内部バッファのサイズ。
    size_t get_size() const { return buffer.size(); }

  protected:
    std::vector<char> buffer;
  };

}

#endif

#endif

#ifndef EASY_SERIALIZER_H_INCLUDED

#define EASY_SERIALIZER_H_INCLUDED

#include "learn.h"

#ifdef EVAL_LEARN_MPI

#include

namespace EasySerializer

{

// 超簡易なシリアライザ実装。Boost::Serializerの代わり。

class ReadSerializer

{

public:

// ptr = バッファの先頭

ReadSerializer(void* ptr_)

{

ptr = (char*)ptr_;

}

template

ReadSerializer& operator >> (T& n)

{

n = *(T*)ptr;

ptr += sizeof(T);

return *this;

}

// vectorのdeserialize

template

ReadSerializer& operator >> (std::vector& v)

{

size_t vec_size;

*this >> vec_size;

v.resize(vec_size);

size_t size = vec_size * sizeof(T);

memcpy(&v[0], ptr, size);

ptr += size;

return *this;

}

protected:

char* ptr; // 次にreadすべき場所

};

class WriteSerializer

{

public:

template

WriteSerializer& operator << (const T n)

{

size_t last = buffer.size();

buffer.resize(last + sizeof(T));

*(T*)&buffer[last] = n;

return *this;

}

// vectorのserialize

template

WriteSerializer& operator << (const std::vector& v)

{

*this << v.size();

size_t size = buffer.size();

buffer.resize(size + sizeof(T)*v.size());

memcpy(&buffer[size], &v[0], v.size()*sizeof(T));

return *this;

}

// 内部バッファをコピーして返す。受け取り側でこのポインターをdelete[]すること。

void* get_ptr()

{

char* ptr = new char[buffer.size()];

memcpy(ptr, &buffer[0], buffer.size());

return ptr;

}

// 内部バッファのサイズ。

size_t get_size() const { return buffer.size(); }

protected:

std::vector buffer;

};

}

#endif

おまけ2 – MPI通信スレッド用のsend/receiveのwrapper

参考用。自由に使ってもらって構わない。

// どんなサイズのバイナリでも指定のrankに送れるsend
void MPI_negotiator::mpi_send(int dest_rank, MpiTag tag, void* ptr, size_t size)
{
  // MPIのtagの仕組みは他にportingするときに使いにくいので利用しないことにする。
  const int mpi_tag = 0;

  // まずサイズ情報を送信
  MessageHeader mes;
  mes.tag = tag;
  mes.size = size;
  mes.source_rank = rank;
  MPI_Send(&mes, sizeof(MessageHeader), MPI_BYTE, dest_rank, mpi_tag, MPI_COMM_WORLD);

  // そのあと続いて、バイナリを送信
  // sizeはintなので2GBまでしか送れないのか。いまどき、なんちゅー糞設計なんや…。

  // sizeはsize_tで、これは無符号型なので引き算でマイナスになってしまう。
  int64_t s = (int64_t)size;
  while (s > 0)
  {
    MPI_Send((const void*)ptr,(int) std::min(size, size_t(INT64_C(0x40000000))), MPI_BYTE, dest_rank, mpi_tag, MPI_COMM_WORLD);
    s -= 0x40000000; // 糞実装なので1GBずつ送るわ
    ptr = (void*)((char*)ptr + 0x40000000);
  }

  // これにより、どんなサイズのどんな型でも事前の情報なしに交換できる。
}

// どんなサイズのバイナリでも受信できるreceive
void MPI_negotiator::mpi_receive(int source_rank, MessageHeader& mes, char*& ptr)
{
  MPI_Status status;
  
  // MPIのtagの仕組みは他にportingするときに使いにくいので利用しないことにする。
  const int mpi_tag = 0;
  mes.tag = TAG_MESSAGE_NONE;

  int flag;
  MPI_Iprobe(source_rank, mpi_tag, MPI_COMM_WORLD, &flag, &status);
  if (flag)
  {
    // 何かデータがあるので受信する。
    MPI_Recv(&mes, sizeof(mes), MPI_BYTE, source_rank, mpi_tag, MPI_COMM_WORLD,&status);

    // 同じsource rankから受信すれば連続して続きのデータが送られてきていると仮定できる。

    // 何もないので返却不要
    if (mes.size == 0)
    {
      ptr = nullptr;
      return;
    }

    ptr = new char[mes.size];
    auto p = ptr;
    int64_t s = (int64_t)mes.size; // 残り読み込みサイズ
    while (s > 0)
    {
      MPI_Recv((void*)p,(int) std::min(s, INT64_C(0x40000000)), MPI_BYTE, mes.source_rank, 0, MPI_COMM_WORLD, &status);
      p += 0x40000000;
      s -= 0x40000000;
    }
  }
}

// どんなサイズのバイナリでも指定のrankに送れるsend

void MPI_negotiator::mpi_send(int dest_rank, MpiTag tag, void* ptr, size_t size)

{

// MPIのtagの仕組みは他にportingするときに使いにくいので利用しないことにする。

const int mpi_tag = 0;

// まずサイズ情報を送信

MessageHeader mes;

mes.tag = tag;

mes.size = size;

mes.source_rank = rank;

MPI_Send(&mes, sizeof(MessageHeader), MPI_BYTE, dest_rank, mpi_tag, MPI_COMM_WORLD);

// そのあと続いて、バイナリを送信

// sizeはintなので2GBまでしか送れないのか。いまどき、なんちゅー糞設計なんや…。

// sizeはsize_tで、これは無符号型なので引き算でマイナスになってしまう。

int64_t s = (int64_t)size;

while (s > 0)

{

MPI_Send((const void*)ptr,(int) std::min(size, size_t(INT64_C(0x40000000))), MPI_BYTE, dest_rank, mpi_tag, MPI_COMM_WORLD);

s -= 0x40000000; // 糞実装なので1GBずつ送るわ

ptr = (void*)((char*)ptr + 0x40000000);

}

// これにより、どんなサイズのどんな型でも事前の情報なしに交換できる。

}

// どんなサイズのバイナリでも受信できるreceive

void MPI_negotiator::mpi_receive(int source_rank, MessageHeader& mes, char*& ptr)

{

MPI_Status status;

// MPIのtagの仕組みは他にportingするときに使いにくいので利用しないことにする。

const int mpi_tag = 0;

mes.tag = TAG_MESSAGE_NONE;

int flag;

MPI_Iprobe(source_rank, mpi_tag, MPI_COMM_WORLD, &flag, &status);

if (flag)

{

// 何かデータがあるので受信する。

MPI_Recv(&mes, sizeof(mes), MPI_BYTE, source_rank, mpi_tag, MPI_COMM_WORLD,&status);

// 同じsource rankから受信すれば連続して続きのデータが送られてきていると仮定できる。

// 何もないので返却不要

if (mes.size == 0)

{

ptr = nullptr;

return;

}

ptr = new char[mes.size];

auto p = ptr;

int64_t s = (int64_t)mes.size; // 残り読み込みサイズ

while (s > 0)

{

MPI_Recv((void*)p,(int) std::min(s, INT64_C(0x40000000)), MPI_BYTE, mes.source_rank, 0, MPI_COMM_WORLD, &status);

p += 0x40000000;

s -= 0x40000000;

}