【freertos】010-訊息佇列概念及其實現細節

前言

訊息佇列是任務間通訊系列介紹的首篇筆記，因為學習完訊息佇列的原始碼實現後，號誌、互斥量這些任務間通訊機制也相當於學完了，只剩下概念性的內容了。

參考：

• https://www.freertos.org/a00018.html
• 李柱明部落格：https://www.cnblogs.com/lizhuming/p/16344076.html

10.1 訊息佇列概念

訊息佇列實任務間通訊機制中的一種。

其它還有二值號誌、計數號誌、互斥量和遞迴互斥量等等。

一個或多個任務往一個訊息容器裡面發訊息，其它一個或多個任務從這個訊息容器裡面獲取訊息，這樣實現通訊。
/* 該圖源自野火 */

freertos的訊息佇列：

• 支援FIFO、支援LIFO也支援非同步讀寫工作方式。
• 支援超時機制。
• 支援不同長度（在節點長度範圍內）、任意型別的訊息。
• 一個任務可對一個訊息佇列讀、寫。
• 一個訊息佇列支援被多個任務讀、寫。
• 佇列使用一次後自動從訊息佇列中移除。

10.2 訊息佇列的資料傳輸機制

佇列傳輸資料有兩種方式：

1. 拷貝：把資料、把變數的值複製進佇列裡。
2. 參照：把資料、把變數的地址複製進佇列裡。

而freertos的訊息佇列機制就是拷貝，拷貝的方式有以下優點：

• 區域性變數的值可以傳送到佇列中，後續即使函數退出、區域性變數被回收，也不會影響佇列中的資料。
• 無需分配buffer來儲存資料，佇列中有buffer。
• 傳送任務、接收任務解耦：接收任務不需要知道這資料是誰的、也不需要傳送任務來釋放資料。
• 如果資料實在太大，可以選擇傳輸地址（即是拷貝地址），依然能實現傳輸參照的效果。
• 佇列的空間有FreeRTOS核心分配，無需上層應用維護。
• 無需考慮記憶體保護功能，因為拷貝的方式新資料的儲存區是由佇列元件提供的，無需擔心獲取訊息的任務需要許可權存取。

當然對比參照的方式也有劣勢：

1. 拷貝資料相對拷貝參照來說要耗時。
2. 需要更多記憶體，因為需要儲存資料副本。

10.3 訊息佇列的阻塞存取機制

只要拿到佇列控制程式碼，任務和中斷都有許可權存取訊息佇列，但是也有阻塞限制。

寫訊息時，如果訊息佇列已滿，則無法寫入（覆蓋寫入除外），如果使用者設定的阻塞時間不為0，則任務會進入阻塞，直到該佇列有空閒空間給當前任務寫入訊息或阻塞時間超時才解除阻塞。

上面說的「該佇列有空閒空間給當前任務寫入訊息」是因為就算當前佇列有空間空間，也會優先安排阻塞在等待寫連結串列中的最高優先順序任務先寫入。如果任務優先順序相同，則先安排給最早開始等待的那個任務先寫。

讀訊息時，機制和寫訊息一樣，只是阻塞的條件是佇列裡面沒有訊息。

資料傳輸和阻塞存取機制都會在分析原始碼時闡述。

10.4 訊息佇列使用場景

訊息佇列可以應用於傳送不定長訊息的場合，包括任務與任務間的訊息交換。

佇列是FreeRTOS主要的任務間通訊方式，可以在任務與任務間、中斷和任務間傳送資訊。

傳送到佇列的訊息是通過拷貝方式實現的，這意味著佇列儲存的資料是原資料，而不是原資料的參照。

10.5 訊息佇列控制塊

訊息佇列的控制塊也是佇列控制塊，這個控制塊的資料結構除了被訊息佇列使用，還被使用到二值號誌、計數號誌、互斥量和遞迴互斥量。

FreeRTOS的訊息佇列控制塊由多個元素組成，當訊息佇列被建立時，系統會為控制塊分配對應的記憶體空間，用於儲存訊息佇列的一些資訊，包括資料區位置、佇列狀態等等。

10.5.1 佇列控制塊原始碼

佇列控制塊struct QueueDefinition原始碼：

/*
 * Definition of the queue used by the scheduler.
 * Items are queued by copy, not reference.  See the following link for the
 * rationale: http://www.freertos.org/Embedded-RTOS-Queues.html
 */
typedef struct QueueDefinition
{
	int8_t *pcHead;					/*< Points to the beginning of the queue storage area. */
	int8_t *pcTail;					/*< Points to the byte at the end of the queue storage area.  Once more byte is allocated than necessary to store the queue items, this is used as a marker. */
	int8_t *pcWriteTo;				/*< Points to the free next place in the storage area. */

	union						/* Use of a union is an exception to the coding standard to ensure two mutually exclusive structure members don't appear simultaneously (wasting RAM). */
	{
		int8_t *pcReadFrom;			/*< Points to the last place that a queued item was read from when the structure is used as a queue. */
		UBaseType_t uxRecursiveCallCount;/*< Maintains a count of the number of times a recursive mutex has been recursively 'taken' when the structure is used as a mutex. */
	} u;

	List_t xTasksWaitingToSend;		/*< List of tasks that are blocked waiting to post onto this queue.  Stored in priority order. */
	List_t xTasksWaitingToReceive;	/*< List of tasks that are blocked waiting to read from this queue.  Stored in priority order. */

	volatile UBaseType_t uxMessagesWaiting;/*< The number of items currently in the queue. */
	UBaseType_t uxLength;			/*< The length of the queue defined as the number of items it will hold, not the number of bytes. */
	UBaseType_t uxItemSize;			/*< The size of each items that the queue will hold. */

	volatile int8_t cRxLock;		/*< Stores the number of items received from the queue (removed from the queue) while the queue was locked.  Set to queueUNLOCKED when the queue is not locked. */
	volatile int8_t cTxLock;		/*< Stores the number of items transmitted to the queue (added to the queue) while the queue was locked.  Set to queueUNLOCKED when the queue is not locked. */

	#if( ( configSUPPORT_STATIC_ALLOCATION == 1 ) && ( configSUPPORT_DYNAMIC_ALLOCATION == 1 ) )
		uint8_t ucStaticallyAllocated;	/*< Set to pdTRUE if the memory used by the queue was statically allocated to ensure no attempt is made to free the memory. */
	#endif

	#if ( configUSE_QUEUE_SETS == 1 )
		struct QueueDefinition *pxQueueSetContainer;
	#endif

	#if ( configUSE_TRACE_FACILITY == 1 )
		UBaseType_t uxQueueNumber;
		uint8_t ucQueueType;
	#endif

} xQUEUE;

