FreeRTOS 12 · セマフォとミューテックス — 中身はキュー、違いは「所有者」

Chapter 12

セマフォとミューテックス — 中身はキュー、違いは「所有者」

この章がなぜ必要なのか——この 2 つの混同が、バグの温床になっている.

「バイナリセマフォとミューテックスは同じようなもの」と説明されることがある。 これは誤りである。 使い分けを間違えると、 優先度逆転(13 章)が起き、システムが止まる。

FreeRTOS では実装が共通なので、なおさら混同しやすい。 何が同じで、何が決定的に違うのかをはっきりさせる。

この章で使う既出の用語(定義は各リンク先). TCB(02 章 2 節)、タスク(02 章 2 節)、Blocked(04 章 1 節)、Ready(04 章 1 節)、FreeRTOS(06 章 4 節)、追加(07 章 2 節)

1. 実装はすべてキュー

/* semphr.h より(マクロで queue.c の関数に落ちる) */
#define xSemaphoreCreateBinary()                                  \
    xQueueGenericCreate( 1, semSEMAPHORE_QUEUE_ITEM_LENGTH, queueQUEUE_TYPE_BINARY_SEMAPHORE )

#define xSemaphoreCreateCounting( uxMaxCount, uxInitialCount )    \
    xQueueCreateCountingSemaphore( ( uxMaxCount ), ( uxInitialCount ) )

#define xSemaphoreCreateMutex()                                   \
    xQueueCreateMutex( queueQUEUE_TYPE_MUTEX )

#define xSemaphoreTake( xSemaphore, xBlockTime )                  \
    xQueueSemaphoreTake( ( QueueHandle_t ) ( xSemaphore ), ( xBlockTime ) )

#define xSemaphoreGive( xSemaphore )                              \
    xQueueGenericSend( ( QueueHandle_t ) ( xSemaphore ), NULL, semGIVE_BLOCK_TIME, queueSEND_TO_BACK )

semSEMAPHORE_QUEUE_ITEM_LENGTH は 0 である。

#define semSEMAPHORE_QUEUE_ITEM_LENGTH  ( ( uint8_t ) 0U )

つまりセマフォとは——

「アイテムサイズ 0 のキュー」=「個数だけを数えるキュー」

である。データを持たないので、バッファも不要である (prvCopyDataToQueue は uxItemSize == 0 のとき何もコピーしない)。

オブジェクトキュー長アイテムサイズ追加の情報
キューnm バイト—
バイナリセマフォ10—
カウンティングセマフォn0—
ミューテックス10所有者 xMutexHolder
再帰ミューテックス10所有者 + uxRecursiveCallCount

最後の行だけが本質的な違いである。

2. バイナリセマフォ — 「事象が起きた」を伝える

SemaphoreHandle_t xSem = xSemaphoreCreateBinary();

/* 待つ側 */
for (;;) {
    xSemaphoreTake( xSem, portMAX_DELAY );    /* 事象が起きるまで寝る */
    handle_event();
}

/* 起こす側(ISR から) */
void UART_IRQHandler( void )
{
    BaseType_t woken = pdFALSE;
    xSemaphoreGiveFromISR( xSem, &woken );
    portYIELD_FROM_ISR( woken );
}

これが最も典型的な使い方である。「割り込みが起きたことをタスクに知らせる」。

「取る前に与える」ことができる

xSemaphoreGive( xSem );          /* 誰も待っていなくてもよい */
...
xSemaphoreTake( xSem, 0 );       /* 後から取れる(1 個だけ覚えている) */

キュー長が 1 なので、1 個だけ記憶できる。2 回 Give しても 1 回分しか残らない。

「取りこぼし」に注意.

ISR: Give (カウント 0→1)
ISR: Give (カウント 1→1、増えない)     ← 2 回目のイベントが消える
Task: Take (カウント 1→0)               ← 1 回しか処理されない

割り込みが速い場合、これが起きる。対策は 2 つある。

  1. カウンティングセマフォを使う(次節)
  2. タスク通知を使う(14 章。カウント方式にできる)

あるいは、そもそも「何回起きたか」を数えるより、 「起きたら、溜まっている分を全部処理する」設計にする方がよいことも多い。

for (;;) {
    xSemaphoreTake( xSem, portMAX_DELAY );
    while( uart_has_data() ) { process( uart_read() ); }   /* 溜まった分を全部 */
}

生成直後は「空」

SemaphoreHandle_t xSem = xSemaphoreCreateBinary();
xSemaphoreTake( xSem, 0 );      /* pdFALSE が返る(まだ Give されていない) */

