条件変数の使用方法

この節では条件変数の使用方法を説明します。表 4-6 にそのための関数を示します。

条件変数の初期化

pthread_cond_init(3THR)

pthread_cond_init(3THR) は、cv が指す条件変数をデフォルト値 (cattr が NULL) に初期化します。また、pthread_condattr_init() ですでに設定してある条件変数の属性を指定することもできます。cattr を NULL にするのは、デフォルト条件変数属性オブジェクトのアドレスを渡すのと同じですが、メモリーのオーバーヘッドがありません。(Solaris スレッドについては、「cond_init(3THR)」を参照)。

プロトタイプ:
int	pthread_cond_init(pthread_cond_t *cv,
    const pthread_condattr_t *cattr);

#include <pthread.h>

pthread_cond_t cv;
pthread_condattr_t cattr;
int ret;

/* 条件変数をデフォルト値に初期化する */
ret = pthread_cond_init(&cv, NULL);

/* 条件変数を初期化する */
ret = pthread_cond_init(&cv, &cattr); 

静的に定義された条件変数は、マクロ PTHREAD_COND_INITIALIZER で、デフォルト属性をもつように直接初期化できます。この効果は、NULL 属性を指定して pthread_cond_init() を動的に割り当てるのと同じです。エラーチェックは行われません。

複数のスレッドで同じ条件変数を同時に初期化または再初期化しないでください。条件変数を再初期化または削除する場合、アプリケーションでその条件変数が現在使用されていないことを確認しなければなりません。

戻り値

正常終了時は 0 です。それ以外の戻り値は、エラーが発生したことを示します。以下のいずれかの条件が検出されると、この関数は失敗し、対応する値を返します。

EINVAL

cattr で指定された値が無効です。

EBUSY

その条件変数は現在使用されています。

EAGAIN

必要なリソースが利用できません。

ENOMEM

メモリー不足のため条件変数を初期化できません。

条件変数によるブロック

pthread_cond_wait(3THR)

pthread_cond_wait(3THR) は、mp が指す相互排他ロックを原子操作により解放し、cv が指す条件変数で呼び出しスレッドをブロックします。(Solaris スレッドについては、「cond_wait(3THR)」を参照)。

プロトタイプ:
int	pthread_cond_wait(pthread_cond_t *cv,pthread_mutex_t *mutex);

#include <pthread.h>

pthread_cond_t cv;
pthread_mutex_t mp;
int ret;

/* 条件変数でブロック */
ret = pthread_cond_wait(&cv, &mp); 

ブロックされたスレッドを呼び起こすには、pthread_cond_signal() か pthread_cond_broadcast() を使います。また、スレッドはシグナルの割り込みによっても呼び起こされます。

pthread_cond_wait() が戻ったからといって、条件変数に対応する条件の値が変化したと判断することはできません。このため、条件をもう一度評価しなければなりません。

pthread_cond_wait() が戻るときは、たとえエラーを戻したときでも、常に mutex は呼び出しスレッドがロックし保持している状態にあります。

pthread_cond_wait() は、指定の条件変数にシグナルが送られてくるまでブロック状態になります。pthread_cond_wait() は原子的操作により、対応する mutex ロックを解除してからブロック状態に入り、ブロック状態から戻る前にもう一度原子的操作によりロックを獲得します。

通常の用法は次のとおりです。mutex ロックの保護下で条件式を評価します。条件式が偽のとき、スレッドは条件変数でブロック状態に入ります。別のスレッドが条件の値を変更すると、条件変数にシグナルが送られます。その条件変数でブロックされていた (1 つまたは全部の) スレッドは、そのシグナルによってブロックが解除され、もう一度 mutex ロックを獲得しようとします。

呼び起こされたスレッドが pthread_cond_wait() から戻る前に条件が変更されることもあるので、mutex ロックを獲得する前に、待ち状態の原因となった条件をもう一度評価しなければなりません。条件チェックを while() ループに入れ、そこで pthread_cond_wait() を呼び出すようにすることをお勧めします。

    pthread_mutex_lock();
        while(condition_is_false)
            pthread_cond_wait();
    pthread_mutex_unlock();

条件変数で複数のスレッドがブロックされているとき、それらのスレッドが、どの順番でブロックが解除されるかは不定です。

pthread_cond_wait() は取り消しポイントです。保留状態になっている取り消しがあって、呼び出しスレッドが取り消しを有効 (使用可能) にしている場合、そのスレッドは終了し、ロックしている間にクリーンアップハンドラの実行を開始します。

戻り値

正常終了時は 0 です。それ以外の戻り値は、エラーが発生したことを示します。以下の条件が検出されると、この関数は失敗し、次の値を返します。

EINVAL

cv または mp で指定された値が無効です。

