The Queue in real-time operating systems (RTOS) such as FreeRTOS is a very important communication mechanism, mainly used for 1. Inter-Process Communication (IPC) and data transfer between tasks and interrupts.2. It can facilitate task synchronization and resource management. General queues can protect and utilize resources by implementing semaphores and mutexes.This article, as the first part of the queue module, will focus on the process of implementing task communication through queues. The next article will discuss how queues solvetask synchronization and resource management issues.1. The Significance of Queues in Task Communication
In an RTOS, multiple tasks (threads) may run simultaneously, and they need to exchange data safely. Queues provide a thread-safe way for tasks to:
1. Send data (xQueueSend)2. Receive data (xQueueReceive)3. Wait for data (blocking or non-blocking)
Example Scenario: Producer-Consumer Model
- Task A (Producer) generates data and sends it to the queue.
- Task B (Consumer) reads data from the queue and processes it. If the queue is empty, the consumer task can block and wait until data arrives. In this case, the queue can act as a data buffer, solving the problem of mismatched speeds between the producer and consumer.
- If the producer is faster than the consumer, the queue can temporarily store data to prevent data loss.
- If the consumer is faster than the producer, the queue can make the consumer wait, avoiding busy waiting.
2. The Essence of QueuesIn RTOS, queues consist of contiguous memory blocks, divided into control blocks (management area) and data storage areas. The following is a detailed analysis:1. Physical layout (as shown)
2. Organization of the data storage area
<span><span>When uxItemSize > 0, it is just a normal queue; when uxItemSize = 0, there is no actual storage area,</span></span> pcHead is forced to point to the control block itself (<span><span>pxNewQueue-pcHead(int8_t*)pxNewQueue</span></span>), and only through<span><span>uxMessagesWaiting</span></span> counting (semaphore/mutex).
3. Key operations and their relation to the physical structure
| Operation | Physical Structure Impact |
|---|---|
| xQueueSend | 1. Data is copied to the location pointed to by<span><span>pcWriteTo</span></span>2. <span><span>pcWriteTo</span></span> advances by<span><span>uxItemSize</span></span> bytes |
| xQueueReceive | 1. Data is read from<span><span>u.xQueue.pcReadFrom</span></span>2. Read pointer advances by<span><span>uxItemSize</span></span> bytes |
| Queue Full Check | <span><span>uxMessagesWaiting == uxLength</span></span> (no need to traverse the storage area, directly compare the counter) |
| Queue Empty Check | <span><span>uxMessagesWaiting == 0</span></span> |
Next, I will introduce how queues specifically implement these functions.3. Implementation Functions for Task Communication via Queues
This diagram shows the relationship between task communication and implementation functions, with task communication mainly achieved through the following functions.
1. Sending Data to the Queue (Producer Task)
xQueueSend() / <span><span>xQueueSendToBack()</span></span>
- Function: Send data to the end of the queue (FIFO mode)
- Applicable Scenario: General tasks sending data to the queue
- Parameters
<span><span>xQueue Queue handle</span></span>pvItemToQueue Pointer to the data to be sentxTicksToWait Blocking time (<span><span>portMAX_DELAY</span></span>indicates infinite wait) - Return ValuepdPASS Send successful.
<span><span>errQUEUE_FULL Queue is full (in non-blocking mode)</span></span>
2. Receiving Data from the Queue (Consumer Task)
(1) <span><span>xQueueReceive()</span></span>
- Function: Remove data from the head of the queue (FIFO mode)
- Applicable Scenario: General tasks reading data from the queue
- ParametersxQueue Queue handle
<span><span>pvBuffer Buffer for receiving data</span></span><span><span>xTicksToWait Blocking time (portMAX_DELAY indicates infinite wait)</span></span> - Return ValuepdPASS Receive successful.
<span><span>errQUEUE_EMPTY Queue is empty (in non-blocking mode)</span></span>
(2) <span><span>xQueuePeek()</span></span>
- Function: View the data at the head of the queue without removing it (similar to Receive but does not delete data)
- Applicable Scenario: Check queue data without consuming it
3. Communication Between Tasks and Interrupts
<span><span>xQueueSendFromISR()</span></span>
- Function: Send data to the queue from an interrupt service routine (ISR)
- Parameters
<span><span>xQueue Queue handle</span></span><span><span>pvItemToQueue Data pointer</span></span><span><span>pxHigherPriorityTaskWoken </span></span>Used to wake up higher priority tasks (can be set to<span><span>NULL</span></span>)
4. Queue Set Processing
xQueueAddToSet()/xQueueRemoveFromSet()
- Function: Add/remove a queue from the queue set
Next, I will explain the important parts of this series of functions.
4. Specific Function Implementations1. Queue Creation and Initialization:Creation:
QueueHandle_t xQueueGenericCreate( const UBaseType_t uxQueueLength, const UBaseType_t uxItemSize, const uint8_t ucQueueType ) { Queue_t * pxNewQueue; size_t xQueueSizeInBytes; uint8_t * pucQueueStorage; configASSERT( uxQueueLength > ( UBaseType_t ) 0 ); /* Allocate enough space to hold the maximum number of items that * can be in the queue at any time. It is valid for uxItemSize to be * zero in the case the queue is used as a semaphore. */ xQueueSizeInBytes = ( size_t ) ( uxQueueLength * uxItemSize ); /*lint !e961 MISRA exception as the casts are only redundant for some ports. */ /* Check for multiplication overflow. */ configASSERT( ( uxItemSize == 0 ) || ( uxQueueLength == ( xQueueSizeInBytes / uxItemSize ) ) ); /* Check for addition overflow. */ configASSERT( ( sizeof( Queue_t ) + xQueueSizeInBytes ) > xQueueSizeInBytes ); /* Allocate the queue and storage area. Justification for MISRA * deviation as follows: pvPortMalloc() always ensures returned memory * blocks are aligned per the requirements of the MCU stack. In this case * pvPortMalloc() must return a pointer that is guaranteed to meet the * alignment requirements of the Queue_t structure - which in this case * is an int8_t *. Therefore, whenever the stack alignment requirements * are greater than or equal to the pointer to char requirements the cast * is safe. In other cases alignment requirements are not strict (one or * two bytes). */ pxNewQueue = ( Queue_t * ) pvPortMalloc( sizeof( Queue_t ) + xQueueSizeInBytes ); /*lint !e9087 !e9079 see comment above. */ if( pxNewQueue != NULL ) { /* Jump past the queue structure to find the location of the queue * storage area. */ pucQueueStorage = ( uint8_t * ) pxNewQueue; pucQueueStorage += sizeof( Queue_t ); /*lint !e9016 Pointer arithmetic allowed on char types, especially when it assists conveying intent. */ #if ( configSUPPORT_STATIC_ALLOCATION == 1 ) { /* Queues can be created either statically or dynamically, so * note this task was created dynamically in case it is later * deleted. */ pxNewQueue->ucStaticallyAllocated = pdFALSE; } #endif /* configSUPPORT_STATIC_ALLOCATION */ prvInitialiseNewQueue( uxQueueLength, uxItemSize, pucQueueStorage, ucQueueType, pxNewQueue ); } else { traceQUEUE_CREATE_FAILED( ucQueueType ); mtCOVERAGE_TEST_MARKER(); } return pxNewQueue; }Initialization
/*-----------------------------------------------------------*/static void prvInitialiseNewQueue( const UBaseType_t uxQueueLength, const UBaseType_t uxItemSize, uint8_t * pucQueueStorage, const uint8_t ucQueueType, Queue_t * pxNewQueue ){ /* Remove compiler warnings about unused parameters should * configUSE_TRACE_FACILITY not be set to 1. */ ( void ) ucQueueType; if( uxItemSize == ( UBaseType_t ) 0 ) { /* No RAM was allocated for the queue storage area, but PC head cannot * be set to NULL because NULL is used as a key to say the queue is used as * a mutex. Therefore just set pcHead to point to the queue as a benign * value that is known to be within the memory map. */ pxNewQueue->pcHead = ( int8_t * ) pxNewQueue; } else { /* Set the head to the start of the queue storage area. */ pxNewQueue->pcHead = ( int8_t * ) pucQueueStorage; } /* Initialise the queue members as described where the queue type is * defined. */ pxNewQueue->uxLength = uxQueueLength; 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 );}/*-----------------------------------------------------------*/For the xQueueGenericCreate() creation function, we need to input the length of the queue and the size of each item (in bytes). This function calculates the total memory size required = queue control block (Queue_t) + queue storage area (uxLength * uxItemSize) and calls<span><span>pvPortMalloc</span></span> to allocate memory, calculate the position of the storage area, and call<span><span>prvInitialiseNewQueue</span></span> for actual initialization.
For the prvInitialiseNewQueue() function, it will be called byxQueueGenericCreate(). This function is responsible for initializing the newly created queue structure. This function has several core processes:
(1) Set the queue head pointer:
- If the item size is 0 (used as a semaphore/mutex), set
<span><span>pcHead</span></span>to point to the queue structure itself - Otherwise, set
<span><span>pcHead</span></span>to point to the start of the allocated storage area.
(2) Set basic parameters:
- Store the queue length (
<span><span>uxLength</span></span>) - Store the item size (
<span><span>uxItemSize</span></span>)
(3) Call<span><span>xQueueGenericReset</span></span>:
- Reset the queue state (set write/read positions, counters, etc.)
- The second parameter
<span><span>pdTRUE</span></span>indicates that this is the initialization at queue creation
Once the setup is complete, this area will become a truly usable queue.
2. Sending to the queue using 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 ); configASSERT( !( ( pvItemToQueue == NULL ) && ( pxQueue->uxItemSize != ( UBaseType_t ) 0U ) ) ); configASSERT( !( ( xCopyPosition == queueOVERWRITE ) && ( pxQueue->uxLength != 1 ) ) ); #if ( ( INCLUDE_xTaskGetSchedulerState == 1 ) || ( configUSE_TIMERS == 1 ) ) { configASSERT( !( ( xTaskGetSchedulerState() == taskSCHEDULER_SUSPENDED ) && ( xTicksToWait != 0 ) ) ); } #endif /*lint -save -e904 This function relaxes the coding standard somewhat to * allow return statements within the function itself. This is done in the * interest of execution time efficiency. */ for( ; ; ) { taskENTER_CRITICAL(); { /* Is there room on the queue now? The running task must be the highest priority task wanting to access the queue. If the head item in the queue is to be overwritten then it does not matter if the queue is full. */ if( ( pxQueue->uxMessagesWaiting < pxQueue->uxLength ) || ( xCopyPosition == queueOVERWRITE ) ) { traceQUEUE_SEND( pxQueue ); #if ( configUSE_QUEUE_SETS == 1 ) { const UBaseType_t uxPreviousMessagesWaiting = pxQueue->uxMessagesWaiting; xYieldRequired = prvCopyDataToQueue( pxQueue, pvItemToQueue, xCopyPosition ); if( pxQueue->pxQueueSetContainer != NULL ) { if( ( xCopyPosition == queueOVERWRITE ) && ( uxPreviousMessagesWaiting != ( UBaseType_t ) 0 ) ) { /* Do not notify the queue set as an existing item * was overwritten in the queue so the number of items * in the queue has not changed. */ mtCOVERAGE_TEST_MARKER(); } else if( prvNotifyQueueSetContainer( pxQueue ) != pdFALSE ) { /* The queue is a member of a queue set, and posting * to the queue set caused a higher priority task to * unblock. A context switch is required. */ queueYIELD_IF_USING_PREEMPTION(); } else { mtCOVERAGE_TEST_MARKER(); } } else { /* If there was a task waiting for data to arrive on the * queue then unblock it now. */ if( listLIST_IS_EMPTY( &( pxQueue->xTasksWaitingToReceive ) ) == pdFALSE ) { if( xTaskRemoveFromEventList( &( pxQueue->xTasksWaitingToReceive ) ) != pdFALSE ) { /* The unblocked task has a priority higher than * our own so yield immediately. Yes it is ok to * do this from within the critical section - the * kernel takes care of that. */ queueYIELD_IF_USING_PREEMPTION(); } else { mtCOVERAGE_TEST_MARKER(); } } else if( xYieldRequired != pdFALSE ) { /* This path is a special case that will only get * executed if the task was holding multiple mutexes * and the mutexes were given back in an order that is * different to that in which they were taken. */ queueYIELD_IF_USING_PREEMPTION(); } else { mtCOVERAGE_TEST_MARKER(); } } } #else /* configUSE_QUEUE_SETS */ { xYieldRequired = prvCopyDataToQueue( pxQueue, pvItemToQueue, xCopyPosition ); /* If there was a task waiting for data to arrive on the * queue then unblock it now. */ if( listLIST_IS_EMPTY( &( pxQueue->xTasksWaitingToReceive ) ) == pdFALSE ) { if( xTaskRemoveFromEventList( &( pxQueue->xTasksWaitingToReceive ) ) != pdFALSE ) { /* The unblocked task has a priority higher than * our own so yield immediately. Yes it is ok to * do this from within the critical section - the * kernel takes care of that. */ queueYIELD_IF_USING_PREEMPTION(); } else { mtCOVERAGE_TEST_MARKER(); } } else if( xYieldRequired != pdFALSE ) { /* This path is a special case that will only get * executed if the task was holding multiple mutexes and * the mutexes were given back in an order that is * different to that in which they were taken. */ queueYIELD_IF_USING_PREEMPTION(); } else { mtCOVERAGE_TEST_MARKER(); } } #endif /* configUSE_QUEUE_SETS */ taskEXIT_CRITICAL(); return pdPASS; } else { if( xTicksToWait == ( TickType_t ) 0 ) { /* The queue was full and no block time is specified (or * the block time has expired) so leave now. */ taskEXIT_CRITICAL(); /* Return to the original privilege level before exiting * the function. */ traceQUEUE_SEND_FAILED( pxQueue ); return errQUEUE_FULL; } else if( xEntryTimeSet == pdFALSE ) { /* The queue was full and a block time was specified so * configure the timeout structure. */ vTaskInternalSetTimeOutState( &xTimeOut ); xEntryTimeSet = pdTRUE; } else { /* Entry time was already set. */ mtCOVERAGE_TEST_MARKER(); } } } taskEXIT_CRITICAL(); /* Interrupts and other tasks can send to and receive from the queue * now the critical section has been exited. */ vTaskSuspendAll(); prvLockQueue( pxQueue ); /* Update the timeout state to see if it has expired yet. */ if( xTaskCheckForTimeOut( &xTimeOut, &xTicksToWait ) == pdFALSE ) { if( prvIsQueueFull( pxQueue ) != pdFALSE ) { traceBLOCKING_ON_QUEUE_SEND( pxQueue ); vTaskPlaceOnEventList( &( pxQueue->xTasksWaitingToSend ), xTicksToWait ); /* Unlocking the queue means queue events can effect the * event list. It is possible that interrupts occurring now * remove this task from the event list again - but as the * scheduler is suspended the task will go onto the pending * ready last instead of the actual ready list. */ prvUnlockQueue( pxQueue ); /* Resuming the scheduler will move tasks from the pending * ready list into the ready list - so it is feasible that this * task is already in a ready list before it yields - in which * case the yield will not cause a context switch unless there * is also a higher priority task in the pending ready list. */ if( xTaskResumeAll() == pdFALSE ) { portYIELD_WITHIN_API(); } } else { /* Try again. */ prvUnlockQueue( pxQueue ); ( void ) xTaskResumeAll(); } } else { /* The timeout has expired. */ prvUnlockQueue( pxQueue ); ( void ) xTaskResumeAll(); traceQUEUE_SEND_FAILED( pxQueue ); return errQUEUE_FULL; } } /*lint -restore */}This function checks if the queue storage area is full when receiving a data pointer:(1) If the storage area is not full, it schedules<span><span>prvCopyDataToQueue()</span></span> to copy the data to the queue storage area and update the write pointer. If there are receiving tasks waiting, it will prioritize waking up higher priority tasks (which may switch) to receive tasks.(2) If the storage area is full, it will return a send error, record the current timestamp if blocking is allowed, add the task to the send waiting list, and return an error after timeout.It is important to note that tasks sent in this way cannot specify a receiver. If a specific receiver is desired, the target task needs to establish a dedicated queue. Additionally, we can use task management’s task notifications to achieve point-to-point instant sending.This function also has an interrupt API: xQueueGenericSendFromISR()xQueueGenericSendFromISR() has an internal implementation similar to the normal send function, with the difference that calling the interrupt send function will disable interrupts rather than just entering a critical section to achieve atomicity. Additionally, due to the nature of interrupts that cannot wait, if the queue is full, it will directly fail to send.3. Receiving from the queue 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; /* Check the pointer is not NULL. */ configASSERT( ( pxQueue ) ); /* The buffer into which data is received can only be NULL if the data size * is zero (so no data is copied into the buffer). */ configASSERT( !( ( ( pvBuffer ) == NULL ) && ( ( pxQueue )->uxItemSize != ( UBaseType_t ) 0U ) ) ); /* Cannot block if the scheduler is suspended. */ #if ( ( INCLUDE_xTaskGetSchedulerState == 1 ) || ( configUSE_TIMERS == 1 ) ) { configASSERT( !( ( xTaskGetSchedulerState() == taskSCHEDULER_SUSPENDED ) && ( xTicksToWait != 0 ) ) ); } #endif /*lint -save -e904 This function relaxes the coding standard somewhat to * allow return statements within the function itself. This is done in the * interest of execution time efficiency. */ for( ; ; ) { taskENTER_CRITICAL(); { const UBaseType_t uxMessagesWaiting = pxQueue->uxMessagesWaiting; /* Is there data in the queue now? To be running the calling task * must be the highest priority task wanting to access the queue. */ if( uxMessagesWaiting > ( UBaseType_t ) 0 ) { /* Data available, remove one item. */ prvCopyDataFromQueue( pxQueue, pvBuffer ); traceQUEUE_RECEIVE( pxQueue ); pxQueue->uxMessagesWaiting = uxMessagesWaiting - ( UBaseType_t ) 1; /* There is now space in the queue, were any tasks waiting to * post to the queue? If so, unblock the highest priority waiting * task. */ 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 ) { /* The queue was empty and no block time is specified (or * the block time has expired) so leave now. */ taskEXIT_CRITICAL(); traceQUEUE_RECEIVE_FAILED( pxQueue ); return errQUEUE_EMPTY; } else if( xEntryTimeSet == pdFALSE ) { /* The queue was empty and a block time was specified so * configure the timeout structure. */ vTaskInternalSetTimeOutState( &xTimeOut ); xEntryTimeSet = pdTRUE; } else { /* Entry time was already set. */ mtCOVERAGE_TEST_MARKER(); } } } taskEXIT_CRITICAL(); /* Interrupts and other tasks can send to and receive from the queue * now the critical section has been exited. */ vTaskSuspendAll(); prvLockQueue( pxQueue ); /* Update the timeout state to see if it has expired yet. */ if( xTaskCheckForTimeOut( &xTimeOut, &xTicksToWait ) == pdFALSE ) { /* The timeout has not expired. If the queue is still empty place * the task on the list of tasks waiting to receive from the queue. */ if( prvIsQueueEmpty( pxQueue ) != pdFALSE ) { traceBLOCKING_ON_QUEUE_RECEIVE( pxQueue ); vTaskPlaceOnEventList( &( pxQueue->xTasksWaitingToReceive ), xTicksToWait ); prvUnlockQueue( pxQueue ); if( xTaskResumeAll() == pdFALSE ) { 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 { /* Timed out. If there is no data in the queue exit, otherwise loop * back and attempt to read the data. */ prvUnlockQueue( pxQueue ); ( void ) xTaskResumeAll(); if( prvIsQueueEmpty( pxQueue ) != pdFALSE ) { traceQUEUE_RECEIVE_FAILED( pxQueue ); return errQUEUE_EMPTY; } else { mtCOVERAGE_TEST_MARKER(); } } } /*lint -restore */}This function, in contrast to the queue sending function, is called by tasks to receive data from the target queue. This function will also check the queue status (protected by a critical section)
(1) If the storage area is not empty:
uxReturn = listGET_LIST_ITEM_VALUE(&if(uxMessagesWaiting > 0) { /* Copy data from the queue to the user buffer */ prvCopyDataFromQueue(pxQueue, pvBuffer); traceQUEUE_RECEIVE(pxQueue); pxQueue->uxMessagesWaiting--; // Update queue count /* Wake up any potentially blocked sending tasks */ if(listLIST_IS_EMPTY(&pxQueue->xTasksWaitingToSend) == pdFALSE) { if(xTaskRemoveFromEventList(&pxQueue->xTasksWaitingToSend) != pdFALSE) { queueYIELD_IF_USING_PREEMPTION(); // Trigger task switch (if the awakened task has a higher priority) } } taskEXIT_CRITICAL(); return pdPASS; // Successfully received data}- Data Copy
<span>prvCopyDataFromQueue()</span>copies data from the queue storage area to the user buffer and advances the read pointer. This function does not free the data, but this data will be ignored and can be directly overwritten. - Priority Wakeup If the sending queue is not empty, it wakes up the highest priority sending task, triggering a context switch if necessary.
(2) If the storage area is empty:
else { if(xTicksToWait == 0) { /* Non-blocking mode returns directly */ taskEXIT_CRITICAL(); traceQUEUE_RECEIVE_FAILED(pxQueue); return errQUEUE_EMPTY; } else if(xEntryTimeSet == 0) { /* Initialize timeout structure */ vTaskInternalSetTimeOutState(&xTimeOut); xEntryTimeSet = pdTRUE; }}- Immediate Return
<code><span><span>In non-blocking mode</span></span>xTicksToWait = 0 returns a queue empty error directly. - Timeout Preparation In blocking mode, allows for a timeout, recording the starting timestamp upon first entering the block. The task is added to the
<span>xTasksWaitingToReceive</span>list, sorted by priority. Timeout will retry the receive, and if still unsuccessful, will return an error.
This function also has an interrupt API: xQueueGenericReceiveFromISR()xQueueGenericReceiveFromISR() has an internal implementation similar to the normal receive function, with the difference that calling the interrupt receive function will disable interrupts rather than just entering a critical section to achieve atomicity. Additionally, due to the nature of interrupts that cannot wait, if the queue is empty, it will directly fail to receive.4. Non-deleting Read xQueuePeek()
BaseType_t xQueuePeek( QueueHandle_t xQueue, void * const pvBuffer, TickType_t xTicksToWait ){ BaseType_t xEntryTimeSet = pdFALSE; TimeOut_t xTimeOut; int8_t * pcOriginalReadPosition; Queue_t * const pxQueue = xQueue; /* Check the pointer is not NULL. */ configASSERT( ( pxQueue ) ); /* The buffer into which data is received can only be NULL if the data size * is zero (so no data is copied into the buffer. */ configASSERT( !( ( ( pvBuffer ) == NULL ) && ( ( pxQueue )->uxItemSize != ( UBaseType_t ) 0U ) ) ); /* Cannot block if the scheduler is suspended. */ #if ( ( INCLUDE_xTaskGetSchedulerState == 1 ) || ( configUSE_TIMERS == 1 ) ) { configASSERT( !( ( xTaskGetSchedulerState() == taskSCHEDULER_SUSPENDED ) && ( xTicksToWait != 0 ) ) ); } #endif /*lint -save -e904 This function relaxes the coding standard somewhat to * allow return statements within the function itself. This is done in the * interest of execution time efficiency. */ for( ; ; ) { taskENTER_CRITICAL(); { const UBaseType_t uxMessagesWaiting = pxQueue->uxMessagesWaiting; /* Is there data in the queue now? To be running the calling task * must be the highest priority task wanting to access the queue. */ if( uxMessagesWaiting > ( UBaseType_t ) 0 ) { /* Remember the read position so it can be reset after the data * is read from the queue as this function is only peeking the * data, not removing it. */ pcOriginalReadPosition = pxQueue->u.xQueue.pcReadFrom; prvCopyDataFromQueue( pxQueue, pvBuffer ); traceQUEUE_PEEK( pxQueue ); /* The data is not being removed, so reset the read pointer. */ pxQueue->u.xQueue.pcReadFrom = pcOriginalReadPosition; /* The data is being left in the queue, so see if there are * any other tasks waiting for the data. */ if( listLIST_IS_EMPTY( &( pxQueue->xTasksWaitingToReceive ) ) == pdFALSE ) { if( xTaskRemoveFromEventList( &( pxQueue->xTasksWaitingToReceive ) ) != pdFALSE ) { /* The task waiting has a higher priority than this task. */ queueYIELD_IF_USING_PREEMPTION(); } else { mtCOVERAGE_TEST_MARKER(); } } else { mtCOVERAGE_TEST_MARKER(); } taskEXIT_CRITICAL(); return pdPASS; } else { if( xTicksToWait == ( TickType_t ) 0 ) { /* The queue was empty and no block time is specified (or * the block time has expired) so leave now. */ taskEXIT_CRITICAL(); traceQUEUE_PEEK_FAILED( pxQueue ); return errQUEUE_EMPTY; } else if( xEntryTimeSet == pdFALSE) { /* The queue was empty and a block time was specified so * configure the timeout structure ready to enter the blocked * state. */ vTaskInternalSetTimeOutState( &xTimeOut ); xEntryTimeSet = pdTRUE; } else { /* Entry time was already set. */ mtCOVERAGE_TEST_MARKER(); } } } taskEXIT_CRITICAL(); /* Interrupts and other tasks can send to and receive from the queue * now the critical section has been exited. */ vTaskSuspendAll(); prvLockQueue( pxQueue ); /* Update the timeout state to see if it has expired yet. */ if( xTaskCheckForTimeOut( &xTimeOut, &xTicksToWait ) == pdFALSE ) { /* Timeout has not expired yet, check to see if there is data in the * queue now, and if not enter the Blocked state to wait for data. */ if( prvIsQueueEmpty( pxQueue ) != pdFALSE ) { traceBLOCKING_ON_QUEUE_PEEK( pxQueue ); vTaskPlaceOnEventList( &( pxQueue->xTasksWaitingToReceive ), xTicksToWait ); prvUnlockQueue( pxQueue ); if( xTaskResumeAll() == pdFALSE ) { portYIELD_WITHIN_API(); } else { mtCOVERAGE_TEST_MARKER(); } } else { /* There is data in the queue now, so don't enter the blocked * state, instead return to try and obtain the data. */ prvUnlockQueue( pxQueue ); ( void ) xTaskResumeAll(); } } else { /* The timeout has expired. If there is still no data in the queue * exit, otherwise go back and try to read the data again. */ prvUnlockQueue( pxQueue ); ( void ) xTaskResumeAll(); if( prvIsQueueEmpty( pxQueue ) != pdFALSE ) { traceQUEUE_PEEK_FAILED( pxQueue ); return errQUEUE_EMPTY; } else { mtCOVERAGE_TEST_MARKER(); } } } /*lint -restore */}This function can read data but will not remove it. It is very similar to xQueueReceive(), but there are still some differences.
pcOriginalReadPosition = pxQueue->u.xQueue.pcReadFrom;...（copy data）pxQueue->u.xQueue.pcReadFrom = pcOriginalReadPosition;In this part, there are two lines of code that are not present in xQueueReceive(), which serve to record the original read pointer position and reset it after calling the copy function, thus removing the deletion nature of xQueueReceive().
if(listLIST_IS_EMPTY(&pxQueue->xTasksWaitingToReceive) == pdFALSE) { if(xTaskRemoveFromEventList(&pxQueue->xTasksWaitingToReceive) != pdFALSE) { queueYIELD_IF_USING_PREEMPTION(); // Trigger task switch (if the awakened task has a higher priority) } }If the array is not empty, it will wake up the highest priority receiving task. This reflects the functionality of this function, which sometimes exists to probe for the receiving function.
This function also has an interrupt version, and the differences with its interrupt version are almost the same as those of the previous functions, which will not be repeated here.
5. Auxiliary Functions
All the functions mentioned above are core functions, and queues also have many auxiliary functions to complete more functionalities. Due to space limitations, I cannot analyze them all here, but I have listed a table for display.
| Function Category | Key Functions | Usage |
|---|---|---|
| Status Query | <span>uxQueueMessagesWaiting()</span><span>uxQueueSpacesAvailable()</span> |
Monitor queue fill status |
| Management Operations | <span>xQueueReset()</span><span>vQueueDelete()</span> |
Clear or destroy the queue |
| Debug Support | <span>pcQueueGetName()</span><span>uxQueueGetQueueNumber()</span> |
Debugging and tracing queues |
| Registry | <span>vQueueAddToRegistry()</span><span>vQueueUnregisterQueue()</span> |
Name queues (for debugging tools) |
| Advanced Operations | <span>xQueueOverwrite()</span><span>xQueuePeekFromISR()</span> |
Overwrite data or interrupt-safe viewing |
| Queue Set | <span>xQueueCreateSet()</span><span><span>xQueueSelectFromSet()</span></span> |
Multi-queue listening |
Among them, the queue set functions are important task communication functions, and I will briefly explain them:
#if ( ( configUSE_QUEUE_SETS == 1 ) && ( configSUPPORT_DYNAMIC_ALLOCATION == 1 ) ) QueueSetHandle_t xQueueCreateSet( const UBaseType_t uxEventQueueLength ) { QueueSetHandle_t pxQueue; pxQueue = xQueueGenericCreate( uxEventQueueLength, ( UBaseType_t ) sizeof( Queue_t * ), queueQUEUE_TYPE_SET ); return pxQueue; }This is the creation function for the queue set. The queue set is essentially a queue filled with queue pointers, and its significance lies in allowing a task to receive information from multiple queues simultaneously. It is created through <span><span>xQueueGenericCreate()</span></span> .
Here is a piece of pseudo-code illustrating the usage process of the queue set:
// 1. Create a queue set (listening to a maximum of 3 queues)QueueSetHandle_t xSet = xQueueCreateSet(3);// 2. Create normal queues and add them to the queue setQueueHandle_t xQueue1 = xQueueCreate(5, sizeof(int));QueueHandle_t xQueue2 = xQueueCreate(5, sizeof(float));xQueueAddToSet(xQueue1, xSet);xQueueAddToSet(xQueue2, xSet);// 3. Task waits for any queue to have datavoid vTask(void *pvParameters) { QueueSetMemberHandle_t xActivated; while (1) { xActivated = xQueueSelectFromSet(xSet, portMAX_DELAY); // Block waiting if (xActivated == xQueue1) { int data; xQueueReceive(xQueue1, &data, 0); // Read data from queue 1 } else if (xActivated == xQueue2) { float data; xQueueReceive(xQueue2, &data, 0); // Read data from queue 2 } }}As can be seen, this task can read data from multiple queues simultaneously using the queue set.
The commonly used functions of the queue set are:
(1) Add a queue to the queue set xQueueAddToSet()
BaseType_t xQueueAddToSet( QueueSetMemberHandle_t xQueueOrSemaphore, // Queue or semaphore handle to be added QueueSetHandle_t xQueueSet // Target queue set handle){ BaseType_t xReturn; // Enter critical section (to prevent multi-task contention) taskENTER_CRITICAL(); { // Check 1: Is the queue/semaphore already part of another queue set? if( ( ( Queue_t * ) xQueueOrSemaphore )->pxQueueSetContainer != NULL ) { xReturn = pdFAIL; // Prevent duplicate addition } // Check 2: Is the queue/semaphore already have data? else if( ( ( Queue_t * ) xQueueOrSemaphore )->uxMessagesWaiting != ( UBaseType_t ) 0 ) { xReturn = pdFAIL; // Prevent non-empty queues from being added } // Check passed, perform the addition operation else { ( ( Queue_t * ) xQueueOrSemaphore )->pxQueueSetContainer = xQueueSet; xReturn = pdPASS; // Addition successful } } taskEXIT_CRITICAL(); // Exit critical section return xReturn;}Only empty queues that do not belong to other queue sets can be inserted into the queue set. The specific approach is to set the queue/semaphore’s <span>pxQueueSetContainer</span> to point to the target queue set, establishing a relationship.
(2) Remove a queue from the queue set xQueueRemoveFromSet()
BaseType_t xQueueRemoveFromSet( QueueSetMemberHandle_t xQueueOrSemaphore, // Queue or semaphore handle to be removed QueueSetHandle_t xQueueSet // Target queue set handle){ BaseType_t xReturn; Queue_t * const pxQueueOrSemaphore = ( Queue_t * ) xQueueOrSemaphore; // Check 1: Is the queue/semaphore part of the target queue set? if( pxQueueOrSemaphore->pxQueueSetContainer != xQueueSet ) { xReturn = pdFAIL; // Does not belong to this queue set } // Check 2: Is the queue/semaphore empty? else if( pxQueueOrSemaphore->uxMessagesWaiting != ( UBaseType_t ) 0 ) { xReturn = pdFAIL; // Non-empty queues cannot be removed } // Check passed, perform the removal operation else { taskENTER_CRITICAL(); // Enter critical section { pxQueueOrSemaphore->pxQueueSetContainer = NULL; // Remove association } taskEXIT_CRITICAL(); // Exit critical section xReturn = pdPASS; // Removal successful } return xReturn;}Only queues that belong to the queue set and are empty can be removed from the queue set. The specific approach is to set the queue/semaphore’s <span>pxQueueSetContainer</span> to point to NULL, canceling the association.
5. Conclusion
That concludes the main content of this article. In summary:
Queues rely on critical section protection and are a crucial communication mechanism in real-time operating systems like FreeRTOS, mainly used for inter-task communication and communication between tasks and interrupts, and can also implement semaphores and mutexes for resource management. This article has detailed the task communication functionality of the FreeRTOS queue mechanism, and the next part will focus on the application of queues in task synchronization and resource management, including how to implement advanced functions such as semaphores and mutexes through queues.