10.5.2 佇列控制塊成員剖析

在成員剖析時預設按訊息佇列的作用去剖析。

int8_t *pcHead;:

/* 該佇列儲存區的起始位置，對應第一個訊息空間。*/
int8_t *pcHead;

int8_t *pcTail;:

/* 訊息佇列儲存區的結尾位置。
 * 結合 pcHead 指標就是整個儲存區合法區域。*/
int8_t *pcHead;

int8_t *pcWriteTo;:

/* 寫指標，指向儲存區中下一個空閒的空間。
 * 佇列下次寫資料的位置，需要入隊時呼叫該指標寫入資料。*/
int8_t *pcWriteTo;

int8_t *pcReadFrom;:

/* 讀指標，指向儲存區中下一個有效資料的空間。
 * 佇列下次讀取資料的位置，需要出隊時呼叫該指標寫入資料。*/
int8_t *pcReadFrom;

UBaseType_t uxRecursiveCallCount;:

/* 遞迴次數。
 * 用於互斥量時使用，與 pcReadFrom 為聯合體。
 * 記錄遞迴互斥量被呼叫的次數。 */
UBaseType_t uxRecursiveCallCount;

List_t xTasksWaitingToSend;:

/* 等待傳送的任務列表。
 * 當佇列儲存區滿時，需要傳送訊息的任務阻塞時記錄到該連結串列。
 * 按任務優先順序排序。 */
List_t xTasksWaitingToSend;

List_t xTasksWaitingToReceive;:

/* 等待接收的任務列表。
 * 當佇列儲存區為空時，需要獲取訊息的任務阻塞時記錄到該連結串列。
 * 按任務優先順序排序。 */
List_t xTasksWaitingToReceive;

volatile UBaseType_t uxMessagesWaiting;:

/* 當前訊息節點的個數。
 * 即是當前有效訊息數量。
 * 二值號誌、互斥號誌時：表示有無號誌可用。
 * 計數號誌時：有效號誌個數。 */
volatile UBaseType_t uxMessagesWaiting;

UBaseType_t uxLength;:

/* 當前佇列最大節點總數。
 * 即是最多能存放多少個訊息。
 * 二值號誌、互斥號誌時：最大為1。
 * 計數號誌時：最大的號誌個數。 */
UBaseType_t uxLength;

UBaseType_t uxItemSize;:

/* 單個節點的大小。
 * 單個訊息的大小。
 * 二值號誌、互斥號誌時：0。
 * 計數號誌時：0。 */
UBaseType_t uxItemSize;

volatile int8_t cRxLock;:

/* 記錄出隊的資料項個數。
 * 即是需要解除多少個阻塞在接收等待列表中的任務。 */
volatile int8_t cRxLock;

volatile int8_t cTxLock;:

/* 記錄入隊的資料項個數。
 * 即是需要解除多少個阻塞在傳送等待列表中的任務。 */
volatile int8_t cTxLock;

10.5.3 cRxLock 和 cTxLock

當中斷服務程式操作佇列並且導致阻塞的任務解除阻塞時。
首先判斷該佇列是否上鎖：

• 如果沒有上鎖，則解除被阻塞的任務，還會根據需要設定上下文切換請求標誌；
• 如果佇列已經上鎖，則不會解除被阻塞的任務，取而代之的是，將xRxLock或xTxLock加1，表示佇列上鎖期間出隊或入隊的數目，也表示有任務可以解除阻塞了。

cRxLock 對應待出隊的個數。
cTxLock 對應待入隊的個數。

10.5.4 佇列控制塊資料結構圖

10.6 建立訊息佇列

建立訊息佇列是在系統上新建一個訊息佇列，申請資源並初始化後返回控制程式碼給使用者，使用者可以使用該佇列控制程式碼存取、操作該佇列。

10.6.1 建立訊息佇列API說明

佇列的建立有兩種方法：靜態分配記憶體、動態分配記憶體。其區別就是佇列的記憶體來源是使用者提供的還是核心分配的。

主要分析動態分配記憶體。

函數原型：

QueueHandle_t xQueueCreate( UBaseType_t uxQueueLength, UBaseType_t uxItemSize );

引數說明：

• uxQueueLength：佇列長度，最多能存放多少個資料(item)。

• uxItemSize：每個資料(item)的大小：以位元組為單位。

• 返回值：

  • 非0：成功，返回控制程式碼，以後使用控制程式碼來操作佇列。
  • NULL：失敗，因為記憶體不足。

10.6.2 建立訊息佇列簡要步驟

1. 引數校驗。
2. 計算本次佇列需要的總記憶體。
3. 分配佇列記憶體空間。
4. 初始化佇列控制塊。
5. 格式化佇列資料區。
6. 返回佇列控制程式碼。

10.6.3 建立訊息佇列原始碼

建立訊息佇列這個API其實就是封裝了建立佇列xQueueGenericCreate()這個通用API，型別為queueQUEUE_TYPE_BASE。

#if ( configSUPPORT_DYNAMIC_ALLOCATION == 1 )
    #define xQueueCreate( uxQueueLength, uxItemSize )    xQueueGenericCreate( ( uxQueueLength ), ( uxItemSize ), ( queueQUEUE_TYPE_BASE ) )
#endif

其中佇列的型別有多種：

/* For internal use only.  These definitions *must* match those in queue.c. */
#define queueQUEUE_TYPE_BASE                  ( ( uint8_t ) 0U )  // 佇列型別
#define queueQUEUE_TYPE_SET                   ( ( uint8_t ) 0U )  // 佇列集合型別
#define queueQUEUE_TYPE_MUTEX                 ( ( uint8_t ) 1U )  // 互斥量型別
#define queueQUEUE_TYPE_COUNTING_SEMAPHORE    ( ( uint8_t ) 2U )  // 計數號誌型別
#define queueQUEUE_TYPE_BINARY_SEMAPHORE      ( ( uint8_t ) 3U )  // 二進位制號誌型別
#define queueQUEUE_TYPE_RECURSIVE_MUTEX       ( ( uint8_t ) 4U )  // 遞迴互斥量型別

