Chapter 11
キュー — FreeRTOS の中核データ構造
この章がなぜ必要なのか——
queue.cは 1 本で 4 つのものを実装している.FreeRTOS におけるキュー・バイナリセマフォ・カウンティングセマフォ・ ミューテックス・再帰ミューテックスは、すべて同じ構造体・同じ関数でできている。
つまりキューを 1 つ理解すれば、残りは全部おまけである。 逆に、キューを曖昧にしたままだと、セマフォもミューテックスも曖昧なままになる。
この章で使う既出の用語（定義は各リンク先）. スタック（02 章 2 節）、タスク（02 章 2 節）、スケジューラ（03 章 2 節）、データ構造（03 章 2 節）、イベントリスト（04 章 4 節）、FreeRTOS（06 章 4 節）、追加（07 章 2 節）、タイムアウト（10 章 3 節）
1. キューとは何か
タスク間・ISR とタスク間で、データを安全に受け渡す FIFO バッファである。
QueueHandle_t q = xQueueCreate( 10, sizeof( int ) ); /* int を 10 個まで */
/* 送る側 */
int value = 42;
xQueueSend( q, &value, pdMS_TO_TICKS( 100 ) );
/* 受ける側 */
int received;
xQueueReceive( q, &received, portMAX_DELAY );FreeRTOS のキューには、3 つの重要な性質がある。
| 性質 | 内容 |
|---|---|
| 値渡し（コピー） | ポインタではなく中身をコピーする |
| ブロックできる | 満杯なら送信側が待ち、空なら受信側が待つ |
| スレッドセーフ | 排他制御が内蔵されている。外側でロックする必要がない |
キューの動作
送信・受信・タイムアウトの様子と、待ちタスクがどう起こされるかを確かめられる。
2. なぜ「値渡し」なのか
これは FreeRTOS の最も重要な設計判断の 1 つである。
xQueueSend( q, &value, 0 ); /* value の「中身」がキューにコピーされる */送った後、value を書き換えても、キューの中のデータは変わらない。
これは大きな安全性をもたらす.
ポインタを渡す設計だと、こういう事故が起きる。
void vSenderTask( void *pv ) { for (;;) { int local = read_sensor(); xQueueSend( q, &local, 0 ); /* ← ポインタ渡しだったら？ */ /* この関数を抜けると local は消える。受信側は壊れたメモリを読む */ } }値渡しなら、送った瞬間にコピーが完了しているので、 送信側は何をしても構わない。所有権の問題が発生しない。
値渡しのコスト
大きなデータを送ると、コピーのコストがかかる。
| データサイズ | キュー長 | メモリ | コピー時間の目安 |
|---|---|---|---|
| 4 バイト | 10 | 40 バイト | 数十 ns |
| 64 バイト | 10 | 640 バイト | 数百 ns |
| 1024 バイト | 10 | 10 KB | 数 µs |
大きなデータは、ポインタを送るのが定石である。
typedef struct { uint8_t *buf; size_t len; } Msg_t;
Msg_t msg = { .buf = pvPortMalloc( 1024 ), .len = 1024 };
fill( msg.buf );
xQueueSend( q, &msg, 0 ); /* 8 バイトのコピーで済む */
/* 受信側が解放の責任を負う */
xQueueReceive( q, &msg, portMAX_DELAY );
process( msg.buf, msg.len );
vPortFree( msg.buf );ポインタを送る場合、「所有権」を設計で決めること.
- 誰が確保するか
- 誰が解放するか
- キューが満杯で送れなかったとき、誰が解放するか（これを忘れるとリークする）
if( xQueueSend( q, &msg, 0 ) != pdPASS ) { vPortFree( msg.buf ); /* 送れなかったので自分で解放 */ }この設計判断を文書化しておかないと、必ずリークかダブルフリーが起きる。
3. キューの内部構造
typedef struct QueueDefinition
{
int8_t *pcHead; /* バッファの先頭 */
int8_t *pcWriteTo; /* 次に書く位置 */
union
{
QueuePointers_t xQueue; /* 通常のキューのとき */
SemaphoreData_t xSemaphore;/* セマフォ／ミューテックスのとき */
} u;
List_t xTasksWaitingToSend; /* 空きを待っているタスク */
List_t xTasksWaitingToReceive; /* データを待っているタスク */
volatile UBaseType_t uxMessagesWaiting; /* いま入っている個数 */
UBaseType_t uxLength; /* 最大個数 */
UBaseType_t uxItemSize; /* 1 個のバイト数 */
volatile int8_t cRxLock; /* ロック中の受信回数 */
volatile int8_t cTxLock; /* ロック中の送信回数 */
...
} xQUEUE;typedef struct QueuePointers
{
int8_t *pcTail; /* バッファの末尾 */
int8_t *pcReadFrom; /* 最後に読んだ位置 */
} QueuePointers_t;
typedef struct SemaphoreData
{
TaskHandle_t xMutexHolder; /* ミューテックスの所有者（[12 章](12_セマフォとミューテックス.md#再帰ミューテックス)） */
UBaseType_t uxRecursiveCallCount; /* 再帰ミューテックスの深さ */
} SemaphoreData_t;
unionに注目してほしい.通常のキューは
pcTailとpcReadFromを使う。 セマフォ／ミューテックスはxMutexHolderとuxRecursiveCallCountを使う。 同時に使うことはないので、unionで重ねてメモリを節約している。セマフォは
uxItemSize = 0なので、バッファもポインタも不要である。 「個数だけを数えるキュー」がセマフォである。 これで実装が完全に共有できる。
リングバッファ
uxLength = 4, uxItemSize = 4 のキュー
pcHead ──→ ┌──────┬──────┬──────┬──────┐ ←── pcTail
│ item │ item │ │ │
└──────┴──────┴──────┴──────┘
↑ ↑
pcReadFrom pcWriteTo
uxMessagesWaiting = 2pcWriteTo が pcTail に達したら pcHead に戻る。単純なリングバッファである。
4. 送信 — xQueueGenericSend()
キューへの送信は、この 1 つの関数に集約されている。
BaseType_t xQueueGenericSend( QueueHandle_t xQueue,
const void * const pvItemToQueue,
TickType_t xTicksToWait,
const BaseType_t xCopyPosition )
{
for( ;; )
{
taskENTER_CRITICAL();
{
/* (A) 空きがあるか？ 上書き指定なら常に書ける */
if( ( pxQueue->uxMessagesWaiting < pxQueue->uxLength )
|| ( xCopyPosition == queueOVERWRITE ) )
{
prvCopyDataToQueue( pxQueue, pvItemToQueue, xCopyPosition );
/* (B) 受信待ちのタスクがいれば起こす */
if( listLIST_IS_EMPTY( &( pxQueue->xTasksWaitingToReceive ) ) == pdFALSE )
{
if( xTaskRemoveFromEventList(
&( pxQueue->xTasksWaitingToReceive ) ) != pdFALSE )
{
queueYIELD_IF_USING_PREEMPTION(); /* 高優先度なら即切替 */
}
}
taskEXIT_CRITICAL();
return pdPASS;
}
else
{
/* (C) 満杯。待たないなら失敗を返す */
if( xTicksToWait == 0 ) {
taskEXIT_CRITICAL();
return errQUEUE_FULL;
}
else if( xEntryTimeSet == pdFALSE ) {
vTaskInternalSetTimeOutState( &xTimeOut );
xEntryTimeSet = pdTRUE;
}
}
}
taskEXIT_CRITICAL();
/* (D) 待つ */
vTaskSuspendAll();
prvLockQueue( pxQueue );
if( xTaskCheckForTimeOut( &xTimeOut, &xTicksToWait ) == pdFALSE )
{
if( prvIsQueueFull( pxQueue ) != pdFALSE )
{
vTaskPlaceOnEventList( &( pxQueue->xTasksWaitingToSend ), xTicksToWait );
prvUnlockQueue( pxQueue );
if( xTaskResumeAll() == pdFALSE ) { portYIELD_WITHIN_API(); }
}
else {
prvUnlockQueue( pxQueue );
( void ) xTaskResumeAll();
}
}
else
{
prvUnlockQueue( pxQueue );
( void ) xTaskResumeAll();
return errQUEUE_FULL; /* タイムアウト */
}
}
}構造を整理する。
| 段階 | 処理 |
|---|---|
| A | クリティカルセクションで空きを確認し、あればコピー |
| B | 受信待ちのタスクを起こす（最高優先度のもの 1 つ） |
| C | 満杯なら、タイムアウト 0 で失敗、そうでなければタイムアウトの基準時刻を設定 |
| D | イベントリストに登録して寝る。起きたらループの先頭に戻ってやり直す |
for(;;)で囲まれ、起きたら「やり直す」構造になっている.なぜなら、起こされてもう一度見たとき、まだ満杯かもしれないからである。
TaskA(低優先度) が満杯を待っている TaskB が 1 個取り出す → TaskA を起こす しかしその前に TaskC(高優先度) が 1 個入れてしまう → TaskA が起きたときには、また満杯これを「偽の目覚め (spurious wakeup)」と呼ぶ。 POSIX の条件変数でも同じ問題があり、対策も同じ—— 「起きたら必ず条件をもう一度確認する」である。
だから待ちは
ifではなくwhile/for(;;)で書く。これは一般則である。
送信の 3 つのバリエーション
xQueueSend( q, &item, timeout ); /* = xQueueSendToBack。末尾に追加 */
xQueueSendToBack( q, &item, timeout ); /* 通常の FIFO */
xQueueSendToFront( q, &item, timeout ); /* 先頭に割り込む（LIFO）*/
xQueueOverwrite( q, &item ); /* 長さ 1 のキュー専用。常に上書き */xQueueSendToFront は「緊急メッセージを先に処理させたい」ときに使う。 xQueueOverwrite は「最新の値だけあればよい」センサ値の受け渡しに便利である （必ず長さ 1 のキューに対して使うこと）。
5. キューのロック — cRxLock / cTxLock
上のコードに prvLockQueue() / prvUnlockQueue() が出てきた。これは何か。
段階 D では vTaskSuspendAll() でスケジューラを止めているが、 割り込みは止めていない。だから ISR から xQueueSendFromISR() が呼ばれうる。
そのとき ISR がイベントリストを直接触ると、 「タスク側が今まさに触っているリスト」を壊す危険がある。
そこで、「ロック中は、イベントリストを触らず、回数だけ数える」という仕組みにする。
#define prvLockQueue( pxQueue ) \
taskENTER_CRITICAL(); \
{ \
if( pxQueue->cRxLock == queueUNLOCKED ) { pxQueue->cRxLock = queueLOCKED_UNMODIFIED; } \
if( pxQueue->cTxLock == queueUNLOCKED ) { pxQueue->cTxLock = queueLOCKED_UNMODIFIED; } \
} \
taskEXIT_CRITICAL()ISR 側（xQueueGenericSendFromISR）:
if( pxQueue->cTxLock == queueUNLOCKED ) {
/* ロックされていない → その場で起こす */
if( listLIST_IS_EMPTY( &( pxQueue->xTasksWaitingToReceive ) ) == pdFALSE ) {
if( xTaskRemoveFromEventList( &( pxQueue->xTasksWaitingToReceive ) ) != pdFALSE ) {
*pxHigherPriorityTaskWoken = pdTRUE;
}
}
}
else {
/* ロック中 → 回数を数えるだけ */
pxQueue->cTxLock = ( int8_t ) ( cTxLock + 1 );
}prvUnlockQueue() で、溜まった回数の分だけまとめて起こす。
static void prvUnlockQueue( Queue_t * const pxQueue )
{
taskENTER_CRITICAL();
{
int8_t cTxLock = pxQueue->cTxLock;
while( cTxLock > queueLOCKED_UNMODIFIED ) {
if( listLIST_IS_EMPTY( &( pxQueue->xTasksWaitingToReceive ) ) == pdFALSE ) {
if( xTaskRemoveFromEventList( &( pxQueue->xTasksWaitingToReceive ) ) != pdFALSE ) {
vTaskMissedYield();
}
} else { break; }
--cTxLock;
}
pxQueue->cTxLock = queueUNLOCKED;
}
taskEXIT_CRITICAL();
/* cRxLock も同様 */
}これは「作業を遅延させる」という古典的な同期技法である.
Linux のソフト割り込み (softirq) や タスクレット、 あるいは NAPI の「割り込みを止めてポーリングに切り替える」—— どれも同じ発想である。
危険な区間では作業を積み上げておいて、安全になったらまとめて処理する。 こうすると、危険な区間で割り込みを止め続ける必要がなくなる。 割り込みレイテンシを悪化させずに、データ構造の整合性を守れる。
6. 受信 — xQueueReceive()
送信とほぼ対称である。違いは 1 点だけ。
prvCopyDataFromQueue( pxQueue, pvBuffer );
--( pxQueue->uxMessagesWaiting );
/* 送信待ちのタスクがいれば起こす（空きができたので） */
if( listLIST_IS_EMPTY( &( pxQueue->xTasksWaitingToSend ) ) == pdFALSE ) {
if( xTaskRemoveFromEventList( &( pxQueue->xTasksWaitingToSend ) ) != pdFALSE ) {
queueYIELD_IF_USING_PREEMPTION();
}
}xQueuePeek() — 読むが消さない
xQueuePeek( q, &buf, timeout ); /* 先頭を読むが、キューからは消えない */prvCopyDataFromQueue() の後に pcReadFrom を元に戻すだけの実装である。
/* xQueuePeek の中 */
pcOriginalReadPosition = pxQueue->u.xQueue.pcReadFrom;
prvCopyDataFromQueue( pxQueue, pvBuffer );
pxQueue->u.xQueue.pcReadFrom = pcOriginalReadPosition; /* 戻す */
/* uxMessagesWaiting は減らさない */「複数のタスクが同じ値を見たい」「値があるか確認だけしたい」ときに使う。
7. ISR 版 API
BaseType_t xHigherPriorityTaskWoken = pdFALSE;
void UART_IRQHandler( void )
{
uint8_t c = UART->DR;
xQueueSendFromISR( q, &c, &xHigherPriorityTaskWoken );
portYIELD_FROM_ISR( xHigherPriorityTaskWoken );
}| 違い | タスク版 | ISR 版 |
|---|---|---|
| ブロック | できる | できない（タイムアウト引数がない） |
| 切り替え | 内部で行う | xHigherPriorityTaskWoken を返すだけ |
| 排他 | taskENTER_CRITICAL() | portSET_INTERRUPT_MASK_FROM_ISR() |
なぜ ISR 版が切り替えを自分で行わないのかは 18 章で詳しく扱う。 簡単に言えば、割り込みハンドラの一番最後で切り替えたいからである。
xHigherPriorityTaskWokenの初期化を忘れないこと.BaseType_t xHigherPriorityTaskWoken = pdFALSE; /* ← 必ず pdFALSE で初期化 */初期化を忘れると、スタック上のゴミが入っていて、 毎回不要な切り替えが起きるか、必要な切り替えが起きない。 前者は性能劣化、後者は応答の遅れになる。どちらも発見しにくい。
8. キューセット — 複数のキューを同時に待つ
QueueSetHandle_t xQueueSet = xQueueCreateSet( 10 );
xQueueAddToSet( q1, xQueueSet );
xQueueAddToSet( q2, xQueueSet );
for (;;) {
QueueSetMemberHandle_t x = xQueueSelectFromSet( xQueueSet, portMAX_DELAY );
if( x == q1 ) { xQueueReceive( q1, &a, 0 ); }
else if( x == q2 ) { xQueueReceive( q2, &b, 0 ); }
}POSIX の select() に相当する機能である。
キューセットはあまり推奨されない.
- メモリを食う（キューセット自体がキューであり、追加のコピーが発生する）
xQueueSelectFromSet()は「どのキューにデータが来たか」を返すだけで、 もう一度xQueueReceive()を呼ぶ必要がある（2 段階になる）configUSE_QUEUE_SETSを有効にすると、全キューの処理が少し重くなる多くの場合、「1 本のキューに、タグ付きの構造体を流す」方が単純で速い。
typedef struct { enum { MSG_SENSOR, MSG_BUTTON, MSG_TIMEOUT } type; union { int sensor_value; int button_id; } data; } Event_t; /* 1 本のキューで全部受ける */ xQueueReceive( xEventQueue, &ev, portMAX_DELAY ); switch( ev.type ) { ... }これは「イベント駆動タスク」の定番パターンであり、設計としても見通しがよい。
9. この章のまとめ
| ポイント | 内容 |
|---|---|
| 値渡し | 中身をコピー。所有権の問題が発生しない |
| 大きなデータ | ポインタを送る。所有権を設計で決めること |
| 内部構造 | リングバッファ ＋ 送信待ち／受信待ちの 2 つのイベントリスト |
union | セマフォと通常キューでフィールドを共有 |
| 待ちの構造 | for(;;) でやり直す。偽の目覚めに備える |
| キューのロック | 危険区間では回数だけ数え、後でまとめて起こす |
xQueuePeek | 読み位置を戻すだけ。個数は減らさない |
| ISR 版 | ブロックできない。xHigherPriorityTaskWoken を必ず初期化 |
| キューセット | 使わずに、タグ付き構造体を 1 本のキューに流す方が良いことが多い |
次章では、この同じ実装から作られる セマフォとミューテックスを見る。新しい実装はほとんど出てこない。