FreeRTOS 19 · 排他制御の設計 — デッドロック・レース・再入

Chapter 19

排他制御の設計 — デッドロック・レース・再入

この章がなぜ必要なのか——RTOS を導入すると、新しい種類のバグが生まれる.

01 章で「OS は無料ではない」と書いた。その最大のコストがこれである。

スーパーループには存在しなかったバグ—— レースコンディション、デッドロック、再入問題——が発生するようになる。

これらのバグには共通の性質がある。再現しない。 タイミングに依存するので、デバッガを繋いだ途端に消える。 だから発生してから直すのではなく、発生しない設計をするしかない。

この章で使う既出の用語(定義は各リンク先). コンテキスト(01 章 6 節)、タスク(02 章 2 節)、ハードウェア(03 章 2 節)、自分自身(04 章 6 節)、FreeRTOS(06 章 4 節)、タイムアウト(10 章 3 節)、ブロック(13 章 1 節)、キュー(14 章 7 節)、不可(16 章 4 節)、ローカル変数(17 章 2 節)

1. レースコンディション

複数の実行の流れが、共有データに同時にアクセスして、結果が実行順序に依存する現象。

int counter = 0;

/* タスク A */              /* タスク B */
counter++;                  counter++;

counter++ は機械語で 3 命令になる。

LDR  r0, [counter]     ; 読む
ADD  r0, r0, #1        ; 足す
STR  r0, [counter]     ; 書く

この間に切り替わると——

時刻  タスク A              タスク B           counter
 1    LDR r0 ← 0                               0
 2    (切り替え)
 3                        LDR r0 ← 0           0
 4                        ADD r0 = 1           0
 5                        STR counter ← 1      1
 6    (切り替え)
 7    ADD r0 = 1                               1
 8    STR counter ← 1                          1  ← 2 になるはずが 1

2 回インクリメントしたのに 1 しか増えていない。

02 章で「並行だけでも問題は起きる」と書いたのはこれである. 単一コアでも、切り替えが「読む」と「書く」の間に入れば同じことが起きる。 並列(マルチコア)でなくても、レースは起きる。

レースが起きる 3 条件

条件内容
1共有されているデータがある
2少なくとも 1 つが書き込む
3アクセスが原子的でない(または順序の保証がない)

どれか 1 つを消せば、レースは起きない。

条件を消す方法手段
1 を消す共有しない(各タスクが自分のデータを持つ。キューで渡す)
2 を消す読み取り専用にする(const、初期化後は不変)
3 を消すミューテックス・クリティカルセクション・原子操作

「1 を消す」が最も強力である.

ミューテックスを正しく使うのは難しい。忘れる、順序を間違える、 持ったままブロックする——人間はミスをする。

共有しなければ、そもそも間違えようがない。

FreeRTOS のキューが「値渡し」なのは、まさにこのためである(11 章)。 送った瞬間に所有権が移る。共有状態が存在しない。

2. どこまでが原子的か

操作32 bit CPU で原子的か
uint8_t / uint16_t / uint32_t の単純な読み原子的
同上の単純な書き原子的
uint64_t の読み書き原子的でない(2 命令)
構造体の読み書き原子的でない
x++ / x += 1原子的でない(read-modify-write)
x = y (両方 32 bit)原子的
ビットフィールドの書き込み原子的でない(隣のビットも読み書きする)
float の読み書き32 bit なら原子的だが、FPU の状態に依存

ビットフィールドは特に危険である.

struct { unsigned a : 1; unsigned b : 1; } flags;

/* タスク A */          /* タスク B */
flags.a = 1;            flags.b = 1;

両方とも「バイト全体を読んで、ビットを変えて、書き戻す」ので、 片方の更新が消える。

「別のフィールドだから安全」と思いがちだが、まったく安全でない。 同じことは、隣接する uint8_t の配列要素にも(CPU によっては)起こりうる。

3. 排他の道具と使い分け

道具割り込みを止めるかタスク切り替えを止めるか適した区間
taskENTER_CRITICAL()止める止める数マイクロ秒。ISR と共有
vTaskSuspendAll()止めない止める数十マイクロ秒〜。タスク間のみ
ミューテックス止めない止めない長い区間。優先度継承あり
キュー/通知——そもそも共有しない(最善)

選ぶ手順

1. キュー/通知で「渡す」設計にできないか?   → できるなら、それが最善
2. ISR とも共有するか?
     はい → taskENTER_CRITICAL()(区間は数 µs に)
     いいえ → 3 へ
3. 区間は短いか(数十 µs 以下)?
     はい → vTaskSuspendAll()
     いいえ → ミューテックス