建立佇列函數原始碼xQueueGenericCreateStatic()：

#if ( configSUPPORT_DYNAMIC_ALLOCATION == 1 )
    QueueHandle_t xQueueGenericCreate( const UBaseType_t uxQueueLength,
                                       const UBaseType_t uxItemSize,
                                       const uint8_t ucQueueType )
    {
        Queue_t * pxNewQueue = NULL;
        size_t xQueueSizeInBytes;
        uint8_t * pucQueueStorage;

        if( ( uxQueueLength > ( UBaseType_t ) 0 ) &&
            /* 檢查需要的資料區size是否溢位限定範圍 */
            ( ( SIZE_MAX / uxQueueLength ) >= uxItemSize ) &&
            /* 檢查本次佇列建立需要的空間是否溢位限定範圍 */
            ( ( SIZE_MAX - sizeof( Queue_t ) ) >= ( uxQueueLength * uxItemSize ) ) )
        {
            /* 計算資料區空間。
               如果佇列建立的是不帶資料的，如號誌、互斥量，則傳入引數時uxItemSize值應該被置為0。 */
            xQueueSizeInBytes = ( size_t ) ( uxQueueLength * uxItemSize );

            /* 一次性分配佇列所需要的空間，包括佇列控制塊和資料區 */
            pxNewQueue = ( Queue_t * ) pvPortMalloc( sizeof( Queue_t ) + xQueueSizeInBytes );

            if( pxNewQueue != NULL )
            {
                /* 找出資料區起始地址 */
                pucQueueStorage = ( uint8_t * ) pxNewQueue;
                pucQueueStorage += sizeof( Queue_t );

                #if ( configSUPPORT_STATIC_ALLOCATION == 1 )
                    {
                        /* 如果系統使能了靜態建立功能，就需要標記當前佇列是動態建立，記憶體有核心管理，以防使用者刪除。 */
                        pxNewQueue->ucStaticallyAllocated = pdFALSE;
                    }
                #endif /* configSUPPORT_STATIC_ALLOCATION */
                /* 初始化這個佇列 */
                prvInitialiseNewQueue( uxQueueLength, uxItemSize, pucQueueStorage, ucQueueType, pxNewQueue );
            }
            else
            {
                traceQUEUE_CREATE_FAILED( ucQueueType );
                mtCOVERAGE_TEST_MARKER();
            }
        }
        else
        {
            configASSERT( pxNewQueue );
            mtCOVERAGE_TEST_MARKER();
        }
        /* 返回佇列起始地址，便是佇列控制程式碼 */
        return pxNewQueue;
    }
#endif /* configSUPPORT_STATIC_ALLOCATION */

初始化佇列函數原始碼prvInitialiseNewQueue()：

• 小筆記：初始化佇列，看原始碼實現就知道控制塊和資料區實體記憶體是可以分開的，但是在建立訊息佇列這個API裡面實現是連續的。

static void prvInitialiseNewQueue( const UBaseType_t uxQueueLength,
                                   const UBaseType_t uxItemSize,
                                   uint8_t * pucQueueStorage,
                                   const uint8_t ucQueueType,
                                   Queue_t * pxNewQueue )
{
    /* 防止編譯時警告未使用 */
    ( void ) ucQueueType;

    if( uxItemSize == ( UBaseType_t ) 0 )
    {
        /* 如果沒有資料區（如號誌、互斥量等等），就需要把佇列中的pcHead指回當前佇列控制塊起始地址，表明當前佇列不含資料區。 */
        pxNewQueue->pcHead = ( int8_t * ) pxNewQueue;
    }
    else
    {
        /* 如果當前佇列含有資料區，則把 */
        pxNewQueue->pcHead = ( int8_t * ) pucQueueStorage;
    }

    /* 儲存當前佇列成員數量 */
    pxNewQueue->uxLength = uxQueueLength;
    /* 儲存當前佇列每個成員的最大size */
    pxNewQueue->uxItemSize = uxItemSize;
    /* 佇列格式化。（組成一個介面是因為不僅僅在這裡用到重置佇列的功能） */
    ( void ) xQueueGenericReset( pxNewQueue, pdTRUE );

    #if ( configUSE_TRACE_FACILITY == 1 )
        {
            /* 記錄當前佇列型別。一般用於偵錯、查棧使用。 */
            pxNewQueue->ucQueueType = ucQueueType;
        }
    #endif /* configUSE_TRACE_FACILITY */

    #if ( configUSE_QUEUE_SETS == 1 )
        {
            pxNewQueue->pxQueueSetContainer = NULL;
        }
    #endif /* configUSE_QUEUE_SETS */

    traceQUEUE_CREATE( pxNewQueue );
}

重置佇列函數xQueueGenericReset()：

• 專門用於函資料區的佇列，如訊息佇列。

