韋東山freeRTOS系列教程之互斥量(mutex)(7)

文章目錄

• 系列教程總目錄
• 概述
• 7.1 互斥量的使用場合
• 7.2 互斥量函數(shù)
  • 7.2.1 創(chuàng)建
  • 7.2.2 其他函數(shù)
• 7.3 示例15: 互斥量基本使用
• 7.4 示例16: 誰上鎖就由誰解鎖？
• 7.5 示例17: 優(yōu)先級反轉
• 7.6 示例18: 優(yōu)先級繼承
• 7.7 遞歸鎖
  • 7.7.1 死鎖的概念
  • 7.7.2 自我死鎖
  • 7.7.3 函數(shù)
  • 7.7.4 示例19: 遞歸鎖
• 7.8 常見問題

需要獲取更好閱讀體驗的同學，請訪問我專門設立的站點查看，地址：http://rtos.100ask.net/

系列教程總目錄

本教程連載中，篇章會比較多，為方便同學們閱讀，點擊這里可以查看文章的 目錄列表，目錄列表頁面地址：https://blog.csdn.net/thisway_diy/article/details/121399484

概述

怎么獨享廁所？自己開門上鎖，完事了自己開鎖。

你當然可以進去后，讓別人幫你把門：但是，命運就掌握在別人手上了。

使用隊列、信號量，都可以實現(xiàn)互斥訪問，以信號量為例：

• 信號量初始值為1
• 任務A想上廁所，"take"信號量成功，它進入廁所
• 任務B也想上廁所，"take"信號量不成功，等待
• 任務A用完廁所，"give"信號量；輪到任務B使用

這需要有2個前提：

• 任務B很老實，不撬門(一開始不"give"信號量)
• 沒有壞人：別的任務不會"give"信號量

可以看到，使用信號量確實也可以實現(xiàn)互斥訪問，但是不完美。

使用互斥量可以解決這個問題，互斥量的名字取得很好：

• 量：值為0、1
• 互斥：用來實現(xiàn)互斥訪問

它的核心在于：誰上鎖，就只能由誰開鎖。

很奇怪的是，FreeRTOS的互斥鎖，并沒有在代碼上實現(xiàn)這點：

• 即使任務A獲得了互斥鎖，任務B竟然也可以釋放互斥鎖。
• 誰上鎖、誰釋放：只是約定。

本章涉及如下內容：

為什么要實現(xiàn)互斥操作

怎么使用互斥量

互斥量導致的優(yōu)先級反轉、優(yōu)先級繼承

7.1 互斥量的使用場合

在多任務系統(tǒng)中，任務A正在使用某個資源，還沒用完的情況下任務B也來使用的話，就可能導致問題。

比如對于串口，任務A正使用它來打印，在打印過程中任務B也來打印，客戶看到的結果就是A、B的信息混雜在一起。

這種現(xiàn)象很常見：

訪問外設：剛舉的串口例子

讀、修改、寫操作導致的問題
對于同一個變量，比如int a，如果有兩個任務同時寫它就有可能導致問題。
對于變量的修改，C代碼只有一條語句，比如：a=a+8;，它的內部實現(xiàn)分為3步：讀出原值、修改、寫入。

我們想讓任務A、B都執(zhí)行add_a函數(shù)，a的最終結果是1+8+8=17。
假設任務A運行完代碼①，在執(zhí)行代碼②之前被任務B搶占了：現(xiàn)在任務A的R0等于1。
任務B執(zhí)行完add_a函數(shù)，a等于9。
任務A繼續(xù)運行，在代碼②處R0仍然是被搶占前的數(shù)值1，執(zhí)行完②③的代碼，a等于9，這跟預期的17不符合。

對變量的非原子化訪問
修改變量、設置結構體、在16位的機器上寫32位的變量，這些操作都是非原子的。也就是它們的操作過程都可能被打斷，如果被打斷的過程有其他任務來操作這些變量，就可能導致沖突。

函數(shù)重入
“可重入的函數(shù)"是指：多個任務同時調用它、任務和中斷同時調用它，函數(shù)的運行也是安全的?？芍厝氲暮瘮?shù)也被稱為"線程安全”(thread safe)。
每個任務都維持自己的棧、自己的CPU寄存器，如果一個函數(shù)只使用局部變量，那么它就是線程安全的。
函數(shù)中一旦使用了全局變量、靜態(tài)變量、其他外設，它就不是"可重入的"，如果改函數(shù)正在被調用，就必須阻止其他任務、中斷再次調用它。

