• 執行緒間同步方式
  • 引言
  • 互斥鎖
    • 探究底層，實現一個鎖
      • 測試並加鎖(TAS)
      • 比較並交換(CAS)
      • 另一個問題，過多的自旋？
    • 回到互斥鎖
  • 訊號量
    • 有名訊號量
    • 無名訊號量
    • 總結
  • 條件變數
    • 什麼是條件變數？
    • 相關函式
      • 1. 初始化
      • 2. 等待條件
      • 3. 通知條件
    • 用法與思考
  • 實踐——讀寫者鎖

文章已收錄至我的倉庫：Java學習筆記與免費書籍分享

執行緒間同步方式

引言

不同執行緒間對臨界區資源的訪問可能會引起常見的併發問題，我們希望執行緒原子式的執行一系列指令，但由於單處理器上的中斷，我們必須想一些其他辦法以同步各執行緒，本文就來介紹一些執行緒間的同步方式。

互斥鎖

互斥鎖(又名互斥量)，強調的是資源的訪問互斥：互斥鎖是用在多執行緒多工互斥的，當一個執行緒佔用了某一個資源，那麼別的執行緒就無法訪問，直到這個執行緒unlock，其他的執行緒才開始可以利用這個資源。

int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_trylock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);

注意理解trylock函式，這與普通的lock不一樣，普通的lock函式在資源被鎖住時會被堵塞，直到鎖被釋放。

trylock函式是非阻塞呼叫模式，也就是說如果互斥量沒被鎖住，trylock函式將把互斥量加鎖，並獲得對共享資源的訪問許可權; 如果互斥量被鎖住了，trylock函式將不會阻塞等待而直接返回EBUSY，表示共享資源處於忙狀態，這樣就可以避免死鎖或餓死等一些極端情況發生。

探究底層，實現一個鎖

實現一個鎖必須需要硬體的支援，因為我們必須要保證鎖也是併發安全的，這就需要硬體支援以保證鎖內部是原子實現的。

很容易想到維護一個全域性變數flag，當該變數為0時，允許執行緒加鎖，並設定flag為1；否則，執行緒必須掛起等待，直到flag為0.

typedef struct lock_t {
    int flag;
}lock_t;

void init(lock_t &mutex) {
    mutex->flag = 0;
}

void lock(lock_t &mutex) {
    while (mutex->flag == 1) {;} //自旋等待變數為0才可進入
    mutex->flag = 1;
}

void unlock(lock_t &mutex) {
    mutex->flag = 0;
}
 

這是基於軟體的初步實現，初始化變數為0，執行緒自旋等待變數為0才可進入，這看上去似乎並沒有什麼毛病，但是仔細思考，這是有問題的：

當執行緒恰好通過while判定時陷入中斷，此時並未設定flag為1，另一個執行緒闖入，此時flag仍然為0，通過while判定進入臨界區，此時中斷，回到原執行緒，原執行緒繼續執行，也進入臨界區，這就造成了同步問題。

在while迴圈中，僅僅設定mutex->flag == 1是不夠的，儘管他是一個原語，我們必須有更多的程式碼，同時，當我們引入更多程式碼時，我們必須保證這些程式碼也是原子的，這就意味著我們需要硬體的支援。

我們思考上面程式碼為什麼會失敗？原因是當退出while迴圈時，在這一時刻flag仍然為0，這就給了其他執行緒搶入臨界區的機會。

解決辦法也很直觀 —— 在退出while時，藉助硬體支援保證flag被設定為1。

測試並加鎖(TAS)

我們編寫如下函式：

int TestAndSet(int *old_ptr, int new) {
    int old = *old_ptr;
    *old_ptr = new;
    return old;
}

同時重新設定while迴圈：

void lock(lock_t &mutex) {
    while (TestAndSet(mutex->flag， 1) == 1) {;} //自旋等待變數為0才可進入
    mutex->flag = 1;
}

這裡，我們藉助硬體，保證TestAndSet函式是原子執行的，現在鎖可以正確的使用了。當flag為0時，我們通過while測試時已經將flag設定為1了，其他執行緒已經無法進入臨界區。

比較並交換(CAS)

我們編寫如下函式：

int CompareAndSwap(int *ptr, int expected, int new) {
    int actual = *ptr;
    if (actual == expected) {
        *ptr = new;
    }
    return actual;
}

同樣的，硬體也應該支援CAS原語以保證CAS內部也是安全的，現在重新設定while：