BaseType_t xQueueGenericReset( QueueHandle_t xQueue,
                               BaseType_t xNewQueue )
{
    BaseType_t xReturn = pdPASS;
    Queue_t * const pxQueue = xQueue;
    /* 引數校驗 */
    configASSERT( pxQueue );

    if( ( pxQueue != NULL ) &&
        ( pxQueue->uxLength >= 1U ) && /* 佇列成員數不能小於1，要不然算引數校驗失敗 */
        ( ( SIZE_MAX / pxQueue->uxLength ) >= pxQueue->uxItemSize ) ) /* 佇列size溢位檢查 */
    {
        taskENTER_CRITICAL(); /* 進入任務臨界 */
        {
            /* 儲存整個佇列尾部的地址，和pxQueue->pcHead結合看，就是這個佇列的合法空間首尾 */
            pxQueue->u.xQueue.pcTail = pxQueue->pcHead + ( pxQueue->uxLength * pxQueue->uxItemSize );
            /* 重置當前有效訊息數量 */
            pxQueue->uxMessagesWaiting = ( UBaseType_t ) 0U;
            /* 重置寫指標，指向第一個佇列成員 */
            pxQueue->pcWriteTo = pxQueue->pcHead;
            /* 重置讀指標，指向最後一個佇列成員。因為下次讀前要先偏移讀指標。 */
            pxQueue->u.xQueue.pcReadFrom = pxQueue->pcHead + ( ( pxQueue->uxLength - 1U ) * pxQueue->uxItemSize );
            /* 重置訊息佇列讀鎖：開鎖狀態 */
            pxQueue->cRxLock = queueUNLOCKED;
            /* 重置訊息佇列寫鎖：開鎖狀態 */
            pxQueue->cTxLock = queueUNLOCKED;

            if( xNewQueue == pdFALSE ) /* 重置已經在使用的佇列 */
            {
                /* 因為重置佇列相當於清空佇列裡面的資料，佇列有空位可寫入，所以可以解除一個寫阻塞任務 */
                /* 如果有任務因為當前佇列而寫阻塞的，可以解除 */
                if( listLIST_IS_EMPTY( &( pxQueue->xTasksWaitingToSend ) ) == pdFALSE )
                {
                    /* 解除一個寫阻塞任務 */
                    if( xTaskRemoveFromEventList( &( pxQueue->xTasksWaitingToSend ) ) != pdFALSE )
                    {
                        /* 如果內部解鎖了個比當前優先順序還高的任務，就觸發一次工作切換。（當然，實際自信還是在退出任務臨界才會執行） */
                        queueYIELD_IF_USING_PREEMPTION();
                    }
                    else
                    {
                        mtCOVERAGE_TEST_MARKER();
                    }
                }
                else
                {
                    mtCOVERAGE_TEST_MARKER();
                }
            }
            else /* 重置的是一個新的佇列 */
            {
                /* 直接初始化寫阻塞任務連結串列和讀阻塞任務連結串列. */
                vListInitialise( &( pxQueue->xTasksWaitingToSend ) );
                vListInitialise( &( pxQueue->xTasksWaitingToReceive ) );
            }
        }
        taskEXIT_CRITICAL(); /* 退出任務臨界 */
    }
    else
    {
        xReturn = pdFAIL;
    }
    /* 前面if裡面引數校驗失敗會直接斷言 */
    configASSERT( xReturn != pdFAIL );

    return xReturn;
}

10.6.4 訊息佇列資料結構圖

10.7 傳送訊息

任務或者中斷服務程式都可以給訊息佇列傳送訊息。

中斷中傳送訊息不可阻塞。要麼直接返回，要麼覆蓋寫入。

任務傳送訊息時，如果佇列未滿或者允許覆蓋入隊，FreeRTOS會將訊息拷貝到訊息佇列隊尾或佇列頭，否則，會根據使用者指定的阻塞超時時間進行阻塞。直到該佇列有空閒空間給當前任務寫入訊息或阻塞時間超時才解除阻塞。

傳送訊息的API分任務和中斷專屬，中斷專用的API都帶FromISR字尾。

因為本系列筆記主要記錄原始碼實現，API的使用不會詳細列舉。

10.7.1 傳送訊息API

/* 往佇列尾部寫入資料。等同於xQueueSendToBack */
BaseType_t xQueueSend(QueueHandle_t xQueue, const void *pvItemToQueue, TickType_t xTicksToWait);

/* 往佇列尾部寫入資料。等同於xQueueSend */
BaseType_t xQueueSendToBack(QueueHandle_t xQueue, const void *pvItemToQueue, TickType_t xTicksToWait);

/* 往佇列尾部寫入資料。中斷專用 */
BaseType_t xQueueSendToBackFromISR(QueueHandle_t xQueue, const void *pvItemToQueue, BaseType_t *pxHigherPriorityTaskWoken);

/* 往佇列頭部寫入資料 */
BaseType_t xQueueSendToFront(QueueHandle_t xQueue, const void *pvItemToQueue, TickType_t xTicksToWait);

/* 往佇列頭部寫入資料。中斷專用 */
BaseType_t xQueueSendToFrontFromISR(QueueHandle_t xQueue, const void *pvItemToQueue, BaseType_t *pxHigherPriorityTaskWoken);

引數說明：

• xQueue：佇列控制程式碼。

• pvItemToQueue：資料指標，這些資料的值會被複制進佇列。

• xTicksToWait：最大阻塞時間，單位Tick Count。

  • 如果被設為0，無法寫入資料時函數會立刻返回；
  • 如果被設為portMAX_DELAY，則會一直阻塞直到有空間可寫

• 返回值：

  • pdPASS：資料成功寫入了佇列
  • errQUEUE_FULL：寫入失敗，因為佇列滿了。

10.7.2 傳送訊息實現簡要步驟

1. 引數校驗。

2. 檢查當前佇列是否有空閒空間可寫入。

   1. 進入臨界。

   2. 有空間可寫入：

      1. 直接寫入。
      2. 檢查下是否有任務阻塞在當前佇列寫阻塞連結串列中，有就解鎖一個最高優先順序、最早開始等待的任務。
      3. 退出臨界。

   3. 沒空間可寫入：進入阻塞處理。

      1. 不需要阻塞，就退出臨界並返回。
      2. 開始阻塞超時計時。
      3. 退出臨界。（可能會切到其它任務或中斷）
      4. 掛起排程器。
      5. 再次檢查下是否有空間可寫，是否超時。
      6. 需要阻塞就計算下當前任務的喚醒時間，記錄到任務事件狀態節點資訊中，把當前任務從就緒連結串列抽離，插入到延時連結串列和當前佇列的寫阻塞任務連結串列中。
      7. 切走任務，等待喚醒。

10.7.3 傳送訊息原始碼分析

往佇列裡發訊息的API（中斷專用除外），都是封裝xQueueGenericSend()函數而來的，所以我們直接分析該函數實現即可。

需要注意的是，如果傳送訊息前，排程器被掛起了，則這個訊息不能設定為阻塞式的，因為如果掛起排程器後使用阻塞式寫入佇列，會觸發斷言。

在這裡可以拓展下，如果沒有這個斷言校驗，佇列已滿，則會在當前任務一直死迴圈，直至有中斷服務恢復排程器或讀取當前佇列的訊息，當前任務才能跑出這個坑。

xQueueGenericSend()：