上述問題的解決方法是：任務A訪問這些全局變量、函數(shù)代碼時，獨占它，就是上個鎖。這些全局變量、函數(shù)代碼必須被獨占地使用，它們被稱為臨界資源。

互斥量也被稱為互斥鎖，使用過程如下：

• 互斥量初始值為1
• 任務A想訪問臨界資源，先獲得并占有互斥量，然后開始訪問
• 任務B也想訪問臨界資源，也要先獲得互斥量：被別人占有了，于是阻塞
• 任務A使用完畢，釋放互斥量；任務B被喚醒、得到并占有互斥量，然后開始訪問臨界資源
• 任務B使用完畢，釋放互斥量

正常來說：在任務A占有互斥量的過程中，任務B、任務C等等，都無法釋放互斥量。

但是FreeRTOS未實現(xiàn)這點：任務A占有互斥量的情況下，任務B也可釋放互斥量。

7.2 互斥量函數(shù)

7.2.1 創(chuàng)建

互斥量是一種特殊的二進制信號量。

使用互斥量時，先創(chuàng)建、然后去獲得、釋放它。使用句柄來表示一個互斥量。

創(chuàng)建互斥量的函數(shù)有2種：動態(tài)分配內存，靜態(tài)分配內存，函數(shù)原型如下：

/* 創(chuàng)建一個互斥量，返回它的句柄。
 * 此函數(shù)內部會分配互斥量結構體 
 * 返回值: 返回句柄，非NULL表示成功
 */
SemaphoreHandle_t xSemaphoreCreateMutex( void );

/* 創(chuàng)建一個互斥量，返回它的句柄。
 * 此函數(shù)無需動態(tài)分配內存，所以需要先有一個StaticSemaphore_t結構體，并傳入它的指針
 * 返回值: 返回句柄，非NULL表示成功
 */
SemaphoreHandle_t xSemaphoreCreateMutexStatic( StaticSemaphore_t *pxMutexBuffer );

要想使用互斥量，需要在配置文件FreeRTOSConfig.h中定義：

#define configUSE_MUTEXES 1

7.2.2 其他函數(shù)

要注意的是，互斥量不能在ISR中使用。

各類操作函數(shù)，比如刪除、give/take，跟一般是信號量是一樣的。

/*
 * xSemaphore: 信號量句柄，你要刪除哪個信號量, 互斥量也是一種信號量
 */
void vSemaphoreDelete( SemaphoreHandle_t xSemaphore );

/* 釋放 */
BaseType_t xSemaphoreGive( SemaphoreHandle_t xSemaphore );

/* 釋放(ISR版本) */
BaseType_t xSemaphoreGiveFromISR(
                       SemaphoreHandle_t xSemaphore,
                       BaseType_t *pxHigherPriorityTaskWoken
                   );

/* 獲得 */
BaseType_t xSemaphoreTake(
                   SemaphoreHandle_t xSemaphore,
                   TickType_t xTicksToWait
               );
/* 獲得(ISR版本) */
xSemaphoreGiveFromISR(
                       SemaphoreHandle_t xSemaphore,
                       BaseType_t *pxHigherPriorityTaskWoken
                   );

7.3 示例15: 互斥量基本使用

本節(jié)代碼為： FreeRTOS_15_mutex 。

使用互斥量時有如下特點：

• 剛創(chuàng)建的互斥量可以被成功"take"
• “take"互斥量成功的任務，被稱為"holder”，只能由它"give"互斥量；別的任務"give"不成功
• 在ISR中不能使用互斥量

本程序創(chuàng)建2個發(fā)送任務：故意發(fā)送大量的字符。可以做2個實驗：

• 使用互斥量：可以看到任務1、任務2打印的字符串沒有混雜在一起
• 不使用互斥量：任務1、任務2打印的字符串混雜在一起

main函數(shù)代碼如下：

/* 互斥量句柄 */
SemaphoreHandle_t xMutex;

