Chapter 10

ブロックと遅延リスト — 待っているタスクを O(1) で起こす

この章がなぜ必要なのか——「待つ」は OS の最大の価値だった.

01 章で「途中で待てること」が OS の決定打だと述べた。 その「待つ」を、カーネルはどう実装しているのか。

難しいのは起こす側である。 毎ティック全タスクを調べていたら、タスク数に比例した時間がかかる。 それではリアルタイム OS を名乗れない。

FreeRTOS はソート済みリスト＋次回起床時刻のキャッシュでこれを解く。 そして 49.7 日ごとに来るティックの折り返しを、リスト 2 本の交換で乗り切る。

この章で使う既出の用語（定義は各リンク先）. TCB（02 章 2 節）、タスク（02 章 2 節）、スケジューラ（03 章 2 節）、Ready（04 章 1 節）、Suspended（04 章 1 節）、イベントリスト（04 章 4 節）、イベント待ち（04 章 4 節）、時間待ち（04 章 4 節）、FreeRTOS（06 章 4 節）

1. 遅延リスト — 起床時刻でソートされたリスト

static List_t xDelayedTaskList1;
static List_t xDelayedTaskList2;
static List_t * volatile pxDelayedTaskList;              /* いま使っている方 */
static List_t * volatile pxOverflowDelayedTaskList;      /* 折り返し後の方 */

遅延リストには、時間待ちのタスクが起床時刻（絶対ティック）の昇順で並ぶ。

xTickCount = 1000 のとき

pxDelayedTaskList:
  [TaskA: 起床 1005] → [TaskB: 起床 1010] → [TaskC: 起床 1100] → [番兵: MAX]
       ↑
   次に起きるのはこれ。先頭を見るだけで分かる

なぜソートするのか

08 章で見た xTaskIncrementTick() の判定を思い出してほしい。

if( xConstTickCount >= xNextTaskUnblockTime ) {
    /* 起こす処理 */
}

xNextTaskUnblockTime は遅延リストの先頭の起床時刻である。

• 大多数のティックでは、この比較が偽 → リストを一切触らない
• 真になったときだけ、先頭から順に、時刻が来たものだけを起こす

for( ;; ) {
    pxTCB = listGET_OWNER_OF_HEAD_ENTRY( pxDelayedTaskList );
    xItemValue = listGET_LIST_ITEM_VALUE( &( pxTCB->xStateListItem ) );

    if( xConstTickCount < xItemValue ) {
        xNextTaskUnblockTime = xItemValue;    /* 次回のためにキャッシュ更新 */
        break;                                 /* 以降は全部まだ先 */
    }
    /* 起こす */
}

ソートされているので、「先頭がまだなら、以降も全部まだ」と言い切れる。 これが break を可能にしている。

計算量の設計を整理する.

操作	頻度	計算量
遅延リストへの挿入（寝るとき）	タスクが寝るたび	O(n) ソート挿入
ティックごとの判定	毎ティック	O(1) 比較 1 回
起床（時刻が来たとき）	起きるタスク数だけ	O(起きる数)

頻度の最も高い「毎ティックの判定」を O(1) にした。 挿入が O(n) なのは受け入れる。これがトレードオフの取り方である。

遅延リストの動作

タスクを寝かせたり、時間を進めたりして、リストの並びと起床の様子を確かめられる。

2. ティックの折り返し — リスト 2 本の交換

xTickCount は 32 bit なので、1 kHz なら 49.7 日で 0 に戻る。

このとき、遅延リストはどうなるか。

xTickCount = 0xFFFFFFF0 (折り返し直前)

TaskA が vTaskDelay(5)  → 起床時刻 = 0xFFFFFFF5   （折り返さない）
TaskB が vTaskDelay(30) → 起床時刻 = 0x0000000E   （折り返す！）

同じリストに入れると、0x0000000E は 0xFFFFFFF5 より小さいので、 ソートの結果 TaskB が先頭に来てしまう。 すると xTickCount = 0xFFFFFFF1 の時点で「TaskB の起床時刻を過ぎている」と判定され、 29 ティック早く起こされる。

解決: リストを 2 本使う

