说明

rtos内核基本上除了任务之间的调度以外，就剩下任务通信了。如果再广一点区分，任务通信也算任务调度了，它其实是为了提高内核效率而衍生出来的内核组件，为了将这些更规范，故而大家取名为信号量，邮箱，互斥量，任务通知等等，它们并不神秘。

代码分析

这节主要讲消息队列机制的代码分析。

首先是队列的创建，调用函数xQueueGenericCreate()：

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 thatcan be in the queue at any time.  It is valid for uxItemSize to bezero 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. *//* Allocate the queue and storage area.  Justification for MISRAdeviation as follows:  pvPortMalloc() always ensures returned memoryblocks are aligned per the requirements of the MCU stack.  In this casepvPortMalloc() must return a pointer that is guaranteed to meet thealignment requirements of the Queue_t structure - which in this caseis an int8_t *.  Therefore, whenever the stack alignment requirementsare greater than or equal to the pointer to char requirements the castis safe.  In other cases alignment requirements are not strict (one ortwo 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 queuestorage 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, sonote this task was created dynamically in case it is laterdeleted. */pxNewQueue->ucStaticallyAllocated = pdFALSE;}#endif /* configSUPPORT_STATIC_ALLOCATION */prvInitialiseNewQueue( uxQueueLength, uxItemSize, pucQueueStorage, ucQueueType, pxNewQueue );}else{traceQUEUE_CREATE_FAILED( ucQueueType );mtCOVERAGE_TEST_MARKER();}return pxNewQueue;}

首先是分配内存，然后是定位到消息队列头于消息体。后面调用prvInitialiseNewQueue( uxQueueLength, uxItemSize, pucQueueStorage, ucQueueType, pxNewQueue );

做基本的初始化，代码如下：

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 shouldconfigUSE_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 cannotbe set to NULL because NULL is used as a key to say the queue is used asa mutex.  Therefore just set pcHead to point to the queue as a benignvalue 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 isdefined. */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 );}

最终需要调用xQueueGenericReset()做更深入的初始化，代码如下：

