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調用的時候，就會喚醒在阻塞隊列中的線程，繼續執行下面的代碼。