特定のスレッドのブロック解除

pthread_cond_signal(3THR)

pthread_cond_signal(3THR) は、cv が指す条件変数でブロックされている 1 つのスレッドのブロックを解除します。(Solaris スレッドについては、「cond_signal(3THR)」を参照)。

プロトタイプ:
int	pthread_cond_signal(pthread_cond_t *cv);

#include <pthread.h>

pthread_cond_t cv;
int ret;

/* ある条件変数がシグナルを送る */
ret = pthread_cond_signal(&cv); 

pthread_cond_signal() は、シグナルを送ろうとしている条件変数で使用されたものと同じ mutex ロックを獲得した状態で呼び出してください。そうしないと、関連する条件が評価されてから pthread_cond_wait() でブロック状態に入るまでの間に、条件変数にシグナルが送られる可能性があり、その場合 pthread_cond_wait() は永久に待ち続けることになります。

スケジューリング方針は、ブロックされたスレッドがどのように呼び起こされるかを決定します。SCHED_OTHER の場合、スレッドは優先順位に従って呼び起こされます。

スレッドがブロックされていない条件変数に対して pthread_cond_signal() を実行しても無視されます。

戻り値

正常終了時は 0 です。それ以外の戻り値は、エラーが発生したことを示します。以下の条件が検出されると、この関数は失敗し、次の値を戻します。

EINVAL

cv が指すアドレスが正しくありません。

例 4-8 pthread_cond_wait() と pthread_cond_signal() の使用例

pthread_mutex_t count_lock;
pthread_cond_t count_nonzero;
unsigned count;

decrement_count()
{
    pthread_mutex_lock(&count_lock);
    while (count == 0)
        pthread_cond_wait(&count_nonzero, &count_lock);
    count = count - 1;
    pthread_mutex_unlock(&count_lock);
}

increment_count()
{
    pthread_mutex_lock(&count_lock);
    if (count == 0)
        pthread_cond_signal(&count_nonzero);
    count = count + 1;
    pthread_mutex_unlock(&count_lock);
}

時刻指定のブロック

pthread_cond_timedwait(3THR)

プロトタイプ:
int	pthread_cond_timedwait(pthread_cond_t *cv,
    pthread_mutex_t *mp, const struct timespec *abstime);

#include <pthread.h>
#include <time.h>

pthread_cond_t cv;
pthread_mutex_t mp;
timestruct_t abstime;
int ret;

/* 条件変数で指定した時刻までブロック */
ret = pthread_cond_timedwait(&cv, &mp, &abstime);

pthread_cond_timedwait(3THR) は、abstime で指定した時刻を過ぎるとブロック状態を解除する点を除いて、pthread_cond_wait() と同じ動作をします。pthread_cond_timedwait() が戻るときは、たとえエラーを戻したときでも、常に mutex は呼び出しスレッドがロックして保持している状態です。(Solaris スレッドについては、「cond_timedwait(3THR)」を参照)。

pthread_cond_timedwait() のブロック状態が解除されるのは、条件変数にシグナルが送られてきたときか、一番最後の引数で指定した時刻を過ぎたときです。

pthread_cond_timedwait() は、取り消しポイントでもあります。

戻り値

正常終了時は 0 です。それ以外の戻り値は、エラーが発生したことを示します。以下の条件が検出されると、この関数は失敗し、次の値を戻します。

EINVAL

cv または abstime が不当なアドレスを指しています。

ETIMEDOUT

abstime で指定された時刻を過ぎています。

時間切れの指定は時刻で行うため、時間切れ時刻を再計算する必要がないので、効率的に条件を再評価できます (詳細は、例 4-9 を参照してください)。

例 4-9 時刻指定のブロック

pthread_timestruc_t to;
pthread_mutex_t m;
pthread_cond_t c;
...
pthread_mutex_lock(&m);
to.tv_sec = time(NULL) + TIMEOUT;
to.tv_nsec = 0;
while (cond == FALSE) {
    err = pthread_cond_timedwait(&c, &m, &to);
    if (err == ETIMEDOUT) {
        /* 時間切れの場合の処理 */
        break;
    }
}
pthread_mutex_unlock(&m);

全スレッドのブロック解除

pthread_cond_broadcast(3THR)

プロトタイプ:
int	pthread_cond_broadcast(pthread_cond_t *cv);

#include <pthread.h>

pthread_cond_t cv;
int ret;

/* 条件変数すべてがシグナルを受ける */
ret = pthread_cond_broadcast(&cv); 

pthread_cond_broadcast(3THR) は、cv (pthread_cond_wait() で指定された) が指す条件変数でブロックされている、すべてのスレッドのブロックを解除します。スレッドがブロックされていない条件変数に対して pthread_cond_broadcast() を実行しても無視されます。(Solaris スレッドについては、「cond_broadcast(3THR)」を参照)。

