Boost

複数のスレッドから1つの変数にアクセスする際、システム側のスレッドスケジューリング次第で、予期せぬ書き換えが起こってしまう場合があります。その為、ある1つのスレッドが変数にアクセスしている際は他のスレッドをブロックする排他制御やスレッドの同期を行う必要があります。C++でJavaのsynchronizedメソッド/ブロックと同じような記法でクリティカルセクションを実装する方法の1つにboost::threadライブラリのmutexとconditionクラスがあります。

mutex クラスの使い方

スレッドの排他制御を実現できます。具体的な使い方は、mutexインスタンスをmutex::scoped_lockクラスのコンストラクタの引数に指定し、そのインスタンスを取得することでロックをかけられます。あるスレッドが上の処理を以ってmutexインスタンスにロックをかけた場合、その他のスレッドは再度同一のmutexインスタンスにロックをかけられないようになっており、その他のスレッドはscoped_lockのコンストラクタ途中で待たされます。

mutexインスタンスのロック解除は、ロックをかけたスレッドがscoped_lockインスタンスのデストラクタを実行することで完了します。

condition クラスの使い方

次に、複数のスレッドを同期させる処理について。例えば、あるスレッドが排他的にある変数にアクセスしようとするが、その前にif文を用いて「Wait! そのアクセスちょっと待った！しばらく他のスレッドの処理を待て！」みたいな処理を行いたい場合。そのスレッドは既にmutex変数でロックを取得していますが、「待ち」に入る際はロックを解除する必要があります。

上述のような処理を実現するのがconditionクラスです。メンバ関数のwait()に引数にmutexインスタンスを指定することで、そのスレッドはロックを解除し一時停止します。他のスレッドがnotify_all()を実行すると、一時停止中の全てのスレッドが実行可能状態になります。これによって、スレッドの同期を実現します。

実際にどんな挙動か確かめよう

上の説明は以下のサンプルと実行結果を先に確認した後の方がしっくりくるかもしれません。

ソースコード

#include <iostream>
#include <string>
#include <boost/thread.hpp>
#include <boost/bind.hpp>
using namespace std;
using namespace boost;

class HandleData {
 public:
  HandleData()
    : index_(0), iarr_len_(sizeof(iarr_)/sizeof(iarr_[0])), num_data_(0){}

  // データの追加
  void putData(int n) {
    // mutexインスタンスにロックをかける
    mutex::scoped_lock look(thread_sync_);
    while (index_ >= iarr_len_) {
      printf("[putData] waitn");
      // 引数のmutexインスタンスのロックを解除する。
      // notify_all()などが呼ばれるまでこのスレッドを一時停止する
      thread_state_.wait(thread_sync_);
    }
    printf("put: %d in iarr_[%d]n", n, index_);
    iarr_[index_++] = n;
    int loop = 1000000;
    while(loop--) ; // 空回し用
    thread_state_.notify_all();
  } // lockのデストラクタが呼ばれてロック解除(lookインスタンスのスコープ切れ)

  // データの取得
  int getData() {
    // putData()と同じくiarr_にスレッドセーフでアクセスする為にロックをかける
    mutex::scoped_lock look(thread_sync_);
    while (index_ <= 0) { // putData側のthread_state_.notify_all();が実行されるまで待つ
      printf("[getData] waitn");
      thread_state_.wait(thread_sync_);
    }
    num_data_ = iarr_[--index_];
    printf("get: %d in iarr_[%d]n", num_data_, index_);
    thread_state_.notify_all();
    return num_data_;
  } // lookのデストラクタでロック解除

 private:
  mutex thread_sync_;
  condition thread_state_;
  volatile int index_;
  int iarr_[100]; // putData()、getData()からアクセスされる変数
  const unsigned int iarr_len_;
  int num_data_;
};

void threadPut(HandleData *hd) {
  const unsigned int NUM_LOOP = 100;
  for (int i = 0; i < NUM_LOOP; i++) {
    hd->putData(i);
  }
}

void threadGet(HandleData *hd) {
  const unsigned int NUM_LOOP = 100;
  for (int i = 0; i < NUM_LOOP; i++) {
    hd->getData();
  }
}

int main()
{
  HandleData hd;
  thread thr_put(bind(&threadPut, &hd));
  thread thr_get(bind(&threadGet, &hd));
  thr_put.join();
  thr_get.join();
  return 0;
}

実行結果の一例

実行する環境によって、出力結果は変化します(格納・出力順序など)。

