FreeRTOS的队列介绍以及怎么实现互斥访问，休眠唤醒以及保存数据（环形缓冲区）

前言前面介绍完了FreeRTOS的一些核心功能如任务切换创建任务等等并将煮包从ARM内核以及内存的视角的相关思考进行了分享从这里开始介绍FreeRTOS的另外一个板块就是任务间通信机制如队列、信号量、互斥量、事件组这些今天分享的是FreeRTOS队列的学习笔记是关于标题中的互斥访问、休眠唤醒以及环形缓冲区都会在文章中一一介绍到~~~~一、FreeRTOS队列的基础知识FreeRTOS队列的核心关于队列数据结构里面的描述就不做介绍啦~最重要的就是队列是FIFO结构即数据先进先出对于FreeRTOS的队列它有如下几个核心知识点1. 关中断对应标题的实现互斥访问2.环形缓冲区用于存放队列的数据3.链表分为等待发送链表和等待获取链表具体的后面细说介绍队列结构体含有ringbuffer下面对队列的结构体进行一些介绍typedef struct QueueDefinition { int8_t * pcHead; int8_t * pcWriteTo; union { QueuePointers_t xQueue; SemaphoreData_t xSemaphore; } u; List_t xTasksWaitingToSend; List_t xTasksWaitingToReceive; volatile UBaseType_t uxMessagesWaiting; UBaseType_t uxLength UBaseType_t uxItemSize; volatile int8_t cRxLock; volatile int8_t cTxLock; #if ( ( configSUPPORT_STATIC_ALLOCATION 1 ) ( configSUPPORT_DYNAMIC_ALLOCATION 1 ) ) uint8_t ucStaticallyAllocated; #endif #if ( configUSE_QUEUE_SETS 1 ) struct QueueDefinition * pxQueueSetContainer; #endif #if ( configUSE_TRACE_FACILITY 1 ) UBaseType_t uxQueueNumber; uint8_t ucQueueType; #endif } xQUEUE;通过上面两个图可以看到FreeRTOS的队列结构体里面含有头尾指针以及写指针和读指针那么pcReadFrom和pcWriteTo两个指针就构成环形缓冲区即ringbuffer用来存放队列项可以上网查一下环形缓冲区是啥嘿嘿~这里就不介绍啦~~~下面就是两个链表一个是xTasksWaitingToSend用来存放当队列满时如果有任务想要写队列就会将该任务挂载到此链表一个是xTasksWaitingToReceive用来存放当队列空时如果有任务想读队列就会将该任务挂载到此链表然后就是这三个uxMessagesWaiting是一个计数值用来表示队列里面Item的数量uxLength是指的队列的长度uxItemSize是队列每一个队列项的大小由此可知队列环形缓冲区的总大小为uxLength*uxItemSize其他参数就是关于上锁和可以配置的参数暂时不介绍了因为有很多其他博主写的非常完善和全面嘿嘿二、FreeRTOS队列的源码解析创建队列:其实队列的创建也很简单就是先在FreeRTOS管理的堆栈空间中申请一片内存包含了队列结构体头和环形缓冲区的大小然后就是给队列结构体头赋初始值让我们详细看一下这个队列创建的函数~~~#if ( configSUPPORT_DYNAMIC_ALLOCATION 1 ) 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 ); 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 ); pxNewQueue ( Queue_t * ) pvPortMalloc( sizeof( Queue_t ) xQueueSizeInBytes ); /*lint !e9087 !e9079 see comment above. */ 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(); } return pxNewQueue; } #endif /* configSUPPORT_STATIC_ALLOCATION */可以看到这两句就是计算了环形缓冲区的大小然后从堆中malloc一片内存~1. 计算环形缓冲区大小xQueueSizeInBytes ( size_t ) ( uxQueueLength * uxItemSize ); 2. 从堆中申请内存包括队列结构体头和ringbufferpxNewQueue ( Queue_t * ) pvPortMalloc( sizeof( Queue_t ) xQueueSizeInBytes );如果内存申请成功也就是pxNewQueue不为NULL则让pucQueueStorage指向环形缓冲区的开头并调用prvInitialiseNewQueue来初始化其他结构体头的字段否则执行else部分的测试呀或者其他的通常功能需要我们自己实现那我们来看一下prvInitialiseNewQueue这个函数具体做了什么吧~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 ) { pxNewQueue-pcHead ( int8_t * ) pxNewQueue; } else { pxNewQueue-pcHead ( int8_t * ) pucQueueStorage; } 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 ); }这个函数也很简单主要就是判断每个队列项的大小是否是0如果是就把pcHead指向结构体头的起始位置如果不为0则pcHead指向环形缓冲区的起始位置然后调用( void ) xQueueGenericReset( pxNewQueue, pdTRUE );函数此函数如下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 ); 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 { /* Ensure the event queues start in the correct state. */ vListInitialise( ( pxQueue-xTasksWaitingToSend ) ); vListInitialise( ( pxQueue-xTasksWaitingToReceive ) ); } } taskEXIT_CRITICAL(); return pdPASS; }这个函数也很简单主要还是对结构体头里面的字段进行初始化~值得注意的是这里在进行初始化操作之前需要先关中断也就是实现了标题里面的互斥访问~1. 阅读代码很容易可以理解pcTail 指向pcHead偏移环形缓冲区大小的位置若每个队列项的大小是0则pcTail和pcHead都指向结构体头的位置若不为0则pcHead指向ringbuffer的起始位置pcTail指向ringbuffer的结束位置2. uxMessagesWaiting参数初始化为03. pcWriteTo初始化是指向pcHead的也就是当uxItemSize 为0时pcWriteTo和pcHead和pcTail都指向结构体头的位置若不为零pcWriteTo指向ringbuffer的起始位置也就是我们是从ringbuffer的开头开始写入数据逻辑也合理嘿嘿4. pcReadFrom初始化指向队列最后一个队列项的起始位置也就是我们从ringbuffer的最后一项开始读取数据代码如下pxQueue-u.xQueue.pcReadFrom pxQueue-pcHead ( ( pxQueue-uxLength - 1U ) * pxQueue-uxItemSize );5. 最后给两个上锁的参数初始值就是queueUNLOCKED-1最后初始化两个链表到此整个队列的创建工作就完成了~~是不是和创建任务及其相似哈哈哈还比创建任务简单最后附图煮包在正点原子上截图的一张图片队列读队列读取也是比较简单的就分为三步先关中断实现互斥再分有数据和无数据两种情况有数据就copy数据然后如果有任务在等待写队列就唤醒队列没有数据就返回ERR或者将任务休眠接下来我们来看一下FreeRTOS的源码时怎么实现的吧~由于FreeRTOS的队列获取函数实在太长这里就放函数定义然后截取关键的函数实现部分来分析啦~BaseType_t xQueueReceive( QueueHandle_t xQueue, void * const pvBuffer, TickType_t xTicksToWait )从定义看我们调用这个接口后就会从xQueue队列中读取数据并存放到pvBuffer里面如果队列为空就会将读取队列的任务阻塞xTicksToWait 长的时间然后看一下主要的函数实现部分有数据以下这个for循环里面就实现了队列读的有数据时的功能嗯~~~应该能看懂吧~哈哈哈就是判断一下uxMessagesWaiting 是否大于0是就是有数据copy数据到pvBuffer 然后uxMessagesWaiting减1for( ; ; ) { 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 { /* Entry time was already set. */ mtCOVERAGE_TEST_MARKER(); } } } taskEXIT_CRITICAL();如果等待写这个队列的链表不为空就唤醒一下链表里面的第一个任务从等待写队列链表中移除添加到对应优先级的就绪链表具体的实现就是下面这个函数里面的如图所示内容( void ) uxListRemove( ( pxUnblockedTCB-xEventListItem ) );更加详细的可以下载FreeRTOS源码进行查看啊哈哈哈哈~~~无数据实现真正的任务休眠当队列中没有数据的时候呢就看以下代码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(); } }当任务传入的等待参数为0的时候直接返回errQUEUE_EMPTY如果不是0则执行如下代码将当前任务插入到队列的xTasksWaitingToReceive链表最后再将当前任务添加到阻塞链表阻塞时间就是传入的xTicksToWaitvoid vTaskPlaceOnEventList( List_t * const pxEventList, const TickType_t xTicksToWait ) { configASSERT( pxEventList ); vListInsert( pxEventList, ( pxCurrentTCB-xEventListItem ) ); prvAddCurrentTaskToDelayedList( xTicksToWait, pdTRUE ); }当任务被添加到阻塞链表后就实现了真正的任务休眠啦~休眠位置如下具体来说就是执行任务切换的地方当后续任务被唤醒后又会从这里开始往下执行最后回到for循环开头再判断uxMessagesWaiting有没有数据此时肯定是有的啦因为有数据才会唤醒相应的任务呀~队列写队列写和队列读基本是完全对称的行为首先就是关中断其次分为有空间或者无空间有空间直接往队列里面copy数据copy完后判断一下有没有任务等待读取数据有就把它唤醒唤醒和队列读基本一样就是将任务从xTasksWaitingToReceive移除再添加到对应优先级的就绪列表如果没有空间了就将当前任务挂载到队列的xTasksWaitingToSend链表然后将当前任务从就绪链表中移动到阻塞链表因为基本和队列读一样这里就只放一段源码作为标记了~~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 */ }好啦今天煮包分享的内容就到这里啦~~~希望对大家有帮助~