ESP32×FreeRTOS：StreamBufferの使い方と設計ガイド（ESP32 / ESP-IDF）

Uncategorized

この記事でわかること

  • StreamBufferで何ができるか（バイト単位の連続通信）
  • Queue / MessageBuffer との決定的な違い
  • ESP32（ESP-IDF）での動く最小サンプル
  • 主要APIの引数・戻り値とTrigger Levelの役割
  • 設計のコツ（1 writer/1 reader・オーバーヘッド・詰まり対策）

こんな方におすすめ

  • UARTやSPIなどの「境界のない生バイト列」を効率よくタスクに渡したい
  • MessageBufferを使っているが、実はメッセージの区切り（パケット管理）を必要としていない
  • 「一定量溜まってからタスクを起こしたい」というバッファリング制御をしたい
  • メモリ消費を極限まで抑えてタスク間通信をしたい

StreamBufferとは

FreeRTOSの StreamBuffer は、タスク間で「連続したバイト列」を高速に受け渡しする仕組みです。

  • 送信側： 好きなタイミングで、好きなバイト数を書き込む。
  • 受信側： 「指定したバイト数（トリガーレベル）」が溜まるまで待機し、一気に読み出す。
  • 境界なし： MessageBufferと違い、「どこまでが1回分の送信か」という情報は保持されません。

Queue / StreamBuffer / MessageBuffer の違い

設計判断に迷ったらこの比較表を活用してください。

手段何が得意？境界サイズ典型用途
Queue固定サイズの“要素”をFIFOあり（要素単位）固定（item size）コマンド/構造体/小さな値
StreamBufferバイト“ストリーム”なし可変（読み出し側次第）UARTの生バイト、連続データ
MessageBuffer可変長“メッセージ”あり可変（メッセージ単位）フレーム（JSON/行/ヘッダ付き）

ポイント: StreamBufferは、データごとに「長さ情報」を付与しません。そのため、1バイトでも無駄にしたくないストリーム処理に最適です。

内部イメージと「Trigger Level」

StreamBufferを理解する上で最も重要なのが Trigger Level（トリガーレベル） です。

  • 内部構造: シンプルなリングバッファです。
  • オーバーヘッド: MessageBufferのような「各データごとのヘッダ（通常4〜8バイト）」がありません。
  • トリガーレベル: 「バッファに何バイト溜まったら、受信タスクを起床させるか」という閾値です。
    • 例：10バイトに設定すると、1〜9バイト書き込まれた時点では受信側はBlock（スリープ）し続け、10バイト目が入った瞬間に動き出します。

【最小サンプル】生バイトを送って、溜まったら受け取る（ESP32）

やりたいこと

  • SenderTask：A～Zまで1バイトずつ送る。
  • ReceiverTask：10バイト溜まるまで待って、まとめて処理する。
#include <cstdio>
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "freertos/stream_buffer.h"

static StreamBufferHandle_t s_buf;

static void SenderTask(void *pv)
{
    char data = 'A';
    for (;;) {
        // 1バイトずつ送信
        xStreamBufferSend(s_buf, &data, 1, portMAX_DELAY);
        
        // A -> Z を繰り返す
        data = (data == 'Z') ? 'A' : data + 1;
        vTaskDelay(pdMS_TO_TICKS(100)); // 0.1秒おきに1バイト送る
    }
}

static void ReceiverTask(void *pv)
{
    uint8_t rx[20];
    for (;;) {
        // トリガーレベル（10バイト）溜まるまでここでブロック
        // 起床後は、バッファにある分を最大 sizeof(rx) まで一気に読み出す
        size_t n = xStreamBufferReceive(s_buf, rx, sizeof(rx), portMAX_DELAY);

        if (n > 0) {
            printf("[Receiver] Read %u bytes: ", (unsigned)n);
            for(int i=0; i<n; i++) putchar(rx[i]);
            printf("\n");
        }
    }
}

extern "C" void app_main(void)
{
    // 第1引数：バッファ容量（100バイト）
    // 第2引数：トリガーレベル（10バイト溜まったら起床）
    s_buf = xStreamBufferCreate(100, 10);

    if (s_buf == NULL) {
        printf("Create Failed\n");
        return;
    }

    xTaskCreate(SenderTask, "Sender", 2048, NULL, 5, NULL);
    xTaskCreate(ReceiverTask, "Receiver", 2048, NULL, 5, NULL);
}

以下出力結果です。
10文字ずつ出力されていることがわかります。

主要API解説

xStreamBufferCreate()（作成する）