void lock(lock_t &mutex) {
    while (CompareAndSwap(mutex->flag， 0， 1) == 1) {;} //自旋等待變數為0才可進入
    mutex->flag = 1;
}

現在鎖可以正確的使用了，當flag為0時，我們通過while測試時已經將flag設定為1了，其他執行緒已經無法進入臨界區。

此外你可能發現CAS所需要更多的暫存器，在將來研究synchronozation時，你會發現它的妙處。

另一個問題，過多的自旋？

你可能發現了，儘管一個執行緒未能獲得鎖，其仍然在不斷while迴圈以佔用CPU資源，一個辦法就是當執行緒未能獲得鎖，進入休眠以釋放CPU資源(條件變數)，當一個執行緒釋放鎖時，喚醒一個正在休眠的執行緒。不過這樣也有缺點，進入休眠與喚醒一個鎖也是需要時間的，當一個執行緒很快就能釋放鎖時，多等等是比陷入休眠更好的選擇。

Linux下采用兩階段鎖，第一階段執行緒自旋一定時間或次數等待鎖的釋放，當達到一定時間或一定次數時，進入第二階段，此時執行緒進入休眠。

回到互斥鎖

互斥鎖提供了併發安全的基本保證，互斥鎖用於保證對臨界區資源的安全訪問，但何時需要訪問資源並不是互斥鎖應該考慮的事情，這可能是條件變數該考慮的事情。

如果執行緒頻繁的加鎖和解鎖，效率是非常低效的，這也是我們必須要考慮到的一個點。

訊號量

訊號量並不用來傳送資源，而是用來保護共享資源，理解這一點是很重要的，訊號量 s 的表示的含義為同時允許訪問資源的最大執行緒數量，它是一個全域性變數。

在程序中也可以使用訊號量，對於訊號量的理解程序中與執行緒中並無太大差異，都是用來保護資源，關於更多訊號量的理解參見這篇文章: JavaLearningNotes/程序間通訊方式。

來考慮一個上面簡單的例子：兩個執行緒同時修改而造成錯誤，我們不考慮讀者而僅僅考慮寫者程序，在這個例子中共享資源最多允許一個執行緒修改資源，因此我們初始化 s 為1。

開始時，A率先寫入資源，此時A呼叫P(s)，將 s 減一，此時 s = 0，A進入共享區工作。

此時，執行緒B也想進入共享區修改資源，它呼叫P(s)發現此時s為0，於是掛起執行緒，加入等待佇列。

A工作完畢，呼叫V(s)，它發現s為0並檢測到等待佇列不為空，於是它隨機喚醒一個等待執行緒，並將s加1，這裡喚醒了B。

B被喚醒，繼續執行P操作，此時s不為0，B成功執行將s置為0並進入工作區。

此時C想要進入工作區......

可以發現，在無論何時只有一個執行緒能夠訪問共享資源，這就是訊號量做的事情，他控制進入共享區的最大程序數量，這取決於初始化s的值。此後，在進入共享區之前呼叫P操作，出共享區後呼叫V操作，這就是訊號量的思想。

有名訊號量

有名訊號量以檔案的形式存在，即時是不同程序間的執行緒也可以訪問該訊號量，因此可以用於不同程序間的多執行緒間的互斥與同步。

建立開啟有名訊號量

sem_t *sem_open(const char *name, int oflag);
sem_t *sem_open(const char *name, int oflag, mode_t mode, unsigned int value);
//成功返回訊號量指標；失敗返回SEM_FAILED，設定errno

name是檔案路徑名，value設定為訊號量的初始值。

關閉訊號量，程序終止時，會呼叫它

int sem_close(sem_t *sem);	//成功返回0；失敗返回-1，設定errno

刪除訊號量，立即刪除訊號量名字，當其他程序都關閉它時，銷燬它

int sem_unlink(const char *name);

等待訊號量，測試訊號量的值，如果其值小於或等於0，那麼就等待（阻塞）；一旦其值變為大於0就將它減1，並返回

int sem_wait(sem_t *sem);
int sem_trywait(sem_t *sem);
//成功返回0；失敗返回-1，設定errno

當訊號量的值為0時，sem_trywait立即返回，設定errno為EAGAIN。如果被某個訊號中斷，sem_wait會過早地返回，設定errno為EINTR

發出訊號量，給它的值加1，然後喚醒正在等待該訊號量的程序或執行緒

int sem_post(sem_t *sem);

成功返回0；失敗返回-1，不會改變它的值，設定errno，該函式是非同步訊號安全的，可以在訊號處理程式裡呼叫它