古い API(vSemaphoreCreateBinary、廃止済み)は生成直後に「満」だった。 移植のときに混乱の元になるので覚えておくこと。

3. カウンティングセマフォ — 資源の個数を数える

/* 最大 5 個、初期値 5 個 */
SemaphoreHandle_t xSem = xSemaphoreCreateCounting( 5, 5 );

xSemaphoreTake( xSem, portMAX_DELAY );    /* 1 個確保(5→4) */
...
xSemaphoreGive( xSem );                    /* 1 個返す(4→5) */

2 つの使い方がある。

使い方初期値意味
資源のプール最大値「使える資源が n 個ある」
イベントのカウント0「イベントが n 回起きた」
/* 資源のプール: DMA チャネルが 3 本 */
SemaphoreHandle_t xDmaSem = xSemaphoreCreateCounting( 3, 3 );

/* イベントのカウント: 割り込み回数を取りこぼさない */
SemaphoreHandle_t xEventSem = xSemaphoreCreateCounting( 10, 0 );

カウンティングセマフォは「取りこぼさない」が「順序は分からない」.

「何回起きたか」は分かるが、「どの順で、何が起きたか」は分からない。 それが要るならキューを使うこと。

4. ミューテックス — 「所有者がいる」ロック

SemaphoreHandle_t xMutex = xSemaphoreCreateMutex();

xSemaphoreTake( xMutex, portMAX_DELAY );    /* ロック */
    /* クリティカルセクション: 共有資源を触る */
xSemaphoreGive( xMutex );                    /* アンロック */

見た目はバイナリセマフォと同じだが、内部に決定的な違いが 2 つある。

違い 1: 所有者を記録する

/* xQueueSemaphoreTake の中(ミューテックスのとき) */
if( pxQueue->uxQueueType == queueQUEUE_IS_MUTEX ) {
    pxQueue->u.xSemaphore.xMutexHolder = pvTaskIncrementMutexHeldCount();
}

「いま誰がこのミューテックスを持っているか」を TCB へのポインタで記録する。

これによって 2 つのことが可能になる。

  1. 優先度継承(13 章)——所有者が誰か分かるから、その優先度を上げられる
  2. 所有者チェック——他人が Give することを防げる
/* xQueueGenericSend の中 */
configASSERT( pxQueue->u.xSemaphore.xMutexHolder == xTaskGetCurrentTaskHandle() );

バイナリセマフォには所有者の概念がない.

だから「タスク A が Take して、タスク B が Give する」ことができる。 これが正しい使い方になる場面もある(次節の使い分けを見よ)。

だがロックとして使うなら、これは危険である。 他人が勝手にアンロックできてしまう。

違い 2: 生成直後は「解放済み」

SemaphoreHandle_t xMutex = xSemaphoreCreateMutex();
xSemaphoreTake( xMutex, 0 );      /* pdTRUE が返る。すぐ取れる */

内部的には、生成時に xQueueGenericSend( NULL ) を 1 回呼んで「満」にしている。

/* prvInitialiseMutex の中 */
( void ) xQueueGenericSend( pxNewQueue, NULL, ( TickType_t ) 0U, queueSEND_TO_BACK );

バイナリセマフォは「空」、ミューテックスは「満」で始まる。 これは意味を考えれば自然である。ロックは最初は誰も持っていない。

5. 使い分け — これを間違えるとシステムが止まる

目的使うもの理由
共有資源の保護(ロック)ミューテックス優先度継承が要る
ISR からタスクへの通知バイナリセマフォ/タスク通知所有者が異なる。ミューテックスでは不可
タスク間の同期バイナリセマフォ同上
資源プールの管理カウンティングセマフォ個数を数える
イベント回数の記録カウンティングセマフォ/タスク通知取りこぼさない

ISR からミューテックスを Give してはいけない.

void ISR( void ) {
    xSemaphoreGiveFromISR( xMutex, &woken );    /* ← 間違い */
}

そもそも xSemaphoreGiveFromISR() はミューテックスに対して使えない (configASSERT で捕まる)。理由は 2 つある。

  1. ISR は「タスク」ではないので、所有者になれない
  2. 優先度継承の解除処理が、ISR コンテキストでは行えない

逆に言えば、ミューテックスはタスク間でしか使えない。 ISR と同期したいならバイナリセマフォかタスク通知を使う。

なぜロックにバイナリセマフォを使ってはいけないのか

