Stream Buffer와 Message Buffer — FreeRTOS 10의 Lock-Free SPSC

#한 줄 요약

“Single producer + single consumer면 lock이 필요 없습니다.” — Stream Buffer는 그 가정을 받아들이는 대신 queue보다 훨씬 빠른 IPC를 제공합니다.

#어떤 문제를 푸는가

FreeRTOS의 xQueue 계열은 general-purpose IPC입니다. Multi-producer·multi-consumer를 모두 지원하기 위해 내부적으로 critical section과 task notification을 두루 씁니다. 한 번의 send/receive에 수백 ns에서 μs 가 들어갑니다.

그런데 임베디드에서 흔한 패턴은 훨씬 단순합니다.

• UART RX ISR → 데이터를 어떤 task에 넘기는 한 방향 스트림
• ADC DMA complete → consumer task 하나가 처리
• Sensor producer task → logger task 하나가 기록

생산자가 하나, 소비자가 하나입니다. 이 조건에서는 lock이 필요 없는 ring buffer만으로 충분합니다. Head와 tail 두 인덱스만 atomic하게 갱신하면 됩니다.

FreeRTOS 10(2017)이 추가한 Stream Buffer와 Message Buffer가 정확히 이 시장을 노립니다. SPSC(Single-Producer Single-Consumer) 가정을 받아들이는 대신 queue보다 10배 가까이 빠른 처리량을 보여 줍니다.

이 글에서는 두 buffer의 차이, 내부 SPSC 구조, ISR-safe API, 측정 결과, 그리고 SPSC 가정을 깨면 어떤 일이 벌어지는지를 살펴봅니다.

#Stream Buffer vs Message Buffer

Stream Buffer는 바이트가 그냥 흘러가는 채널입니다. 100 byte를 한 번에 write하고 receiver가 30 byte씩 세 번 read해도 자연스럽게 동작합니다.

Message Buffer는 메시지 경계를 보존합니다. 100 byte짜리 메시지를 보내면 receiver는 정확히 100 byte 한 단위로 받습니다. 내부적으로는 길이를 4 byte prefix로 저장해 두고 receive 때 그만큼만 잘라 줍니다.

1
Stream Buffer write 3회: [10B][20B][15B]
2
Stream Buffer read 1회:  [45B 한 덩어리] — OK
3
Message Buffer write 3회: [10B][20B][15B]
4
Message Buffer read 1회:  [10B] — 첫 메시지만, 다음 receive에서 다음 메시지

#SPSC Ring Buffer — 내부 구조

핵심은 head와 tail 두 인덱스가 서로 다른 task에서만 갱신된다는 점입니다.

1
typedef struct {
2
    uint8_t *buffer;          // 원형 저장소
3
    size_t   size;            // buffer 크기
4
    volatile size_t head;     // producer만 갱신, consumer는 read만
5
    volatile size_t tail;     // consumer만 갱신, producer는 read만
6
    // ... wait list 등
7
} StreamBuffer_t;

Producer는 head만 쓰고 tail은 읽기만 합니다. Consumer는 그 반대입니다. Cortex-M처럼 word-aligned int read/write가 atomic인 아키텍처에서는 이 패턴 자체로 안전합니다. Lock이 필요 없습니다.

1
// 단순화한 send (실제 코드 아님 — 개념만)
2
size_t stream_send(StreamBuffer_t *sb, const void *data, size_t n) {
3
    size_t space = sb->size - (sb->head - sb->tail);   // free space
4
    size_t to_copy = (n < space) ? n : space;
5
    /* head 위치부터 wrap-around 고려해서 copy */
6
    memcpy(sb->buffer + (sb->head % sb->size), data, to_copy);
7
    sb->head += to_copy;            // ← 마지막에 한 번 publish
8
    /* consumer task가 wait 중이면 notify */
9
    return to_copy;
10
}
11

12
size_t stream_receive(StreamBuffer_t *sb, void *out, size_t n) {
13
    size_t available = sb->head - sb->tail;
14
    size_t to_copy = (n < available) ? n : available;
15
    memcpy(out, sb->buffer + (sb->tail % sb->size), to_copy);
16
    sb->tail += to_copy;            // ← 마지막에 한 번 publish
17
    return to_copy;
18
}

head를 마지막에 한 번 갱신하는 게 핵심입니다. Producer가 데이터 복사를 마친 뒤에야 head가 진행하므로, consumer는 head < new_head 영역의 데이터를 항상 valid한 상태로 봅니다.

메모리 ordering이 약한 아키텍처(Cortex-A 등)에서는 head publish 전에 DMB가 필요합니다. FreeRTOS 포트가 자동으로 처리합니다.

#API 한눈에

1
// 생성
2
StreamBufferHandle_t xStreamBufferCreate(
3
    size_t xBufferSizeBytes,
4
    size_t xTriggerLevelBytes);
5

6
MessageBufferHandle_t xMessageBufferCreate(
7
    size_t xBufferSizeBytes);
