Linux–多執行緒（二）

• 2022 年 10 月 30 日
• 筆記
• Linux高並發伺服器

執行緒的同步和互斥

基本概念

概述：現在作業系統基本都是多任務的作業系統，同時有大量可以調度的實體在運行。在多任務作業系統當中，同時運行的多個任務可能：

• 都需要訪問/使用同一種資源
• 多個任務之間有依賴關係，某個任務的運行依賴於另一個任務

同步和互斥就是用來解決上述兩個問題的。

同步和互斥的概念：

• 互斥是要求兩個任務不能同時佔用資源，會相互排序，必須等待一個執行緒運行完畢，另外一個執行緒才能過來使用資源。
• 同步是一種更為複雜的互斥，在互斥的基礎上，要求兩個任務的執行存在先後順序。

其他相關概念：

• 臨界資源： 多執行緒執行流共享的資源就叫做臨界資源
• 臨界區： 每個執行緒內部，訪問臨界資源的程式碼，就叫做臨界區
• 原子性： 不會被任何調度機制打斷的操作，該操作只有兩態（無中間態，即使被打斷，也不會受影響），要麼完成，要麼未完成

互斥量mutex

概念： 多個執行緒對一個共享變數進行操控時，會引發數據不一致的問題。此時就引入了互斥量（也叫互斥鎖）的概念，來保證共享數據操作的完整性。在被加鎖的任一時刻，臨界區的程式碼只能被一個執行緒訪問。

互斥鎖是一種簡單的加鎖的方法來控制對共享資源的訪問，互斥鎖只有兩種狀態，即加鎖（lock）和解鎖（unlock）。

程式碼的要求：

• 程式碼必須要有互斥行為：當程式碼進入臨界區執行時，不允許其他執行緒進入該臨界區。
• 如果多個執行緒同時要求執行臨界區的程式碼，並且臨界區沒有執行緒在執行，那麼只能允許一個執行緒進入該臨界區。
• 如果執行緒不在臨界區中執行，那麼該執行緒不能阻止其他執行緒進入臨界區。

互斥量的介面

互斥量其實就是一把鎖，是一個類型為pthread_mutex_t 的變數，使用前需要進行初始化操作，使用完之後需要對鎖資源進行釋放。

• 初始化互斥量

int pthread_mutex_init(pthread_mutex_t *restrict mutex, const pthread_mutexattr_t *restrict attr); 
功能：
	初始化一個互斥鎖
參數：
	mutex：互斥鎖地址，類型是pthread_mutex_t
	attr：設置互斥量的屬性，通常可採取默認屬性，即可將attr改為NULL
	可以使用宏pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER靜態初始化互斥鎖

這種方法等價於使用NULL指定的attr參數調用pthread_mutex_init()來完成動態初始化，不同之處在於PTHREAD_MUTEX_INITIALIZER宏不進行錯誤檢查

返回值：
	成功：0 成功申請的鎖默認是打開的
	失敗：非0 錯誤碼

注意：restrict是C語言中的一種類型限定符，用於告訴編譯器，對象已經被指針引用，不能通過除該指針外所有其他直接或者間接的方式修改該對象的內容。

• 加鎖

int pthread_mutex_lock(pthread—mutex—t *mutex)；
功能：
	對互斥鎖上鎖，若互斥鎖已經上鎖，則調用者阻塞，直到互斥鎖解鎖後再上鎖。
參數：
	mutex：互斥鎖地址。
返回值：
	成功：0
	失敗：非0錯誤碼
	
int pthread_mutex_trylock(pthread_mutex_t *mutex)；
調用該函數時，若互斤鎖未加鎖，則上鎖，返回0；
若互斥鎖已加鎖，則函數直接返回失敗，即EBUSY

• 解鎖

int pthread_mutex_unlock(pthread_mutex_t *mutex)；
功能：
	對指定的互斥鎖解鎖
參數：
	mutex：互斥鎖地址
返回值：
	成功：0
	失敗：非0錯誤碼

• 銷毀互斥量

int pthread_mutex_destroy(pthread_mdtex_t *mutex)；
功能：
	銷毀指定的一個互斥鎖。互斥鎖在使用完畢後，必須要對互斥鎖進行銷毀，以釋放資源
參數：
	mutex：互斥鎖地址
