C++11における同期処理(std::mutex, std::unique_guard, std::lock_guard, std::condition_variable)

前回は非同期処理についてまとめたが、
今回は並行(concurrent)処理中の同期が必要な処理をC++11で実行するために必要な知識をまとめていく。
ThreadPoolを実装するために必要な知識として、

• mutexによるロック
• 条件変数の使い方

をまとめる。
ThreadPoolはまた次回に持ち越しである。

mutexを用いたロック: std::unique_lock or std::lock_guard ?

スレッド間でもプロセス間でも相互排他処理、
つまりある操作を同時に実行するスレッド/プロセスが一つである事を保証する必要がある場合がある。
このような排他的に実行する必要のある処理をクリティカルセッションと呼ぶ。
相互排他処理を実現するための同期機構としてmutexというものがある。
Wikipediaによれば相互排他(MUTual EXclusion)の省略形が語源だそうだ。

C++11ではmutexを簡単に扱うためヘッダ<mutex>が用意されている。
以下のクラスがここで定義されている。

• std::mutex: mutexの本体。単独でも使えるが、自動でロックを解除しないので以下を使う事が推奨される。
• std::lock_guard<Mutex>: 単純なScoped Locking Patternを実装する。
  つまりコンストラクタでmutexをロックして他のスレッドがクリティカルセッションに入るのを防止し、
  デストラクタでロックを開放する。
• std::unique_lock<Mutex>: Scoped Locking Patternを実装する高機能なロッククラス。
  std::lock_guard<Mutex>がコンストラクタでしかロックできないのに対し、こちらは任意のタイミングでロックを取得できる。
  さらにロックの所有権を移譲するような処理が書ける。
  std::lock_guard<Mutex>の上位互換だが、その分実行時コストがある。

使い方

基本的なScoped Locking Pattern:

std::mutex mtx;
{
    std::lock_guard<std::mutex> lock(mtx); // mtxを使ってロックする
    do_critical_session(); // この部分を実行している間は
                           // 他のスレッドでmtxを使ってロックできない
} // デストラクタでロックが開放される

単独で使用する分にはstd::lock_guard<Mutex>で十分なように思う。
std::unique_lock<Mutex>は条件変数の項で扱う。

注意

• std::mutexの初期化について(mutexのconstexprコンストラクタより転載, 改行を調整)：

これにより非ローカルなstd::mutex型変数の初期化は静的初期化(static initialization)として扱われ、
任意の他スレッド開始（＝std::thread型変数の動的初期化）よりも前にmutexオブジェクトが初期化済みである事が保証される。

参考文献

• std::unique_lockstd::mutex or std::lock_guardstd::mutex? - Stack Overflow
• std::unique_lockはミューテックスインタフェースを持つ - Faith and Brave - C++で遊ぼう
• std::unique_lock - cppreference.com
• std::unique_lock - cpprefjp
• std::lock_guard - cpprefjp

条件変数の使い方 : std::condition_variable

条件変数は元々POSIXスレッドにおける用語だったが、他の実装でも同様の機能に対して同じ名前を用いる。
以下の説明がわかりやすい:

同期処理を行う際、ある共有情報が条件を満たしたらスレッドが動いて欲しい、という時があります。
それまではサスペンドしてて欲しいですが、条件が満たされた時に他のスレッドから処理を始めるよう合図する仕組みが必要です。
条件変数は上記のような仕組みを提供します。
混乱しないように述べますが、mutexは共有情報への同時多重アクセスを回避するための機能で、
条件変数はもっと単純で、スレッドの停止・再開を指示する信号のような機能です。

(pthreadについて（条件変数・モデル） - CodeZine より転載, 改行を調整)

基本的に、スレッドの実行を一時停止して、他のスレッドの実行を待つ場合に使用する。

使い方

以下にごく単純な例を示す。
メッセージが出力されるタイミングに注目してほしい。

condition_variable.cpp

#include <mutex>
#include <condition_variable>
#include <atomic>
#include <iostream>
#include <thread>

std::mutex mtx;
std::condition_variable cv;
bool is_ready = false; // for spurious wakeup

void do_preparing_process(){
  std::cout << "Start Preparing" << std::endl;
  // preparing
  // ... σ(^_^;)ｱｾｱｾ...
  std::this_thread::sleep_for(std::chrono::seconds(3));
  // finish preparing
  std::cout << "Finish Preparing" << std::endl;;
  {
    std::lock_guard<std::mutex> lock(mtx);
    is_ready = true;
  }
  cv.notify_one();
}

void do_main_process(){
  std::this_thread::sleep_for(std::chrono::seconds(1));
  std::cout << "Start Main Thread" << std::endl;
  {
    std::unique_lock<std::mutex> uniq_lk(mtx); // ここでロックされる
    cv.wait(uniq_lk, []{ return is_ready;});
    // 1. uniq_lkをアンロックする
    // 2. 通知を受けるまでこのスレッドをブロックする
    // 3. 通知を受けたらuniq_lkをロックする

    /* ここではuniq_lkはロックされたまま */

  } // デストラクタでアンロックする
  std::cout << "Finish Main Thread" << std::endl;
}

int main(int argc, char const* argv[])
{
  std::thread th_prepare([&]{ do_preparing_process(); });
  std::thread th_main([&]{ do_main_process(); });

  th_prepare.join();
  th_main.join();
  
  return 0;
}

実行結果は以下の通り

clang++ -std=c++11 condition_variable.cpp -pthread

Start Preparing     # ここで1秒まつ
Start Main Process  # ここで3秒まつ
Finish Preparing
Finish Main Thread

