C++でスレッドの停止機構と同期をやってみる

さて、本日は久々にC++のお話です。 
っていってもC++の仕様がどうとかいうのは書くつもりはないのですが、 会社で色々書いていてスレッドを大量に扱わなければならなかったので、 その基礎的な技術を備忘録兼ねてここに載せようと思います。

結構色々な処理抜いてるっていうか、 骨組みだけの内容なので、 ご利用は計画的に♥ 

ソースコード

何のことはない、ただフラグで制御しているループの中で関数オブジェクトを呼ぶだけのクラスです。

ただし、C++ の同期の仕組みとして提供されている stl の futer パターンを使用して、スレッドの終了時に値を返してくるのを待てるようにしてあります。

基本的には以下のように、テンプレートの機構を使って多様な関数オブジェクトを取り実行しています。
 可変引数テンプレートを加えて、

<class T_function_object, class... T_argments >

このようにすれば、関数呼び出し時に引数を与えるようにも作れるのですが、下手にインターフェースを便利にすると大域変数とか与えかねないので今回は省いています。

スレッドの本体 

スレッドの動作内容と停止自体は非常に簡単です。 
this ポインタを介してラムダ式て定義した関数内のループを制御しているだけです。

もちろんコールバックした関数オブジェクトの中で暴走させられるとどうしようもないんですが、、、 

ラムダ式で構築された本来無名の関数は STL の function クラスを使って、呼び出せる形で格納しています。

std::function< void() > async_process_ = 
 [this]( ){
  do{
      this->callback_();
  }
  while( continue_flag_ );

  this->promise_.set_value( std::move( callback_ ) );
 };

 T_function_object        callback_;
 bool          continue_flag_;

スレッドの開始  

機構的にはコンストラクタで実行したい関数オブジェクトが与えられればそれでスレッドが走り出します。

 同時に future は promise オブジェクトを受取ります。

promiseには実行した関数オブジェクトそのものを返させています。

他の開発者に対して使用するパラメータや状態などは必ずオブジェクトの中で完結させてやり取りしてねという意思表示です。通じるかどうかは別として。 

※コンストラクタで例外起こるような処理するなって思う方はどうぞご自分でお書きください。 
リソースの生存期間はコンストラクタ終了後からなので私はあんまそこにこだわる意味を感じないです。

template
class my_thread_base
{
 public:
 my_thread_base( T_function_object& callback_function_ ):
 callback_( callback_function_ ),
 promise_(),
 future_( promise_.get_future() ),
 continue_flag_( true ),
 thread_runner_( async_process_ )
 {}

 my_thread_base( T_function_object& callback_function_ , bool one_shot):
 callback_( callback_function_ ),
 promise_(),
 future_( promise_.get_future() ),
 continue_flag_( !one_shot ),
 thread_runner_( async_process_ )
 {}

スレッドの破棄  

もちろんデストラクタでは、スレッドを停止させるためにフラグを折ります。

また thread クラスや future パターンが値を保持し続けた状態とならないよう、後片付けをしています。

thread クラスは join または detach でスレッドを指さなくなります。
不要な参照を抱えたまま破棄されてしまうとリソースリークしてしまうので、きっちり開放してやりましょう。

future も同様に get で promise への参照を手放しますので最後にしっかり開放します。

ただし既に一度 get を実行していた場合、存在しない参照を get で取りに行こうとしてコアダンプとなってしまいます。
ここでもしっかりエラー処理をしましょう。
promise オブジェクトを抱えているかどうかは、valid で確認することが出来ます。

~my_thread_base()
 {
  continue_flag_ = false;
  if( thread_runner_.joinable() ) thread_runner_.join();
  if( future_.valid() ) future_.get();
 }

使用方法 

関数オブジェクト与えて適当に呼んでください。
それだけです。

ムーブコンストラクタ定義してればムーブでも呼べます。(このサンプル起こしたときに何か抜けてなければ…‥) このサンプルは業務で作ったフレームワークに使った技術を抜粋してダーッと書いてダーッと実行したので、右辺値参照は未確認です。

//関数オブジェクトのインスタンスを渡す場合
 print_and_add paa(100, 200);
 my_thread_base< print_and_add >    thread_test( std::move( paa ), true );   // 一回だけ実行
 my_thread_base< print_and_add >    thread_test( std::move( paa ), false );  // 止まるか死ぬまで実行
 my_thread_base< print_and_add >    thread_test( std::move( paa ) );    // 止まるか死ぬまで実行

//というか function クラスを実行できるようにしてあるから大抵何でも呼べる
 std::function< void () > lamda_driver = [](){
  std::cout  << "lamda executed. " << std::endl;
 };
 my_thread_base< decltype( lamda_driver ) >  thread_test5( std::move( lamda_driver ), true );

ところで lamda driverってどっかで聞いたことあるけど何だっけ、まあどうでもいいか。 

実行結果

実際に冒頭のソースコードをフルでコピって実行すると以下のような結果が得られます。

初期状態
add  answer @ main = 0
mul  answer @ main = 0
sum  answer @ main = 0
sum2 answer @ main = 0
スレッド実行中
100 + 200 = 300
100 x 200 = 20000
Sum from 0 to 100 = 5050
Sum from 0 to 1000 = 500500
スレッド終了後
add  answer @ main = 300
mul  answer @ main = 20000
sum  answer @ main = 5050
sum2 answer @ main = 500500

定義時のワンショットフラグを外して実行すれば、 signal_red 関数を呼び出されるか、スコープを抜けてオブジェクトが開放されるまで与えられた関数オブジェクトを実行し続けます。

最後に

signal_green 関数はこのサンプルじゃあんま使いみち無かったかな…

ここのところはこのサンプルのような構造に mutex やらバリア同期の真似事やらごちゃごちゃ機構を突っ込んだスレッドを、大量に使うフレームワークを作っているわけです。

この例は例のためのプログラムなのでご利用（する人いんのか？）は自己責任でお願いします。

っていうかサンプル処理もうちょいいいの思いつかないのか自分・・・