int main( void )
{
	prvSetupHardware();
	
    /* 創(chuàng)建互斥量 */
    xMutex = xSemaphoreCreateMutex( );


	if( xMutex != NULL )
	{
		/* 創(chuàng)建2個任務: 都是打印
		 * 優(yōu)先級相同
		 */
		xTaskCreate( vSenderTask, "Sender1", 1000, (void *)1, 1, NULL );
		xTaskCreate( vSenderTask, "Sender2", 1000, (void *)2, 1, NULL );

		/* 啟動調度器 */
		vTaskStartScheduler();
	}
	else
	{
		/* 無法創(chuàng)建互斥量 */
	}

	/* 如果程序運行到了這里就表示出錯了, 一般是內存不足 */
	return 0;
}

發(fā)送任務的函數(shù)如下：

static void vSenderTask( void *pvParameters )
{
	const TickType_t xTicksToWait = pdMS_TO_TICKS( 10UL );	
	int cnt = 0;
	int task = (int)pvParameters;
	int i;
	char c;
	
	/* 無限循環(huán) */
	for( ;; )
	{	
		/* 獲得互斥量: 上鎖 */
		xSemaphoreTake(xMutex, portMAX_DELAY);
		
		printf("Task %d use UART count: %d, ", task, cnt++);
		c = (task == 1 ) ? 'a' : 'A';
		for (i = 0; i < 26; i++)
			printf("%c", c + i);
		printf("rn");
		
		/* 釋放互斥量: 開鎖 */
		xSemaphoreGive(xMutex);
		
		vTaskDelay(xTicksToWait);
	}
}

可以做兩個實驗：vSenderTask函數(shù)的for循環(huán)中xSemaphoreTake和xSemaphoreGive這2句代碼保留、不保留

• 保留：實驗現(xiàn)象如下圖左邊，任務1、任務2的打印信息沒有混在一起
• 不保留：實驗現(xiàn)象如下圖右邊，打印信息混雜在一起

程序運行結果如下圖所示：

7.4 示例16: 誰上鎖就由誰解鎖？

互斥量、互斥鎖，本來的概念確實是：誰上鎖就得由誰解鎖。

但是FreeRTOS并沒有實現(xiàn)這點，只是要求程序員按照這樣的慣例寫代碼。

本節(jié)代碼為： FreeRTOS_16_mutex_who_give 。

main函數(shù)創(chuàng)建了2個任務：

• 任務1：高優(yōu)先級，一開始就獲得互斥鎖，永遠不釋放。
• 任務2：任務1阻塞時它開始執(zhí)行，它先嘗試獲得互斥量，失敗的話就監(jiān)守自盜(釋放互斥量、開鎖)，然后再上鎖

代碼如下：

int main( void )
{
	prvSetupHardware();
	
    /* 創(chuàng)建互斥量 */
    xMutex = xSemaphoreCreateMutex( );

	if( xMutex != NULL )
	{
		/* 創(chuàng)建2個任務: 一個上鎖, 另一個自己監(jiān)守自盜(開別人的鎖自己用)
		 */
		xTaskCreate( vTakeTask, "Task1", 1000, NULL, 2, NULL );
		xTaskCreate( vGiveAndTakeTask, "Task2", 1000, NULL, 1, NULL );

		/* 啟動調度器 */
		vTaskStartScheduler();
	}
	else
	{
		/* 無法創(chuàng)建互斥量 */
	}

	/* 如果程序運行到了這里就表示出錯了, 一般是內存不足 */
	return 0;
}

兩個任務的代碼和執(zhí)行流程如下圖所示：

• A：任務1的優(yōu)先級高，先運行，立刻上鎖
• B：任務1阻塞
• C：任務2開始執(zhí)行，嘗試獲得互斥量(上鎖)，超時時間設為0。根據(jù)返回值打印出：上鎖失敗
• D：任務2監(jiān)守自盜，開鎖，成功！
• E：任務2成功獲得互斥量
• F：任務2阻塞

可見，任務1上的鎖，被任務2解開了。所以，F(xiàn)reeRTOS并沒有實現(xiàn)"誰上鎖就得由誰開鎖"的功能。

程序運行結果如下圖所示：

7.5 示例17: 優(yōu)先級反轉

假設任務A、B都想使用串口，A優(yōu)先級比較低：

• 任務A獲得了串口的互斥量
• 任務B也想使用串口，它將會阻塞、等待A釋放互斥量
• 高優(yōu)先級的任務，被低優(yōu)先級的任務延遲，這被稱為"優(yōu)先級反轉"(priority inversion)

如果涉及3個任務，可以讓"優(yōu)先級反轉"的后果更加惡劣。

