Chapter 14

タスク通知とイベントグループ — キューより 45 % 速い通知路

この章がなぜ必要なのか——キューは万能だが、重い.

11 章で見たとおり、キューは オブジェクト本体 ＋ バッファ ＋ 2 本のイベントリスト ＋ ロック機構を持つ。 送信のたびにクリティカルセクションに入り、コピーし、リストを操作する。

だが用途によっては、そこまで必要ない。 「ISR がタスクを 1 つ起こす」だけなら、もっと軽い方法がある。 それがタスク通知であり、FreeRTOS 公式が推奨する方法である。

この章で使う既出の用語（定義は各リンク先）. TCB（02 章 2 節）、タスク（02 章 2 節）、Ready（04 章 1 節）、イベントリスト（04 章 4 節）、FreeRTOS（06 章 4 節）、追加（07 章 2 節）、タイムアウト（10 章 3 節）

1. タスク通知 — TCB に直接書き込む

#define configUSE_TASK_NOTIFICATIONS  1        /* 既定値は 1 */

各タスクの TCB に、通知用の変数が最初から埋め込まれている。

#if ( configUSE_TASK_NOTIFICATIONS == 1 )
    volatile uint32_t ulNotifiedValue[ configTASK_NOTIFICATION_ARRAY_ENTRIES ];
    volatile uint8_t  ucNotifyState[  configTASK_NOTIFICATION_ARRAY_ENTRIES ];
#endif

configTASK_NOTIFICATION_ARRAY_ENTRIES（既定 1）で、通知の「チャネル数」を増やせる。

これが速さの理由である.

• オブジェクトを作る必要がない（TCB に埋め込み済み）
• キューもリストもバッファもない
• 通知の実体は「32 bit 変数への書き込み ＋ 状態フラグの更新」だけ

FreeRTOS 公式の測定では、 キューでの通知より 45 % 速く、メモリを 60 バイト以上節約するとされている （構成による。ARM Cortex-M での GCC 最適化ありの測定値）。

通知の使い方（最も単純な形）

/* 待つ側 */
void vTask( void *pv )
{
    for (;;) {
        ulTaskNotifyTake( pdTRUE, portMAX_DELAY );    /* 通知が来るまで寝る */
        handle();
    }
}

/* 起こす側（ISR から） */
void UART_IRQHandler( void )
{
    BaseType_t woken = pdFALSE;
    vTaskNotifyGiveFromISR( xTaskHandle, &woken );
    portYIELD_FROM_ISR( woken );
}

バイナリセマフォと同じ形だが、オブジェクトの生成が不要である。

2. 通知の 5 つのモード

xTaskNotify() の第 3 引数 eAction で、通知値の扱い方を選ぶ。

BaseType_t xTaskNotify( TaskHandle_t xTaskToNotify,
                        uint32_t ulValue,
                        eNotifyAction eAction );

5 つのモードで、キュー・セマフォ・イベントグループのほとんどを代替できる。

使い分けの例

/* (a) 単純な起床通知 = バイナリセマフォ */
vTaskNotifyGive( xTask );                          /* = xTaskNotify(t, 0, eIncrement) */
ulTaskNotifyTake( pdTRUE, portMAX_DELAY );         /* pdTRUE = 取ったら 0 にクリア */

/* (b) 回数を数える = カウンティングセマフォ */
vTaskNotifyGive( xTask );                          /* 呼ぶたびに +1 */
ulTaskNotifyTake( pdFALSE, portMAX_DELAY );        /* pdFALSE = 1 だけ減らす */

/* (c) 複数の事象をビットで = イベントグループ */
#define EVT_RX   (1 << 0)
#define EVT_TX   (1 << 1)
#define EVT_ERR  (1 << 2)

xTaskNotify( xTask, EVT_RX, eSetBits );

uint32_t ulNotified;
xTaskNotifyWait( 0x00,            /* 入る前にクリアするビット */
                 ULONG_MAX,       /* 出るときにクリアするビット（全部） */
                 &ulNotified,     /* 受け取った値 */
                 portMAX_DELAY );
if( ulNotified & EVT_RX ) { ... }

/* (d) 値を 1 個渡す = 長さ 1 のキュー */
xTaskNotify( xTask, sensor_value, eSetValueWithOverwrite );

タスク通知 vs キューのコスト

同じ通知をキュー・セマフォ・タスク通知で行ったときの メモリ使用量と処理時間を比較できる。

3. タスク通知の制約

速い代わりに、できないことがある。

「送信側がブロックできない」が実務では効いてくる.

キューなら、満杯のとき送信側が待って流量が自然に調整される（バックプレッシャ）。 通知にはそれがないので、受信側が遅いと通知が失われる。