StreamBufferHandle_t sb = xStreamBufferCreate(1024, 10);
  • 役割：StreamBuffer を作成する（バイトストリーム用のバッファ）
  • 引数
    • 第1引数：xBufferSizeBytes（確保するバッファ容量 [byte]）
      • 例：1024
    • 第2引数：xTriggerLevelBytes（受信タスクが起床する閾値 [byte]）
      • 例：10（10バイト溜まるまで受信側はブロックされる）
  • 戻り値
    • 成功：StreamBufferHandle_t（バッファのハンドル）
    • 失敗：NULL（メモリ不足など）

xStreamBufferSend()（送信する）

size_t sent = xStreamBufferSend(sb, txData, dataLen, portMAX_DELAY);
  • 役割：StreamBuffer にデータを書き込む
  • 引数
    • 第1引数：対象の StreamBuffer ハンドル
    • 第2引数：送信データ先頭ポインタ
    • 第3引数：送信バイト数
    • 第4引数：待ち時間（Tick）
      • 0：待たない、portMAX_DELAY：空きが出るまで待つ
  • 戻り値
    • 成功：実際に書き込めたバイト数
    • 失敗：0（タイムアウト、空き不足など）

xStreamBufferReceive()（受信する）

size_t rxLen = xStreamBufferReceive(sb, rxBuffer, sizeof(rxBuffer), portMAX_DELAY);
  • 役割：StreamBuffer からデータを読み出す
  • 引数
    • 第1引数：対象の StreamBuffer ハンドル
    • 第2引数：受信バッファ先頭ポインタ
    • 第3引数：受信バッファサイズ（最大で受け取れるサイズ）
    • 第4引数：待ち時間（Tick）
  • 戻り値
    • 成功：実際に読み出せたバイト数
    • 失敗：0（未到着、タイムアウトなど）

xStreamBufferSendFromISR()（ISRから送信する）

BaseType_t hpw = pdFALSE;
size_t sent = xStreamBufferSendFromISR(sb, txData, dataLen, &hpw);
if (hpw) portYIELD_FROM_ISR();
  • 役割：割り込み（ISR）からデータを書き込む
  • 引数
    • 第1引数：対象の StreamBuffer ハンドル
    • 第2引数：送信データ先頭ポインタ
    • 第3引数：送信バイト数
    • 第4引数：pxHigherPriorityTaskWoken（高優先度タスク起床フラグのポインタ）
  • 戻り値
    • 成功：実際に書き込めたバイト数
    • 失敗：0（空き不足など）

xStreamBufferReceiveFromISR()（ISRから受信する）

BaseType_t hpw = pdFALSE;
size_t rxLen = xStreamBufferReceiveFromISR(sb, rxBuffer, sizeof(rxBuffer), &hpw);
if (hpw) portYIELD_FROM_ISR();
  • 役割：割り込み（ISR）からデータを読み出す
  • 引数
    • 第1引数：対象の StreamBuffer ハンドル
    • 第2引数：受信バッファ先頭ポインタ
    • 第3引数：受信バッファサイズ
    • 第4引数：pxHigherPriorityTaskWoken（高優先度タスク起床フラグのポインタ）
  • 戻り値
    • 成功：実際に読み出せたバイト数
    • 失敗：0（未到着など）

設計のコツと注意点

注意1：1 writer / 1 reader 原提

StreamBufferは、複数タスクからの同時書き込み・読み出しに対して、内部でセマフォ等による排他制御を行っていません。

  • 黄金パターン: UART受信ISR (1 writer) → 処理タスク (1 reader)
  • もし複数タスクから送りたい場合は、Mutexでラップするか、送信専用タスクを用意してください。

注意2：トリガーレベルの「罠」

「100バイト溜まったら処理したい」とトリガーレベルを100に設定したが、送信側が50バイト送ったきり沈黙した場合、受信側は永遠に待ち続けます。

  • 対策1: xTicksToWait を有限にし、タイムアウト時に残りを処理する。
  • 対策2: xStreamBufferSetTriggerLevel() を使って、送信側が「これで終わり」と分かったタイミングで強制的にトリガーを下げる。

注意3：MessageBufferへの移行は容易

StreamBufferを使ってみて「やっぱりデータの区切り（パケット境界）が欲しいな」となったら、APIを xMessageBuffer... に書き換えるだけでほぼ対応可能です。

内部的には、MessageBufferは「データの先頭に長さを書いてStreamBufferに投げている」というラッパー構造だからです。

まとめ

  • StreamBufferは、境界を気にせず「生データ」を高速に流すための仕組み。
  • メモリ効率が最高で、管理用ヘッダの消費がない。
  • トリガーレベルを調整することで、不必要なコンテキストスイッチ（タスクの起床）を減らし、CPU負荷を下げられる。

コメント

タイトルとURLをコピーしました