Windows の C で Guarded Suspension パターンを実現する (条件変数を実装する)

マルチスレッドプログラミングをしているときに、「ある条件が満たされているかどうか調べ、満たされていなければ条件が満たされるまで待機する」 という処理が必要になることがあります。 このような処理はパターン化されており、『増補改訂版 Java言語で学ぶデザインパターン入門 マルチスレッド編』 では Guarded Suspension パターンと呼ばれています。

POSIX のマルチスレッドが使える環境では、条件変数 (condition variable) pthread_cond を使って Guarded Suspension パターンを実現することができます。 しかし、Windows のスレッド機能には条件変数に相当するものはありません。

本記事では、Windows が提供している CriticalSection と Event を使って条件変数を実装し、Windows の C で Guarded Suspension パターンを実現する方法を説明します。

基本的な考え方

条件変数を使って Guarded Suspension パターンを実現する際の基本的な流れを説明します。

  • ミューテックスのロックを取得する
  • 条件が満たされているかどうか調べる (*)
    • 条件が満たされていれば次の処理に移る
    • 条件が満たされていなければ 条件変数の上で待つ → 他のスレッドから信号が来たらスレッド再開し (*) の処理に戻る
  • 何らかの処理をする
  • ミューテックスのロックを解放する

条件というのは、何らかの処理をするために満たされている必要がある条件です。 例えば、「変数 count が 0 以上である」 というような条件です。 当然ながら、条件に含まれる変数 (この例では count) は複数のスレッドから読み書きされるものなので、ミューテックスによる排他制御をしなければいけません。 上の説明で出てくるミューテックスは、条件に含まれる変数を保護するためのものです。

なお、条件変数の上で待つ際に、ミューテックスのロックを解放しなければ他のスレッドが条件に含まれる変数の値を変更できなくなってしまいデッドロックしてしまいます。 そうならないように、条件変数の上で待つときに自動的にミューテックスのロックを解放し、再開するときにミューテックスのロックを取得するようになっています。

条件変数の実装

さて、それでは Windows で条件変数を実装する手法を説明します。 条件変数を実装する上で重要なのは次の処理です。

  • 他のスレッドからシグナルが送られてくるのを待つ処理
  • 他のスレッドからシグナルが送られてくるのを待つときに、ミューテックスのロックを解放し、スレッドを再開するときにロックを取得する処理
  • 条件に含まれる変数を書き換えたときにシグナルを送信する処理

Windows ではシグナルのやりとりは Event を使って行います。 ミューテックスは CriticalSection を使用します *1。 また、pthread_cond では、条件変数の上で待機しているスレッドの中から 1 つだけ再開させるのか、待機しているスレッドを全て再開させるのか選ぶことができます。 今回はこの処理も同じように実装します。

まず条件変数を表す構造体ですが、シグナルのやり取りのための Event だけを扱えればいいので次のようになります。 Event が 2 つありますが、これは待機しているスレッドの中から 1 つだけ再開させるためのシグナルと、全てのスレッドを再開させるシグナルに対応しています。

typedef struct {
    HANDLE hEventsWakeup[2];
} COND_VARIABLE;

構造体 COND_VARIABLE の初期化処理と終了処理のための関数は次のようになります。

// 初期化処理を行う関数
void CondVariable_init( COND_VARIABLE *self ) {
    // 1 つのスレッドを起こすためのイベント
    self->hEventsWakeup[0] = CreateEvent( NULL, FALSE, FALSE, NULL );
    // 全てのスレッドを起こすためのイベント
    self->hEventsWakeup[1] = CreateEvent( NULL, TRUE,  FALSE, NULL );
}
// 終了処理を行う関数
void CondVariable_final( COND_VARIABLE *self ) {
    CloseHandle( self->hEventsWakeup[0] );
    CloseHandle( self->hEventsWakeup[1] );
}

また、条件が満たされていない場合に待機するための関数は次のようになります。 第 2 引数はロック済みのミューテックスをあらわします。 第 3 引数は最大の待ち時間です。 他のスレッドが起こすまでいつまでも待ち続けるならば INFINITE を渡します。