これが 13 章の主題である。先に結論だけ述べる。

低優先度タスク L がバイナリセマフォを Take(資源を占有)
    ↓
高優先度タスク H が同じセマフォを Take しようとして Blocked
    ↓
中優先度タスク M が Ready になる → L より優先度が高いので L を止める
    ↓
L は走れない → セマフォを Give できない → H も走れない
    ↓
【結果】M が終わるまで H が待たされる。優先度の意味が逆転している

ミューテックスなら、L の優先度を一時的に H まで引き上げる(優先度継承)ので、 M に邪魔されず、L はすぐに Give できる。

この 1 点が、ミューテックスとバイナリセマフォの決定的な違いである。

6. 再帰ミューテックス

#define configUSE_RECURSIVE_MUTEXES  1

SemaphoreHandle_t xMutex = xSemaphoreCreateRecursiveMutex();

xSemaphoreTakeRecursive( xMutex, portMAX_DELAY );    /* 1 回目: count=1 */
    xSemaphoreTakeRecursive( xMutex, portMAX_DELAY ); /* 2 回目: count=2。デッドロックしない */
    xSemaphoreGiveRecursive( xMutex );                /* count=1 */
xSemaphoreGiveRecursive( xMutex );                    /* count=0。ここで実際に解放 */

同じタスクが何度も Take できる。 通常のミューテックスなら 2 回目でデッドロックする。

/* xQueueTakeMutexRecursive の中 */
if( pxMutex->u.xSemaphore.xMutexHolder == xTaskGetCurrentTaskHandle() ) {
    ( pxMutex->u.xSemaphore.uxRecursiveCallCount )++;
    xReturn = pdPASS;
}
else {
    xReturn = xQueueSemaphoreTake( pxMutex, xTicksToWait );
    if( xReturn != pdFAIL ) { ( pxMutex->u.xSemaphore.uxRecursiveCallCount )++; }
}

再帰ミューテックスが必要になったら、設計を疑うこと.

「同じロックを再帰的に取る」状況は、たいてい 「ロックを取る関数が、ロックを取る別の関数を呼んでいる」ことを意味する。

これは階層が混乱している兆候である。理想的には——

という規約にすれば、再帰ミューテックスは不要になる。

/* 公開 API: ロックを取る */
void buffer_push( int v ) {
    xSemaphoreTake( xMutex, portMAX_DELAY );
    buffer_push_locked( v );
    xSemaphoreGive( xMutex );
}

/* 内部関数: ロックを取らない。呼ぶ側が持っていること */
static void buffer_push_locked( int v ) { ... }

ただし既存コードを RTOS 化するときなど、現実的な選択肢として有用ではある。 Linux カーネルは再帰ロックを提供していない(意図的に禁止している)が、 Java の synchronized や C++ の std::recursive_mutex は提供している。

7. セマフォとミューテックスの実装比較

セマフォ/ミューテックスの動作比較

Take/Give の順序を変えて、カウント・所有者・待ちタスクの状態を確かめられる。

8. 静的確保

StaticSemaphore_t xMutexBuffer;
SemaphoreHandle_t xMutex = xSemaphoreCreateMutexStatic( &xMutexBuffer );

StaticQueue_t xQueueBuffer;
uint8_t ucQueueStorage[ 10 * sizeof( int ) ];
QueueHandle_t q = xQueueCreateStatic( 10, sizeof( int ), ucQueueStorage, &xQueueBuffer );
#define configSUPPORT_STATIC_ALLOCATION  1

静的確保を強く推奨する.

configSUPPORT_DYNAMIC_ALLOCATION を 0 にすれば、 動的確保 API がコンパイルエラーになるので、うっかり混ざることも防げる。

9. この章のまとめ

ポイント内容
実装すべてアイテムサイズ 0 のキュー。個数を数えるだけ
バイナリセマフォ長さ 1。生成直後は空。所有者なし
カウンティング長さ n。資源プール(初期値=最大)/イベント計数(初期値=0)
ミューテックス長さ 1。生成直後は満。所有者を記録
決定的な違い所有者の有無。これが優先度継承を可能にする
ロック用途必ずミューテックス。バイナリセマフォは優先度逆転を起こす
ISR との同期バイナリセマフォ/タスク通知。ミューテックスは使えない
再帰ミューテックス便利だが、設計の混乱の兆候でもある
静的確保推奨。ヒープ不足の可能性が消える

次章では、ミューテックスの存在理由—— 優先度逆転という現象と、それがどう解決されるかを見る。