戻り値

正常終了時は 0 です。それ以外の戻り値は、エラーが発生したことを示します。以下の条件が検出されると、この関数は失敗し、次の値を戻します。

EINVAL

cv が指すアドレスが正しくありません。

条件変数に対するブロードキャストの例

条件変数でブロックされていたすべてのスレッドが、もう一度相互排他ロックを争奪するようになるので慎重に使用してください。たとえば、pthread_cond_broadcast() を使用すると可変量のリソースに対して、そのリソースが解放される時にスレッド間で争奪させることができます (詳細は、例 4-10 を参照してください)。

例 4-10 条件変数に対するブロードキャスト

pthread_mutex_t rsrc_lock;
pthread_cond_t rsrc_add;
unsigned int resources;

get_resources(int amount)
{
    pthread_mutex_lock(&rsrc_lock);
    while (resources < amount) {
        pthread_cond_wait(&rsrc_add, &rsrc_lock);
    }
    resources -= amount;
    pthread_mutex_unlock(&rsrc_lock);
}

add_resources(int amount)
{
    pthread_mutex_lock(&rsrc_lock);
    resources += amount;
    pthread_cond_broadcast(&rsrc_add);
    pthread_mutex_unlock(&rsrc_lock);
}

上記のコード例の add_resources() で、次の点に注意してください。相互排他ロックの範囲内では、resources の更新と pthread_cond_broadcast() の呼び出しはどちらを先に行なってもかまいません。

pthread_cond_broadcast() は、シグナルを送ろうとしている条件変数で使用されたものと同じ相互排他ロックを獲得した状態で呼び出してください。そうしないと、関連する条件変数が評価されてから pthread_cond_wait() でブロック状態に入るまでの間に条件変数にシグナルが送られる可能性があり、その場合 pthread_cond_wait() は永久に待ち続けることになります。

条件変数の削除

pthread_cond_destroy(3THR)

pthread_cond_destroy(3T) は、cv が指す条件変数を削除します。(Solaris スレッドについては、「cond_destroy(3THR)」を参照)。

プロトタイプ:
int	pthread_cond_destroy(pthread_cond_t *cv);

#include <pthread.h>

pthread_cond_t cv;
int ret;

/* 条件変数を削除する */
ret = pthread_cond_destroy(&cv); 

条件変数の記憶領域は解放されません。

戻り値

正常終了時は 0 です。それ以外の戻り値は、エラーが発生したことを示します。以下の条件が検出されると、この関数は失敗し、次の値を戻します。

EINVAL

cv で指定された値が無効です。

「呼び起こし忘れ」問題

pthread_cond_signal() または pthread_cond_broadcast() を呼び出すとき、スレッドが条件変数に関連する相互排他ロックを保持していないと「呼び起こし忘れ」(lost wake-up) という問題が生じることがあります。

次のすべての条件に該当する場合は、そのシグナルには効果がないので「呼び起こし忘れ」が発生します。

• あるスレッドが pthread_cond_signal() または pthread_cond_broadcast() を呼び出す

• 別のスレッドが条件の評価と pthread_cond_wait() 呼び出しの間にある

• 待ち状態のスレッドがない

「生産者 / 消費者」問題

「生産者 / 消費者」問題は、並行プログラミングに関する問題の中でも一般によく知られているものの 1 つです。この問題は次のように定式化されます。サイズが有限の 1 個のバッファと 2 種類のスレッドが存在します。一方のスレッドを生産者、もう一方のスレッドを消費者と呼びます。生産者がバッファにデータを入れ、消費者がバッファからデータを取り出します。

生産者は、バッファに空きができるまでデータを入れることができません。消費者は、バッファが空の間はデータを取り出すことができません。

特定の条件のシグナルを待つスレッドの待ち行列を条件変数で表すことにします。

例 4-11 では、そうした待ち行列として less と more の 2 つを使用しています。less はバッファ内の未使用スロットを待つ生産者のための待ち行列で、more は情報が格納されたバッファスロットを待つ消費者のための待ち行列です。また、バッファが同時に複数のスレッドによってアクセスされないようにするために、相互排他ロック (mutex ロック) も使用しています。

例 4-11 「生産者 / 消費者」問題と条件変数

typedef struct {
    char buf[BSIZE];
    int occupied;
    int nextin;
    int nextout;
    pthread_mutex_t mutex;
    pthread_cond_t more;
    pthread_cond_t less;
} buffer_t;

buffer_t buffer;

例 4-12 は、生産者側の処理です。最初に、mutex をロックしてバッファデータ構造 (buffer) を保護します。次に、これから作成するデータのための空きがあるか確認します。空きがない場合は、pthread_cond_wait() を呼び出して、「バッファ内に空きがある」を表す条件 less にシグナルが送られてくるのを待つスレッドの待ち行列に入ります。