// 条件変数の上で待つための関数
void CondVariable_wait( COND_VARIABLE *self, CRITICAL_SECTION *cs_ptr, DWORD dwMilliseconds ) {
    ResetEvent( self->hEventsWakeup[0] );
    ResetEvent( self->hEventsWakeup[1] );
    LeaveCriticalSection( cs_ptr );
    WaitForMultipleObjects( 2, self->hEventsWakeup, FALSE, dwMilliseconds );
    EnterCriticalSection( cs_ptr );
}

条件に含まれる変数の値を変更した際に待機しているスレッドに通知を行うための関数は次のようになります。

// 待機しているスレッドのうち 1 つのスレッドを起こす関数
void CondVariable_notify( COND_VARIABLE *self ) {
    SetEvent( self->hEventsWakeup[0] );
}
// 待機しているスレッド全てを起こす関数
void CondVariable_notifyAll( COND_VARIABLE *self ) {
    SetEvent( self->hEventsWakeup[1] );
}

使用例

「変数 shCount が 0 より大きい」 というのを条件にして、条件が満たされていれば shCount をデクリメントして "process..." という文字列を出力するスレッドを 5 個動かすプログラムです。 メインスレッドでは約 1 秒ごとに shCount をインクリメントするので、約 1 秒ごとに "process..." という文字列が出力されます。

#include <stdio.h>
#include <process.h>
#include <windows.h>

// VARIABLE_COND の構造体と関数
typedef struct {
    HANDLE hEventsWakeup[2];
} COND_VARIABLE;
void CondVariable_init( COND_VARIABLE *self ) {
    // notify のためのイベント
    self->hEventsWakeup[0] = CreateEvent( NULL, FALSE, FALSE, NULL );
    // notifyAll のためのイベント
    self->hEventsWakeup[1] = CreateEvent( NULL, TRUE,  FALSE, NULL );
}
void CondVariable_final( COND_VARIABLE *self ) {
    CloseHandle( self->hEventsWakeup[0] );
    CloseHandle( self->hEventsWakeup[1] );
}
void CondVariable_wait( COND_VARIABLE *self, CRITICAL_SECTION *cs_ptr, DWORD dwMilliseconds ) {
    ResetEvent( self->hEventsWakeup[0] );
    ResetEvent( self->hEventsWakeup[1] );
    LeaveCriticalSection( cs_ptr );
    WaitForMultipleObjects( 2, self->hEventsWakeup, FALSE, dwMilliseconds );
    EnterCriticalSection( cs_ptr );
}
void CondVariable_notify( COND_VARIABLE *self ) {
    SetEvent( self->hEventsWakeup[0] );
}
void CondVariable_notifyAll( COND_VARIABLE *self ) {
    SetEvent( self->hEventsWakeup[1] );
}

// スレッド間で共有されるデータ
int shCount = 0;
// スレッド間で共有されるデータを保護するための排他処理, 同期機構
CRITICAL_SECTION shCS;
COND_VARIABLE    shCV;

// 子スレッドで行う処理
static unsigned __stdcall run( void *arg ) {
    // ミューテックスのロックを取得
    EnterCriticalSection( &shCS );
    // 条件が満たされているかどうかチェック
    while( shCount <= 0 ) {
        printf( "wait...\n" );
        fflush( stdout );
        // 条件が満たされていないので待機する
        CondVariable_wait( &shCV, &shCS, INFINITE );
        printf( "wakeup!\n" );
    }
    shCount--;
    printf( "decrease shCount: %d\n", shCount );
    printf( "process...\n" );
    fflush( stdout );
    // ミューテックスのロックを解放
    LeaveCriticalSection( &shCS );
    _endthreadex(0);
    return 0; // コンパイラの警告を殺す
}

int main( int argc, char **argv ) {
    int numSign = 5;
    HANDLE hThreads[numSign];
    InitializeCriticalSection( &shCS );
    CondVariable_init( &shCV );
    for( int i = 0; i < numSign; i++ ) {
        hThreads[i] = (HANDLE)_beginthreadex( NULL, 0, run, NULL, 0, NULL );
    }
    // スレッドの数だけカウンタをインクリメントする
    for( int i = 0; i < numSign; i++ ) {
        Sleep( 1000 );
        // 共有変数をいじるのでミューテックスのロックを取得
        EnterCriticalSection( &shCS );
        shCount++;
        printf( "increase shCount: %d\n", shCount );
        fflush( stdout );
        // 条件に含まれる変数を変更したので待機しているスレッドを起こす
        CondVariable_notify( &shCV );
        // ミューテックスのロックを解放
        LeaveCriticalSection( &shCS );
    }
    // 全てのスレッドが終了するまで待つ
    WaitForMultipleObjects( numSign, hThreads, TRUE, INFINITE );
    DeleteCriticalSection( &shCS );
    CondVariable_final( &shCV );
    return 0;
}