返回值：
	成功：0
	失敗：非0錯誤碼

注意：

• 使用 PTHREAD_ MUTEX_ INITIALIZER 初始化的互斥量不需要銷毀
• 不要銷毀一個已經加鎖的互斥量
• 已經銷毀的互斥量，要確保後面不會有執行緒再嘗試加鎖
• 加鎖的粒度要夠小

程式碼示例：寫了一個搶票的小程式，用全局變數ticket代表現有票數，五個執行緒分別執行搶票的操作，也就是對ticket進行減減的操作，直到票數為0就停止搶票

#include <stdio.h>
#include <pthread.h>
#include <unistd.h>

pthread_mutex_t mutex;// 創建鎖變數
//全局變數，所有執行緒共享
int ticket = 10;

void* get_tickets(void* arg)
{
	long id = (long)arg;
	while (1){
		usleep(1000);
		// 加鎖
		pthread_mutex_lock(&mutex);
		if (ticket > 0){
			// 有票
			--ticket;
			printf("執行緒%ld獲得一張票，剩餘%d張票\n",id,ticket);
			// 解鎖
			pthread_mutex_unlock(&mutex);
		}else{
			// 無票，退出
			// 解鎖
			pthread_mutex_unlock(&mutex);
			break;
		}
	}
}

int main()
{
	pthread_t t[5];
	// 初始化鎖
	pthread_mutex_init(&mutex, NULL);
	// 創建5個執行緒
	long i = 0;
	for (; i < 5; ++i)
	{
		 pthread_create(t+i, NULL, get_tickets, (void*)(i+1));
	}
	// 釋放5個執行緒
	for (i = 0; i < 5; ++i)
	{
		pthread_join(t[i], NULL);
	}
	// 銷毀鎖
	pthread_mutex_destroy(&mutex);
	return 0;
}

運行結果如下：

總結幾點並回答幾個問題：

鎖的作用： 對臨界區進行保護，所有的執行流執行緒都必須遵守這個規則：lock——>訪問臨界區——>unlock

需要注意的點：

• 所有的執行緒必須看到同一把鎖，鎖本身就是臨界資源，所以鎖本身需要先保證自身安全申請鎖的過程不能出現中間態，必須保證原子性
• 任一執行緒持有鎖之後，其它執行緒如果還想申請鎖時申請不到的，保證互斥性

執行緒申請不到鎖此時會做什麼？

進入等待隊列進行等待，從運行隊列轉移到等待隊列，狀態由R變成S，持有鎖的執行緒unlock之後，需要喚醒等待隊列中的第一個執行緒

struct mutex
{ 	int lock;// 0 1 	
     // ... 	
     sturct wait_queue;//鎖下的等待隊列 
} 

互斥量的原理

大多數體系結構都提供了swap或exchange指令，該指令的作用是把暫存器和記憶體單元的數據相交換,由於只有一條指令，保證了原子性，即使是多處理器平台,訪問記憶體的匯流排周期也有先後，一個處理器上的交換指令執行時另一個處理器的交換指令只能等待匯流排周期。
下面是lock和unlock的偽程式碼

lock:
	movb $0, %a1     # 把0值放進暫存器a1里
	xchgb %a1, mutex # 交換a1暫存器的內容和鎖的值（無執行緒使用鎖時，metux的值為1） 
	if (%a1 > 0)
		return 0; # 得到鎖
	else
		掛起等待;
	goto lock;
unlock:
	movb $1 mutex  #把1賦給鎖	
	喚醒等待的執行緒;
	return 0;

在上述加鎖的偽程式碼中演示了上步驟：

1. 對暫存器的內容進行清0
2. 把mutex的值（被使用值為0，未被使用值為1）和暫存器的內容進行交換
3. 暫存器的內容為1代表得到了鎖，為0代表未得到鎖，要掛起等待

解鎖的偽程式碼步驟（只有有鎖的執行緒才可以執行到這段程式碼）：

1. 把mutex的值改為1
2. 喚醒等待鎖的執行緒

死鎖

概念： 死鎖是指兩個或兩個以上的進程在執行過程中，由於競爭資源或者由於彼此通訊而造成的一種阻塞的現象，若無外力作用，它們都將無法推進下去。此時稱系統處於死鎖狀態或系統產生了死鎖，這些永遠在互相等待的進程稱為死鎖進程。

