큐와 메시지 패싱 — Producer-Consumer·Ring Buffer·전달 의미

1
SemaphoreHandle_t mtx;
2
SharedBuffer_t shared;
3

4
void producer(void *arg) {
5
    while (1) {
6
        Data_t data = read_sensor();
7
        xSemaphoreTake(mtx, portMAX_DELAY);
8
        shared.data = data;            // ← race condition 위험
9
        shared.ready = 1;
10
        xSemaphoreGive(mtx);
11
    }
12
}
13

14
void consumer(void *arg) {
15
    while (1) {
16
        xSemaphoreTake(mtx, portMAX_DELAY);
17
        if (shared.ready) {
18
            Data_t data = shared.data;
19
            shared.ready = 0;
20
        }
21
        xSemaphoreGive(mtx);
22
    }
23
}

1
QueueHandle_t q = xQueueCreate(10, sizeof(Data_t));
2

3
void producer(void *arg) {
4
    while (1) {
5
        Data_t data = read_sensor();
6
        xQueueSend(q, &data, portMAX_DELAY);
7
    }
8
}
9

10
void consumer(void *arg) {
11
    Data_t data;
12
    while (1) {
13
        xQueueReceive(q, &data, portMAX_DELAY);  // 자동 block
14
        process(data);
15
    }
16
}

1
typedef struct QueueDefinition {
2
    int8_t *pcHead;             // buffer 시작
3
    int8_t *pcTail;             // buffer 끝
4
    int8_t *pcWriteTo;          // 다음 write 위치
5
    int8_t *pcReadFrom;         // 다음 read 위치
6
    UBaseType_t uxMessagesWaiting;  // 현재 항목 수
7
    UBaseType_t uxLength;       // 최대 항목 수
8
    UBaseType_t uxItemSize;     // 각 항목 byte
9
    List_t xTasksWaitingToSend;     // queue full 대기
10
    List_t xTasksWaitingToReceive;  // queue empty 대기
11
} Queue_t;

1
BaseType_t xQueueSend(QueueHandle_t q, const void *item, TickType_t timeout) {
2
    portENTER_CRITICAL();
3
    if (q->uxMessagesWaiting < q->uxLength) {
4
        memcpy(q->pcWriteTo, item, q->uxItemSize);
5
        q->pcWriteTo += q->uxItemSize;
6
        if (q->pcWriteTo >= q->pcTail) q->pcWriteTo = q->pcHead;  // wrap
7
        q->uxMessagesWaiting++;
8
        // Receive waiter 깨움
9
        wake_first_waiter(&q->xTasksWaitingToReceive);
10
        portEXIT_CRITICAL();
11
        return pdTRUE;
12
    }
13
    // Full — block
14
    add_to_wait_list(&q->xTasksWaitingToSend, current_task);
15
    portEXIT_CRITICAL();
16
    block_with_timeout(timeout);
17
    /* ... 재시도 또는 timeout return */
18
}

1
typedef struct {
2
    uint8_t buf[1024];          // 1 KB
3
    int size;
4
} BigMsg_t;
5

6
// 매 send마다 1024 byte memcpy
7
QueueHandle_t q = xQueueCreate(5, sizeof(BigMsg_t));

1
QueueHandle_t q = xQueueCreate(5, sizeof(BigMsg_t *));
2

3
void producer(void *arg) {
4
    BigMsg_t *msg = pool_alloc();   // 메모리 풀에서
5
    fill(msg);
6
    xQueueSend(q, &msg, portMAX_DELAY);   // pointer만
7
}
8

9
void consumer(void *arg) {
10
    BigMsg_t *msg;
11
    while (1) {
12
        xQueueReceive(q, &msg, portMAX_DELAY);
13
        process(msg);
14
        pool_free(msg);                  // 수신 후 해제
15
    }
16
}

1
StreamBufferHandle_t sb = xStreamBufferCreate(256, 1);
2
xStreamBufferSend(sb, data, len, timeout);
3
xStreamBufferReceive(sb, buf, sizeof(buf), timeout);

1
MessageBufferHandle_t mb = xMessageBufferCreate(1024);
2
xMessageBufferSend(mb, "Hello", 5, timeout);
3
xMessageBufferSend(mb, "World!", 6, timeout);
4
// Receive 시 정확히 5 byte, 그 다음 6 byte

1
xQueueOverwrite(q, &data);   // 큐 full이어도 OK, 옛것 덮어씀
2
xQueuePeek(q, &data, 0);     // 비파괴 read

1
// 작은 데이터 (sizeof(Data_t) ≤ 32 byte)
2
xQueueSend(q, &data, ...);   // memcpy 32 byte ≈ 8 cycle on 32-bit MCU
3

4
// 큰 데이터 (1 KB)
5
xQueueSend(q, &big, ...);    // memcpy 1 KB ≈ 250 cycle

1
// ISR
2
BaseType_t woken = pdFALSE;
3
xQueueSendFromISR(q, &data, &woken);
4
portYIELD_FROM_ISR(woken);

1
StaticQueue_t q_buf;
2
uint8_t q_storage[10 * sizeof(Data_t)];
3

4
QueueHandle_t q = xQueueCreateStatic(
5
    10,                  // length
6
    sizeof(Data_t),      // item size
7
    q_storage,           // storage buffer
8
    &q_buf               // queue control block
9
);

1
xQueueSend(q, &ptr, ...);
2
free(ptr);   // ← 수신자가 read 전에 free → crash