無名訊號量

無名訊號量存在於程序內的虛擬空間中，對於其他程序是不可見的，因此無名訊號量用於一個程序體內各執行緒間的互斥和同步,使用如下API：

（1）sem_init 功能：用於建立一個訊號量，並初始化訊號量的值。 函式原型：

int sem_init (sem_t* sem, int pshared, unsigned int value);

函式傳入值： sem:訊號量。pshared:決定訊號量能否在幾個程序間共享。由於目前LINUX還沒有實現程序間共享資訊量，所以這個值只能取0。

（2）其他函式

int sem_wait       (sem_t* sem);
int sem_trywait   (sem_t* sem);
int sem_post       (sem_t* sem);
int sem_getvalue (sem_t* sem);
int sem_destroy   (sem_t* sem);

功能：

sem_wait和sem_trywait相當於P操作，它們都能將訊號量的值減一，兩者的區別在於若訊號量的值小於零時，sem_wait將會阻塞程序，而sem_trywait則會立即返回。

sem_post相當於V操作，它將訊號量的值加一，同時發出喚醒的訊號給等待的執行緒。

sem_getvalue 得到訊號量的值。

sem_destroy 摧毀訊號量。

如果某個基於記憶體的訊號量是在不同程序間同步的，該訊號燈必須存放在共享記憶體區中，這要只要該共享記憶體區存在，該訊號燈就存在。

總結

無名訊號量存在於記憶體中，有名訊號量是存在於磁碟上的，因此無名訊號量的速度更快，但只適用於一個獨立程序內的各執行緒；有名訊號量可以速度欠缺，但可以使不同程序間的執行緒同步，這是通過共享記憶體實現的，共享記憶體是程序間的一種通訊方式。

你可能發現了，當訊號量的值s為1時，訊號量的作用於互斥鎖的作用是一樣的，互斥鎖只能允許一個執行緒進入臨界區，而訊號量允許更多的執行緒進入臨界區，這取決於訊號量的值為多少。

條件變數

什麼是條件變數？

在互斥鎖中，執行緒等待flag為0才能進入臨界區；訊號量中P操作也要等待s不為0......在多執行緒中，一個執行緒等待某個條件是很常見的，互斥鎖實現一節中，我們採用自旋是否有一個更專門、更高效的方式實現條件的等待？

它就是條件變數！條件變數(condition variable)是利用執行緒間共享的全域性變數進行同步的一種機制，主要包括兩個動作：一個執行緒等待某個條件為真，而將自己掛起；另一個執行緒設定條件為真，並通知等待的執行緒繼續。

由於某個條件是全域性變數，因此條件變數常使用互斥鎖以保護(這是必須的，是被強制要求的)。

條件變數與互斥量一起使用時，允許執行緒以無競爭的方式等待特定的條件發生。

執行緒可以使用條件變數來等待某個條件為真，注意理解並不是等待條件變數為真，條件變數(cond)是在多執行緒程式中用來實現"等待-->喚醒"邏輯常用的方法，用於維護一個條件(與是條件變數不同的概念)，並不是說等待條件變數為真或為假。條件變數是一個顯式的佇列，當條件不滿足時，執行緒將自己加入等待佇列，同時釋放持有的互斥鎖；當一個執行緒改變條件時，可以喚醒一個或多個等待執行緒(注意此時條件不一定為真)。

在條件變數上有兩種基本操作：

• 等待（wait）：一個執行緒處於等待佇列中休眠，此時執行緒不會佔用互斥量，當執行緒被喚醒後，重新獲得互斥鎖(可能是多個執行緒競爭)，並重新獲得互斥量。
• 通知（signal/notify）：當條件更改時，另一個執行緒傳送通知以喚醒等待佇列中的執行緒。

相關函式

1. 初始化

條件變數採用的資料型別是pthread_cond_t,，在使用之前必須要進行初始化,，這包括兩種方式:

靜態: 直接設定條件變數cond為常量PTHREAD_COND_INITIALIZER。

動態: pthread_cond_init函式, 是釋放動態條件變數的記憶體空間之前, 要用pthread_cond_destroy對其進行清理。

int pthread_cond_init(pthread_cond_t *restrict cond, pthread_condattr_t *restrict attr);
int pthread_cond_destroy(pthread_cond_t *cond);
//成功則返回0, 出錯則返回錯誤編號.

注意：條件變數佔用的空間並未被釋放。