本節(jié)代碼為： FreeRTOS_17_mutex_inversion 。

互斥量可以通過"優(yōu)先級繼承"，可以很大程度解決"優(yōu)先級反轉"的問題，這也是FreeRTOS中互斥量和二級制信號量的差別。

本節(jié)程序使用二級制信號量來演示"優(yōu)先級反轉"的惡劣后果。

main函數(shù)創(chuàng)建了3個任務：LPTask/MPTask/HPTask(低/中/高優(yōu)先級任務)，代碼如下：

/* 互斥量/二進制信號量句柄 */
SemaphoreHandle_t xLock;

int main( void )
{
	prvSetupHardware();
	
    /* 創(chuàng)建互斥量/二進制信號量 */
    xLock = xSemaphoreCreateBinary( );


	if( xLock != NULL )
	{
		/* 創(chuàng)建3個任務: LP,MP,HP(低/中/高優(yōu)先級任務)
		 */
		xTaskCreate( vLPTask, "LPTask", 1000, NULL, 1, NULL );
		xTaskCreate( vMPTask, "MPTask", 1000, NULL, 2, NULL );
		xTaskCreate( vHPTask, "HPTask", 1000, NULL, 3, NULL );

		/* 啟動調度器 */
		vTaskStartScheduler();
	}
	else
	{
		/* 無法創(chuàng)建互斥量/二進制信號量 */
	}

	/* 如果程序運行到了這里就表示出錯了, 一般是內存不足 */
	return 0;
}

LPTask/MPTask/HPTask三個任務的代碼和運行過程如下圖所示：

• A：HPTask優(yōu)先級最高，它最先運行。在這里故意打印，這樣才可以觀察到flagHPTaskRun的脈沖。
• HP Delay：HPTask阻塞
• B：MPTask開始運行。在這里故意打印，這樣才可以觀察到flagMPTaskRun的脈沖。
• MP Delay：MPTask阻塞
• C：LPTask開始運行，獲得二進制信號量，然后故意打印很多字符
• D：HP Delay時間到，HPTask恢復運行，它無法獲得二進制信號量，一直阻塞等待
• E：MP Delay時間到，MPTask恢復運行，它比LPTask優(yōu)先級高，一直運行。導致LPTask無法運行，自然無法釋放二進制信號量，于是HPTask用于無法運行。

總結：

• LPTask先持有二進制信號量，
• 但是MPTask搶占LPTask，是的LPTask一直無法運行也就無法釋放信號量，
• 導致HPTask任務無法運行
• 優(yōu)先級最高的HPTask竟然一直無法運行！

程序運行的時序圖如下：

7.6 示例18: 優(yōu)先級繼承

本節(jié)代碼為： FreeRTOS_18_mutex_inheritance 。

示例17的問題在于，LPTask低優(yōu)先級任務獲得了鎖，但是它優(yōu)先級太低而無法運行。

如果能提升LPTask任務的優(yōu)先級，讓它能盡快運行、釋放鎖，"優(yōu)先級反轉"的問題不就解決了嗎？

把LPTask任務的優(yōu)先級提升到什么水平？

優(yōu)先級繼承：

• 假設持有互斥鎖的是任務A，如果更高優(yōu)先級的任務B也嘗試獲得這個鎖
• 任務B說：你既然持有寶劍，又不給我，那就繼承我的愿望吧
• 于是任務A就繼承了任務B的優(yōu)先級
• 這就叫：優(yōu)先級繼承
• 等任務A釋放互斥鎖時，它就恢復為原來的優(yōu)先級
• 互斥鎖內部就實現(xiàn)了優(yōu)先級的提升、恢復

本節(jié)源碼是在FreeRTOS_17_mutex_inversion 的代碼上做了一些簡單修改：