BaseType_t xQueueGenericSend( QueueHandle_t xQueue,
                              const void * const pvItemToQueue,
                              TickType_t xTicksToWait,
                              const BaseType_t xCopyPosition )
{
    BaseType_t xEntryTimeSet = pdFALSE, xYieldRequired;
    TimeOut_t xTimeOut;
    Queue_t * const pxQueue = xQueue;

    /* 傳入的佇列控制程式碼不能為空 */
    configASSERT( pxQueue );
    /* 如果寫入佇列的資料為空，就說明呼叫當前API的不是一個訊息佇列，而是不含資料區的號誌、互斥量這些IPC，所以佇列成員size必須為0 */
    configASSERT( !( ( pvItemToQueue == NULL ) && ( pxQueue->uxItemSize != ( UBaseType_t ) 0U ) ) );
    /* 如果是覆蓋寫入，這個功能預設只能在佇列成員只有1個的情況下使用 */
    configASSERT( !( ( xCopyPosition == queueOVERWRITE ) && ( pxQueue->uxLength != 1 ) ) );
    #if ( ( INCLUDE_xTaskGetSchedulerState == 1 ) || ( configUSE_TIMERS == 1 ) )
        {
            /* 如果排程器被掛起，則不能進入阻塞。 */
            configASSERT( !( ( xTaskGetSchedulerState() == taskSCHEDULER_SUSPENDED ) && ( xTicksToWait != 0 ) ) );
        }
    #endif

    /* 使用迴圈邏輯，是為了解除阻塞後能檢查一下能否可以寫入。 */
    for( ; ; )
    {
        taskENTER_CRITICAL(); /* 進入臨界，因為下面操作可能會涉及到全域性資源，如那幾個任務連結串列 */
        {
            /* 佇列有空閒空間或需要強制寫入佇列，方可寫入 */
            if( ( pxQueue->uxMessagesWaiting < pxQueue->uxLength ) || ( xCopyPosition == queueOVERWRITE ) )
            {
                traceQUEUE_SEND( pxQueue );

                    {
                        const UBaseType_t uxPreviousMessagesWaiting = pxQueue->uxMessagesWaiting;
                        /* 拷貝資料到佇列裡 */
                        xYieldRequired = prvCopyDataToQueue( pxQueue, pvItemToQueue, xCopyPosition );

                        if( pxQueue->pxQueueSetContainer != NULL ) /* 佇列集合 */
                        {
                            if( ( xCopyPosition == queueOVERWRITE ) && ( uxPreviousMessagesWaiting != ( UBaseType_t ) 0 ) )
                            {
                                /* 如果當前佇列裡面有資料，且本次寫入是覆蓋寫入，就不需要通知佇列集了，因為佇列集已經被通知過。 */
                                mtCOVERAGE_TEST_MARKER();
                            }
                            else if( prvNotifyQueueSetContainer( pxQueue ) != pdFALSE ) /* 通知佇列集當前佇列有資料了 */
                            {
                                /* 觸發工作切換。只是觸發，實際切換需要到退出臨界後才執行。 */
                                queueYIELD_IF_USING_PREEMPTION();
                            }
                            else
                            {
                                mtCOVERAGE_TEST_MARKER();
                            }
                        }
                        else /* 不是佇列集 */
                        {
                            /* 有任務阻塞在讀阻塞連結串列，現在佇列有資料了，需要解鎖一個任務 */
                            if( listLIST_IS_EMPTY( &( pxQueue->xTasksWaitingToReceive ) ) == pdFALSE )
                            {
                                /* 從讀阻塞連結串列中解除一個最高優先順序且最先進入阻塞的任務。
                                    即把這個解除阻塞的任務的事件節點從當前佇列的阻塞連結串列中抽離，把狀態節點從掛起連結串列或延時連結串列重新插入到就緒連結串列或掛起的就緒連結串列中。
                                    如果解除阻塞的任務比當前在跑任務優先順序還高，就返回pdTRUE */
                                if( xTaskRemoveFromEventList( &( pxQueue->xTasksWaitingToReceive ) ) != pdFALSE )
                                {
                                    /* 觸發工作切換。只是觸發，實際切換需要到退出臨界後才執行。 */
                                    queueYIELD_IF_USING_PREEMPTION();
                                }
                                else
                                {
                                    mtCOVERAGE_TEST_MARKER();
                                }
                            }
                            else if( xYieldRequired != pdFALSE )
                            {
                                /* 如果是釋放互斥量時優先順序繼承機制觸發當前任務優先順序回落，就緒連結串列中有更高優先順序的任務，則觸發工作切換。只是觸發，實際切換需要到退出臨界後才執行。 */
                                queueYIELD_IF_USING_PREEMPTION();
                            }
                            else
                            {
                                mtCOVERAGE_TEST_MARKER();
                            }
                        }
                    }
                taskEXIT_CRITICAL(); /* 退出臨界 */
                return pdPASS; /* 返回成功 */
            }
            else /* 本次不能寫入，則檢查、準備進入阻塞處理 */
            {
                if( xTicksToWait == ( TickType_t ) 0 ) /* 不需要阻塞 */
                {
                    taskEXIT_CRITICAL(); /* 退出臨界 */
                    traceQUEUE_SEND_FAILED( pxQueue );
                    return errQUEUE_FULL; /* 返回寫入失敗 */
                }
                else if( xEntryTimeSet == pdFALSE ) /* 需要阻塞。第一次迴圈，需要記錄當前時間，開始計時阻塞超時。 */
                {
                    /* 備份當前系統節拍 */
                    vTaskInternalSetTimeOutState( &xTimeOut );
                    xEntryTimeSet = pdTRUE; /* 標記已開始記錄了 */
                }
                else
                {
                    /* 進入時間已經設定 */
                    mtCOVERAGE_TEST_MARKER();
                }
            }
        }
        taskEXIT_CRITICAL(); /* 退出臨界 */

        /* 退出臨界後系統會先處理在臨界期觸發的被遮蔽的中斷服務，如工作切換的中斷服務、其它中斷服務等等。 */

        vTaskSuspendAll(); /* 又排程回到當前任務了，掛起排程器，繼續幹活 */
        prvLockQueue( pxQueue ); /* 當前佇列上鎖，以免有中斷服務操作當前佇列，打亂老子的節奏 */

        /* 檢查阻塞時間是否已經超時 */
        if( xTaskCheckForTimeOut( &xTimeOut, &xTicksToWait ) == pdFALSE ) /* 還沒超時呢，繼續等唄 */
        {
            if( prvIsQueueFull( pxQueue ) != pdFALSE ) /* 納尼，佇列還是滿的，寫不進去啊 */
            {
                traceBLOCKING_ON_QUEUE_SEND( pxQueue );
                /* 我還是去這個佇列裡面的寫阻塞連結串列裡面排個隊吧。還可以插隊，不過只能插到比自己優先順序低的任務前面。 */
                vTaskPlaceOnEventList( &( pxQueue->xTasksWaitingToSend ), xTicksToWait );

                /* 放開當前佇列的控制權 */
                prvUnlockQueue( pxQueue );

                /* 恢復排程器 */
                if( xTaskResumeAll() == pdFALSE )
                {
                    /* 如果在恢復排程器裡面沒有觸發過排程，那這裡需要觸發一次排程，因為當前任務已經處於阻塞態了，怎麼滴也要觸發一次排程切走。 */
                    portYIELD_WITHIN_API();
                }
            }
            else /* 佇列有空位，趕緊寫 */
            {
                /* 解鎖當前佇列，進入下一個迴圈，看看能不能搶到寫入許可權 */
                prvUnlockQueue( pxQueue );
                ( void ) xTaskResumeAll(); /* 恢復排程器 */
            }
        }
        else /* 超時都沒等到寫入的許可權 */
        {
            /* 解鎖佇列 */
            prvUnlockQueue( pxQueue );
            ( void ) xTaskResumeAll(); /* 恢復排程器 */

            traceQUEUE_SEND_FAILED( pxQueue );
            return errQUEUE_FULL; /* 寫入失敗 */
        }
    } /*lint -restore */
}

