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 になるはずが 12 回インクリメントしたのに 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()の中でやってはいけないこと.
- ブロックする API を呼ぶ(切り替えられないので永久に止まる)
- 長い処理(割り込みレイテンシが悪化する)
printfなどライブラリ関数(何をするか分からない)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 である。組み込みでは大きすぎることが多い。より現実的な対策:
printf系をログタスク 1 つに集約する(17 章)newlib-nanoを使う(_reentが小さい)- 標準ライブラリを使わない軽量実装に差し替える
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 は内部で原子的 |
次章では、ここまでの設計が「本当に間に合っているか」を 計算で確かめる方法を扱う。