舉個例子：

這裡執行緒1先申請資源1，申請到了之後，資源1被鎖死（資源1會永遠被執行緒1申請，因為只有申請到資源2執行完臨界程式碼，才會釋放掉資源1，此時執行緒1被卡在申請資源2的點，根本走不到釋放資源1的程式碼，所以會一直被執行緒1佔有），執行緒2無法申請，執行緒2先申請資源2，同樣資源2也被鎖死，這樣當執行緒1繼續向下申請資源2的時候，就被阻塞在那裡，執行緒2在向下申請資源1的時候，也被阻塞在那裡，這就形成了死鎖，永遠解不了鎖。

死鎖引起的原因：

• 競爭不可搶佔資源引起死鎖：這就是上述情況，都在等待對方佔有的不可搶佔的資源
• 競爭可消耗資源引起的死鎖：有p1,p2,p3三個進程，p1向p2發送消息並接受p3發送的消息，p2向p3發送消息並接收p1的消息，p3向p1發送消息並接收p2的消息，如果設置時先接收消息後發送消息，則所有的資訊都不能發送，這就造成死鎖

死鎖產生的四個必要條件：

• 互斥條件：一個資源每次只能被一個執行流使用
• 請求與保持條件：一個執行流因請求資源而阻塞時，對已獲得的資源保持不放
• 不剝奪條件:一個執行流已獲得的資源，在末使用完之前，不能強行剝奪
• 循環等待條件:若干執行流之間形成一種頭尾相接的循環等待資源的關係

避免死鎖：

• 破壞請求和保持條件
  • 協議1：所有進程開始前，必須一次性地申請所需的所有資源，這樣運行期間就不會再提出資源的需求，破壞了請求條件，即使有一種資源不能滿足需求，也不會給它分配正在空閑的資源，這樣它就沒有資源，就破壞了保持條件，從而預防死鎖
  • 協議2：允許一個進程只獲得初期的資源就開始運行，然後再把運行完的資源釋放出來，然後再請求新的資源
• 破壞不可搶佔條件
  • 當一個已經保持了某種不可搶佔資源的進程，提出新資源請求不能被滿足的時候，它必須釋放已經保持的所有資源，以後需要的時候再申請
• 破壞循環等待條件
  • 對系統中的所有資源類型進行線性排序，然後規定每個進程必須按序列號遞增的順序請求資源。加入進程請求到了一些序列號較高的資源，然後請求一個序列號較低的資源時，必須先釋放相同的更高序號的資源後才能申請低序列號的資源，多個同類資源必須一起請求
  • 將所有資源進行線性排序，每個進程申請資源的順序保持一致

實例演示：

#include <stdio.h>
#include <pthread.h>
#include <unistd.h>
//執行緒的兩個互斥量
pthread_mutex_t mutex1;
pthread_mutex_t mutex2;
//執行緒1處理函數
void *fun1(void *arg)
{
    //執行緒1先申請資源1，再申請資源2
    //加鎖
    pthread_mutex_lock(&mutex1);
    printf("執行緒1加鎖資源1ok....\n");
    pthread_mutex_lock(&mutex2);
    printf("執行緒1加鎖資源2ok....\n");
    printf("執行緒1執行臨界程式碼");
    //解鎖
    pthread_mutex_unlock(&mutex1);
    pthread_mutex_unlock(&mutex2);
    return NULL;
  }
//執行緒2處理函數
void *fun2(void* arg)
{
    //執行緒2先申請資源2，再申請資源1
    //加鎖
    pthread_mutex_lock(&mutex2);
    printf("執行緒2加鎖資源1ok....\n");
    pthread_mutex_lock(&mutex1);
    printf("執行緒2加鎖資源2ok....\n");
    printf("執行緒2執行臨界區程式碼....\n");
    //解鎖
    pthread_mutex_unlock(&mutex2);
    pthread_mutex_unlock(&mutex1);
    return NULL;
}
//演示死鎖
int main()
{
    int ret = -1;
    int ret1 = -1;
    pthread_t tid1,tid2;
    //初始化互斥量
    pthread_mutex_init(&mutex1,NULL);
    pthread_mutex_init(&mutex2,NULL);
    //創建兩個執行緒
    pthread_create(&tid1,NULL,fun1,NULL);
    pthread_create(&tid2,NULL,fun2,NULL);
    //回收資源
    ret = pthread_join(tid1,NULL);
    ret = pthread_join(tid2,NULL);
    if(0!=ret)
    {
      printf("執行緒1資源回收失敗\n");
      return 1;
    }
    if(0!=ret1)
    {
      printf("執行緒2資源回收失敗\n");
      return 1;
    }
    //銷毀互斥鎖
    pthread_mutex_destroy(&mutex1);
    pthread_mutex_destroy(&mutex2);
    return 0;
}