寫入佇列的API原始碼prvCopyDataToQueue():（臨界中呼叫）

static BaseType_t prvCopyDataToQueue( Queue_t * const pxQueue,
                                      const void * pvItemToQueue,
                                      const BaseType_t xPosition )
{
    BaseType_t xReturn = pdFALSE;
    UBaseType_t uxMessagesWaiting;

    /* 當前函數需要在臨界裡被呼叫 */

    uxMessagesWaiting = pxQueue->uxMessagesWaiting;

    if( pxQueue->uxItemSize == ( UBaseType_t ) 0 ) /* 非隊型別 */
    {
        #if ( configUSE_MUTEXES == 1 )
            {
                if( pxQueue->uxQueueType == queueQUEUE_IS_MUTEX ) /* 互斥量 */
                {
                    /* 互斥量型別呼叫該函數就是釋放互斥量的意思 */
                    /* 釋放互斥量，需要處理優先順序繼承機制，回落到基優先順序 */
                    xReturn = xTaskPriorityDisinherit( pxQueue->u.xSemaphore.xMutexHolder );
                    /* 標記互斥量已經被解鎖 */
                    pxQueue->u.xSemaphore.xMutexHolder = NULL;
                }
                else
                {
                    mtCOVERAGE_TEST_MARKER();
                }
            }
        #endif /* configUSE_MUTEXES */
    }
    else if( xPosition == queueSEND_TO_BACK ) /* 往佇列尾部寫入 */
    {
        /* 按佇列屬性寫入 */
        ( void ) memcpy( ( void * ) pxQueue->pcWriteTo, pvItemToQueue, ( size_t ) pxQueue->uxItemSize );
        /* 更新佇列寫指標 */
        pxQueue->pcWriteTo += pxQueue->uxItemSize;

        if( pxQueue->pcWriteTo >= pxQueue->u.xQueue.pcTail )
        {
            /* 如果本次寫入的資料時佇列最後一個成員，就需要把當前佇列寫指標重置回首個佇列成員。和ringbuffer原理類似 */
            pxQueue->pcWriteTo = pxQueue->pcHead;
        }
        else
        {
            mtCOVERAGE_TEST_MARKER();
        }
    }
    else /* 往有效佇列頭部寫入 */
    {
        /* 按列屬性寫入，讀指標就是有效佇列頭 */
        ( void ) memcpy( ( void * ) pxQueue->u.xQueue.pcReadFrom, pvItemToQueue, ( size_t ) pxQueue->uxItemSize );
        /* 讀指標往前推 */
        pxQueue->u.xQueue.pcReadFrom -= pxQueue->uxItemSize;

        if( pxQueue->u.xQueue.pcReadFrom < pxQueue->pcHead )
        {
            /* 讀指標往前推時溢位後需要回溯到佇列最後一個成員 */
            pxQueue->u.xQueue.pcReadFrom = ( pxQueue->u.xQueue.pcTail - pxQueue->uxItemSize );
        }
        else
        {
            mtCOVERAGE_TEST_MARKER();
        }

        if( xPosition == queueOVERWRITE )
        {
            if( uxMessagesWaiting > ( UBaseType_t ) 0 )
            {
                /* 如果是覆蓋寫入，那當前佇列有效成員數量維持不變 */
                --uxMessagesWaiting;
            }
            else
            {
                mtCOVERAGE_TEST_MARKER();
            }
        }
        else
        {
            mtCOVERAGE_TEST_MARKER();
        }
    }
    /* 更新當前佇列有效成員數量 */
    pxQueue->uxMessagesWaiting = uxMessagesWaiting + ( UBaseType_t ) 1;

    return xReturn;
}

優先順序繼承機制概念、實現原理及其原始碼在互斥量章節的筆記講解。

10.7.5 中斷專用的傳送訊息API

中斷專用的傳送訊息API比普通的傳送訊息API佛繫了。

區別就是：中斷專用的沒有阻塞機制。

如果佇列有空閒空間，或本次是強制寫入，就把資料寫入。

• 寫入後如果佇列沒有上鎖，就更新當前佇列資訊，解鎖阻塞在讀阻塞佇列的最高優先順序、最早等待的一個任務。

• 如果佇列上鎖了，就用佇列中的pxQueue->cTxLock記錄當前佇列入隊了一個資料，在呼叫prvUnlockQueue()解鎖時更新當前佇列資訊，解鎖阻塞在讀阻塞佇列的最高優先順序、最早等待的一個任務。

如果佇列沒有空閒空間，又不是強制寫入，就直接退出。

10.8 接收訊息

當任務從佇列中讀取訊息時，如果佇列中有訊息，可以讀取並返回。

如果佇列中沒有訊息，需要進入阻塞處理，在阻塞超時前，有其他任務或中斷服務往這個佇列裡面寫訊息了，且當前任務時這個佇列中阻塞在讀阻塞連結串列中的最高優先順序、最先等待的任務，就解鎖該任務，否則還會一直阻塞到超時才喚醒當前任務。

10.8.1 接收訊息API

