マルチスレッドのプログラミング

セマフォ

セマフォは、E.W. ダイクストラ (Dijkstra) が 1960 年代の終わりごろに考案したプログラミング手法です。ダイクストラのセマフォモデルは、鉄道線路の運行をモデル化したものです。一度に一本の列車しか走れない単線の鉄道線路を思い浮かべてください。

この鉄道線路を保護するのがセマフォです。列車は単線区間に入るとき、セマフォの状態が進行許可状態になるのを待たなければなりません。列車が単線区間に入るとセマフォの状態は、他の列車が単線区間に入るのを禁止する状態に変化します。単線区間から出る列車は、セマフォの状態を進行許可状態に戻して他の列車が単線区間に入ることができるようにしなければなりません。

コンピュータ内のセマフォは、単一の整数で表現されます。スレッドは進行が許可されるのを待ち、その後進行したことを知らせるためにセマフォに対して P 操作を実行します。

この操作をもう少し具体的に説明しましょう。スレッドは、セマフォの値が正になるのを待たなければなりません。処理を完了したセマフォは、V 操作を実行します。この操作は 1 を加えることでセマフォの値を変更します。ここで必ず守らなければならないことがあります。これらの各操作を不可分操作により行うことです。P 操作では、1 を引く直前のセマフォの値が正でなければなりません (結果的に、引いた後の値が負にならないことと、その値が引く前の値よりも 1 だけ小さいことが保証されます)。

P 操作と V 操作のどちらの演算操作でも干渉が生じないようにしなければなりません。たとえば、同じセマフォに対して 2 つの V 操作が同時に行われた場合、そのセマフォの新しい値は最初よりも 2 だけ大きくなっていなければなりません。

ダイクストラがオランダ人だったこともあり、P と V の記号的な意味は現在ではほとんど忘れられています。参考までに、P はオランダ語の「prolagen」という単語を表します。その語源は「proberen te verlagen」で、「小さくする」という意味です。また、V は「verhogen」を表し、「大きくする」という意味です。このことは、ダイクストラのテクニカルノート『EWD 74』で説明されています。

sem_wait(3RT) と sem_post(3RT) は、ダイクストラの P 操作と V 操作にそれぞれ対応しています。また、sem_trywait(3RT) は、P 操作の条件付きの形式です。この関数は、呼び出しスレッドがセマフォの値を差し引くために待たなければならない場合は、ただちに 0 以外の値を返します。

セマフォは、大きく 2 つに分類できます。1 つは 2 値型セマフォで、0 および 1 以外の値はとりません。もう 1 つは計数型セマフォで、0 以上の任意の値をとります。2 値型セマフォは、論理的に mutex と同じです。

必須要件ではありませんが、mutex のロックは、ロックを保持しているスレッドがそのロックを解放するべきです。ただし、セマフォには「スレッドがセマフォを保持している」という概念がないため、任意のスレッドが V 操作 (すなわち sem_post(3RT)) を実行できます。

計数型セマフォは、mutex とともに使用される条件変数と同等の能力があります。多くの場合、条件変数よりも計数型セマフォを使用した方がコードが簡素化されます (後述の例を参照してください)。

しかし、mutex と条件変数をいっしょに使用すれば、自然と 1 つのまとまりとなり、プログラム内で保護されている場所が明確になります。セマフォは強力ですが、構造化されないあいまいな方法で使用してしまいがちです。「並行プログラミングにおける go to」と呼ばれることもあります。

計数型セマフォ

セマフォの概念は、0 以上の整数カウントです。通常は、リソースに対するアクセスの調整をはかる目的で、次のように使用されます。最初に、使用可能なリソースの数をセマフォに初期設定します。スレッドは、リソースが追加されると不可分操作的にカウントを 1 加算し、リソースが削除されると不可分操作的に 1 減算します。

この場合、セマフォの値を 1 減らそうとすると、スレッドはセマフォの値が 0 より大きくなるまでブロックされます。

表 4–7 セマフォに関するルーチン


操作	参照先
セマフォの初期化	sem_init(3RT)
セマフォの加算	sem_post(3RT)
セマフォの値によるブロック	sem_wait(3RT)
セマフォの減算	sem_trywait(3RT)
セマフォの削除	sem_destroy(3RT)

セマフォは、その獲得と解放を同じスレッドで行う必要がないため、シグナルハンドラで行われているような非同期のイベント通知を実現できます。また、セマフォ自身が状態を持っているため、条件変数を使用する場合と違って相互排他ロックを獲得しなくても非同期で使用できます。ただし、セマフォは相互排他ロックほど効率的ではありません。

デフォルトでは、複数のスレッドがセマフォを待機している場合、ブロックを解除する順序はあらかじめ定義されていません。

セマフォは、使用する前に初期化されている必要がありますが、属性はありません。

セマフォの初期化

sem_init(3RT)

プロトタイプ:
int	sem_init(sem_t *sem, int pshared, unsigned int value);

#include <semaphore.h>

sem_t sem;
int pshared;
int ret;
int value;

/* セマフォの初期化 */
pshared = 0;
value = 1;
ret = sem_init(&sem, pshared, value);

sema_init(3THR) は、sem が指すセマフォ変数を value の値に初期設定します。pshared の値が 0 なら、そのセマフォはプロセス間で共有できません。pshared の値が 0 以外なら、そのセマフォはプロセス間で共有できます。(Solaris スレッドについては、sema_init(3THR)を参照)。

複数のスレッドから同じセマフォを初期化してはいけません。

一度初期化したセマフォは他のスレッドが使用している可能性があるので、再初期化してはいけません。

戻り値

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