int main( void )
{
	prvSetupHardware();
	
    /* 創(chuàng)建互斥量/二進制信號量 */
    //xLock = xSemaphoreCreateBinary( );
	xLock = xSemaphoreCreateMutex( );

運行時序圖如下圖所示：

• A：HPTask執(zhí)行xSemaphoreTake(xLock, portMAX_DELAY);，它的優(yōu)先級被LPTask繼承
• B：LPTask搶占MPTask，運行
• C：LPTask執(zhí)行xSemaphoreGive(xLock);，它的優(yōu)先級恢復為原來值
• D：HPTask得到互斥鎖，開始運行
• 互斥鎖的"優(yōu)先級繼承"，可以減小"優(yōu)先級反轉"的影響

7.7 遞歸鎖

7.7.1 死鎖的概念

日常生活的死鎖：我們只招有工作經(jīng)驗的人！我沒有工作經(jīng)驗怎么辦？那你就去找工作??！

假設有2個互斥量M1、M2，2個任務A、B：

• A獲得了互斥量M1
• B獲得了互斥量M2
• A還要獲得互斥量M2才能運行，結果A阻塞
• B還要獲得互斥量M1才能運行，結果B阻塞
• A、B都阻塞，再無法釋放它們持有的互斥量
• 死鎖發(fā)生！

7.7.2 自我死鎖

假設這樣的場景：

• 任務A獲得了互斥鎖M
• 它調用一個庫函數(shù)
• 庫函數(shù)要去獲取同一個互斥鎖M，于是它阻塞：任務A休眠，等待任務A來釋放互斥鎖！
• 死鎖發(fā)生！

7.7.3 函數(shù)

怎么解決這類問題？可以使用遞歸鎖(Recursive Mutexes)，它的特性如下：

• 任務A獲得遞歸鎖M后，它還可以多次去獲得這個鎖
• "take"了N次，要"give"N次，這個鎖才會被釋放

遞歸鎖的函數(shù)根一般互斥量的函數(shù)名不一樣，參數(shù)類型一樣，列表如下：

函數(shù)原型如下：

/* 創(chuàng)建一個遞歸鎖，返回它的句柄。
 * 此函數(shù)內部會分配互斥量結構體 
 * 返回值: 返回句柄，非NULL表示成功
 */
SemaphoreHandle_t xSemaphoreCreateRecursiveMutex( void );


/* 釋放 */
BaseType_t xSemaphoreGiveRecursive( SemaphoreHandle_t xSemaphore );

/* 獲得 */
BaseType_t xSemaphoreTakeRecursive(
                   SemaphoreHandle_t xSemaphore,
                   TickType_t xTicksToWait
               );

7.7.4 示例19: 遞歸鎖

本節(jié)代碼為： FreeRTOS_19_mutex_recursive 。

遞歸鎖實現(xiàn)了：誰上鎖就由誰解鎖。

本程序從FreeRTOS_16_mutex_who_give修改得來，它的main函數(shù)里創(chuàng)建了2個任務

• 任務1：高優(yōu)先級，一開始就獲得遞歸鎖，然后故意等待很長時間，讓任務2運行
• 任務2：低優(yōu)先級，看看能否操作別人持有的鎖

main函數(shù)代碼如下：

/* 遞歸鎖句柄 */
SemaphoreHandle_t xMutex;

int main( void )
{
	prvSetupHardware();
	
    /* 創(chuàng)建遞歸鎖 */
    xMutex = xSemaphoreCreateRecursiveMutex( );

	if( xMutex != NULL )
	{
		/* 創(chuàng)建2個任務: 一個上鎖, 另一個自己監(jiān)守自盜(看看能否開別人的鎖自己用)
		 */
		xTaskCreate( vTakeTask, "Task1", 1000, NULL, 2, NULL );
		xTaskCreate( vGiveAndTakeTask, "Task2", 1000, NULL, 1, NULL );

		/* 啟動調度器 */
		vTaskStartScheduler();
	}
	else
	{
		/* 無法創(chuàng)建遞歸鎖 */
	}

	/* 如果程序運行到了這里就表示出錯了, 一般是內存不足 */
	return 0;
}

兩個任務經(jīng)過精細設計，代碼和運行流程如下圖所示：

A：任務1優(yōu)先級最高，先運行，獲得遞歸鎖

B：任務1阻塞，讓任務2得以運行

C：任務2運行，看看能否獲得別人持有的遞歸鎖：不能

D：任務2故意執(zhí)行"give"操作，看看能否釋放別人持有的遞歸鎖：不能

E：任務2等待遞歸鎖

F：任務1阻塞時間到后繼續(xù)運行，使用循環(huán)多次獲得、釋放遞歸鎖

遞歸鎖在代碼上實現(xiàn)了：誰持有遞歸鎖，必須由誰釋放。

程序運行結果如下圖所示：

7.8 常見問題

使用互斥量的兩個任務是相同優(yōu)先級時的注意事項。