Queue 내부 구현 추적 — Ring Buffer·2 Wait Lists·Atomic Send/Receive

1
typedef struct QueueDefinition {
2
    int8_t *pcHead;                  // buffer 시작
3
    int8_t *pcWriteTo;               // 다음 write 위치
4
    union {
5
        int8_t *pcReadFrom;          // Queue 일반
6
        UBaseType_t uxRecursiveCallCount;   // Mutex 시
7
    } u;
8

9
    List_t xTasksWaitingToSend;      // queue full 대기
10
    List_t xTasksWaitingToReceive;   // queue empty 대기
11

12
    volatile UBaseType_t uxMessagesWaiting;
13
    UBaseType_t uxLength;            // max items
14
    UBaseType_t uxItemSize;          // bytes/item
15

16
    volatile int8_t cRxLock;         // ISR 안전성용
17
    volatile int8_t cTxLock;
18

19
    uint8_t ucStaticallyAllocated;
20
    uint8_t ucQueueType;
21
} Queue_t;

1
BaseType_t xQueueGenericSend(QueueHandle_t q, const void *item,
2
                              TickType_t xTicksToWait, BaseType_t xCopyPosition) {
3
    portENTER_CRITICAL();
4

5
    if (q->uxMessagesWaiting < q->uxLength || xCopyPosition == queueOVERWRITE) {
6
        /* Space available — copy item */
7
        prvCopyDataToQueue(q, item, xCopyPosition);
8

9
        /* Wake highest-priority receiver if any */
10
        if (listLIST_IS_EMPTY(&q->xTasksWaitingToReceive) == pdFALSE) {
11
            if (xTaskRemoveFromEventList(&q->xTasksWaitingToReceive) != pdFALSE) {
12
                /* Higher priority — yield */
13
                queueYIELD_IF_USING_PREEMPTION();
14
            }
15
        }
16
        portEXIT_CRITICAL();
17
        return pdPASS;
18
    }
19

20
    /* Queue full */
21
    if (xTicksToWait == 0) {
22
        portEXIT_CRITICAL();
23
        return errQUEUE_FULL;
24
    }
25

26
    /* Block on send */
27
    vTaskPlaceOnEventList(&q->xTasksWaitingToSend, xTicksToWait);
28
    portEXIT_CRITICAL();
29
    portYIELD_WITHIN_API();
30
    /* ... 깨어난 후 재시도 ... */
31
}

1
static void prvCopyDataToQueue(Queue_t *q, const void *src, BaseType_t pos) {
2
    if (q->uxItemSize == 0) {
3
        /* Semaphore mode — counter만 사용 */
4
        if (q->ucQueueType == queueQUEUE_TYPE_MUTEX) {
5
            /* Mutex give */
6
            xTaskPriorityDisinherit(q->pxMutexHolder);
7
            q->pxMutexHolder = NULL;
8
        }
9
    } else if (pos == queueSEND_TO_BACK) {
10
        memcpy(q->pcWriteTo, src, q->uxItemSize);
11
        q->pcWriteTo += q->uxItemSize;
12
        if (q->pcWriteTo >= q->pcTail) {
13
            q->pcWriteTo = q->pcHead;   // wrap
14
        }
15
    } else if (pos == queueSEND_TO_FRONT) {
16
        q->u.pcReadFrom -= q->uxItemSize;
17
        if (q->u.pcReadFrom < q->pcHead) {
18
            q->u.pcReadFrom = q->pcTail - q->uxItemSize;   // wrap
19
        }
20
        memcpy(q->u.pcReadFrom, src, q->uxItemSize);
21
    } else if (pos == queueOVERWRITE) {
22
        /* Mailbox mode */
23
        memcpy(q->pcWriteTo, src, q->uxItemSize);
24
        /* uxMessagesWaiting 증가시키지 않음 */
25
        return;
26
    }
27
    q->uxMessagesWaiting++;
28
}

1
BaseType_t xQueueReceive(QueueHandle_t q, void *buffer, TickType_t xTicksToWait) {
2
    portENTER_CRITICAL();
3

4
    if (q->uxMessagesWaiting > 0) {
5
        prvCopyDataFromQueue(q, buffer);
6
        q->uxMessagesWaiting--;
7

8
        /* Sender waiter wake */
9
        if (!list_empty(&q->xTasksWaitingToSend)) {
10
            xTaskRemoveFromEventList(&q->xTasksWaitingToSend);
11
        }
12
        portEXIT_CRITICAL();
13
        return pdPASS;
14
    }
15

16
    /* Empty */
17
    if (xTicksToWait == 0) {
18
        portEXIT_CRITICAL();
19
        return errQUEUE_EMPTY;
20
    }
21

22
    vTaskPlaceOnEventList(&q->xTasksWaitingToReceive, xTicksToWait);
23
    portEXIT_CRITICAL();
24
    portYIELD_WITHIN_API();
25
    /* ... */
26
}

1
volatile int8_t cRxLock;   // -1 = unlocked, 0+ = ISR queued sends count
2
volatile int8_t cTxLock;

1
xQueueSendFromISR(...) {
2
    /* ... copy item ... */
3
    if (q->cTxLock == queueUNLOCKED) {
4
        /* Scheduler running — wake 직접 수행 */
5
        wake_receiver();
6
    } else {
7
        /* Locked — count up */
8
        q->cTxLock++;
9
    }
10
}

1
queueSEND_TO_BACK    // 일반 FIFO
2
queueSEND_TO_FRONT   // LIFO 동작
3
queueOVERWRITE       // Mailbox (1 slot, 덮어쓰기)

1
Send / Receive (empty wait list):
2
- portENTER_CRITICAL: 5 cycle
3
- check queue state: 10 cycle
4
- memcpy item: itemSize / 4 × 1 cycle
5
- wake check: 10 cycle
6
- portEXIT_CRITICAL: 5 cycle
7

8
Total ≈ 30 + itemSize/4 cycle

1
StreamBufferHandle_t sb = xStreamBufferCreate(256, 1);
2
xStreamBufferSend(sb, data, len, timeout);

1
xMessageBufferSend(mb, msg, msg_len, timeout);