エラー処理など

この記事のコードは勉強用に書いたものなので、エラー処理等はちゃんとしていません。 実際に Windows で条件変数を使う場合はもうちょっとエラー処理等をちゃんとしないとだめだと思います。

*1:Windows には CriticalSection のほかに Mutex がありますが、プロセス間の排他処理をしないのであれば CriticalSection を使ったほうが高速なので良いようです。

『増補改訂版 Java 言語で学ぶデザインパターン入門 マルチスレッド編』 結城 浩 著

マルチスレッドの基本を勉強するための本を探していたところ、知り合いに本書を薦められたので読んでみました。 もともと本書の存在は知っていて、デザインパターンの本ということでマルチスレッドの勉強には向かないんじゃないかと敬遠してたんですが、実際にはマルチスレッドの入門書としてすごく良い本でした。

マルチスレッドプログラミングで気をつけるべきところを体系的に学べる実践的入門書

本書はマルチスレッドに関するデザインパターンを説明した書籍です。 デザインパターンというと入門書よりもちょっと敷居が高い感じがしますが、そんなことはありません。 本書ではマルチスレッドプログラミングを行う際に 必ず使うであろう基本的な処理をパターン化したもの (具体的には、クリティカルセクションを同時に読み書きしないように排他処理をする、ということなど) から順に説明を行っていますので、マルチスレッドプログラミングの入門書として最適です。

マルチスレッドプログラミングをする際に気をつけるべきことと、実際にどう実装すればいいのか、ということをパターン化しており、実践的な内容となっています。 また、「なぜそれをする必要があるのか」 ということや 「もしそれをしなかったらどういう問題が起こるのか」 といったことを丁寧に説明していますので、マルチスレッドプログラミングの経験がない人でも理解しやすいと思います。 サンプルプログラムのソースコードのうち重要な部分を網掛けなどで強調しているのもわかりやすくて良かったです。

Java で使えるのはもちろんのこと Java 以外の言語でも有用

本書では Java を題材にしていますので、Java の基本的な部分は知っておかないと読み進めるのは困難です。 しかし、本書の内容は Java でしか使えないものではなく、どんな言語でもマルチスレッドプログラミングをする際には参考になるものです。 本書では 「Java ではどう実装するか」 を説明していますので、他の言語でどのように実装するかは自分で調べる必要がありますが、「マルチスレッドプログラミングをする際にどういうことを気をつければいいのか」 ということは本書を読んで学ぶことができます。

Java ではどう実装すればよいかという実践的な内容を説明しながら、言語に縛られないマルチスレッドに関する原則的な説明もしているのが本書の特徴だと思います。

java.util.concurrent パッケージの説明も

J2SE 5.0 ではマルチスレッドで動作させる際に便利なクラスライブラリ java.util.concurrent パッケージが追加されました。 本書ではこのクラスライブラリに含まれるクラスの使い方をサンプルプログラムにからめて説明しています。 (サンプルプログラムを java.util.concurrent パッケージのクラスを使って書き直すとどうなるか、など。)

本書を読んだだけで java.util.concurrent パッケージを使いこなせるようになるわけではありませんが、パッケージの一部のクラスの使い方を知れるだけでも意味のあることだと思います。

マルチスレッドプログラミングの入門者から初心者にオススメです

本書は 「デザインパターン入門」 と銘打たれているので、マルチスレッドプログラミングの経験者がデザインパターンを学ぶための本かと思ってしまいますが、実際にはマルチスレッドプログラミングの入門書にふさわしい内容になっています。 マルチスレッドプログラミングなんて経験したことがないという入門者や、マルチスレッドプログラミングはちょっとやったことがあるけどいまいちよくわかっていない、という初心者の人にオススメします。