Zenn
🐍

SpresenseでNuttXのpthreadsを使用しマルチスレッドを学ぶ #6　〜バリアで同期して各スレッドの処理の足並みを揃える〜

2022/03/28に公開
spresense
pthreads
バリア同期
tech

これは何?

前回に引き続きSpresenseでpthreadsインターフェースを使って動作確認します。

今回のテーマ

今回は複数のスレッドが特定の地点に到達したら処理を行うバリア同期を確認します。
今回使うpthreadsの関数は次のとおりです。

  • pthread_barrier_init:バリア変数の初期化
  • pthread_barrier_wait:バリア同期
  • pthread_barrier_destroy:バリア変数の破棄

確認環境

シリーズ #1と同じです。
※今回はAPS学習ボードのスイッチ、LEDを使わないためSpresenseメインボードのみで確認可能です。

テストコード説明

今回のテストコードはGitHub[1]におきました。
ディレクトリ・実行ファイル名称はpthread_barrier_waitです。

3つのスレッドでsleep時間を変えています。スレッド0は3秒、スレッド1は6秒、スレッド2は9秒のsleep時間です。
スレッドの中ではsleepが終了した後にバリアしています。バリア変数はカウント3で初期化しているため、スレッド0・スレッド1はスレッド2の9秒のsleepが終わるまで待機します。

複数スレッドの初期化時間が異なるが初期化以降の処理開始タイミングを複数スレッド間で揃えたい、という状況を想定しテストコードを書きました。

バリア変数の定義

条件変数を定義します。

pthread_barrier_wait_main.c // 該当部分を抽出し記載
// 初期化完了同期用バリア変数
pthread_barrier_t initialization_completion_barrier;

バリア変数の初期化

条件変数を利用するにはpthread_barrier_initで初期化します。

pthread_barrier_wait_main.c main関数 // 該当部分を抽出し記載
#define THREAD_NUM   (3)  // スレッドの数

int main(int argc, FAR char *argv[])
{

  if (pthread_barrier_init(&initialization_completion_barrier, NULL, THREAD_NUM) != 0) {
      printf("Error; pthread_barrier_init.\n");
      exit(1);
  }
  
}

バリア同期

バリア同期はスレッド：thread_funcのpthread_barrier_waitで実現します。

pthread_barrier_wait_main.c thread_func関数 // 該当部分を抽出し記載
void* thread_func(void* arg) {
  int thread_index = (uint32_t)arg;
  int ret;
#if 1
  unsigned int initialize_time = 3 * (thread_index + 1);
#else
  unsigned int initialize_time;

  if (thread_index == 0) {
    initialize_time = 9;    
  } else if (thread_index == 1) {
    initialize_time = 6;    
  } else {
    initialize_time = 3;
  }
#endif

  printf("thread index[%d] start.initialize_time = %d(sec).\n", thread_index, initialize_time);

  sleep(initialize_time);

  printf("thread index[%d] initialize end.\n", thread_index);

  ret = pthread_barrier_wait(&initialization_completion_barrier);
  if (!(ret == 0 || ret == PTHREAD_BARRIER_SERIAL_THREAD)) {
    printf("Error; thread index[%d] pthread_barrier_wait ret = %d.\n", thread_index, ret);
    exit(1);
  }

  printf("thread index[%d] barrier exit.ret = %d.\n", thread_index, ret);

  return (void*)thread_index;
}

バリア変数破棄

バリア変数が不要になったらpthread_barrier_destroyで削除します。

pthread_barrier_wait_main.c main関数 // 該当部分を抽出し記載

int main(int argc, FAR char *argv[])
{

  if (pthread_barrier_destroy(&initialization_completion_barrier) != 0) {
      printf("Error; pthread_barrier_destroy.\n");
      exit(1);
  }

  printf("Bye.\n");

  return 0;
}

動作確認結果

printf出力の文字情報のみだと動作がよくわからないので動画を撮りました。
https://youtu.be/yDxJMUViTc8

プログラムを実行すると各スレッドはsleep時間を表示した後に指定時間sleepします。
sleep後はsleep終了のメッセージを出力しバリア同期の地点に到達します。3スレッドがこのバリアに到達するまで待ち状態になります。
3スレッドのsleepが終了し、バリアに到達するとpthread_barrier_waitから制御が戻り、pthread_barrier_waitの戻り値を出力し(★印以降)スレッドを終了します。
次の実行結果はスレッドのsleep時間が次の場合です。

  • スレッド0(thread index[0]):sleep時間 = 3秒
  • スレッド1(thread index[1]):sleep時間 = 6秒
  • スレッド2(thread index[2]):sleep時間 = 9秒
nsh> pthread_barrier_wait
thread index[0] start.initialize_time = 3(sec).
thread index[1] start.initialize_time = 6(sec).
thread index[2] start.initialize_time = 9(sec). 
thread index[0] initialize end.
thread index[1] initialize end.
thread index[2] initialize end.
thread index[2] barrier exit.ret = 4096.　// ★
thread index[0] barrier exit.ret = 0.
thread index[1] barrier exit.ret = 0.
main:thread index[0] exit.return value = 0
main:thread index[1] exit.return value = 1
main:thread index[2] exit.return value = 2
Bye.

pthread_barrier_waitの戻り値は次のとおりです。

  • スレッド0・1: 0
  • スレッド2: 4096 == PTHREAD_BARRIER_SERIAL_THREAD(0x1000)

スレッド2の戻り値だけは4096でした。ソースコードを読むとバリアに到達した最終スレッドの戻り値は4096 == PTHREAD_BARRIER_SERIAL_THREAD(0x1000)になるようです。

スレッド0・３のsleep時間を逆にして実行結果を確認します。

  • スレッド0(thread index[0]):sleep時間 = 3秒→９秒
  • スレッド1(thread index[1]):sleep時間 = 6秒
  • スレッド2(thread index[2]):sleep時間 = 9秒→3秒
nsh> pthread_barrier_wait
thread index[0] start.initialize_time = 9(sec).
thread index[1] start.initialize_time = 6(sec).
thread index[2] start.initialize_time = 3(sec).
thread index[2] initialize end.
thread index[1] initialize end.
thread index[0] initialize end.
thread index[0] barrier exit.ret = 4096.
thread index[2] barrier exit.ret = 0.
thread index[1] barrier exit.ret = 0.
main:thread index[0] exit.return value = 0
main:thread index[1] exit.return value = 1
main:thread index[2] exit.return value = 2
Bye.

バリアに到達した最終スレッド0の戻り値が4096になりました。

動作確認方法

シリーズ #1と同じです。
今回のアプリケーション名はpthread_barrier_waitです。

今回の感想

バリアで複数スレッドの処理を同期することを確認できました。
バリアは【複数スレッドの処理時間が異なるが複数スレッドの処理は同じタイミングで開始したい】という場合に有効と思いました。

今回も最後まで読んでいただきありがとうございました。

脚注

  1. GitHubリンク ↩︎

Discussion