準備(do_preparing_process())が終了するまでメインスレッド(do_main_process())が待機している事が出力結果から読みとれる。

spurious wakeup

ここで問題なのは、何故is_ready変数が必要なのかという事だ。
do_preparing_process()はcv.notify_one()によって準備が終了した事を通知している筈である。
という事はis_readyなんて変数は必要ないのではないか？と思える。

これは大抵のライブラリがwait()状態から処理を再開する際に、
notify_one()あるいは他の通知を受けとった事を保証していないからである。
つまりどのスレッドも通知を条件変数に送っていないのにwait()状態が解除される事がある。
これをspurious wakeupと呼ぶ。日本語が確定していないようだが、 見せ掛けの目覚め である。
なのでこの場合、準備が確実に終了した事を通知する必要がある。
そのために用いているのがis_readyである。

また

cv.wait(uniq_lk, []{ return is_ready;});

の第二引数のラムダ式は、上述のspurious wakeupに対応するためのoverloadで、
この関数がfalseを返す限り、wait()状態が解けないようになっている。
(2014/5/30 追加) 下記の@yohhoyさんのコメントも参照のこと。
(2014/5/30 修正)
因みにこの関数の評価は第一引数のロックを使用する。

template <typename _Lock> void wait(_Lock &__lock) {
  unique_lock<mutex> __my_lock(_M_mutex);
  _Unlock<_Lock> __unlock(__lock);
  // _M_mutex must be unlocked before re-locking __lock so move
  // ownership of _M_mutex lock to an object with shorter lifetime.
  unique_lock<mutex> __my_lock2(std::move(__my_lock));
  _M_cond.wait(__my_lock2);
}

template <typename _Lock, typename _Predicate>
void wait(_Lock &__lock, _Predicate __p) {
  while (!__p())
    wait(__lock);
}

少なくとも私の環境の/usr/include/c++/4.8.2/condition_variableでの実装ではロックしてないように見える。

notify_one()がwait()より先に呼ばれたらどうなるのか？

(2014/5/30 修正, @yohhoy さんに感謝)
基本的にnotify_one()は無視される。
ではプログラマがwait()がnotify_one()より先に呼ばれるのを保証しないといけないかというと、そうではない:

condition_variable2.cpp

#include <mutex>
#include <condition_variable>
#include <atomic>
#include <iostream>
#include <thread>

std::mutex mtx;
std::condition_variable cv;
bool is_ready = false; // for spurious wakeup

void do_preparing_process(){
  std::cout << "Start Preparing" << std::endl;
  // preparing
  // ... σ(^_^;)ｱｾｱｾ...
  std::this_thread::sleep_for(std::chrono::seconds(3));
  // finish preparing
  std::cout << "Finish Preparing" << std::endl;;
  {
    std::lock_guard<std::mutex> lock(mtx);
    is_ready = true;
  }
  cv.notify_one();
}

void do_main_process(){
  std::this_thread::sleep_for(std::chrono::seconds(1));
  std::cout << "Start Main Thread" << std::endl;
  /* ここで準備がいらない操作が可能 */
  std::cout << "Doing task without Preparing..." << std::endl;
  std::this_thread::sleep_for(std::chrono::seconds(5));
  {
    std::unique_lock<std::mutex> uniq_lk(mtx); // ここでロックされる
    cv.wait(uniq_lk, []{ return is_ready;});
    // 1. uniq_lkをアンロックする
    // 2. 通知を受けるまでこのスレッドをブロックする
    // 3. 通知を受けたらuniq_lkをロックする

    /* ここではuniq_lkはロックされたまま */

  } // デストラクタでアンロックする
  std::cout << "Finish Main Thread" << std::endl;
}

int main(int argc, char const* argv[])
{
  std::thread th_prepare([&]{ do_preparing_process(); });
  std::thread th_main([&]{ do_main_process(); });

  th_prepare.join();
  th_main.join();
  
  return 0;
}

Start Preparing
Start Main Thread
Doing task without Preparing...
Finish Preparing
Finish Main Thread

調子にのって他の作業に熱中している間に、準備スレッドで準備が終わってしまった。
つまりwait()する前にnotify_one()が来ている事になる。
wait()する前に来たnotify_one()は全て無視される。
ただしwait(lock, pred)のoverloadで呼びだした場合、
最初にもpred()を評価し既にtrueであればそもそもwait()しないので問題ない。
つまり準備は別スレッドにまかせたまま、いくらでも作業できる。

注意事項

• mutexのロックはstd::unique_lock<std::mutex>を使用しなければならない
• それ以外のロック型を使用する場合はstd::condition_variable_anyを用いる
• ほとんどの実装で条件変数をwaitしているスレッドが、
  notifyされていないのに間違って起動する現象(spurious wakeup)が起る。

参考文献

• pthreadについて（条件変数・モデル） - CodeZine
• ミューテックス、セマフォ、条件変数、違いを整理してみよう
• 条件変数とspurious wakeup - yohhoyの日記
• std::condition_variable - cppreference.com
• condition_variable (C++11) - cpprefjp

補足

この記事を書いていてcpprefjp - C++ Library Referenceの存在を知った。
日本語でC++の標準ライブラリの解説が読める。
しかもサンプルコードが各項目にあり、Creative Commonsで公開されている。
非常に素晴しい。

変更履歴

• 2014/5/30
  • sleepをstd::this_thread::sleep_forに変更
  • spurious wakeup のセクションに嘘が書いてあったので修正
  • "notify_one()がwait()より先に呼ばれたらどうなるのか？"の節を書き換え