運行結果如下：

兩個進程都想獲得對方的鎖，造成死鎖。

條件變數

概念

利用執行緒間共享的全局變數進行同步的一種機制，主要包括兩個動作：一個執行緒等待”條件變數的條件成立”而掛起；另一個執行緒使「條件成立」（給出條件成立訊號）。為了防止競爭，條件變數的使用總是和一個互斥鎖結合在一起。

同步： 在保證數據安全的前提下，讓執行緒能夠按照某種特定的順序訪問臨界資源，從而避免飢餓問題，叫做同步

為什麼存在執行緒同步？

執行緒同步使得每個執行緒都能夠訪問臨界資源，多個執行緒協同高效完成某些任務。

條件變數如何與互斥鎖結合使用？

條件變數是包含一個等待隊列的。多個執行緒可以去競爭一把鎖，沒有得到鎖資源的執行緒會在鎖上繼續掛起等待，當擁有鎖的執行緒條件變數滿足時，會先釋放鎖資源，然後進入到條件變數的等待隊列去等待（等待其他執行緒喚醒），這樣其他執行緒就可以獲得鎖資源，如果此時喚醒的條件變數滿足，該執行緒可以去喚醒等待隊列中的第一個執行緒，自己釋放鎖資源，然後讓第一個執行緒重新擁有鎖資源，依次如此，多個執行緒就是順序地執行工作。這樣就可以實現執行緒同步的操作。

與互斥鎖不同的是，條件變數是用來等待而不是用來上鎖的，條件變數本身就不是鎖！

條件變數用來自動阻塞一個執行緒，直到某種特殊情況發生為止，通常和互斥鎖一起使用。

條件變數的兩個動作：

• 條件不滿，阻塞執行緒
• 條件滿足，通知阻塞的執行緒開始工作

條件變數的類型：pthread_cond_t

條件變數的介面

條件變數是一個類型為pthread_cond_t的條件變數，課通過定義變數的方式來定義一個條件變數

• 條件變數初始化

int pthread_cond_init(pthread_cond_t *restrict cond,  const pthread_condattr_t *restrict attr);
功能：
	初始化一個條件變數
參數：
	cond：指向要初始化的條件變數指針
	attr：條件變數屬性，通常為默認值，傳入NULL即可
		  也可以使用靜態初始化的方法，初始化條件變數：pthread_cond_t cond = PTHREAD_COND_INITIALIZER	
返回值：
	成功：0
	失敗：非0錯誤號

• 條件變數的銷毀

int pthread_cond_destroy(pthread_cond_t *cond)；
功能：
	銷毀一個條件變數
參數：
	cond：指向要始化的條件變數指針
返回值：
	成功：0
	失敗：非0錯誤號

• 等待條件變數滿足

int pthread_cond_wait(pthread_cond_t *restrict  cond,pthread_mutex_t *restrict mutex)；
功能:
	阻塞等待一個條件變數
	a)阻塞等待條件變數cond（參1）滿足
	b)釋放已掌握的互斥鎖（解鎖互斥量）相當於pthread_mutex_unlock(&mutex)；
		a)b)兩步為一個原子操作
	c)當被喚醒，pthread_cond_wait函數返回時，解除阻塞並重新申請獲取互斥鎖pthread_mutex_lock(&mutex)；
參數：
	cond：指向要初始化的條件變數指針
	mutex：互斥鎖
返回值：
	成功：0
	失敗：非0錯誤號

為什麼pthread_cond_wait需要互斥量？