EINVAL: value の値が SEM_VALUE_MAX を超えています。
ENOSPC: そのセマフォを初期化するのに必要なリソースが使い果たされています。セマフォの制限 SEM_NSEMS_MAX に達しています。
EPERM: そのセマフォを初期化するのに必要な特権をそのプロセスがもっていません。

プロセス間スコープでセマフォを初期化する

pshared の値が 0 の場合は、そのプロセス内のスレッドだけがそのセマフォを使用できます。

#include <semaphore.h>

sem_t sem;
int ret;
int count = 4;

/* このプロセスでのみ使用 */
ret = sem_init(&sem, 0, count);

プロセス間スコープでセマフォを初期化する

pshared の値が 0 以外の場合は、他のプロセスによってそのセマフォは共有されます。

#include <semaphore.h>

sem_t sem;
int ret;
int count = 4;

/* プロセス間で共有 */
ret = sem_init(&sem, 1, count);

名前付きセマフォ

sem_open(3RT)、sem_getvalue(3RT)、sem_close(3RT)、sem_unlink(3RT) の各関数が、名前付きセマフォを開く、取得する、閉じる、削除するのにそれぞれ使用できます。sem_open() では、ファイルシステムの名前空間で名前が定義されたセマフォを生成できます。

名前付きセマフォはプロセス間で共有されるセマフォに似ていますが、pshared 値ではなくパス名で参照される点が異なります。

名前付きセマフォの詳細は、sem_open(3RT)、sem_getvalue(3RT)、sem_close(3RT)、sem_unlink(3RT) のマニュアルページを参照してください。

セマフォの加算

sem_post(3RT)

プロトタイプ:
int	sem_post(sem_t *sem);

#include <semaphore.h>

sem_t sem;
int ret;

ret = sem_post(&sem); /* セマフォを加算する */

sema_post(3THR) は、sem が指すセマフォの値を不可分操作によって 1 増やします。そのセマフォでブロックされているスレッドがある場合は、そのスレッドのうちの 1 つのスレッドがブロック解除されます (Solaris スレッドについては、sema_post(3THR)を参照)。

戻り値

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

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

セマフォの値によるブロック

sem_wait(3RT)

プロトタイプ:
int	sem_wait(sem_t *sem);

#include <semaphore.h>

sem_t sem;
int ret;

ret = sem_wait(&sem); /* セマフォの値の変化を待つ */

sema_wait(3THR) は、sem が指すセマフォの値が 0 より大きくなるまで呼び出しスレッドをブロックし、0 より大きくなったらセマフォの値を不可分操作によって 1 減らします。

戻り値

EINVAL: sem が指すアドレスが正しくありません。
EINTR: この関数にシグナルが割り込みを行いました。

セマフォの減算

sem_trywait(3RT)

プロトタイプ:
int	sem_trywait(sem_t *sem);

#include <semaphore.h>

sem_t sem;
int ret;

ret = sem_trywait(&sem); /* セマフォの値の変化を待つ */

sem_trywait(3RT) は、sem が指すセマフォの値が 0 より大きい場合は不可分操作によって 1 減らします。この関数はブロックしない点を除いて、sem_wait() と同じ働きをします。つまり、失敗した場合にはすぐに戻ります。

戻り値

EINVAL: sem が指すアドレスが正しくありません。
EINTR: この関数にシグナルが割り込みを行いました。
EAGAIN: そのセマフォはすでにロックされているので、sem_trywait() でただちにロックできません。

セマフォの削除

sem_destroy(3RT)

プロトタイプ:
int	sem_destroy(sem_t *sem);

#include <semaphore.h>

sem_t sem;
int ret;

ret = sem_destroy(&sem); /* セマフォを削除する */

sem_destroy(3RT) は、sem が指すセマフォを削除します。セマフォの記憶領域は解放されません (Solaris スレッドについては、sem_destroy(3THR)を参照)。

戻り値

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

「生産者 / 消費者」問題 — セマフォを使った例

例 4–14 のデータ構造は、条件変数による「生産者 / 消費者」問題のコード例 (例 4–11参照) のデータ構造と似ています。 2 つのセマフォでそれぞれ、いっぱいになったバッファ数と未使用バッファ数を表します。これらのセマフォは、未使用バッファができるまで生産者を待たせ、バッファがいっぱいになるまで消費者を待たせます。

例 4–14 「生産者 / 消費者」問題 — セマフォを使った例

typedef struct {
    char buf[BSIZE];
    sem_t occupied;
    sem_t empty;
    int nextin;
    int nextout;
    sem_t pmut;
    sem_t cmut;
} buffer_t;

buffer_t buffer;

sem_init(&buffer.occupied, 0, 0);
sem_init(&buffer.empty,0, BSIZE);
sem_init(&buffer.pmut, 0, 1);
sem_init(&buffer.cmut, 0, 1);
buffer.nextin = buffer.nextout = 0;

ここでは、もう一組の (バイナリ) セマフォを使用しています。これは 2 値型セマフォで、相互排他ロック (mutex ロック) と同じ働きをします。本来このような場合では mutex を使用すべきですが、セマフォの使用例を示すために特に使用しています。

例 4–15 「生産者 / 消費者」問題 — 生産者

void producer(buffer_t *b, char item) {
    sem_wait(&b->empty);
    sem_wait(&b->pmut);

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

    sem_post(&b->pmut);
    sem_post(&b->occupied);
}

例 4–16 「生産者 / 消費者」問題 — 消費者

char consumer(buffer_t *b) {
    char item;

    sem_wait(&b->occupied);
   
    sem_wait(&b->cmut);

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

    sem_post(&b->cmut);

    sem_post(&b->empty);

    return(item);
}