/* prvAddCurrentTaskToDelayedList() の中 */
xTimeToWake = xConstTickCount + xTicksToWait;

if( xTimeToWake < xConstTickCount ) {
    /* 折り返した → オーバーフロー用のリストへ */
    vListInsert( pxOverflowDelayedTaskList, &( pxCurrentTCB->xStateListItem ) );
}
else {
    /* 折り返さない → 通常のリストへ */
    vListInsert( pxDelayedTaskList, &( pxCurrentTCB->xStateListItem ) );

    if( xTimeToWake < xNextTaskUnblockTime ) {
        xNextTaskUnblockTime = xTimeToWake;
    }
}

判定は xTimeToWake < xConstTickCount の一行である。 符号なし整数の加算が折り返すと、結果は元より小さくなる。それが検出条件になる。

交換

xTickCount が 0 に戻った瞬間、2 本のリストのポインタを入れ替える。

#define taskSWITCH_DELAYED_LISTS()                                     \
{                                                                       \
    List_t *pxTemp;                                                     \
    configASSERT( ( listLIST_IS_EMPTY( pxDelayedTaskList ) ) );         \
    pxTemp = pxDelayedTaskList;                                         \
    pxDelayedTaskList = pxOverflowDelayedTaskList;                      \
    pxOverflowDelayedTaskList = pxTemp;                                 \
    xNumOfOverflows++;                                                  \
    prvResetNextTaskUnblockTime();                                      \
}

折り返し前:
  pxDelayedTaskList         → List1  [起床 0xFFFFFFF5 の TaskA]
  pxOverflowDelayedTaskList → List2  [起床 0x0000000E の TaskB]

xTickCount が 0 に戻る → ポインタを交換

折り返し後:
  pxDelayedTaskList         → List2  [起床 0x0000000E の TaskB]  ← 正しく先頭
  pxOverflowDelayedTaskList → List1  （空）

configASSERT( listLIST_IS_EMPTY( pxDelayedTaskList ) ) に注目.

交換する時点で、古いリストは空になっているはずである。 なぜなら、そこにいたタスクの起床時刻はすべて 0xFFFFFFFF 以下であり、 xTickCount がそこまで進んだ時点で全員起こされているからである。

もし空でなければ、それはカーネルの状態が壊れているということである。 だからここで止める。

たった 3 行のポインタ交換で、49.7 日ごとの折り返しを完全に扱えている。 これは非常にうまい設計である。「特別な日を、特別扱いせずに済ませる」—— リストを 2 本用意しておくだけで、比較のロジックは一切変えなくてよい。

3. イベント待ち — 2 つのリストに同時登録

04 章で触れたとおり、xQueueReceive(q, &buf, timeout) のような待ちは、 イベント（データ到着）と時間（タイムアウト）の両方を待つ。

TCB_t
├── xStateListItem  ──→ 遅延リスト（起床時刻 = now + timeout）
└── xEventListItem  ──→ キューの xTasksWaitingToReceive リスト

起きる経路が 2 つある

どちらの経路でも、両方のリストから外す必要がある。 片方だけ外すと、リストに幽霊のような項目が残り、後で暴走する。

これが 07 章の pvContainer が効くところである。

/* ティック割り込みの中（タイムアウト経路） */
( void ) uxListRemove( &( pxTCB->xStateListItem ) );

if( listLIST_ITEM_CONTAINER( &( pxTCB->xEventListItem ) ) != NULL ) {
    ( void ) uxListRemove( &( pxTCB->xEventListItem ) );   /* 探索不要で O(1) */
}

/* xTaskRemoveFromEventList()（イベント経路） */
pxUnblockedTCB = listGET_OWNER_OF_HEAD_ENTRY( pxEventList );
( void ) uxListRemove( &( pxUnblockedTCB->xEventListItem ) );

if( uxSchedulerSuspended == ( UBaseType_t ) pdFALSE ) {
    ( void ) uxListRemove( &( pxUnblockedTCB->xStateListItem ) );
    prvAddTaskToReadyList( pxUnblockedTCB );
}
else {
    /* スケジューラ停止中 → 保留リストへ */
    vListInsertEnd( &( xPendingReadyList ), &( pxUnblockedTCB->xEventListItem ) );
}