BaseType_t xQueueGenericReset( QueueHandle_t xQueue, BaseType_t xNewQueue ){Queue_t * const pxQueue = xQueue;configASSERT( pxQueue );taskENTER_CRITICAL();{pxQueue->u.xQueue.pcTail = pxQueue->pcHead + ( pxQueue->uxLength * pxQueue->uxItemSize ); /*lint !e9016 Pointer arithmetic allowed on char types, especially when it assists conveying intent. */pxQueue->uxMessagesWaiting = ( UBaseType_t ) 0U;pxQueue->pcWriteTo = pxQueue->pcHead;pxQueue->u.xQueue.pcReadFrom = pxQueue->pcHead + ( ( pxQueue->uxLength - 1U ) * pxQueue->uxItemSize ); /*lint !e9016 Pointer arithmetic allowed on char types, especially when it assists conveying intent. */pxQueue->cRxLock = queueUNLOCKED;pxQueue->cTxLock = queueUNLOCKED;if( xNewQueue == pdFALSE ){/* If there are tasks blocked waiting to read from the queue, thenthe tasks will remain blocked as after this function exits the queuewill still be empty.  If there are tasks blocked waiting to write tothe queue, then one should be unblocked as after this function exitsit will be possible to write to it. */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{/* Ensure the event queues start in the correct state. */vListInitialise( &( pxQueue->xTasksWaitingToSend ) );vListInitialise( &( pxQueue->xTasksWaitingToReceive ) );}}taskEXIT_CRITICAL();/* A value is returned for calling semantic consistency with previousversions. */return pdPASS;}

最终形成这样的结构图：

说明一下，为什么pcReadFrom为什么不等于pcWriteTo。因为后面代码会执行一次pcReadFrom=pcReadFrom+1的操作。

之后就是一个任务负责发送消息，另外一个任务则负责接收任务了。

任务消息发送使用函数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 toallow return statements within the function itself.  This is done in theinterest of execution time efficiency. */for( ;; ){taskENTER_CRITICAL();{/* Is there room on the queue now?  The running task must be thehighest priority task wanting to access the queue.  If the head itemin the queue is to be overwritten then it does not matter if thequeue 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 itemwas overwritten in the queue so the number of itemsin the queue has not changed. */mtCOVERAGE_TEST_MARKER();}else if( prvNotifyQueueSetContainer( pxQueue ) != pdFALSE ){/* The queue is a member of a queue set, and postingto the queue set caused a higher priority task tounblock. 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 thequeue then unblock it now. */if( listLIST_IS_EMPTY( &( pxQueue->xTasksWaitingToReceive ) ) == pdFALSE ){if( xTaskRemoveFromEventList( &( pxQueue->xTasksWaitingToReceive ) ) != pdFALSE ){/* The unblocked task has a priority higher thanour own so yield immediately.  Yes it is ok todo this from within the critical section - thekernel 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 getexecuted if the task was holding multiple mutexesand the mutexes were given back in an order that isdifferent 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 thequeue then unblock it now. */if( listLIST_IS_EMPTY( &( pxQueue->xTasksWaitingToReceive ) ) == pdFALSE ){if( xTaskRemoveFromEventList( &( pxQueue->xTasksWaitingToReceive ) ) != pdFALSE ){/* The unblocked task has a priority higher thanour own so yield immediately.  Yes it is ok to dothis from within the critical section - the kerneltakes care of that. */queueYIELD_IF_USING_PREEMPTION();}else{mtCOVERAGE_TEST_MARKER();}}else if( xYieldRequired != pdFALSE ){/* This path is a special case that will only getexecuted if the task was holding multiple mutexes andthe mutexes were given back in an order that isdifferent 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 (orthe block time has expired) so leave now. */taskEXIT_CRITICAL();/* Return to the original privilege level before exitingthe function. */traceQUEUE_SEND_FAILED( pxQueue );return errQUEUE_FULL;}else if( xEntryTimeSet == pdFALSE ){/* The queue was full and a block time was specified soconfigure 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 queuenow 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 theevent list.  It is possible that interrupts occurring nowremove this task from the event list again - but as thescheduler is suspended the task will go onto the pendingready last instead of the actual ready list. */prvUnlockQueue( pxQueue );/* Resuming the scheduler will move tasks from the pendingready list into the ready list - so it is feasible that thistask is already in a ready list before it yields - in whichcase the yield will not cause a context switch unless thereis 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 */}

如果任务A调用xQueueGenericSend()函数，而调用这个函数的时候，如果消息队列未满或者允许覆盖写入，则调用prvCopyDataToQueue()将数据拷贝到指定位置。当指定queueSEND_TO_BACK时，表示插入尾部，其实就是pcWriteTo的位置，否则可以插入到头部，也就是在读取的时候，直接就能读到这个拷贝过去的消息。

拷贝完数据之后，如果有任务在等待这个消息，那就需要调用函数xTaskRemoveFromEventList()将该任务从事件列表删除，并恢复等待该消息的任务，让其进入就绪态，如果它比当前任务优先级高，则需要切换过去。

假如，任务A调用xQueueGenericSend()函数而队列已满。此时，如果延时时间xTicksToWait为0，则函数需要返回。否则设置的延时时间大于0，那需要调用vTaskInternalSetTimeOutState()进行的简单初始化结构体TimeOut_t。之后是确实队列已经满了的处理，则调用vTaskPlaceOnEventList()，将当前任务的xEventListItem加入pxQueue->xTasksWaitingToSend中，并调用prvAddCurrentTaskToDelayedList()添加到延时表里面。之后调用xTaskResumeAll()恢复调度器运行。

再看看调用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 sizeis 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 toallow return statements within the function itself.  This is done in theinterest 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 taskmust 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 topost to the queue?  If so, unblock the highest priority waitingtask. */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 (orthe 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 soconfigure 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 queuenow 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 placethe 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 thedata. */prvUnlockQueue( pxQueue );( void ) xTaskResumeAll();}}else{/* Timed out.  If there is no data in the queue exit, otherwise loopback 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 */}

如果队列有数据(uxMessagesWaiting>0)，说明队列可读，则需要调用prvCopyDataFromQueue()拷贝数据。如果pxQueue->xTasksWaitingToSend有item，则会调用xTaskRemoveFromEventList()将pxQueue->xTasksWaitingToSend中的一项item删除，而且也会将其加入到就就绪表中。如果队列为空，则跟前面发送消息队列数据类似，则需要在pxQueue->xTasksWaitingToReceive中加入pxCurrentTCB->xEventListItem，并且添加到延时表中。