同時に、pthread_cond_wait() の呼び出しによって、スレッドは mutex のロックを解除します。生産者スレッドは、条件が真になって消費者スレッドがシグナルを送ってくれるのを待ちます (詳細は、例 4-12 を参照してください)。条件にシグナルが送られてくると、less を待っている一番目のスレッドが呼び起こされます。しかし、そのスレッドは pthread_cond_wait() が戻る前に、mutex ロックを再び獲得する必要があります。

このようにして、バッファデータ構造への相互排他アクセスが保証されます。その後、生産者スレッドはバッファに本当に空きがあるか確認しなければなりません。空きがある場合は、最初の未使用スロットにデータを入れます。

このとき、バッファにデータが入れられるのを消費者スレッドが待っている可能性があります。そのスレッドは、条件変数 more で待ち状態となっています。生産者スレッドはバッファにデータを入れると、pthread_cond_signal() を呼び出して、待ち状態の最初の消費者を呼び起こします (待ち状態の消費者がいないときは、この呼び出しは無視されます)。

最後に、生産者スレッドは mutex ロックを解除して、他のスレッドがバッファデータ構造を操作できるようにします。

例 4-12 「生産者 / 消費者」問題 − 生産者

void producer(buffer_t *b, char item)
{
    pthread_mutex_lock(&b->mutex);
   
    while (b->occupied >= BSIZE)
        pthread_cond_wait(&b->less, &b->mutex);

    assert(b->occupied < BSIZE);

    b->buf[b->nextin++] = item;

    b->nextin %= BSIZE;
    b->occupied++;

    /* 現在の状態:  「b->occupied < BSIZE かつ b->nextin はバッファ内
       の次の空きスロットのインデックス」または「b->occupied == BSIZE 
       かつ b->nextin は次の (占有されている) スロットのインデックス。これは
       消費者によって空にされる (例: b->nextin == b->nextout) 」 */

    pthread_cond_signal(&b->more);

    pthread_mutex_unlock(&b->mutex);
}

上記のコード例の assert() 文の用法に注意してください。コンパイル時に NDEBUG を定義しなければ、assert() は次のように動作します。すなわち、引数が真 (0 以外の値) のときは何も行わず、引数が偽 (0) のときはプログラムを強制的に終了させます。このように、実行時に発生した問題をただちに指摘できる点がマルチスレッドプログラムに特に適しています。assert() はデバッグのための有用な情報も与えてくれます。

/* 現在の状態: ... で始まるコメント部分も assert() で表現した方がよいかもしれません。しかし、論理式で表現するには複雑すぎるので、ここでは文章で表現しています。

上記の assert() やコメント部分の論理式は、どちらも不変式の例です。不変式は、あるスレッドが不変式中の変数を変更している瞬間を除いて、プログラムの実行により偽の値に変更されない論理式です (もちろん assert() の論理式は、どのスレッドがいつ実行した場合でも常に真であるべきです)。

不変式は非常に重要な手法です。プログラムテキストとして明示的に表現しなくても、プログラムを分析するときは不変式に置き換えて問題を考えることが大切です。

上記の生産者コード内のコメントで表現された不変式は、スレッドがそのコメントを含むコード部分を処理中には常に真となります。しかし、それを mutex_unlock() のすぐ後ろに移動すると、必ずしも常に真とはなりません。assert() のすぐ後ろに移動した場合は、真となります。

つまり、この不変式は、生産者または消費者がバッファの状態を変更しようとしているとき以外は、常に真となるような特性を表現しています。スレッドは mutex の保護下でバッファを操作しているとき、この不変式の値を一時的に偽にしてもかまいません。しかし、処理が完了したら不変式の値を再び真に戻さなければなりません。

例 4-13 は、消費者の処理です。この処理の流れは生産者の場合と対称的です。

例 4-13 「生産者 / 消費者」問題 − 消費者

char consumer(buffer_t *b)
{
    char item;
    pthread_mutex_lock(&b->mutex);
    while(b->occupied <= 0)
        pthread_cond_wait(&b->more, &b->mutex);

    assert(b->occupied > 0);

    item = b->buf[b->nextout++];
    b->nextout %= BSIZE;
    b->occupied--;

    /* 現在の状態:  「b->occupied > 0 かつ b->nextout はバッファ内の
       最初の占有されているスロットのインデックス」または「b->occupied == 0 
       かつ b-> nextout は次の (未使用) スロットのインデックス。これは生産者側
       によっていっぱいにされる (例: b->nextout == b->nextin) 」 */

    pthread_cond_signal(&b->less);
    pthread_mutex_unlock(&b->mutex);

    return(item);
}