xTaskRemoveFromEventList() の戻り値が重要である.

BaseType_t xTaskRemoveFromEventList( const List_t * const pxEventList );

「起こしたタスクが、現在のタスクより優先度が高いか」を返す。 呼び出し側（queue.c など）はこれを見て、切り替えが必要かを判断する。

if( xTaskRemoveFromEventList( &( pxQueue->xTasksWaitingToReceive ) ) != pdFALSE ) {
    queueYIELD_IF_USING_PREEMPTION();
}

これが「キューに送った瞬間、待っていた高優先度タスクに切り替わる」の実体である。

4. xPendingReadyList — スケジューラ停止中の起床

vTaskSuspendAll() の区間では、切り替えができない（06 章）。 だが割り込みは動いているので、その中でタスクが Ready になることはありうる。

そのとき、Ready リストに直接入れると危険である。 なぜなら vTaskSuspendAll() の目的の 1 つが「Ready リストを安全に触ること」だからである。

そこで、保留用のリストにいったん退避する。

static List_t xPendingReadyList;

xTaskResumeAll() で、まとめて Ready リストへ移す。

BaseType_t xTaskResumeAll( void )
{
    --uxSchedulerSuspended;

    if( uxSchedulerSuspended == 0 ) {
        /* 保留されていたタスクを全部 Ready にする */
        while( listLIST_IS_EMPTY( &xPendingReadyList ) == pdFALSE ) {
            pxTCB = listGET_OWNER_OF_HEAD_ENTRY( &xPendingReadyList );
            ( void ) uxListRemove( &( pxTCB->xEventListItem ) );
            ( void ) uxListRemove( &( pxTCB->xStateListItem ) );
            prvAddTaskToReadyList( pxTCB );

            if( pxTCB->uxPriority >= pxCurrentTCB->uxPriority ) {
                xYieldPending = pdTRUE;
            }
        }

        /* 停止中に溜まったティックをまとめて処理する */
        if( xPendedTicks > 0 ) {
            do {
                if( xTaskIncrementTick() != pdFALSE ) { xYieldPending = pdTRUE; }
                --xPendedTicks;
            } while( xPendedTicks > 0 );
        }

        if( xYieldPending != pdFALSE ) {
            taskYIELD_IF_USING_PREEMPTION();
            xAlreadyYielded = pdTRUE;
        }
    }
    return xAlreadyYielded;
}

xPendedTicks の処理に注目.

vTaskSuspendAll() 中もティック割り込みは来る。だがスケジューラが止まっているので、 xTaskIncrementTick() は ++xPendedTicks するだけで帰る（08 章）。

xTaskResumeAll() で、溜まった分をまとめて処理する。 だから時間は失われない。ただし遅れる。

つまり——vTaskSuspendAll() の区間が長いと、全タスクの起床が遅れる。 ここが 06 章で「長い区間には使うが、長すぎてはいけない」と言った理由である。

5. 待つ側の全体像

タスクが vTaskDelay() するときの流れを、最初から最後まで追う。

void vTaskDelay( const TickType_t xTicksToDelay )
{
    BaseType_t xAlreadyYielded = pdFALSE;

    if( xTicksToDelay > ( TickType_t ) 0U )
    {
        configASSERT( uxSchedulerSuspended == 0 );    /* 停止中は禁止 */

        vTaskSuspendAll();
        {
            prvAddCurrentTaskToDelayedList( xTicksToDelay, pdFALSE );
        }
        xAlreadyYielded = xTaskResumeAll();
    }

    if( xAlreadyYielded == pdFALSE ) {
        portYIELD_WITHIN_API();
    }
}

prvAddCurrentTaskToDelayedList() の中身:

static void prvAddCurrentTaskToDelayedList( TickType_t xTicksToWait,
                                            const BaseType_t xCanBlockIndefinitely )
{
    const TickType_t xConstTickCount = xTickCount;

    /* (1) いま Ready リストにいるので、そこから外す */
    if( uxListRemove( &( pxCurrentTCB->xStateListItem ) ) == 0 ) {
        portRESET_READY_PRIORITY( pxCurrentTCB->uxPriority, uxTopReadyPriority );
    }

    /* (2) 無期限待ちなら Suspended リストへ（最適化） */
    #if ( INCLUDE_vTaskSuspend == 1 )
    {
        if( ( xTicksToWait == portMAX_DELAY ) && ( xCanBlockIndefinitely != pdFALSE ) ) {
            vListInsertEnd( &xSuspendedTaskList, &( pxCurrentTCB->xStateListItem ) );
            return;
        }
    }
    #endif

    /* (3) 起床時刻を計算 */
    xTimeToWake = xConstTickCount + xTicksToWait;
    listSET_LIST_ITEM_VALUE( &( pxCurrentTCB->xStateListItem ), xTimeToWake );

    /* (4) 折り返すかどうかで入れるリストを選ぶ */
    if( xTimeToWake < xConstTickCount ) {
        vListInsert( pxOverflowDelayedTaskList, &( pxCurrentTCB->xStateListItem ) );
    }
    else {
        vListInsert( pxDelayedTaskList, &( pxCurrentTCB->xStateListItem ) );

        if( xTimeToWake < xNextTaskUnblockTime ) {
            xNextTaskUnblockTime = xTimeToWake;
        }
    }
}

5 ステップに整理できる。

portRESET_READY_PRIORITY に注目.

uxListRemove() の戻り値は残りの項目数である。 それが 0 なら、その優先度の Ready リストが空になったということなので、 ビットマップの該当ビットを落とす（07 章）。

ビットマップ最適化を使っていない場合、このマクロは何もしない。 uxTopReadyPriority は「ヒント」なので、下げなくても正しく動くからである（07 章）。

6. 待ちのタイムアウト管理 — xTaskCheckForTimeOut

キューやセマフォは、「待っている途中で一度起きて、また待ち直す」ことがある。 このとき、残りのタイムアウト時間を正しく減らす必要がある。

TimeOut_t xTimeOut;
TickType_t xTicksToWait = pdMS_TO_TICKS( 100 );

vTaskSetTimeOutState( &xTimeOut );      /* 開始時刻を記録 */

for( ;; ) {
    /* 何か待つ */
    if( xTaskCheckForTimeOut( &xTimeOut, &xTicksToWait ) != pdFALSE ) {
        break;      /* タイムアウトした */
    }
    /* xTicksToWait は残り時間に更新されている */
}

xTaskCheckForTimeOut() の中身は、折り返しを考慮した引き算である。

BaseType_t xTaskCheckForTimeOut( TimeOut_t * const pxTimeOut,
                                 TickType_t * const pxTicksToWait )
{
    const TickType_t xConstTickCount = xTickCount;
    const TickType_t xElapsedTime = xConstTickCount - pxTimeOut->xTimeOnEntering;

    if( *pxTicksToWait == portMAX_DELAY ) {
        xReturn = pdFALSE;      /* 無期限は絶対にタイムアウトしない */
    }
    else if( ( xNumOfOverflows != pxTimeOut->xOverflowCount ) &&
             ( xConstTickCount >= pxTimeOut->xTimeOnEntering ) ) {
        /* 折り返しをまたいで、しかも追い越した → タイムアウト */
        xReturn = pdTRUE;
        *pxTicksToWait = 0;
    }
    else if( xElapsedTime < *pxTicksToWait ) {
        /* まだ残っている */
        *pxTicksToWait -= xElapsedTime;
        vTaskInternalSetTimeOutState( pxTimeOut );    /* 基準時刻を更新 */
        xReturn = pdFALSE;
    }
    else {
        *pxTicksToWait = 0;
        xReturn = pdTRUE;
    }
    return xReturn;
}

xNumOfOverflows は、折り返しが何回起きたかのカウンタである（taskSWITCH_DELAYED_LISTS() で加算）。 これを比較することで、「折り返しをまたいだかどうか」を確実に判定できる。

7. この章のまとめ

次章では、この「待つ・起こす」の仕組みを使って作られた、 FreeRTOS で最も重要なオブジェクト——キューを見る。