インテルTBBの同期メカニズム

mutex ひとめぐり

recursive_mutex

　当座預金口座のサンプルに手を加え、入金／払出にともなうデータの読み出し／書き込みの回数を勘定してみましょう。読み出し／書き込みカウンタを埋め込んだクラスを用意し、入金／払出のふたつのスレッドで同時にアクセスします。

#include <tbb/tbb.h>
#include <iostream>

using namespace std;

template<typename T>
class usage_counted_value {
  T value;
  mutable unsigned nget;
  mutable unsigned nset;
  typedef tbb::spin_mutex mutex_type;
  mutable mutex_type mtx;
public:
  usage_counted_value(T init) : value(init), nget(0), nset(0) {}
  T get() const { mutex_type::scoped_lock lock(mtx); ++nget; return value; }
  void set(T v) { mutex_type::scoped_lock lock(mtx); ++nset; value = v;    }
  unsigned get_count() const { return nget; }
  unsigned set_count() const { return nset; }
};

void run() {
  const int N = 1000000;
  usage_counted_value<long long> value(0);
  tbb::tick_count cnt = tbb::tick_count::now();
  tbb::tbb_thread add([&]() { for ( int i = 0; i < N; ++i ) { value.set(value.get()+i); }});
  tbb::tbb_thread sub([&]() { for ( int i = 0; i < N; ++i ) { value.set(value.get()-i); }});
  add.join();
  sub.join();
  cout << (tbb::tick_count::now() - cnt).seconds() << "[sec]\t";
  cout << " get: " << value.get_count()
       << " set: " << value.set_count();
  cout << " value: " << value.get() << endl;
}

int main() {
  for ( int i = 0; i< 10; ++i ) {
    run();
  }
}

　結果が狂ってしまいました。get()／set()はそれぞれちゃんとscoped_lockでガードされていますが、value.set(value.get()+i)の部分で割り込めてしまうからです。加算メソッドadd()を用意し、各スレッドからはadd()を呼ぶよう書き換えます。

#include <tbb/tbb.h>
#include <iostream>

using namespace std;

template<typename T>
class usage_counted_value {
  T value;
  mutable unsigned nget;
  mutable unsigned nset;
  typedef tbb::spin_mutex mutex_type;
  mutable mutex_type mtx;
public:
  usage_counted_value(T init) : value(init), nget(0), nset(0) {}
  T get() const { mutex_type::scoped_lock lock(mtx); ++nget; return value; }
  void set(T v) { mutex_type::scoped_lock lock(mtx); ++nset; value = v;    }
  void add(T v) { mutex_type::scoped_lock lock(mtx); set(get()+v);         } // 追加
  unsigned get_count() const { return nget; }
  unsigned set_count() const { return nset; }
};

void run() {
  const int N = 1000000;
  usage_counted_value<long long> value(0);
  tbb::tick_count cnt = tbb::tick_count::now();
  tbb::tbb_thread add([&]() { for ( int i = 0; i < N; ++i ) { value.add(i);  }});
  tbb::tbb_thread sub([&]() { for ( int i = 0; i < N; ++i ) { value.add(-i); }});
  add.join();
  sub.join();
  cout << (tbb::tick_count::now() - cnt).seconds() << "[sec]\t";
  cout << " get: " << value.get_count()
       << " set: " << value.set_count();
  cout << " value: " << value.get() << endl;

}

int main() {
  for ( int i = 0; i< 10; ++i ) {
    run();
  }
}

　今度は「だんまり」になってしまいました。add()で実行権を取得済なので、その中で呼んだget()／set()で実行権が得られず、ブロックが解けることなく無限の待ち状態となります。いわゆるデッドロックですね。

　ここで、tbb::spin_mutexをtbb::recursive_mutexに差し替えるとあっさり解決します。

　recursive_mutexはacquireで実行権が取得されていても、同一スレッドからであれば何度でも重ねて取得できます。recursive_mutex内にカウンタが仕込んであり、aquire時に+1,release時に-1して、0となったら他スレッドのaquireが許されます。