put: 0 in iarr_[0]
get: 0 in iarr_[0]
put: 1 in iarr_[0]
get: 1 in iarr_[0]
＜中略＞
get: 12 in iarr_[1]
get: 11 in iarr_[0]
[getData] wait
put: 13 in iarr_[0]
put: 14 in iarr_[1]
get: 14 in iarr_[1]
get: 13 in iarr_[0]
[getData] wait
put: 15 in iarr_[0]
put: 16 in iarr_[1]
＜中略＞
get: 95 in iarr_[0]
[getData] wait
put: 97 in iarr_[0]
put: 98 in iarr_[1]
get: 98 in iarr_[1]
get: 97 in iarr_[0]
[getData] wait
put: 99 in iarr_[0]
get: 99 in iarr_[0]

volatile 修飾子の必要性

52行目のメンバ変数のindexの宣言

volatile int index_;

volatile(揮発性)接頭語がつくと、コンパイラによる最適化を防ぐことができます。これによって、プログラムはindex_を読み取る際は必ずメモリから読みにいきます。これは変数が複数スレッドから常に書き換えられても、必ずそれが反映されるということです。

このサンプルでは付けなくても最適化によるバグの発生はなさそうかな...?

リファレンス

• Boost C++ Libraries - Chapter 17. Thread
  • Boost C++ Libraries - Synchronization

同じような処理を行うスレッドが複数ある場合は、それらをスレッドグループでまとめると、スレッドへの操作がやり易くなります。スレッドグループへの登録には、boost::thread ライブラリの thread_group クラスを用いて、メンバ関数 create_thread() の引数にマルチスレッドで実行したい関数のアドレスを指定します。

それでは下記のサンプルで実際にその過程と実行結果を確認してみましょう。

ソースコード

#include <iostream>
#include <boost/bind.hpp>
#include <boost/thread.hpp>
using namespace std;
using namespace boost;

// マルチスレッドで実行する関数
void run(const char thr_name) {
  int count = 0;
  while(1) {
    if (++count % 100000000 == 0) // 空回り用の if 文
      cout << "Thread-" << thr_name << ": No." << count << endl;
  }
}

int main()
{
  const char chs[] = {'A', 'B', 'C', 'D'}; // 生成スレッドの名前の配列
  const int NUM_THREAD = sizeof(chs) / sizeof(chs[0]); // 配列の要素数の算出

  // スレッドグループの生成と実行開始
  thread_group thr_grp;
  for (int i = 0; i < NUM_THREAD; ++i) {
    thr_grp.create_thread(bind(&run, chs[i])); // create_thread()でrun()を別スレッドで実行
  }
  // join_all()で全スレッドの終了を待つ
  thr_grp.join_all();
  return 0;
}

実行結果の一例

実行する環境によって、出力結果は変化します(順序など)。

Thread-B: No.100000000
Thread-A: No.100000000
Thread-C: No.100000000
Thread-D: No.100000000
Thread-B: No.200000000
Thread-A: No.200000000
Thread-C: No.200000000
Thread-D: No.200000000
Thread-B: No.300000000
Thread-A: No.300000000
Thread-C: No.300000000
Thread-D: No.300000000
Thread-B: No.400000000
Thread-A: No.400000000
＜略＞

リファレンス

• Boost C++ Libraries - Chapter 17. Thread
  • Boost C++ Libraries - Thread Management

C/C++でスレッドを扱う場合は、プラットフォームによって使用するライブラリが違います。 Windows なら Windows API の thread で、 UNIX や Linux 系ならば pthread ライブラリ等を使用します。プラットフォーム依存するコードは可搬性に難があり、解決策の1つとしてプリコンパイルで依存部分をプラットフォームに合わせたライブラリを選択してコンパイルする方法があります。

boost ライブラリの boost::thread は、上のような処理をラップして共通のインターフェイスとして実装されています。

boost/thread.hppの一部

#if defined(BOOST_THREAD_PLATFORM_WIN32)
#include <boost/thread/win32/thread.hpp>
#elif defined(BOOST_THREAD_PLATFORM_PTHREAD)
#include <boost/thread/pthread/thread.hpp>
#else
#error "Boost threads unavailable on this platform"
#endif

これによって、 boost がインストールされている OS ならば、同じコードでコンパイルすることが出来ます。さて、それでは boost::thread を使って基本的なスレッドの生成と実行を、以下の三つのパターンで確認してみます。

• 引数なし関数をマルチスレッドで実行
• 引数付き関数をマルチスレッドで実行
• メンバ関数をマルチスレッドで実行

引数なし関数をマルチスレッドで実行

ソースコード

#include <iostream>
#include <boost/thread.hpp>
using namespace std;
using namespace boost;

// 引数なし関数をマルチスレッドで実行

const int kCount =  100000000;
const int kInterval = 1000000;

void PrintHello() {
  for(int i = 0; i != kCount; ++i)
    if (!(i % kInterval)) cout << "Hello ";
}