8

9
// 송신 (task 컨텍스트)
10
size_t xStreamBufferSend(
11
    StreamBufferHandle_t xStreamBuffer,
12
    const void *pvTxData,
13
    size_t xDataLengthBytes,
14
    TickType_t xTicksToWait);
15

16
// 송신 (ISR 컨텍스트)
17
size_t xStreamBufferSendFromISR(
18
    StreamBufferHandle_t xStreamBuffer,
19
    const void *pvTxData,
20
    size_t xDataLengthBytes,
21
    BaseType_t *pxHigherPriorityTaskWoken);
22

23
// 수신
24
size_t xStreamBufferReceive(
25
    StreamBufferHandle_t xStreamBuffer,
26
    void *pvRxData,
27
    size_t xBufferLengthBytes,
28
    TickType_t xTicksToWait);

xTriggerLevelBytes가 흥미롭습니다. trigger level만큼 데이터가 모이기 전에는 consumer를 깨우지 않습니다. UART 1-byte 인터럽트가 매번 task wake-up을 시키면 overhead가 크니, 64 byte 모일 때까지 모았다가 한 번에 깨우는 식의 batching이 가능합니다.

#코드 — UART RX → Stream Buffer

가장 흔한 패턴입니다. UART 수신 ISR이 stream buffer에 byte를 밀어 넣고, consumer task가 batch로 받습니다.

1
StreamBufferHandle_t uart_rx_stream;
2

3
void app_init(void) {
4
    /* 1024 byte buffer, 64 byte trigger */
5
    uart_rx_stream = xStreamBufferCreate(1024, 64);
6
    xTaskCreate(uart_consumer_task, "rx", 512, NULL, 3, NULL);
7
}
8

9
// UART RX ISR
10
void USART1_IRQHandler(void) {
11
    uint8_t byte;
12
    BaseType_t higher_woken = pdFALSE;
13
    while (uart_rx_available()) {
14
        byte = uart_rx_read();
15
        xStreamBufferSendFromISR(uart_rx_stream, &byte, 1, &higher_woken);
16
    }
17
    portYIELD_FROM_ISR(higher_woken);
18
}
19

20
// Consumer task
21
void uart_consumer_task(void *arg) {
22
    uint8_t buf[128];
23
    for (;;) {
24
        size_t n = xStreamBufferReceive(uart_rx_stream, buf,
25
                                         sizeof(buf), portMAX_DELAY);
26
        process(buf, n);
27
    }
28
}

Trigger level 64 덕분에 ISR이 매 byte마다 yield하지 않습니다. 64 byte 모일 때만 consumer가 깨어나서 한 번에 처리합니다.

#코드 — Variable-length 명령 큐

Message Buffer는 가변 길이 명령을 주고받기에 좋습니다. Command parser → executor 같은 구조를 떠올려 보세요.

1
typedef struct {
2
    uint16_t cmd_id;
3
    uint16_t len;
4
    uint8_t  payload[];
5
} cmd_t;
6

7
MessageBufferHandle_t cmd_mb;
8

9
// Producer
10
void send_cmd(uint16_t id, const void *data, uint16_t len) {
11
    uint8_t buf[256];
12
    cmd_t *c = (cmd_t *)buf;
13
    c->cmd_id = id;
14
    c->len = len;
15
    memcpy(c->payload, data, len);
16
    xMessageBufferSend(cmd_mb, buf, sizeof(cmd_t) + len, portMAX_DELAY);
17
}
18

19
// Consumer
20
void executor_task(void *arg) {
21
    uint8_t buf[256];
22
    for (;;) {
23
        size_t n = xMessageBufferReceive(cmd_mb, buf,
24
                                          sizeof(buf), portMAX_DELAY);
25
        if (n > 0) {
26
            cmd_t *c = (cmd_t *)buf;
27
            dispatch(c->cmd_id, c->payload, c->len);
28
        }
29
    }
30
}

Receiver는 한 메시지 단위로 깨끗하게 받습니다. Queue로 같은 구조를 만들려면 고정 크기 item을 가정하거나 메시지를 외부 heap에 두고 포인터만 전달해야 합니다. Message Buffer는 데이터 자체를 들고 다니므로 lifetime 관리가 단순합니다.

#측정 — Queue 대비 처리량

Cortex-M4F @ 168 MHz에서 1-byte send/receive cycle을 측정한 결과입니다.

Stream Buffer가 queue 대비 4배 빠릅니다. Message Buffer도 3.5배 수준입니다. ISR 컨텍스트에서는 차이가 더 커집니다(critical section overhead가 줄어들기 때문).

100 KB/s 수준의 UART는 queue로 처리하면 *CPU 6%*를 소비하지만 stream buffer로 바꾸면 1.5% 미만으로 떨어집니다. Battery 기반 디바이스에서는 의미 있는 절감입니다.

#SPSC 가정 — 절대 깨지 마세요