xTaskNotify( t, v, eSetValueWithoutOverwrite );   /* 未読なら pdFAIL が返る */

eSetValueWithoutOverwrite を使えば失敗を検出できるが、 「待つ」ことはできない。流量制御が要るならキューを使うこと。

4. 通知の内部実装

BaseType_t xTaskGenericNotify( TaskHandle_t xTaskToNotify,
                               UBaseType_t uxIndexToNotify,
                               uint32_t ulValue,
                               eNotifyAction eAction,
                               uint32_t * pulPreviousNotificationValue )
{
    TCB_t * pxTCB = xTaskToNotify;

    taskENTER_CRITICAL();
    {
        if( pulPreviousNotificationValue != NULL ) {
            *pulPreviousNotificationValue = pxTCB->ulNotifiedValue[ uxIndexToNotify ];
        }

        ucOriginalNotifyState = pxTCB->ucNotifyState[ uxIndexToNotify ];
        pxTCB->ucNotifyState[ uxIndexToNotify ] = taskNOTIFICATION_RECEIVED;

        switch( eAction )
        {
            case eSetBits:   pxTCB->ulNotifiedValue[ uxIndexToNotify ] |= ulValue; break;
            case eIncrement: ( pxTCB->ulNotifiedValue[ uxIndexToNotify ] )++;      break;
            case eSetValueWithOverwrite:
                             pxTCB->ulNotifiedValue[ uxIndexToNotify ] = ulValue;  break;
            case eSetValueWithoutOverwrite:
                if( ucOriginalNotifyState != taskNOTIFICATION_RECEIVED ) {
                    pxTCB->ulNotifiedValue[ uxIndexToNotify ] = ulValue;
                } else {
                    xReturn = pdFAIL;      /* 未読の通知があるので失敗 */
                }
                break;
            case eNoAction:  break;
        }

        /* 待っていたなら起こす */
        if( ucOriginalNotifyState == taskWAITING_NOTIFICATION )
        {
            ( void ) uxListRemove( &( pxTCB->xStateListItem ) );
            prvAddTaskToReadyList( pxTCB );

            configASSERT( pxTCB->xEventListItem.pvContainer == NULL );  /* イベントリスト不使用 */

            if( pxTCB->uxPriority > pxCurrentTCB->uxPriority ) {
                taskYIELD_IF_USING_PREEMPTION();
            }
        }
    }
    taskEXIT_CRITICAL();

    return xReturn;
}

configASSERT( pxTCB->xEventListItem.pvContainer == NULL ) に注目.

通知待ちのタスクは、イベントリストを使わない。 「誰から通知が来るか」が既に決まっているので、待ち行列を作る必要がないからである。

だから起こす処理が

uxListRemove( &( pxTCB->xStateListItem ) );
prvAddTaskToReadyList( pxTCB );

の 2 行だけで済む。キューの xTaskRemoveFromEventList() より格段に短い。 これが 45 % 速い理由の本体である。

5. イベントグループ — 複数の事象を組み合わせて待つ

#define configUSE_EVENT_GROUPS  1

EventGroupHandle_t xEventGroup = xEventGroupCreate();

#define BIT_WIFI_UP    ( 1 << 0 )
#define BIT_TIME_SYNC  ( 1 << 1 )
#define BIT_CONFIG_OK  ( 1 << 2 )

/* 立てる側 */
xEventGroupSetBits( xEventGroup, BIT_WIFI_UP );

/* 待つ側: 3 つ全部が揃うまで待つ */
EventBits_t bits = xEventGroupWaitBits(
        xEventGroup,
        BIT_WIFI_UP | BIT_TIME_SYNC | BIT_CONFIG_OK,
        pdFALSE,          /* xClearOnExit: 抜けるときにクリアしない */
        pdTRUE,           /* xWaitForAllBits: pdTRUE = AND、pdFALSE = OR */
        portMAX_DELAY );

タスク通知との決定的な違いはこれである。

イベントグループを使う理由は「AND 待ち」と「複数受信者」の 2 つだけ.

それ以外の用途なら、タスク通知の eSetBits で足りる。 「複数の条件が揃うまで待つ」は、素朴に書くと難しい。

/* イベントグループなしで AND 待ちを書くと… */
for (;;) {
    xSemaphoreTake( xSemA, portMAX_DELAY );
    xSemaphoreTake( xSemB, portMAX_DELAY );   /* ← 順序に依存する。B が先に来ると？ */
    ...
}

イベントグループなら 1 行で済む。これが存在理由である。

ビット数の制限

typedef uint32_t EventBits_t;      /* configUSE_16_BIT_TICKS == 0 のとき */

