FreeRTOS如何解决访问冲突/线程不安全(临界段、互斥锁、挂起调度、看门人任务)

在多任务（多线程）系统中，存在一个隐患，那就是多线程的访问（在FreeRTOS中就是任务）。当一个任务A开始访问一个资源（外设、一块内存等），但是A还没有完成访问，B任务运行了，也开始访问，这就会造成数据破坏、错误等问题。

例如：

两个任务试图写入一个液晶显示器(LCD)。

1任务A执行并开始向LCD写入字符串“Hello world”。

1. 任务A在输出字符串“Hello w”后被任务B抢占。

3.任务B在进入阻塞态前向LCD写入“Abort, Retry, Fail?”

1. 任务A继续从它被抢占的点开始，并完成输出它的字符串“world”的剩余字符。

LCD现在显示字符串是“Hello wAbort, Retry, Fail? world”。这显然不是我们想要的结果。

原文链接：FreeRTOS全解析-8.解决访问冲突/线程不安全(临界段、挂起调度、互斥锁、看门人任务)

1.一些概念

1.1原子和非原子操作

读、修改、写操作

对一个变量PORTA或上0x01，C语言写法：

PORTA |= 0x01;

通过编译转成汇编后：

LOAD R1,[#PORTA] ; Read a value from PORTA into R1MOVE R2,#0x01 ; Move the absolute constant 1 into R2OR R1,R2 ; Bitwise OR R1 (PORTA) with R2 (constant 1)STORE R1,[#PORTA] ; Store the new value back to PORTA

第1句，从PORTA的地址读取数据，保存到R1；（读操作）

第2句，把0x01保存到R2；（读操作）

第3句，R1和R2进行或操作，并存入R1;（修改操作）

第4句，把R1的值保存到PORTA的地址去。（写操作）

这就叫非原子操作，因为他使用了超过一条的汇编指令，并且可以被中断（相反，只用到一条指令的，无法中断的称作原子操作）。更新一个结构体的多个成员，或者更新一个大于CPU结构的字长(例如，在16位机器上更新一个32位变量)的变量，都是非原子操作的例子。如果中断，可能会导致数据丢失或损坏。

考虑以下场景：

1任务A将PORTA的值加载到寄存器中(操作的读部分)。

1. 任务A在完成修改和写入部分之前被任务B抢占。

3.任务B更新PORTA的值，然后进入阻塞态。

1. 任务A继续从它被抢占的点开始。它修改已经保存在寄存器中的PORTA值，然后写入PORTA的地址。

在这个场景中，任务A用到的PORTA的值相当于已经过期了（因为任务B对PORTA进行了修改），这个问题也被叫做数据不一致

1.2可重入函数

如果一个函数可以从多个任务调用，或者从任务和中断调用是安全的，那么这个函数就是“可重入的”。可重入函数被称为“线程安全的”，因为它们可以从多个线程访问，而不会有数据或逻辑操作损坏的风险。

每个任务维护自己的堆栈和自己的处理器(硬件)寄存器集。如果函数不访问存储在堆栈上或保存在寄存器中的数据以外的任何数据，那么函数是可重入的，并且是线程安全的。

如下，这就是可重入的函数，因为，lVar1是通过栈或者寄存器传递的，lVar2是在任务自己的栈中。每个任务访问这段代码时lVar1和lVar2都是不同的地址。

long lAddOneHundred( long lVar1 ){  long lVar2;  lVar2 = lVar1 + 100;  return lVar2;}

如下，这是不可重入的，lVar1是全局变量，lState用了static修饰，保存在数据段上。每个去访问的任务访问到的lVar1和lState都是同一份。

long lVar1;long lNonsenseFunction( void ){  static long lState = 0;  long lReturn;  switch( lState )  {    case 0 : lReturn = lVar1 + 10;      lState = 1;      break;    case 1 : lReturn = lVar1 + 20;      lState = 0;      break;  }}

1.3互斥

为了确保在任何时候都保持数据一致性，必须使用“互斥”来管理任务之间或任务和中断之间共享的资源。技术。目标是确保一旦任务开始访问非可重入且非线程安全的共享资源，同一任务对资源具有独占访问权，直到资源返回到一致状态。

FreeRTOS提供了几个可用于实现互斥的特性，但是最好的互斥方法是(在可能的情况下，因为通常不实用)将应用程序设计成不共享资源的方式，并且每个资源只能从单个任务访问。

2.临界段和挂起调度器

2.1临界段

临界段是分别被调用宏taskENTER_CRITICAL()和taskEXIT_CRITICAL()所包围的代码区域。临界段也称为临界区。

taskENTER_CRITICAL();PORTA |= 0x01;taskEXIT_CRITICAL();

回到写LCD冲突的例子，就可以这样：

void vPrintStringToLCD( const char *pcString ){  taskENTER_CRITICAL();  LCD_printf( "%s", pcString );  fflush( stdout );  taskEXIT_CRITICAL();}

用临界段实现互斥是非常粗糙的方法。它通过完全禁用中断来工作，或者达到configMAX_SYSCALL_INTERRUPT_PRIORITY设置的中断优先级（设置的最高优先级）。

抢占式上下文切换（任务调度）只能在中断内部发生，因此，只要中断保持禁用状态，调用taskENTER_CRITICAL()的任务就保证保持在运行状态，直到临界段退出。

临界段代码必须保持非常短，否则会对中断响应时间产生不利影响。每个对taskENTER_CRITICAL()的调用必须与对taskEXIT_CRITICAL()的调用紧密配对。假如写LCD或者输出会比较慢，就不应该用临界段。

临界段嵌套是安全的，因为内核会记录嵌套深度的计数。只有当嵌套深度返回到零时，临界段才会退出。

调用taskENTER_CRITICAL()和taskEXIT_CRITICAL()是任务改变正在运行FreeRTOS的处理器的中断启用状态的唯一合法方法。通过任何其他方式改变中断启用状态将使宏的嵌套计数失效。

taskENTER_CRITICAL()和taskEXIT_CRITICAL()不以'FromISR'结尾，因此不能从中断服务例程中调用。taskENTER_CRITICAL_FROM_ISR()是taskENTER_CRITICAL()的中断安全版本，taskEXIT_CRITICAL_FROM_ISR()是taskEXIT_CRITICAL()的中断安全版本。中断安全版本只对允许中断嵌套的处理器生效。用法：

void vAnInterruptServiceRoutine( void ){  UBaseType_t uxSavedInterruptStatus;  uxSavedInterruptStatus = taskENTER_CRITICAL_FROM_ISR();  taskEXIT_CRITICAL_FROM_ISR( uxSavedInterruptStatus );}

2.2挂起（锁住）调度器

还可以通过挂起调度程序来创建临界区。挂起调度器有时也称为“锁定”调度器。临界段保护代码区域不被其他任务和中断访问。通过挂起调度器实现的临界区只保护代码区域不被其他任务访问，因为中断仍然是启用的。

如果临界段太长，不能通过简单地禁用中断来实现，则可以通过挂起调度器来实现。然而，恢复(或“取消挂起”)调度器比较慢，因此必须考虑在每种情况下使用哪种方法是最好的。

调度器通过调用vTaskSuspendAll()来挂起。挂起调度程序可以防止发生上下文切换，但会启用中断。如果在调度器挂起时，有切换任务的请求，则该请求将保持挂起状态，并且仅在调度器恢复(未挂起)时执行。当调度程序挂起时，不能调用FreeRTOS API函数。

void vTaskSuspendAll( void )BaseType_t xTaskResumeAll( void );

嵌套调用vTaskSuspendAll()和xTaskResumeAll()是安全的，因为内核保留了嵌套深度的计数。只有当嵌套深度返回0时，调度器才会恢复。

3.互斥锁(和二进制信号量)

不懂信号量的可以看一下这篇FreeRTOS全解析-8.信号量(semaphore)

互斥锁（或者叫互斥量，我用Linux比较多，习惯叫锁，FreeRTOS中叫量比较合适）是一种特殊类型的二进制信号量，用于控制对两个或多个任务之间共享的资源的访问。"Mutex"（互斥锁）这个词起源于"Mutual Exclusion"。（互斥）

FreeRTOSConfig.h中的configUSE_MUTEXES必须设置为1，才能使互斥锁。

互斥锁在需要互斥的场景中使用时，可以将其视为与共享资源相关联的令牌。对于要合法访问资源的任务，它必须首先成功地“获取”令牌(成为令牌持有者)。当令牌持有者使用完资源后，它必须“归还”令牌。只有当令牌已经归还时，另一个任务才能成功获取令牌，然后安全地访问相同的共享资源。除非任务持有令牌，否则不允许访问共享资源。

尽管互斥锁和二进制信号量很像，但还是不一样。主要的区别是信号量被获取后会发生什么:用于互斥的信号量必须始终返还（take后要give）。用于同步的信号量通常被丢弃而不返还（take后不用give）。还有一个区别是互斥锁有优先级继承（本文后面讲）。

互斥锁就像这样使用：获取和释放函数和信号量用的是一样的。

static void prvNewPrintString( const char *pcString ){  xSemaphoreTake( xMutex, portMAX_DELAY );  printf( "%s", pcString );  fflush( stdout );  xSemaphoreGive( xMutex );}

使用前要创建，调用函数：

SemaphoreHandle_t xSemaphoreCreateMutex( void );

比如：

SemaphoreHandle_t xMutex; xMutex = xSemaphoreCreateMutex();

一个完整的使用互斥锁的例子：

static void prvNewPrintString( const char *pcString ){  xSemaphoreTake( xMutex, portMAX_DELAY );  printf( "%s", pcString );  fflush( stdout );  xSemaphoreGive( xMutex );}static void prvPrintTask( void *pvParameters ){  char *pcStringToPrint;  const TickType_t xMaxBlockTimeTicks = 0x20;  pcStringToPrint = ( char * ) pvParameters;  for( ;; )  {    prvNewPrintString( pcStringToPrint );    vTaskDelay( ( rand() % xMaxBlockTimeTicks ) );  }}int main( void ){  xMutex = xSemaphoreCreateMutex();  if( xMutex != NULL )  {    xTaskCreate( prvPrintTask, "Print1", 1000,    "Task 1 ***************************************\r\n", 1, NULL );    xTaskCreate( prvPrintTask, "Print2", 1000,    "Task 2 ---------------------------------------\r\n", 2, NULL );    vTaskStartScheduler();  }  for( ;; );}

3.1优先级翻转

先来看看上面的例子会发生什么

Task1优先级为1，Task2优先级为2.

Task1先运行，获得互斥锁，Task2优先级虽然高，但是因为没有获得互斥锁，进入阻塞态，只有等Task1释放了互斥锁，才有机会运行。

这表现出使用互斥锁来提供互斥的一个潜在缺陷。

高优先级Task 2必须等待低优先级Task 1放弃对互斥锁的控制。高优先级任务被低优先级任务以这种方式延迟称为“优先级反转”。

在这种情况下会加剧：

如图有三个任务LP低优先级任务，MP中等优先级任务，HP高优先级任务。

LP运行，获得互斥锁，HP尝试抢占，但是因为没有获得互斥锁，只能进入阻塞，LP继续运行，但是LP运行过程中，被不需要互斥锁的MP给抢占了。

LP不运行，就无法释放互斥锁，不释放，HP就永远无法运行。结果就变成了，最高优先级的任务在等最低优先级的任务。

优先级反转可能是一个重大问题，但在小型嵌入式系统中，通过考虑如何访问资源，通常可以在系统设计时避免它。

3.2优先级继承

FreeRTOS互斥量和二进制信号量的区别还在于互斥量有“优先级继承”机制，而二进制信号量没有。优先级继承是一种使优先级反转负面影响最小化的方案。它不会“修复”优先级反转，而只是通过确保反转总是有时间限制来减少其影响。然而，优先级继承使系统定时分析复杂化，如果说是依靠它来使系统正常运行，那不太可取。

优先级继承是通过临时将互斥锁持有者的优先级提高到试图获得相同互斥锁的最高优先级任务的优先级来实现的。持有互斥锁的低优先级任务“继承”了等待互斥锁的任务的优先级。互斥锁持有者的优先级在返回互斥锁时自动重置为其原始值。

有了这个机制，前面说到情况就会变成这样：

LP运行，获得互斥锁，HP尝试运行，但是因为没有互斥锁，进入阻塞态，同时因为HP优先级高，LP继承了HP的优先级，不再会被MP抢占。HP就可以在LP释放互斥锁的时候运行了。

正因为优先级继承功能会影响使用互斥锁的任务的优先级。所以不能在中断服务例程中使用互斥锁。

3.3死锁

“死锁”是使互斥锁进行互斥的另一个潜在陷阱。

当两个任务都在等待由另一个任务持有的资源时，就会发生死锁。考虑下面的场景，任务A和任务B都需要获得互斥量X和Y来执行一个操作:

1任务A执行并成功获取互斥量X。

1. 任务A被任务B抢占。

3.任务B在尝试使用互斥量X之前成功地使用了互斥量Y，但互斥量X由任务A持有，因此任务B无法使用。任务B选择进入阻塞状态，等待互斥量X释放。

1. 任务A继续执行。它尝试获取互斥量Y，但互斥量Y由任务B持有，因此任务A无法使用。任务A选择进入阻塞状态，等待互斥量Y释放。

任务A阻塞等待互斥量X，任务B阻塞等待互斥量Y，等待的互斥量都在对方手里，而又都在阻塞态，运行不了，就这么一直等下去，就是死锁。

与优先级反转一样，避免死锁的最佳方法是在设计时充分考虑这个问题，设计系统以确保不会发生死锁。

实际上，死锁在小型嵌入式系统中并不是一个大问题，因为系统设计人员可以很好地理解整个应用程序，因此可以识别并删除可能发生死锁的区域。

3.4递归互斥锁

任务本身也有可能死锁。如果一个任务多次尝试使用同一个互斥锁，而没有首先返回互斥锁，就会发生这种情况。考虑以下场景:

1. 任务成功获取互斥锁A。

2. 当持有互斥锁A时，任务调用一个库函数。

3.库函数里面尝试使用相同的互斥锁A，然后进入阻塞状态，等待互斥锁A。

在这个场景的最后，任务处于阻塞状态，等待互斥锁返回，但该任务已经是互斥锁的持有者。发生死锁是因为任务处于等待自身的阻塞态，就是我等我自己。

这种类型的死锁可以通过使用递归互斥锁来代替标准互斥锁来避免。一个任务可以多次获取（take）同一个互斥锁，不过要记得take几次就要give几次。

创建：

xSemaphoreCreateRecursiveMutex().

获取take变成了taken

xSemaphoreTakeRecursive().

释放give变成了given

xSemaphoreGiveRecursive()

4.看门人任务（Gatekeeper Tasks）

看门人任务提供了一种干净的实现互斥的方法，没有优先级反转或死锁的风险。

看门人任务是对资源拥有唯一所有权的任务。只有看门人任务被允许直接访问资源——任何其他需要访问资源的任务只能通过使用看门人的服务间接访问资源。

如下面例子，思路挺简单的，任务是要打印，输出就是资源，任务不能直接打印，必需通过队列发送到看门人任务，看门人任务进行打印操作。

static void prvStdioGatekeeperTask( void *pvParameters ){  char *pcMessageToPrint;  for( ;; )  {    xQueueReceive( xPrintQueue, &pcMessageToPrint, portMAX_DELAY );    printf( "%s", pcMessageToPrint );    fflush( stdout );  }}static void prvPrintTask( void *pvParameters ){  int iIndexToString;  const TickType_t xMaxBlockTimeTicks = 0x20;  iIndexToString = ( int ) pvParameters;  for( ;; )  {    xQueueSendToBack( xPrintQueue, &( pcStringsToPrint[ iIndexToString ] ), 0 );    vTaskDelay( ( rand() % xMaxBlockTimeTicks ) );  }}static char *pcStringsToPrint[] ={  "Task 1 ****************************************************\r\n",  "Task 2 ----------------------------------------------------\r\n",  "Message printed from the tick hook interrupt ##############\r\n"};QueueHandle_t xPrintQueue;int main( void ){  xPrintQueue = xQueueCreate( 5, sizeof( char * ) );  if( xPrintQueue != NULL )  {    xTaskCreate( prvPrintTask, "Print1", 1000, ( void * ) 0, 1, NULL );    xTaskCreate( prvPrintTask, "Print2", 1000, ( void * ) 1, 2, NULL );    xTaskCreate( prvStdioGatekeeperTask, "Gatekeeper", 1000, NULL, 0, NULL );    vTaskStartScheduler();  }  for( ;; );}

往期精彩：

STM32F4+FreeRTOS+LVGL实现快速开发(缝合怪)