Stream/Message Buffer의 모든 성능은 single producer + single consumer 가정 위에 서 있습니다. 이 가정을 깨면 조용히 데이터가 깨집니다. 컴파일러도 RTOS도 막아 주지 않습니다.

1
// ❌ 절대 금지 — 두 task가 동시에 send
2
void task_a(void *arg) {
3
    xStreamBufferSend(sb, "hello", 5, portMAX_DELAY);
4
}
5
void task_b(void *arg) {
6
    xStreamBufferSend(sb, "world", 5, portMAX_DELAY);
7
}
8
// → head를 동시에 갱신하다가 데이터 corruption

다중 producer가 필요하면 두 가지 선택지가 있습니다.

1. Mutex로 send를 직렬화 — 단, 성능 이점이 사라집니다.
2. Producer마다 stream buffer를 따로 만들고 consumer가 round-robin으로 모음.

대부분의 경우 2번이 깨끗합니다. Per-producer 채널이 디버깅도 더 쉽습니다.

한 가지 예외: ISR 하나 + task 하나는 SPSC로 간주됩니다. ISR이 task를 preempt해도 head/tail 갱신은 atomic합니다.

#ISR-safe Variant — FromISR

ISR 컨텍스트에서는 xStreamBufferSendFromISR()와 xStreamBufferReceiveFromISR()을 씁니다. 일반 버전을 ISR에서 부르면 crash합니다.

1
void DMA_IRQHandler(void) {
2
    BaseType_t higher_woken = pdFALSE;
3
    xStreamBufferSendFromISR(audio_stream, dma_buf, BUF_SIZE, &higher_woken);
4
    portYIELD_FROM_ISR(higher_woken);   // 더 높은 priority task 깨우면 즉시 switch
5
}

higher_woken이 consumer task가 깨어났는지 여부입니다. 그게 true면 portYIELD_FROM_ISR이 ISR 종료 직후 바로 context switch를 trigger합니다. ISR latency가 짧아지고 consumer는 지연 없이 데이터를 받습니다.

#자주 보는 함정과 안티패턴

⚠️ SPSC 가정 위반

위에서 다룬 가장 큰 함정입니다. 조용히 데이터가 깨지므로 코드 리뷰에서 producer/consumer가 정확히 하나씩인지 확인합니다. 새 task가 send 라인을 추가하는 순간 무너집니다.

⚠️ Buffer 크기 부족

Trigger level + 1회 send 크기보다 작은 buffer를 만들면 영원히 trigger되지 않습니다. 보통 trigger × 4 이상으로 잡습니다.

⚠️ Timeout 처리 무시

1
size_t n = xStreamBufferReceive(sb, buf, sizeof(buf), pdMS_TO_TICKS(100));
2
process(buf, n);   // ← n이 0일 수 있음

Timeout이 끝나면 partial 데이터 또는 0이 반환됩니다. 반환값 확인 없이 처리하면 garbage를 만집니다.

⚠️ Message Buffer의 4-byte prefix를 잊음

Buffer 크기를 계산할 때 메시지마다 4 byte 길이 prefix가 추가됩니다. 100 byte 메시지 10개를 담으려면 1040 byte 이상 필요합니다.

⚠️ Task 컨텍스트와 ISR 컨텍스트 혼용

1
// task에서
2
xStreamBufferSend(sb, ...);
3
// ISR에서 — 다른 API!
4
xStreamBufferSendFromISR(sb, ...);

이름이 다르므로 컴파일러가 잡아 줍니다. 하지만 같은 stream에 task와 ISR이 동시에 send하는 건 여전히 SPSC 위반입니다. ISR + task receiver, 또는 task + ISR receiver처럼 방향이 다른 쌍이어야 합니다.

⚠️ Stream Buffer를 frame 단위처럼 사용

Stream Buffer는 경계가 없습니다. 100 byte frame을 보내면 receiver가 50 byte씩 두 번 받을 수 있습니다. 경계가 필요하면 Message Buffer를 씁니다.

#정리

• Stream/Message Buffer는 FreeRTOS 10에서 추가된 SPSC lock-free IPC입니다.
• Stream Buffer는 byte stream, Message Buffer는 경계 보존 variable-length 메시지를 다룹니다.
• 내부는 head/tail 두 인덱스만 갱신하는 ring buffer로, lock 없이 동작합니다.
• 1-byte 처리량에서 queue 대비 3~4배 빠르고 CPU 사용량도 그만큼 줄어듭니다.
• Single producer + single consumer 가정은 절대 위반하면 안 됩니다.
• ISR 컨텍스트에서는 반드시 FromISR variant를 씁니다.
• Trigger level로 ISR wake-up을 batch해서 추가 overhead를 줄일 수 있습니다.

다음 편은 3-12: Task Notification — 가벼운 동기화 원시 타입을 다룹니다.

#관련 항목

• 3-07: Queue 구현
• 3-10: Deadlock
• 4-04: Lock-free Container
• 4-07: Lock-free 자료구조