/* 從佇列中讀取資料。 */
BaseType_t xQueueReceive( QueueHandle_t xQueue, void * const pvBuffer, TickType_t xTicksToWait );
/* 從佇列中讀取資料。中斷專屬 */
BaseType_t xQueueReceiveFromISR(QueueHandle_t xQueue, void *pvBuffer, BaseType_t *pxTaskWoken);

引數說明：

• xQueue：佇列控制程式碼。

• pvBuffer：儲存接收資料的指標，其有效空間需要按照當前佇列屬性設定。

• xTicksToWait：如果佇列空則無法讀出資料，可以讓任務進入阻塞狀態，xTicksToWait表示阻塞的最大時間，單位：Tick Count。

  • 如果被設為0，無法讀出資料時函數會立刻返回；
  • 如果被設為portMAX_DELAY，則會一直阻塞直到有資料可寫。

• 返回值：

  • pdPASS：從佇列讀出資料入；
  • errQUEUE_EMPTY：讀取失敗，因為佇列空了。

10.8.2 接收訊息簡要步驟

10.8.3 接收訊息原始碼

xQueueReceive()：

BaseType_t xQueueReceive( QueueHandle_t xQueue,
                          void * const pvBuffer,
                          TickType_t xTicksToWait )
{
    BaseType_t xEntryTimeSet = pdFALSE;
    TimeOut_t xTimeOut;
    Queue_t * const pxQueue = xQueue;

    /* 佇列控制程式碼不能為空 */
    configASSERT( ( pxQueue ) );

    /* 如果資料區回傳地址為空，只能是不含資料區的IPC（號誌、互斥量等） */
    configASSERT( !( ( ( pvBuffer ) == NULL ) && ( ( pxQueue )->uxItemSize != ( UBaseType_t ) 0U ) ) );

    #if ( ( INCLUDE_xTaskGetSchedulerState == 1 ) || ( configUSE_TIMERS == 1 ) )
        {
            /* 排程器掛起後，不能已阻塞式呼叫當前API */
            configASSERT( !( ( xTaskGetSchedulerState() == taskSCHEDULER_SUSPENDED ) && ( xTicksToWait != 0 ) ) );
        }
    #endif

     /* 使用迴圈邏輯，是為了解除阻塞後能檢查一下能否可以讀取。 */
    for( ; ; )
    {
        taskENTER_CRITICAL(); /* 進入臨界 */
        {
            const UBaseType_t uxMessagesWaiting = pxQueue->uxMessagesWaiting;

            /* 當前佇列有資料可讀 */
            if( uxMessagesWaiting > ( UBaseType_t ) 0 )
            {
                /* 出隊。 */
                prvCopyDataFromQueue( pxQueue, pvBuffer );
                traceQUEUE_RECEIVE( pxQueue );
                /* 有效佇列成員個數更新 */
                pxQueue->uxMessagesWaiting = uxMessagesWaiting - ( UBaseType_t ) 1;

                /* 如果有任務阻塞在當前佇列的寫阻塞連結串列中，就解鎖一個，讓其寫入。 */
                if( listLIST_IS_EMPTY( &( pxQueue->xTasksWaitingToSend ) ) == pdFALSE )
                {
                    /* 把這個解除阻塞的任務從當前佇列的寫阻塞連結串列中解除，並把該任務從延時連結串列或掛起連結串列中恢復到就緒連結串列或掛起的就緒連結串列中 */
                    if( xTaskRemoveFromEventList( &( pxQueue->xTasksWaitingToSend ) ) != pdFALSE )
                    {
                        /* 解鎖的任務比當前任務優先順序更加高，需要觸發任務排程。 */
                        queueYIELD_IF_USING_PREEMPTION();
                    }
                    else
                    {
                        mtCOVERAGE_TEST_MARKER();
                    }
                }
                else
                {
                    mtCOVERAGE_TEST_MARKER();
                }
                /* 退出臨界 */
                taskEXIT_CRITICAL();
                return pdPASS; /* 返回讀取成功 */
            }
            else /* 佇列為空呢 */
            {
                if( xTicksToWait == ( TickType_t ) 0 ) /* 不需要阻塞 */
                {
                    /* 退出臨界並返回讀取失敗 */
                    taskEXIT_CRITICAL();
                    traceQUEUE_RECEIVE_FAILED( pxQueue );
                    return errQUEUE_EMPTY;
                }
                else if( xEntryTimeSet == pdFALSE ) /* 進入阻塞，首次迴圈需要開始計時 */
                {
                    /* 獲取當前系統節拍 */
                    vTaskInternalSetTimeOutState( &xTimeOut );
                    xEntryTimeSet = pdTRUE; /* 標記已經開始計時 */
                }
                else
                {
                    mtCOVERAGE_TEST_MARKER();
                }
            }
        }
        taskEXIT_CRITICAL(); /* 退出臨界 */

        /* 退出臨界後系統會先處理在臨界期觸發的被遮蔽的中斷服務，如工作切換的中斷服務、其它中斷服務等等。 */

        vTaskSuspendAll(); /* 有回到了當前任務。掛起排程器 */
        prvLockQueue( pxQueue ); /* 佇列上鎖 */

        /* 檢查是否已經超時。 */
        if( xTaskCheckForTimeOut( &xTimeOut, &xTicksToWait ) == pdFALSE ) /* 還沒超時 */
        {
            if( prvIsQueueEmpty( pxQueue ) != pdFALSE ) /* 如果佇列還沒有資料，需要繼續阻塞 */
            {
                traceBLOCKING_ON_QUEUE_RECEIVE( pxQueue );
                /* 我還是去這個佇列裡面的讀阻塞連結串列裡面排個隊吧。還可以插隊，不過只能插到比自己優先順序低的任務前面。 */
                vTaskPlaceOnEventList( &( pxQueue->xTasksWaitingToReceive ), xTicksToWait );
                prvUnlockQueue( pxQueue ); /* 解鎖當前佇列 */

                if( xTaskResumeAll() == pdFALSE ) /* 恢復排程器 */
                {
                    /* 如果在恢復排程器時沒有排程過，這裡必須手動觸發一次排程。否則會在當前這個坑裡一直跑，直到有中斷服務往當前佇列裡發訊息，或者有更高優先順序的任務被解除阻塞，或者系統節拍中有同優先順序任務被解鎖（就緒連結串列中還有大於2個及其以上同優先順序的任務）（開啟時間片的前提下）才會跳出這個坑。 */
                    portYIELD_WITHIN_API();
                }
                else
                {
                    mtCOVERAGE_TEST_MARKER();
                }
            }
            else
            {
                /* The queue contains data again.  Loop back to try and read the
                 * data. */
                prvUnlockQueue( pxQueue );
                ( void ) xTaskResumeAll();
            }
        }
        else /* 已經超時了 */
        {
            /* 解鎖佇列 */
            prvUnlockQueue( pxQueue );
            /* 恢復排程器 */
            ( void ) xTaskResumeAll();

            if( prvIsQueueEmpty( pxQueue ) != pdFALSE ) /* 再次判斷下是否真的沒有資料，現在有資料還來得及 */
            {
                /* 真的沒有資料，返回讀取失敗吧。 */
                traceQUEUE_RECEIVE_FAILED( pxQueue );
                return errQUEUE_EMPTY;
            }
            else
            {
                mtCOVERAGE_TEST_MARKER();
            }
        }
    } /*lint -restore */
}