條件變數是實現執行緒同步的一種手段，如果一個執行緒進入等待隊列還不釋放鎖資源，這樣其他執行緒也不能夠得到鎖資源，這樣喚醒執行緒的條件變數永遠不可能滿足，那麼這個執行緒也將一直等待下去。所以一個執行緒進入等待隊列需要釋放自己手中的鎖資源來實現真正地同步

• 喚醒條件變數

int pthread_cond_signal(pthread_cond_t *cond)
功能：
	喚醒阻塞隊列上的第一個執行緒
參數：
	cond指向要初始化的條件變數指針
返回值：
	成功：0
	失敗：非0錯誤號

int pthread_cond_broadcast(pthread_cond_t *cond)
功能:
	喚醒全部阻塞在條件變數上的執行緒
參數：
	cond：指向要初始化的條件變數指針
返回值：
	成功：0
	失敗：非0錯誤號
	
後者是喚醒等待隊列中所有的執行緒，而前者只喚醒等待隊列中的第一個執行緒。後者會帶來一個很不好的效應——驚群效應。多個執行緒同時被喚醒，但是最終只有一個執行緒能夠獲得「控制權」，其他獲得控制權失敗的執行緒可能重新進入休眠狀態。等待獲得控制權的執行緒釋放鎖資源後去通知下一個執行緒，這樣就容易引起OS和CPU的管理調度負擔，所以不建議使用。

實例演示： 創建五個執行緒，四個執行緒執行run1，上來就在條件變數下等待，另一個執行緒執行run2，然後無腦喚醒等待隊列下的執行緒

#include<stdio.h>
#include<pthread.h>
#include<unistd.h>
//創建條件變數
pthread_cond_t cond;
//創建互斥鎖
pthread_mutex_t mutex;
//執行緒處理函數1
void *threadfun1(void *arg)
{
    char* name = (char*)arg;
    while(1)
    {
      pthread_mutex_lock(&mutex);
      pthread_cond_wait(&cond,&mutex);
      printf("%s is waked up...\n",name);
      sleep(1);
      pthread_mutex_unlock(&mutex);
    }
  }
//執行緒處理函數2
void *threadfun2(void *arg)
{
    char *name = (char *)arg;
    while(1)
    {
       sleep(1);
      //喚醒一個等待隊列中的執行緒
      pthread_cond_signal(&cond);
      printf("%s is wakeding up a thread...\n",name);
    }
}
int main()
{
    pthread_t pthread1,pthread2,pthread3,pthread4,pthread5;
    //初始化條件變數
    pthread_cond_init(&cond,NULL);
    //初始化互斥鎖
    pthread_mutex_init(&mutex,NULL);
    //創建五個執行緒
    pthread_create(&pthread1,NULL,threadfun1,(void *)"pthread 1");
    pthread_create(&pthread2,NULL,threadfun1,(void *)"pthread 2");
    pthread_create(&pthread3,NULL,threadfun1,(void *)"pthread 3");
    pthread_create(&pthread4,NULL,threadfun1,(void *)"pthread 4");
    pthread_create(&pthread5,NULL,threadfun2,(void *)"pthread 5");
  
//等待執行緒結束
    pthread_join(pthread1,NULL);
    pthread_join(pthread2,NULL);
    pthread_join(pthread3,NULL);
    pthread_join(pthread4,NULL);
    pthread_join(pthread5,NULL);
  
    pthread_mutex_destroy(&mutex);
    pthread_cond_destroy(&cond);
    return 0;
}

運行結果如下：

值得注意的是pthread_cond_wait在阻塞的時候，會釋放已經掌握的互斥鎖，等到被喚醒的時候，重新上鎖。

舉個例子：

其實pthread_cond_wait內部隱藏一次解鎖的過程，如果是fun1先運行，num被上鎖，會阻塞在第24條語句，但是pthread_cond_wait會先解鎖，釋放掉num資源，但依然阻塞在24行，此時fun2加鎖，改變條件，函數pthread_cond_signal會喚醒pthread_cond_wait函數，此時num會再次被上鎖，然後解鎖，所以pthread_cond_wait其實在內部做了一次解鎖的操作。

條件變數其實很簡單，遇到pthread_cond_wait執行緒就會阻塞在阻塞隊列，當pthread_cond_signal調用的時候，就會喚醒在阻塞隊列中的執行緒，繼續執行下面的程式碼。