taskENTER_CRITICAL() の中でやってはいけないこと.

taskENTER_CRITICAL();
{
    shared_value = compute();      /* compute() が何をするか把握していること */
}
taskEXIT_CRITICAL();

4. デッドロック

複数のタスクが、互いに相手の持つ資源を待って、永久に止まる現象。

/* タスク A */                        /* タスク B */
xSemaphoreTake( mutexX, MAX );        xSemaphoreTake( mutexY, MAX );
xSemaphoreTake( mutexY, MAX );        xSemaphoreTake( mutexX, MAX );   /* ← 両方止まる */
タスク A: X を持ち、Y を待つ
タスク B: Y を持ち、X を待つ
→ 循環待ち。どちらも永久に進まない

デッドロックの 4 条件(Coffman の条件)

4 つ全部が揃ったときだけデッドロックが起きる。

#条件意味
1相互排除資源は同時に 1 つのタスクしか使えない
2保持と待機資源を持ったまま、別の資源を待つ
3横取り不可他人が持っている資源を奪えない
4循環待ち待ちの関係が輪になっている

どれか 1 つを崩せば、デッドロックは起きない。

対策 1: 順序を決める(条件 4 を崩す)— 最も実用的

すべてのミューテックスに番号を振り、必ず番号の小さい順に取る。

/* 規約: mutexX(=1) → mutexY(=2) の順でしか取らない */

/* タスク A */                        /* タスク B */
xSemaphoreTake( mutexX, MAX );        xSemaphoreTake( mutexX, MAX );   /* ← 順序を守る */
xSemaphoreTake( mutexY, MAX );        xSemaphoreTake( mutexY, MAX );

これで循環が発生しえなくなる。最も実用的で、最も広く使われている方法である。

Linux カーネルもこの方式で、ロック順序をドキュメントに明記している (lockdep という実行時検証機構まで持っている)。

対策 2: タイムアウトを使う(条件 2 を緩める)

if( xSemaphoreTake( mutexY, pdMS_TO_TICKS( 100 ) ) != pdTRUE ) {
    xSemaphoreGive( mutexX );        /* 諦めて、持っているものを返す */
    vTaskDelay( pdMS_TO_TICKS( random_backoff() ) );
    goto retry;
}

これは「対症療法」であって、根本解決ではない.

リアルタイムシステムには向かない。 ただし「万一のときに止まらない」保険としては価値がある。

対策 3: ミューテックスをネストしない(条件 2 を消す)— 最善

1 度に 1 つのミューテックスしか持たない設計にする。

/* 悪い: ネストしている */
xSemaphoreTake( mutexX, MAX );
    read_from_x( &data );
    xSemaphoreTake( mutexY, MAX );
        write_to_y( data );
    xSemaphoreGive( mutexY );
xSemaphoreGive( mutexX );

/* 良い: 逐次的に */
xSemaphoreTake( mutexX, MAX );
    read_from_x( &data );          /* ローカル変数にコピー */
xSemaphoreGive( mutexX );

xSemaphoreTake( mutexY, MAX );
    write_to_y( data );
xSemaphoreGive( mutexY );

これができるなら、デッドロックの心配は完全に消える。 13 章の連鎖的優先度継承の限界にも当たらない。

デッドロックを見る

タスクとロックの取得順序を変えて、循環待ちが発生する様子と、 順序規約で防げることを確かめられる。

5. 自己デッドロック

void foo( void ) {
    xSemaphoreTake( xMutex, portMAX_DELAY );
    bar();                                      /* ← bar も同じミューテックスを取る */
    xSemaphoreGive( xMutex );
}

void bar( void ) {
    xSemaphoreTake( xMutex, portMAX_DELAY );    /* ← 自分自身を待つ。永久に止まる */
    ...
}

1 つのタスクだけで起きるデッドロックである。

対策内容
再帰ミューテックスを使う動くが、設計の混乱の兆候(12 章)
ロック規約を作る公開関数はロックを取る/内部関数は取らない(推奨)
/* 公開 API: ロックを取る */
void buffer_push( int v ) {
    xSemaphoreTake( xMutex, portMAX_DELAY );
    buffer_push_locked( v );
    xSemaphoreGive( xMutex );
}

void buffer_flush( void ) {
    xSemaphoreTake( xMutex, portMAX_DELAY );
    while( !empty_locked() ) { buffer_pop_locked(); }    /* 内部関数を呼ぶ */
    xSemaphoreGive( xMutex );
}

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

_locked という命名規約は非常に有効である。 関数名を見ただけで「ロックを持っている前提か」が分かる。 Linux カーネルでも __ プレフィックスなどで同じ区別をしている。