使えるのは下位 24 ビットだけである（上位 8 ビットはカーネルが制御用に使う）。

#define eventEVENT_BITS_CONTROL_BYTES   0xff000000UL

configUSE_16_BIT_TICKS == 1 の場合は uint16_t になり、使えるのは下位 8 ビットになる。

xEventGroupSync() — ランデブー

EventBits_t xEventGroupSync( EventGroupHandle_t xEventGroup,
                             const EventBits_t uxBitsToSet,
                             const EventBits_t uxBitsToWaitFor,
                             TickType_t xTicksToWait );

「自分のビットを立てて、全員のビットが揃うまで待つ」—— 複数タスクの待ち合わせ（バリア同期）である。

#define TASK_A_BIT  (1 << 0)
#define TASK_B_BIT  (1 << 1)
#define TASK_C_BIT  (1 << 2)
#define ALL_BITS    (TASK_A_BIT | TASK_B_BIT | TASK_C_BIT)

/* 各タスクで */
do_my_part();
xEventGroupSync( xEventGroup, TASK_A_BIT, ALL_BITS, portMAX_DELAY );
/* ここに来たときは、3 つ全部が do_my_part() を終えている */

ビットのクリアは自動で行われるので、次のラウンドでそのまま使える。

xEventGroupSetBits() と xEventGroupWaitBits() を素朴に組み合わせても ランデブーにはならない.

xEventGroupSetBits( g, MY_BIT );
xEventGroupWaitBits( g, ALL_BITS, pdTRUE, pdTRUE, portMAX_DELAY );  /* ← 競合する */

「クリアするのは誰か」が競合する。 早く来たタスクが xClearOnExit でクリアすると、 遅れて来たタスクが「まだ揃っていない」と判定してしまう。

xEventGroupSync() はこの競合をカーネル内部で正しく処理している。 バリア同期には必ずこれを使うこと。

6. イベントグループと ISR

イベントグループの ISR 版は、タイマタスク経由で動く。

BaseType_t xEventGroupSetBitsFromISR( EventGroupHandle_t xEventGroup,
                                      const EventBits_t uxBitsToSet,
                                      BaseType_t *pxHigherPriorityTaskWoken )
{
    return xTimerPendFunctionCallFromISR( vEventGroupSetBitsCallback,
                                          ( void * ) xEventGroup,
                                          ( uint32_t ) uxBitsToSet,
                                          pxHigherPriorityTaskWoken );
}

なぜか。ビットを立てる処理は「待っている全タスクを調べて、条件を満たすものを全部起こす」 という長さの読めない処理だからである。ISR の中でやりたくない。

そこで、タイマタスク（デーモンタスク）に処理を委譲する（15 章）。

これには 2 つの帰結がある.

1. configUSE_TIMERS を有効にしないと xEventGroupSetBitsFromISR は使えない
2. ビットが実際に立つのは、タイマタスクが走った後である（遅延がある）

ISR から即座に反応させたいなら、タスク通知を使うこと。 vTaskNotifyGiveFromISR() はその場で処理が完了する。

7. ストリームバッファとメッセージバッファ

v10.0 で追加された、1 対 1 専用の軽量なバイト列転送である。

StreamBufferHandle_t xStream = xStreamBufferCreate( 1024, 10 );
/* サイズ 1024 バイト、トリガレベル 10 バイト */

/* 送る */
xStreamBufferSend( xStream, data, len, timeout );

/* 受ける: 10 バイト以上溜まるか、タイムアウトするまで待つ */
size_t n = xStreamBufferReceive( xStream, buf, sizeof(buf), timeout );

MessageBufferHandle_t xMsg = xMessageBufferCreate( 1024 );
xMessageBufferSend( xMsg, "hello", 5, 0 );
xMessageBufferSend( xMsg, "world!", 6, 0 );

char buf[32];
size_t n = xMessageBufferReceive( xMsg, buf, sizeof(buf), 0 );   /* n = 5, "hello" */
n = xMessageBufferReceive( xMsg, buf, sizeof(buf), 0 );          /* n = 6, "world!" */

重要な制約: 1 対 1 専用である.

• 送信側は1 つのタスク（または 1 つの ISR）だけ
• 受信側は1 つのタスクだけ

この前提のおかげで、ロックなし（lock-free）で実装されている。 だからキューより速い。

複数の送信者がいる場合は、自分でミューテックスをかける必要がある。 「ロックなしで速い」のは前提を守った場合だけである。

適材適所のまとめ

8. この章のまとめ

ここまでで第 III 部は終わりである。 次章から第 IV 部——時間とメモリの管理に入る。