出隊函數prvCopyDataFromQueue()：

static void prvCopyDataFromQueue( Queue_t * const pxQueue,
                                  void * const pvBuffer )
{
    if( pxQueue->uxItemSize != ( UBaseType_t ) 0 ) /* 只有帶資料區的IPC才能呼叫 */
    {
        /* 偏移到下一個佇列成員 */
        pxQueue->u.xQueue.pcReadFrom += pxQueue->uxItemSize;
  
        if( pxQueue->u.xQueue.pcReadFrom >= pxQueue->u.xQueue.pcTail )
        {
            /* 讀指標溢位的話需要回溯 */
            pxQueue->u.xQueue.pcReadFrom = pxQueue->pcHead;
        }
        else
        {
            mtCOVERAGE_TEST_MARKER();
        }
        /* 拷貝出資料 */
        ( void ) memcpy( ( void * ) pvBuffer, ( void * ) pxQueue->u.xQueue.pcReadFrom, ( size_t ) pxQueue->uxItemSize );
    }
}

10.9 窺探訊息

就是唯讀取資料，不刪除該資料。

其原始碼和xQueueReceive()差不多，只是資料不刪除，讀指標也不偏移，有效個數也不減少。

BaseType_t xQueuePeek( QueueHandle_t xQueue,
                       void * const pvBuffer,
                       TickType_t xTicksToWait );

BaseType_t xQueueReceiveFromISR( QueueHandle_t xQueue,
                                 void * const pvBuffer,
                                 BaseType_t * const pxHigherPriorityTaskWoken );

10.10 佇列查詢

佇列查詢主要是操作佇列控制塊中的資訊。

10.10.1 查詢佇列當前有效資料個數

UBaseType_t uxQueueMessagesWaiting( const QueueHandle_t xQueue )
{
    UBaseType_t uxReturn;
    configASSERT( xQueue );
    taskENTER_CRITICAL();
    {
        /* 獲取佇列有效成員個數 */
        uxReturn = ( ( Queue_t * ) xQueue )->uxMessagesWaiting;
    }
    taskEXIT_CRITICAL();

    return uxReturn;
}

10.10.2 查詢佇列當前可以空間個數

UBaseType_t uxQueueSpacesAvailable( const QueueHandle_t xQueue )
{
    UBaseType_t uxReturn;
    Queue_t * const pxQueue = xQueue;
    configASSERT( pxQueue );
    taskENTER_CRITICAL();
    {
        /* 總個數減去有效個數 */
        uxReturn = pxQueue->uxLength - pxQueue->uxMessagesWaiting;
    }
    taskEXIT_CRITICAL();

    return uxReturn;
}

10.11 刪除訊息佇列

佇列刪除函數是根據訊息佇列控制程式碼直接刪除的，刪除之後這個訊息佇列的所有資訊都會被系統回收清空，而且不能再次使用這個訊息佇列了。

直接上原始碼：

void vQueueDelete( QueueHandle_t xQueue )
{
    Queue_t * const pxQueue = xQueue;
    /* 佇列必須存在 */
    configASSERT( pxQueue );
    traceQUEUE_DELETE( pxQueue );

    #if ( configQUEUE_REGISTRY_SIZE > 0 )
        {
            /* 如果開啟了佇列登入檔功能，也需要從佇列登入檔中取出當前佇列的記錄 */
            vQueueUnregisterQueue( pxQueue );
        }
    #endif

    #if ( ( configSUPPORT_DYNAMIC_ALLOCATION == 1 ) && ( configSUPPORT_STATIC_ALLOCATION == 0 ) )
        {
            /* 如果只開啟了動態記憶體功能，就是直接釋放當前佇列資源 */
            vPortFree( pxQueue );
        }
    #elif ( ( configSUPPORT_DYNAMIC_ALLOCATION == 1 ) && ( configSUPPORT_STATIC_ALLOCATION == 1 ) )
        {
            /* 如果動態記憶體和靜態記憶體都開啟了,就需要區分當前佇列的記憶體資源來源 */
            if( pxQueue->ucStaticallyAllocated == ( uint8_t ) pdFALSE ) /* 動態建立 */
            {
                /* 直接回收 */
                vPortFree( pxQueue );
            }
            else /* 靜態記憶體，由使用者回收資源 */
            {
                mtCOVERAGE_TEST_MARKER();
            }
        }
    #else /* if ( ( configSUPPORT_DYNAMIC_ALLOCATION == 1 ) && ( configSUPPORT_STATIC_ALLOCATION == 0 ) ) */
        {
            /* 靜態分配，只能由使用者回收。 */
            ( void ) pxQueue;
        }
    #endif /* configSUPPORT_DYNAMIC_ALLOCATION */
}

10.12 訊息佇列使用注意

在使用freertos提供的訊息佇列元件時，需要注意以下幾點：

1. 使用xQueueSend()、xQueueSendFromISR()、xQueueReceive()等這些函數之前應先建立需訊息佇列，並根據佇列控制程式碼進行操作。
2. 要明白寫入佇列採用的邏輯時FIFO還是LIFO，使用對應的API。
3. 在獲取佇列中的訊息時候，必須要定義一個儲存讀取資料的地方，並且該資料區域大小不小於訊息大小，否則，很可能引發地址非法的錯誤。
4. freertos的資料流是拷貝方式實現的，如果訊息過大，建議使用拷貝參照。
5. 佇列獨立在核心中，不屬於任何一個任務。

小結

學習，重在理解，懂得底層原理，上層特性、特點即可推理。