void PrintWorld() {
  for(int i = 0; i != kCount; ++i)
    if (!(i % kInterval)) cout << "World! ";
}

int main() {
  thread thr_hello(&PrintHello); // PrintHello関数を実行するスレッドの生成、実行
  thread thr_world(&PrintWorld);

  thr_hello.join(); // thr_helloが走らせているスレッドの終了を待つ
  thr_world.join();
  return 0;
}

実行結果の一部抜粋


Hello World! Hello Hello World! Hello World! Hello World! Hello World! World! Hello World! Hello World! Hello World! Hello World! Hello World! World!


13行目: PrintHello() の for ループで kInteral を用いて空回りの負荷をある程度与えることで、スレッドの切り替え(スレッドスケジューリング)を出力に反映させやすくしています。仮に、単なるforループだと、マシン性能が高い場合スレッドの切り替え前に関数の実行が終了してしまう場合もありマルチスレッドか否かの確認がとり難いです。

23行目: thread クラスのインスタンスを生成しそのコンストラクタに、0引数の関数のアドレスを渡すことで、その関数を走らせるスレッドを生成、実行しています。

ここで、関数アドレスを表す表記は&関数名ですが、このアンパサンド「&」は省略できます。実際に確認しますと、

cout <<  PrintHello << endl;
cout << &PrintHello << endl;

の出力結果は、

003B17A8
003B17A8

となります。実行する環境によって上の数値は変わりますが、ここで大事なのは二つの出力が同値である、すなわち「関数名＝＝&関数名＝＝関数のアドレス」ということです。

余談ですが、関数アドレスを格納する関数ポインタを引数として渡すことも勿論可能です。例えば、

void (*fp)(void) = PrintHello; // 関数のアドレスを指す関数ポインタを宣言
thread thr_fp(fp); // ポインタが指すアドレスにある関数を走らせるスレッドを生成、実行

とも書けます。さて、次に引数をとる関数のスレッドの生成、実行方法を確認しましょう。

引数付き関数をマルチスレッドで実行

ソースコード

#include <iostream>
#include <boost/thread.hpp>
#include <boost/bind.hpp>
using namespace std;
using namespace boost;

// 引数付き関数をマルチスレッドで実行

const int kCount =  100000000;
const int kInterval = 1000000;

void PrintString(const char *str) {
  for(int i = 0; i != kCount; ++i)
    if (!(i % kInterval)) cout << str;
}

int main() {
  const char *kHello = "Hello ";
  const char *kWorld = "World! ";

  thread thr_hello(bind(&PrintString, kHello));
  thread thr_world(bind(&PrintString, kWorld));

  thr_hello.join();
  thr_world.join();
  return 0;
}

実行結果の一部抜粋


Hello World! Hello Hello World! Hello World! Hello World! Hello World! World! Hello World! Hello World! Hello World! Hello World! Hello World! World!


21行目: thread クラスのコンストラクタには関数オブジェクトも渡せるので、 boost::bind を用いて、その第二引数にPrintString 関数に渡したい引数を指定して、関数オブジェクトを生成しています。

さて、2つのスレッドの生成法を確認しましたが、 Java や他のオブジェクト指向をサポートしている言語を使い慣れている人には、違和感のある書き方だったかもしれません。私は「引数よりメンバ変数の呼び出しの方がいい」「C++ならオブジェクト指向だよね」と脊髄反射的に思ったり。
ということで最後は、

メンバ関数をマルチスレッドで実行

ソースコード

#include <iostream>
#include <boost/thread.hpp>
#include <boost/bind.hpp>
using namespace std;
using namespace boost;

// メンバ関数をマルチスレッドで実行

class PrintMessage {
 public:
  PrintMessage(const char* msg)
    :kMessage_(msg), kCount_(100000000), kInterval_(1000000) {}
  void run();

 private:
  const int kCount_;
  const int kInterval_;
  const char *kMessage_;
};

void PrintMessage::run() {
  for(int i = 0; i != kCount_; ++i)
    if (!(i % kInterval_)) cout << kMessage_;
}

int main() {
  PrintMessage hello("Hello ");
  PrintMessage world("World! ");

  thread thr_hello(bind(&PrintMessage::run, &hello));
  thread thr_world(bind(&PrintMessage::run, &world));

  thr_hello.join();
  thr_world.join();
  return 0;
}

実行結果の一部抜粋


Hello World! Hello Hello World! Hello World! Hello World! Hello World! World! Hello World! Hello World! Hello World! Hello World! Hello World! World!


おー、盛り上がってきましたね。

リファレンス

Boost C++ Libraries - Boost.Threads - <boost/thread.hpp>

雑感

エラー処理を入れてない...さらっておこう。あと色々なライブラリを使いこなす為に、テンプレートを復習しよう。