6. 再入問題

同じ関数が、複数の実行の流れから同時に呼ばれても正しく動くか。

/* 再入不可 */
char *itoa_bad( int n ) {
    static char buf[16];        /* ← static。全タスクで共有される */
    sprintf( buf, "%d", n );
    return buf;
}

/* 再入可能 */
void itoa_ok( int n, char *buf, size_t len ) {
    snprintf( buf, len, "%d", n );      /* 呼び出し側がバッファを用意 */
}

再入可能の条件

条件内容
static 変数を書かない全タスクで共有されるから
グローバル変数を書かない同上
再入不可な関数を呼ばない伝染する
ハードウェアの状態を仮定しない他のタスクが変えているかもしれない

標準ライブラリの再入性

関数再入性
strlen / memcpy / strcmp再入可能
strtok再入不可(内部状態を持つ)→ strtok_r を使う
rand再入不可 → rand_r を使う
malloc / free実装依存。FreeRTOS では pvPortMalloc を使う
printf / sprintf実装依存。newlib では errno と内部バッファの問題がある
errnoグローバル変数。設定次第
#define configUSE_NEWLIB_REENTRANT  1

これを有効にすると、TCB に struct _reent が埋め込まれ、 コンテキストスイッチのたびに _impure_ptr が切り替わる(06 章の vTaskSwitchContext)。

configUSE_NEWLIB_REENTRANT はメモリを大量に食う.

struct _reent は 1 KB 近い(実装による)。それがタスクごとに必要になる。 タスクが 10 個なら 10 KB である。組み込みでは大きすぎることが多い。

より現実的な対策:

7. TOCTOU — 確認と実行の隙間

Time Of Check to Time Of Use。 「確認したときの状態」が「実行するとき」には変わっている問題。

if( uxQueueMessagesWaiting( q ) > 0 ) {        /* 確認 */
    /* ← ここで他のタスクが取ってしまうかもしれない */
    xQueueReceive( q, &item, 0 );              /* 実行 → 失敗する */
}

正しくは、確認せずに実行して結果を見る。

if( xQueueReceive( q, &item, 0 ) == pdPASS ) {
    /* 取れた */
}

FreeRTOS の API は、内部で確認と実行を原子的に行っている。 外側で確認を足すと、かえって危険になる。

09 章のティックレスアイドルで見た「二重判定」も TOCTOU 対策だった.

xExpectedIdleTime = prvGetExpectedIdleTime();       /* 1 回目 */
if( xExpectedIdleTime >= threshold ) {
    vTaskSuspendAll();
    {
        xExpectedIdleTime = prvGetExpectedIdleTime();  /* 2 回目(排他区間で) */
        if( xExpectedIdleTime >= threshold ) { sleep(); }
    }
    xTaskResumeAll();
}

1 回目は「排他区間に入る価値があるか」の粗い判定、 2 回目は「実際に行動してよいか」の確実な判定である。

これは「ダブルチェックロッキング」と呼ばれるパターンで、 排他区間に入るコストを避けつつ正しさを保つための定石である。

8. 設計チェックリスト

項目確認
共有データを一覧にしたか誰が読み、誰が書くかを表にする
共有をやめられないかキュー/通知で渡す設計にできないか
ロック順序を決めたかミューテックスに番号を振り、文書化する
ミューテックスをネストしていないかしていないのが理想
ミューテックスを持ったままブロックしていないか13 章。最重要
クリティカルセクションは数 µs 以下か割り込みレイテンシに直結
static 変数を持つ関数を、複数タスクから呼んでいないか再入問題
確認と実行を分けていないかTOCTOU
configASSERT を有効にしているか多くのミスがここで捕まる
ISR とタスクの共有に volatile だけで済ませていないか原子性は別問題

9. この章のまとめ

ポイント内容
レースの 3 条件共有・書き込み・非原子性。どれか 1 つを消せばよい
最良の対策共有しない。キューで渡す
原子性32 bit の単純な読み書きのみ。x++ もビットフィールドも非原子的
排他の選択ISR と共有→クリティカル、短い→SuspendAll、長い→ミューテックス
デッドロックの 4 条件相互排除・保持と待機・横取り不可・循環待ち
最も実用的な対策ロック順序を決める(番号順に取る)
最善の対策ミューテックスをネストしない
自己デッドロック_locked 命名規約で防ぐ
再入性static・グローバルを書かない。printf はログタスクに集約
TOCTOU確認せずに実行して結果を見る。API は内部で原子的

次章では、ここまでの設計が「本当に間に合っているか」を 計算で確かめる方法を扱う。