CON38-C. 1 つのスレッドではなく条件変数を待っているすべてのスレッドに通知する

条件変数を待っているスレッド(cnd_wait() または cnd_timedwait())は、単一の操作(cnd_signal())の結果、動作を開始できる。ただし、複数のスレッドが同じ条件変数を待っている場合、これらのスレッドのいずれかがスケジューラによって選択され起動する可能性がある(すべてのスレッドの優先レベルが同じ場合)。

述語によるテストが真の場合にのみ各スレッドが実行されることを保証するために、ユーザは待機条件に関して述語によるテストのループの作成を強制される(IEEE Std 1003.1 の 2001 年以降のリリースでのレコメンデーション [IEEE Std 1003.1-2004])。その結果、述語によるテストが偽であることを発見したスレッドは、再度待機状態に入り、最終的にはデッドロック状況が発生する。

cnd_signal() の使用は、次の条件が満たされる場合にのみ安全である。

すべてのスレッドは、起動後、同じ操作セットを実行する必要がある。これにより、cnd_signal() の単一起動に対応して、任意のスレッドを選択して起動および再開できる。
信号受信後、起動する必要があるスレッドは 1 つだけである。

cnd_signal() の使用は、各スレッドが固有の条件変数を使用する場合にも安全である。

cnd_broadcast() を使用すればこれらの問題を防止できる。なぜなら、この関数は、条件変数に関連付けられているすべてのスレッドを起動するからである。また、すべてのスレッドが述語条件を評価しなおす必要があるため、これらのスレッドのうち 1 つのスレッドがそのテストが真であることを発見する。これによりデッドロックが防止される。

違反コード (`cnd_signal()`)

以下のコード例は、作成時に各スレッドに割り当てられたステップレベルに従って次々に実行される(連続処理)、特定の数のスレッド(5)で構成される。current_step 変数は、現在のステップレベルを格納し、各スレッドがその処理を終了した時点でただちにインクリメントされる。最終的には、別のスレッドが信号を受け取り、次のステップを実行できる。

#include <stdio.h>
#include <stdlib.h>
#include <threads.h>

#define NTHREADS  5

mtx_t mutex;
cnd_t cond;


void *run_step(void *t) {
  static int current_step = 0;
  int my_step = (int)t;
  int result;

  if ((result = mtx_lock(&mutex)) != thrd_success) {
    /* エラー条件の処理 */
  }

  printf("Thread %d has the lock\n", my_step);

  while (current_step != my_step) {
    printf("Thread %d is sleeping...\n", my_step);

    if ((result = cnd_wait(&cond, &mutex)) != thrd_success) {
      /* エラー条件の処理 */
    }

    printf("Thread %d woke up\n", my_step);
  }

  /* 処理を行う... */
  printf("Thread %d is processing...\n", my_step);

  current_step++;

  /* 待機中のタスクに信号を送る */
  if ((result = cnd_signal(&cond)) != thrd_success) {
    /* エラー条件の処理 */
  }

  printf("Thread %d is exiting...\n", my_step);

  if ((result = mtx_unlock(&mutex)) != thrd_success) {
    /* エラー条件の処理 */
  }

  thrd_exit(NULL);
}


int main(int argc, char** argv) {
  int i;
  int result;
  thrd_t threads[NTHREADS];
  int step[NTHREADS];

  if ((result = mtx_init(&mutex, mtx_plain)) != thrd_success) {
    /* エラー条件の処理 */
  }
  if ((result = cnd_init(&cond)) != thrd_success) {
    /* エラー条件の処理 */
  }

  /* スレッドを作成する */
  for (i = 0; i < NTHREADS; i++) {
    step[i] = i;
    if ((result = thrd_create(&threads[i], run_step, (void *)step[i])) != thrd_success) {
      /* エラー条件の処理 */
    }
  }

  /* Wait for all threads to complete */
  for (i = NTHREADS-1; i >= 0; i--) {
    if ((result = thrd_join(threads[i], NULL)) != thrd_success) {
      /* エラー条件の処理 */
    }
  }

  if ((result = mtx_destroy(&mutex)) != thrd_success) {
    /* エラー条件の処理 */
  }
  if ((result = cnd_destroy(&cond)) != thrd_success) {
    /* エラー条件の処理 */
  }

  thrd_exit(NULL);
}

この例では、各スレッドが独自の述語を持つ。これは、次に進む前に current_step が異なる値を持つことを各スレッドが要求するからである。信号操作(pthread_cond_signal())が行われると、待機中のスレッドのいずれかが起動できる。たまたま次のステップ値を持ったスレッドでない場合は、そのスレッドは再度待ち状態に入り(pthread_cond_wait())、それ以降に信号操作が行われないため、デッドロック状態が発生する。

たとえば、次のような例があるとする。

時間	スレッド番号 (`my_step`)	`current_step`	動作
0	3	0	スレッド 3 が初めて実行される。述語は、`FALSE -> wait()`
1	2	0	スレッド 2 が初めて実行される。述語は、`FALSE -> wait()`
2	4	0	スレッド 4 が初めて実行される。述語は、`FALSE -> wait()`
3	0	0	スレッド 0 が初めて実行される。述語は、`TRUE -> current_step++; signal()`
4	1	1	スレッド 1 が初めて実行される。述語は、`TRUE -> current_step++; signal()`
5	3	2	スレッド 3 が起動する(スケジューラの選択)。述語は、`FALSE -> wait()`
6	—	—	デッドロック状況これ以降スレッドが実行されなくなり、残りのスレッドを起動するには信号が必要。