cond：要初始化的條件變數；attr：一般為NULL。

2. 等待條件

int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restric mutex);
int pthread_cond_timedwait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex, const struct timespec *restrict timeout);
//成功則返回0, 出錯則返回錯誤編號.

這兩個函式分別是阻塞等待和超時等待，堵塞等到進入等待佇列休眠直到條件修改而被喚醒；超時等待在休眠一定時間後自動醒來。

進入等待時執行緒釋放互斥鎖，而在被喚醒時執行緒重新獲得鎖。

3. 通知條件

int pthread_cond_signal(pthread_cond_t *cond);
int pthread_cond_broadcast(pthread_cond_t *cond);
//成功則返回0, 出錯則返回錯誤編號.

這兩個函式用於通知執行緒條件已被修改，呼叫這兩個函式向執行緒或條件傳送訊號。

用法與思考

條件變數用法模板：

pthread_cond_t cond;  //條件變數
mutex_t mutex;	//互斥鎖
int flag; //條件

//A執行緒
void threadA() {
    Pthread_mutex_lock(&mutex);  //保護臨界資源，因為執行緒會修改全域性條件flag
    while (flag == 1) //等待某條件成立
        Pthread_cond_wait(&cond, &mutex);  //不成立則加入佇列休眠，並釋放鎖
    ....dosomthing
    ....change flag   //條件被修改
    Pthread_cond_signal(&cond); //傳送訊號通知條件被修改
    Pthread_mutex_unlock(&mutex); //放鬆訊號後儘量快速釋放鎖，因為被喚醒的執行緒會嘗試獲得鎖
}


//B執行緒
void threadB() {
    Pthread_mutex_lock(&mutex);  //保護臨界資源
    while (flag == 0) //等待某條件成立
        Pthread_cond_wait(&cond, &mutex);  //不成立則加入佇列休眠，並釋放鎖
    ....dosomthing
    ....change flag   //條件被修改
    Pthread_cond_signal(&cond); //放鬆訊號後儘量快速釋放鎖，因為被喚醒的執行緒會嘗試獲得鎖
    Pthread_mutex_unlock(&mutex);
}

通過上面的一個例子，應該很好理解條件變數與條件的區別，條件變數是一個邏輯，它並不是while迴圈裡的bool語句，我相信很多初學者都有這麼一個誤區，即條件變數就是執行緒需要等待的條件。條件是條件，執行緒等待條件而不是等待條件變數，條件變數使得執行緒更高效的等待條件成立，是一組等待 — 喚醒 的邏輯。

注意這裡仍然要使用while迴圈等待條件，你可能會認為明明已經上鎖了別的執行緒無法強入。事實上當執行緒A陷入休眠時會釋放鎖，而當其被喚醒時，會嘗試獲得鎖，而正在其嘗試獲得鎖時，另一個執行緒B現在嘗試獲得鎖，並且搶到鎖進入臨界區，然後修改條件，使得執行緒A的條件不再成立，執行緒B返回，此時執行緒A終於獲得鎖了，並進入臨界區，但此時執行緒A的條件根本已經不成立，他不該進入臨界區！

此外，被喚醒也不代表條件成立了，例如上述程式碼執行緒B修改flag = 3，並且喚醒執行緒A，這裡執行緒A的條件根本不符合，所以必須重複判定條件。互斥鎖和條件變數的例子告訴我們：在等待條件時，總是使用while而不是if！

陷入休眠的執行緒必須釋放鎖也是有意義的，如果不釋放鎖，其他執行緒根本無法修改條件，休眠的執行緒永遠都不會醒過來！

實踐——讀寫者鎖

讀取鎖——共享；寫入鎖——獨佔。即：讀執行緒可以加多個，而寫執行緒只能有一個，並且讀者和寫者不能同時工作。

這種情況下由於允許多個讀者共享臨界區效率會高效，我們來考慮實現的問題：只允許一個寫者工作，那麼一定需要一個互斥量或二值訊號量來維護，我們稱為寫者鎖；由於讀者和寫者不能同時工作，第一個讀者必須嘗試獲取寫者鎖，而一旦讀者數量大於1，則後續讀者無須嘗試獲取寫者鎖而可直接進入，注意到這裡存在全域性讀者數量變數，因此讀者也需要一個鎖以維護全域性讀者數量，最後一個退出的讀者必須負責釋放讀者鎖。

知曉原理，快去自己動手實現一個讀寫者鎖把！

Linux下通過pthread_rwlock函式族實現。