このコード例は、liveness 特性に違反している。

適合コード (`cnd_broadcast()` の使用)

この適合コードは、cnd_broadcast() 方式を使用して、無作為の単一スレッドではなく待機中のすべてのスレッドに信号を送る。違反コード例の run_step() スレッドコードだけが、次のように修正されている。

void *run_step(void *t) {
  static int current_step = 0;
  int my_step = (int)t;
  int result;

  if ((result = mtx_lock(&mutex)) != thrd_success) {
    /* エラー条件の処理 */
  }

  printf("Thread %d has the lock\n", my_step);

  while (current_step != my_step) {
    printf("Thread %d is sleeping...\n", my_step);

    if ((result = cnd_wait(&cond, &mutex)) != thrd_success) {
      /* エラー条件の処理 */
    }

    printf("Thread %d woke up\n", my_step);
  }

  /* 処理を行う... */
  printf("Thread %d is processing...\n", my_step);

  current_step++;

  /* 待機中のすべてのタスクに信号を送る */
  if ((result = cnd_broadcast(&cond)) != thrd_success) {
    /* エラー条件の処理 */
  }

  printf("Thread %d is exiting...\n", my_step);

  if ((result = mtx_unlock(&mutex)) != 0) {
    /* エラー条件の処理 */
  }

  thrd_exit(NULL);
}

すべてのスレッドが起動されることで問題が解決される。これは、それぞれのスレッドが最終的にその述語によるテストを実行して、そのうちの 1 つがそのテスト結果が真であることを発見し、終了まで実行が続行されるからである。

適合コード (`cnd_signal()` を使用するが、スレッドごとに固有の条件変数を指定する)

信号の問題を解決するもう 1 つの方法が、スレッドごとに固有の条件変数を使用する方法である(関連付けられた単一ミューテックスを維持)。この場合、信号操作(cnd_signal())により、その信号を待っているスレッドだけが起動する。

注: 信号を受け取るスレッドの述語は、真でなければならない。そうでなければ、デッドロックが発生する可能性がある。

#include <stdio.h>
#include <stdlib.h>
#include <threads.h>

#define NTHREADS  5
mtx_t mutex;
cnd_t cond[NTHREADS];


void *run_step(void *t) {
  static int current_step = 0;
  int my_step = (int)t;
  int result;

  if ((result = mtx_lock(&mutex)) != thrd_success) {
    /* エラー条件の処理 */
  }

  printf("Thread %d has the lock\n", my_step);

  while (current_step != my_step) {
    printf("Thread %d is sleeping...\n", my_step);

    if ((result = cnd_wait(&cond[my_step], &mutex)) != thrd_success) {
      /* エラー条件の処理 */
    }

    printf("Thread %d woke up\n", my_step);
  }

  /* 処理を行う... */
  printf("Thread %d is processing...\n", my_step);

  current_step++;

  /* 次のステップのスレッドに信号を送る */
  if ((my_step + 1) < NTHREADS) {
    if ((result = cnd_signal(&cond[my_step+1])) != thrd_success) {
      /* エラー条件の処理 */
    }
  }

  printf("Thread %d is exiting...\n", my_step);

  if ((result = mtx_unlock(&mutex)) != thrd_success) {
    /* エラー条件の処理 */
  }

  thrd_exit(NULL);
}


int main(int argc, char** argv) {
  int i;
  int result;
  thrd_t threads[NTHREADS];
  int step[NTHREADS];

  if ((result = mtx_init(&mutex, mtx_plain)) != thrd_success) {
    /* エラー条件の処理 */
  }

  for (i = 0; i< NTHREADS; i++) {
    if ((result = cnd_init(&cond[i])) != thrd_success) {
      /* エラー条件の処理 */
    }
  }

  /* スレッドを作成する */
  for (i = 0; i < NTHREADS; i++) {
    step[i] = i;
    if ((result = thrd_create(&threads[i], run_step, (void *)step[i])) != thrd_success) {
      /* エラー条件の処理 */
    }
  }

  /* すべてのスレッドが完了するのを待つ */
  for (i = NTHREADS-1; i >= 0; i--) {
    if ((result = thrd_join(threads[i], NULL)) != thrd_success) {
      /* エラー条件の処理 */
    }
  }

  if ((result = mtx_destroy(&mutex)) != thrd_success) {
    /* エラー条件の処理 */
  }

  for (i = 0; i < NTHREADS; i++) {
    if ((result = cnd_destroy(&cond[i])) != thrd_success) {
      /* エラー条件の処理 */
    }
  }

  thrd_exit(NULL);
}

この適合コードでは、各スレッドが固有の条件変数と結び付いており、この変数はそのスレッドを起動する必要があるときに信号を受け取る。この適合コードは、起動したいスレッドだけが起動されるため、より効果的であることが判明している。