02.28 一文帶你懟明白進程和線程通信原理

進程間通信

進程是需要頻繁的和其他進程進行交流的。例如，在一個 shell 管道中，第一個進程的輸出必須傳遞給第二個進程，這樣沿著管道進行下去。因此，進程之間如果需要通信的話，必須要使用一種良好的數據結構以至於不能被中斷。下面我們會一起討論有關 進程間通信(Inter Process Communication, IPC) 的問題。

關於進程間的通信，這裡有三個問題

• 上面提到了第一個問題，那就是一個進程如何傳遞消息給其他進程。
• 第二個問題是如何確保兩個或多個線程之間不會相互干擾。例如，兩個航空公司都試圖為不同的顧客搶購飛機上的最後一個座位。
• 第三個問題是數據的先後順序的問題，如果進程 A 產生數據並且進程 B 打印數據。則進程 B 打印數據之前需要先等 A 產生數據後才能夠進行打印。

需要注意的是，這三個問題中的後面兩個問題同樣也適用於線程

第一個問題在線程間比較好解決，因為它們共享一個地址空間，它們具有相同的運行時環境，可以想象你在用高級語言編寫多線程代碼的過程中，線程通信問題是不是比較容易解決？

另外兩個問題也同樣適用於線程，同樣的問題可用同樣的方法來解決。我們後面會慢慢討論這三個問題，你現在腦子中大致有個印象即可。

競態條件

在一些操作系統中，協作的進程可能共享一些彼此都能讀寫的公共資源。公共資源可能在內存中也可能在一個共享文件。為了講清楚進程間是如何通信的，這裡我們舉一個例子：一個後臺打印程序。當一個進程需要打印某個文件時，它會將文件名放在一個特殊的後臺目錄(spooler directory)中。另一個進程 打印後臺進程(printer daemon) 會定期的檢查是否需要文件被打印，如果有的話，就打印並將該文件名從目錄下刪除。

假設我們的後臺目錄有非常多的 槽位(slot)，編號依次為 0，1，2，...，每個槽位存放一個文件名。同時假設有兩個共享變量：out，指向下一個需要打印的文件；in，指向目錄中下個空閒的槽位。可以把這兩個文件保存在一個所有進程都能訪問的文件中，該文件的長度為兩個字。在某一時刻，0 至 3 號槽位空，4 號至 6 號槽位被佔用。在同一時刻，進程 A 和 進程 B 都決定將一個文件排隊打印，情況如下

墨菲法則(Murphy) 中說過，任何可能出錯的地方終將出錯，這句話生效時，可能發生如下情況。

進程 A 讀到 in 的值為 7，將 7 存在一個局部變量 next_free_slot 中。此時發生一次時鐘中斷，CPU 認為進程 A 已經運行了足夠長的時間，決定切換到進程 B 。進程 B 也讀取 in 的值，發現是 7，然後進程 B 將 7 寫入到自己的局部變量 next_free_slot 中，在這一時刻兩個進程都認為下一個可用槽位是 7 。

進程 B 現在繼續運行，它會將打印文件名寫入到 slot 7 中，然後把 in 的指針更改為 8 ，然後進程 B 離開去做其他的事情

現在進程 A 開始恢復運行，由於進程 A 通過檢查 next_free_slot也發現 slot 7 的槽位是空的，於是將打印文件名存入 slot 7 中，然後把 in 的值更新為 8 ，由於 slot 7 這個槽位中已經有進程 B 寫入的值，所以進程 A 的打印文件名會把進程 B 的文件覆蓋，由於打印機內部是無法發現是哪個進程更新的，它的功能比較侷限，所以這時候進程 B 永遠無法打印輸出，類似這種情況，即兩個或多個線程同時對一共享數據進行修改，從而影響程序運行的正確性時，這種就被稱為競態條件(race condition)。調試競態條件是一種非常困難的工作，因為絕大多數情況下程序運行良好，但在極少數的情況下會發生一些無法解釋的奇怪現象。不幸的是，多核增長帶來的這種問題使得競態條件越來越普遍。

臨界區

不僅共享資源會造成競態條件，事實上共享文件、共享內存也會造成競態條件、那麼該如何避免呢？或許一句話可以概括說明：禁止一個或多個進程在同一時刻對共享資源（包括共享內存、共享文件等）進行讀寫。換句話說，我們需要一種 互斥(mutual exclusion) 條件，這也就是說，如果一個進程在某種方式下使用共享變量和文件的話，除該進程之外的其他進程就禁止做這種事（訪問統一資源）。上面問題的糾結點在於，在進程 A 對共享變量的使用未結束之前進程 B 就使用它。在任何操作系統中，為了實現互斥操作而選用適當的原語是一個主要的設計問題，接下來我們會著重探討一下。

避免競爭問題的條件可以用一種抽象的方式去描述。大部分時間，進程都會忙於內部計算和其他不會導致競爭條件的計算。然而，有時候進程會訪問共享內存或文件，或者做一些能夠導致競態條件的操作。我們把對共享內存進行訪問的程序片段稱作 臨界區域(critical region) 或 臨界區(critical section)。如果我們能夠正確的操作，使兩個不同進程不可能同時處於臨界區，就能避免競爭條件，這也是從操作系統設計角度來進行的。

儘管上面這種設計避免了競爭條件，但是不能確保併發線程同時訪問共享數據的正確性和高效性。一個好的解決方案，應該包含下面四種條件

1. 任何時候兩個進程不能同時處於臨界區
2. 不應對 CPU 的速度和數量做任何假設
3. 位於臨界區外的進程不得阻塞其他進程
4. 不能使任何進程無限等待進入臨界區

使用臨界區的互斥

從抽象的角度來看，我們通常希望進程的行為如上圖所示，在 t1 時刻，進程 A 進入臨界區，在 t2 的時刻，進程 B 嘗試進入臨界區，因為此時進程 A 正在處於臨界區中，所以進程 B 會阻塞直到 t3 時刻進程 A 離開臨界區，此時進程 B 能夠允許進入臨界區。最後，在 t4 時刻，進程 B 離開臨界區，系統恢復到沒有進程的原始狀態。

忙等互斥

下面我們會繼續探討實現互斥的各種設計，在這些方案中，當一個進程正忙於更新其關鍵區域的共享內存時，沒有其他進程會進入其關鍵區域，也不會造成影響。

屏蔽中斷

在單處理器系統上，最簡單的解決方案是讓每個進程在進入臨界區後立即屏蔽所有中斷，並在離開臨界區之前重新啟用它們。屏蔽中斷後，時鐘中斷也會被屏蔽。CPU 只有發生時鐘中斷或其他中斷時才會進行進程切換。這樣，在屏蔽中斷後 CPU 不會切換到其他進程。所以，一旦某個進程屏蔽中斷之後，它就可以檢查和修改共享內存，而不用擔心其他進程介入訪問共享數據。

這個方案可行嗎？進程進入臨界區域是由誰決定的呢？不是用戶進程嗎？當進程進入臨界區域後，用戶進程關閉中斷，如果經過一段較長時間後進程沒有離開，那麼中斷不就一直啟用不了，結果會如何？可能會造成整個系統的終止。而且如果是多處理器的話，屏蔽中斷僅僅對執行 disable 指令的 CPU 有效。其他 CPU 仍將繼續運行，並可以訪問共享內存。

另一方面，對內核來說，當它在執行更新變量或列表的幾條指令期間將中斷屏蔽是很方便的。例如，如果多個進程處理就緒列表中的時候發生中斷，則可能會發生競態條件的出現。所以，屏蔽中斷對於操作系統本身來說是一項很有用的技術，但是對於用戶線程來說，屏蔽中斷卻不是一項通用的互斥機制。

鎖變量

作為第二種嘗試，可以尋找一種軟件層面解決方案。考慮有單個共享的（鎖）變量，初始為值為 0 。當一個線程想要進入關鍵區域時，它首先會查看鎖的值是否為 0 ，如果鎖的值是 0 ，進程會把它設置為 1 並讓進程進入關鍵區域。如果鎖的狀態是 1，進程會等待直到鎖變量的值變為 0 。因此，鎖變量的值是 0 則意味著沒有線程進入關鍵區域。如果是 1 則意味著有進程在關鍵區域內。我們對上圖修改後，如下所示

這種設計方式是否正確呢？是否存在紕漏呢？假設一個進程讀出鎖變量的值並發現它為 0 ，而恰好在它將其設置為 1 之前，另一個進程調度運行，讀出鎖的變量為0 ，並將鎖的變量設置為 1 。然後第一個線程運行，把鎖變量的值再次設置為 1，此時，臨界區域就會有兩個進程在同時運行。

也許有的讀者可以這麼認為，在進入前檢查一次，在要離開的關鍵區域再檢查一次不就解決了嗎？實際上這種情況也是於事無補，因為在第二次檢查期間其他線程仍有可能修改鎖變量的值，換句話說，這種 set-before-check 不是一種 原子性 操作，所以同樣還會發生競爭條件。

嚴格輪詢法

第三種互斥的方式先拋出來一段代碼，這裡的程序是用 C 語言編寫，之所以採用 C 是因為操作系統普遍是用 C 來編寫的（偶爾會用 C++），而基本不會使用 Java 、Modula3 或 Pascal 這樣的語言，Java 中的 native 關鍵字底層也是 C 或 C++ 編寫的源碼。對於編寫操作系統而言，需要使用 C 語言這種強大、高效、可預知和有特性的語言，而對於 Java ，它是不可預知的，因為它在關鍵時刻會用完存儲器，而在不合適的時候會調用垃圾回收機制回收內存。在 C 語言中，這種情況不會發生，C 語言中不會主動調用垃圾回收回收內存。有關 C 、C++ 、Java 和其他四種語言的比較可以參考

進程 0 的代碼

<code>while(TRUE){  while(turn != 0){    /* 進入關鍵區域 */    critical_region();    turn = 1;    /* 離開關鍵區域 */    noncritical_region();  }}/<code>

進程 1 的代碼

<code>while(TRUE){  while(turn != 1){    critical_region();    turn = 0;    noncritical_region();  }}/<code>

在上面代碼中，變量 turn，初始值為 0 ，用於記錄輪到那個進程進入臨界區，並檢查或更新共享內存。開始時，進程 0 檢查 turn，發現其值為 0 ，於是進入臨界區。進程 1 也發現其值為 0 ，所以在一個等待循環中不停的測試 turn，看其值何時變為 1。連續檢查一個變量直到某個值出現為止，這種方法稱為 忙等待(busywaiting)。由於這種方式浪費 CPU 時間，所以這種方式通常應該要避免。只有在有理由認為等待時間是非常短的情況下，才能夠使用忙等待。用於忙等待的鎖，稱為 自旋鎖(spinlock)。

進程 0 離開臨界區時，它將 turn 的值設置為 1，以便允許進程 1 進入其臨界區。假設進程 1 很快便離開了臨界區，則此時兩個進程都處於臨界區之外，turn 的值又被設置為 0 。現在進程 0 很快就執行完了整個循環，它退出臨界區，並將 turn 的值設置為 1。此時，turn 的值為 1，兩個進程都在其臨界區外執行。

突然，進程 0 結束了非臨界區的操作並返回到循環的開始。但是，這時它不能進入臨界區，因為 turn 的當前值為 1，此時進程 1 還忙於非臨界區的操作，進程 0 只能繼續 while 循環，直到進程 1 把 turn 的值改為 0 。這說明，在一個進程比另一個進程執行速度慢了很多的情況下，輪流進入臨界區並不是一個好的方法。

這種情況違反了前面的敘述 3 ，即 位於臨界區外的進程不得阻塞其他進程，進程 0 被一個臨界區外的進程阻塞。由於違反了第三條，所以也不能作為一個好的方案。

Peterson 解法

荷蘭數學家 T.Dekker 通過將鎖變量與警告變量相結合，最早提出了一個不需要嚴格輪換的軟件互斥算法，關於 Dekker 的算法，參考 鏈接

後來， G.L.Peterson 發現了一種簡單很多的互斥算法，它的算法如下

<code>#define FALSE 0#define TRUE  1#define N     2    /* 進程數量 */int turn;/* 現在輪到誰 */int interested[N];/* 所有值初始化為 0 (FALSE) */void enter_region(int process){/* 進程是 0 或 1 */    int other;/* 另一個進程號 */    other = 1 - process;/* 另一個進程 */  interested[process] = TRUE;/* 表示願意進入臨界區 */  turn = process;  while(turn == process         && interested[other] == true){}         /* 空循環 */  }void leave_region(int process){    interested[process] == FALSE;    /* 表示離開臨界區 */}/<code>

在使用共享變量時（即進入其臨界區）之前，各個進程使用各自的進程號 0 或 1 作為參數來調用 enter_region，這個函數調用在需要時將使進程等待，直到能夠安全的臨界區。在完成對共享變量的操作之後，進程將調用 leave_region 表示操作完成，並且允許其他進程進入。

現在來看看這個辦法是如何工作的。一開始，沒有任何進程處於臨界區中，現在進程 0 調用 enter_region。它通過設置數組元素和將 turn 置為 0 來表示它希望進入臨界區。由於進程 1 並不想進入臨界區，所以 enter_region 很快便返回。如果進程現在調用 enter_region，進程 1 將在此處掛起直到 interested[0] 變為 FALSE，這種情況只有在進程 0 調用 leave_region 退出臨界區時才會發生。

那麼上面討論的是順序進入的情況，現在來考慮一種兩個進程同時調用 enter_region 的情況。它們都將自己的進程存入 turn，但只有最後保存進去的進程號才有效，前一個進程的進程號因為重寫而丟失。假如進程 1 是最後存入的，則 turn 為 1 。當兩個進程都運行到 while 的時候，進程 0 將不會循環並進入臨界區，而進程 1 將會無限循環且不會進入臨界區，直到進程 0 退出位置。

TSL 指令

現在來看一種需要硬件幫助的方案。一些計算機，特別是那些設計為多處理器的計算機，都會有下面這條指令

<code>TSL RX,LOCK/<code>

稱為 測試並加鎖(test and set lock)，它將一個內存字 lock 讀到寄存器 RX 中，然後在該內存地址上存儲一個非零值。讀寫指令能保證是一體的，不可分割的，一同執行的。在這個指令結束之前其他處理器均不允許訪問內存。執行 TSL 指令的 CPU 將會鎖住內存總線，用來禁止其他 CPU 在這個指令結束之前訪問內存。

很重要的一點是鎖住內存總線和禁用中斷不一樣。禁用中斷並不能保證一個處理器在讀寫操作之間另一個處理器對內存的讀寫。也就是說，在處理器 1 上屏蔽中斷對處理器 2 沒有影響。讓處理器 2 遠離內存直到處理器 1 完成讀寫的最好的方式就是鎖住總線。這需要一個特殊的硬件（基本上，一根總線就可以確保總線由鎖住它的處理器使用，而其他的處理器不能使用）

為了使用 TSL 指令，要使用一個共享變量 lock 來協調對共享內存的訪問。當 lock 為 0 時，任何進程都可以使用 TSL 指令將其設置為 1，並讀寫共享內存。當操作結束時，進程使用 move 指令將 lock 的值重新設置為 0 。

這條指令如何防止兩個進程同時進入臨界區呢？下面是解決方案

<code>enter_region:TSL REGISTER,LOCK             | 複製鎖到寄存器並將鎖設為1  CMP REGISTER,#0      | 鎖是 0 嗎？  JNE enter_region  | 若不是零，說明鎖已被設置，所以循環  RET  | 返回調用者，進入臨界區        leave_region:MOVE LOCK,#0  | 在鎖中存入 0   RET  | 返回調用者/<code>

我們可以看到這個解決方案的思想和 Peterson 的思想很相似。假設存在如下共 4 指令的彙編語言程序。第一條指令將 lock 原來的值複製到寄存器中並將 lock 設置為 1 ，隨後這個原來的值和 0 做對比。如果它不是零，說明之前已經被加過鎖，則程序返回到開始並再次測試。經過一段時間後（可長可短），該值變為 0 （當前處於臨界區中的進程退出臨界區時），於是過程返回，此時已加鎖。要清除這個鎖也比較簡單，程序只需要將 0 存入 lock 即可，不需要特殊的同步指令。

現在有了一種很明確的做法，那就是進程在進入臨界區之前會先調用 enter_region，判斷是否進行循環，如果lock 的值是 1 ，進行無限循環，如果 lock 是 0，不進入循環並進入臨界區。在進程從臨界區返回時它調用 leave_region，這會把 lock 設置為 0 。與基於臨界區問題的所有解法一樣，進程必須在正確的時間調用 enter_region 和 leave_region ，解法才能奏效。

還有一個可以替換 TSL 的指令是 XCHG，它原子性的交換了兩個位置的內容，例如，一個寄存器與一個內存字，代碼如下

<code>enter_region:MOVE REGISTER,#1| 把 1 放在內存器中XCHG REGISTER,LOCK| 交換寄存器和鎖變量的內容CMP REGISTER,#0| 鎖是 0 嗎？JNE enter_region| 若不是 0 ，鎖已被設置，進行循環RET| 返回調用者，進入臨界區leave_region:MOVE LOCK,#0| 在鎖中存入 0 RET| 返回調用者/<code>

XCHG 的本質上與 TSL 的解決辦法一樣。所有的 Intel x86 CPU 在底層同步中使用 XCHG 指令。

睡眠與喚醒

上面解法中的 Peterson 、TSL 和 XCHG 解法都是正確的，但是它們都有忙等待的缺點。這些解法的本質上都是一樣的，先檢查是否能夠進入臨界區，若不允許，則該進程將原地等待，直到允許為止。

這種方式不但浪費了 CPU 時間，而且還可能引起意想不到的結果。考慮一臺計算機上有兩個進程，這兩個進程具有不同的優先級，H 是屬於優先級比較高的進程，L 是屬於優先級比較低的進程。進程調度的規則是不論何時只要 H 進程處於就緒態 H 就開始運行。在某一時刻，L 處於臨界區中，此時 H 變為就緒態，準備運行（例如，一條 I/O 操作結束）。現在 H 要開始忙等，但由於當 H 就緒時 L 就不會被調度，L 從來不會有機會離開關鍵區域，所以 H 會變成死循環，有時將這種情況稱為優先級反轉問題(priority inversion problem)。

現在讓我們看一下進程間的通信原語，這些原語在不允許它們進入關鍵區域之前會阻塞而不是浪費 CPU 時間，最簡單的是 sleep 和 wakeup。Sleep 是一個能夠造成調用者阻塞的系統調用，也就是說，這個系統調用會暫停直到其他進程喚醒它。wakeup 調用有一個參數，即要喚醒的進程。還有一種方式是 wakeup 和 sleep 都有一個參數，即 sleep 和 wakeup 需要匹配的內存地址。

生產者-消費者問題

作為這些私有原語的例子，讓我們考慮生產者-消費者(producer-consumer) 問題，也稱作 有界緩衝區(bounded-buffer) 問題。兩個進程共享一個公共的固定大小的緩衝區。其中一個是生產者(producer)，將信息放入緩衝區， 另一個是消費者(consumer)，會從緩衝區中取出。也可以把這個問題一般化為 m 個生產者和 n 個消費者的問題，但是我們這裡只討論一個生產者和一個消費者的情況，這樣可以簡化實現方案。

如果緩衝隊列已滿，那麼當生產者仍想要將數據寫入緩衝區的時候，會出現問題。它的解決辦法是讓生產者睡眠，也就是阻塞生產者。等到消費者從緩衝區中取出一個或多個數據項時再喚醒它。同樣的，當消費者試圖從緩衝區中取數據，但是發現緩衝區為空時，消費者也會睡眠，阻塞。直到生產者向其中放入一個新的數據。

這個邏輯聽起來比較簡單，而且這種方式也需要一種稱作 監聽 的變量，這個變量用於監視緩衝區的數據，我們暫定為 count，如果緩衝區最多存放 N 個數據項，生產者會每次判斷 count 是否達到 N，否則生產者向緩衝區放入一個數據項並增量 count 的值。消費者的邏輯也很相似：首先測試 count 的值是否為 0 ，如果為 0 則消費者睡眠、阻塞，否則會從緩衝區取出數據並使 count 數量遞減。每個進程也會檢查檢查是否其他線程是否應該被喚醒，如果應該被喚醒，那麼就喚醒該線程。下面是生產者消費者的代碼

<code>#define N 100/* 緩衝區 slot 槽的數量 */int count = 0/* 緩衝區數據的數量 */  // 生產者void producer(void){  int item;    while(TRUE){    /* 無限循環 */    item = produce_item()            /* 生成下一項數據 */    if(count == N){      sleep();    /* 如果緩存區是滿的，就會阻塞 */    }        insert_item(item);           /* 把當前數據放在緩衝區中 */    count = count + 1;           /* 增加緩衝區 count 的數量 */    if(count == 1){      wakeup(consumer);           /* 緩衝區是否為空？ */    }  }}// 消費者void consumer(void){    int item;    while(TRUE){     /* 無限循環 */  if(count == 0){         /* 如果緩衝區是空的，就會進行阻塞 */      sleep();    }   item = remove_item();              /* 從緩衝區中取出一個數據 */    count = count - 1         /* 將緩衝區的 count 數量減一 */    if(count == N - 1){            /* 緩衝區滿嘛？ */      wakeup(producer);    }    consumer_item(item);             /* 打印數據項 */  }  }/<code>

為了在 C 語言中描述像是 sleep 和 wakeup 的系統調用，我們將以庫函數調用的形式來表示。它們不是 C 標準庫的一部分，但可以在實際具有這些系統調用的任何系統上使用。代碼中未實現的 insert_item 和 remove_item 用來記錄將數據項放入緩衝區和從緩衝區取出數據等。

現在讓我們回到生產者-消費者問題上來，上面代碼中會產生競爭條件，因為 count 這個變量是暴露在大眾視野下的。有可能出現下面這種情況：緩衝區為空，此時消費者剛好讀取 count 的值發現它為 0 。此時調度程序決定暫停消費者並啟動運行生產者。生產者生產了一條數據並把它放在緩衝區中，然後增加 count 的值，並注意到它的值是 1 。由於 count 為 0，消費者必須處於睡眠狀態，因此生產者調用 wakeup 來喚醒消費者。但是，消費者此時在邏輯上並沒有睡眠，所以 wakeup 信號會丟失。當消費者下次啟動後，它會查看之前讀取的 count 值，發現它的值是 0 ，然後在此進行睡眠。不久之後生產者會填滿整個緩衝區，在這之後會阻塞，這樣一來兩個進程將永遠睡眠下去。

引起上面問題的本質是 喚醒尚未進行睡眠狀態的進程會導致喚醒丟失。如果它沒有丟失，則一切都很正常。一種快速解決上面問題的方式是增加一個喚醒等待位(wakeup waiting bit)。當一個 wakeup 信號發送給仍在清醒的進程後，該位置為 1 。之後，當進程嘗試睡眠的時候，如果喚醒等待位為 1 ，則該位清除，而進程仍然保持清醒。

然而，當進程數量有許多的時候，這時你可以說通過增加喚醒等待位的數量來喚醒等待位，於是就有了 2、4、6、8 個喚醒等待位，但是並沒有從根本上解決問題。

信號量

信號量是 E.W.Dijkstra 在 1965 年提出的一種方法，它使用一個整形變量來累計喚醒次數，以供之後使用。在他的觀點中，有一個新的變量類型稱作 信號量(semaphore)。一個信號量的取值可以是 0 ，或任意正數。0 表示的是不需要任何喚醒，任意的正數表示的就是喚醒次數。

Dijkstra 提出了信號量有兩個操作，現在通常使用 down 和 up（分別可以用 sleep 和 wakeup 來表示）。down 這個指令的操作會檢查值是否大於 0 。如果大於 0 ，則將其值減 1 ；若該值為 0 ，則進程將睡眠，而且此時 down 操作將會繼續執行。檢查數值、修改變量值以及可能發生的睡眠操作均為一個單一的、不可分割的 原子操作(atomic action) 完成。這會保證一旦信號量操作開始，沒有其他的進程能夠訪問信號量，直到操作完成或者阻塞。這種原子性對於解決同步問題和避免競爭絕對必不可少。

原子性操作指的是在計算機科學的許多其他領域中，一組相關操作全部執行而沒有中斷或根本不執行。

up 操作會使信號量的值 + 1。如果一個或者多個進程在信號量上睡眠，無法完成一個先前的 down 操作，則由系統選擇其中一個並允許該程完成 down 操作。因此，對一個進程在其上睡眠的信號量執行一次 up 操作之後，該信號量的值仍然是 0 ，但在其上睡眠的進程卻少了一個。信號量的值增 1 和喚醒一個進程同樣也是不可分割的。不會有某個進程因執行 up 而阻塞，正如在前面的模型中不會有進程因執行 wakeup 而阻塞是一樣的道理。

用信號量解決生產者 - 消費者問題

用信號量解決丟失的 wakeup 問題，代碼如下

<code>#define N 100/* 定義緩衝區槽的數量 */typedef int semaphore;/* 信號量是一種特殊的 int */semaphore mutex = 1;/* 控制關鍵區域的訪問 */semaphore empty = N;    /* 統計 buffer 空槽的數量 */semaphore full = 0;/* 統計 buffer 滿槽的數量 */void producer(void){     int item;      while(TRUE){/* TRUE 的常量是 1 */    item = producer_item();/* 產生放在緩衝區的一些數據 */    down(&empty);/* 將空槽數量減 1  */    down(&mutex);/* 進入關鍵區域  */    insert_item(item);/* 把數據放入緩衝區中 */    up(&mutex);/* 離開臨界區 */    up(&full);    /* 將 buffer 滿槽數量 + 1 */  }}void consumer(void){    int item;    while(TRUE){/* 無限循環 */    down(&full);/* 緩存區滿槽數量 - 1 */    down(&mutex);/* 進入緩衝區 */    item = remove_item();/* 從緩衝區取出數據 */    up(&mutex);/* 離開臨界區 */    up(&empty);/* 將空槽數目 + 1 */    consume_item(item);    /* 處理數據 */  }  }/<code>

為了確保信號量能正確工作，最重要的是要採用一種不可分割的方式來實現它。通常是將 up 和 down 作為系統調用來實現。而且操作系統只需在執行以下操作時暫時屏蔽全部中斷：

使用 TSL 或者 XCHG 來防止幾個 CPU 同時訪問一個信號量，與生產者或消費者使用忙等待來等待其他騰出或填充緩衝區是完全不一樣的。前者的操作僅需要幾個毫秒，而生產者或消費者可能需要任意長的時間。

上面這個解決方案使用了三種信號量：一個稱為 full，用來記錄充滿的緩衝槽數目；一個稱為 empty，記錄空的緩衝槽數目；一個稱為 mutex，用來確保生產者和消費者不會同時進入緩衝區。Full 被初始化為 0 ，empty 初始化為緩衝區中插槽數，mutex 初始化為 1。信號量初始化為 1 並且由兩個或多個進程使用，以確保它們中同時只有一個可以進入關鍵區域的信號被稱為 二進制信號量(binary semaphores)。如果每個進程都在進入關鍵區域之前執行 down 操作，而在離開關鍵區域之後執行 up 操作，則可以確保相互互斥。

現在我們有了一個好的進程間原語的保證。然後我們再來看一下中斷的順序保證

1. 硬件壓入堆棧程序計數器等
2. 硬件從中斷向量裝入新的程序計數器
3. 彙編語言過程保存寄存器的值
4. 彙編語言過程設置新的堆棧
5. C 中斷服務器運行（典型的讀和緩存寫入）
6. 調度器決定下面哪個程序先運行
7. C 過程返回至彙編代碼
8. 彙編語言過程開始運行新的當前進程

在使用信號量的系統中，隱藏中斷的自然方法是讓每個 I/O 設備都配備一個信號量，該信號量最初設置為0。在 I/O 設備啟動後，中斷處理程序立刻對相關聯的信號執行一個 down 操作，於是進程立即被阻塞。當中斷進入時，中斷處理程序隨後對相關的信號量執行一個 up操作，能夠使已經阻止的進程恢復運行。在上面的中斷處理步驟中，其中的第 5 步 C 中斷服務器運行 就是中斷處理程序在信號量上執行的一個 up 操作，所以在第 6 步中，操作系統能夠執行設備驅動程序。當然，如果有幾個進程已經處於就緒狀態，調度程序可能會選擇接下來運行一個更重要的進程，我們會在後面討論調度的算法。

上面的代碼實際上是通過兩種不同的方式來使用信號量的，而這兩種信號量之間的區別也是很重要的。mutex 信號量用於互斥。它用於確保任意時刻只有一個進程能夠對緩衝區和相關變量進行讀寫。互斥是用於避免進程混亂所必須的一種操作。

另外一個信號量是關於同步(synchronization)的。full 和 empty 信號量用於確保事件的發生或者不發生。在這個事例中，它們確保了緩衝區滿時生產者停止運行；緩衝區為空時消費者停止運行。這兩個信號量的使用與 mutex 不同。

互斥量

如果不需要信號量的計數能力時，可以使用信號量的一個簡單版本，稱為 mutex(互斥量)。互斥量的優勢就在於在一些共享資源和一段代碼中保持互斥。由於互斥的實現既簡單又有效，這使得互斥量在實現用戶空間線程包時非常有用。

互斥量是一個處於兩種狀態之一的共享變量：解鎖(unlocked) 和 加鎖(locked)。這樣，只需要一個二進制位來表示它，不過一般情況下，通常會用一個 整形(integer) 來表示。0 表示解鎖，其他所有的值表示加鎖，比 1 大的值表示加鎖的次數。

mutex 使用兩個過程，當一個線程（或者進程）需要訪問關鍵區域時，會調用 mutex_lock 進行加鎖。如果互斥鎖當前處於解鎖狀態（表示關鍵區域可用），則調用成功，並且調用線程可以自由進入關鍵區域。

另一方面，如果 mutex 互斥量已經鎖定的話，調用線程會阻塞直到關鍵區域內的線程執行完畢並且調用了 mutex_unlock 。如果多個線程在 mutex 互斥量上阻塞，將隨機選擇一個線程並允許它獲得鎖。

由於 mutex 互斥量非常簡單，所以只要有 TSL 或者是 XCHG 指令，就可以很容易地在用戶空間實現它們。用於用戶級線程包的 mutex_lock 和 mutex_unlock 代碼如下，XCHG 的本質也一樣。

<code>mutex_lock:TSL REGISTER,MUTEX| 將互斥信號量複製到寄存器，並將互斥信號量置為1CMP REGISTER,#0| 互斥信號量是 0 嗎？JZE ok| 如果互斥信號量為0，它被解鎖，所以返回CALL thread_yield| 互斥信號正在使用；調度其他線程JMP mutex_lock| 再試一次ok: RET| 返回調用者，進入臨界區mutex_unlcok:MOVE MUTEX,#0| 將 mutex 置為 0 RET| 返回調用者/<code>

mutex_lock 的代碼和上面 enter_region 的代碼很相似，我們可以對比著看一下

上面代碼最大的區別你看出來了嗎？

• 根據上面我們對 TSL 的分析，我們知道，如果 TSL 判斷沒有進入臨界區的進程會進行無限循環獲取鎖，而在 TSL 的處理中，如果 mutex 正在使用，那麼就調度其他線程進行處理。所以上面最大的區別其實就是在判斷 mutex/TSL 之後的處理。
• 在（用戶）線程中，情況有所不同，因為沒有時鐘來停止運行時間過長的線程。結果是通過忙等待的方式來試圖獲得鎖的線程將永遠循環下去，決不會得到鎖，因為這個運行的線程不會讓其他線程運行從而釋放鎖，其他線程根本沒有獲得鎖的機會。在後者獲取鎖失敗時，它會調用 thread_yield 將 CPU 放棄給另外一個線程。結果就不會進行忙等待。在該線程下次運行時，它再一次對鎖進行測試。

上面就是 enter_region 和 mutex_lock 的差別所在。由於 thread_yield 僅僅是一個用戶空間的線程調度，所以它的運行非常快捷。這樣，mutex_lock 和 mutex_unlock 都不需要任何內核調用。通過使用這些過程，用戶線程完全可以實現在用戶空間中的同步，這個過程僅僅需要少量的同步。

我們上面描述的互斥量其實是一套調用框架中的指令。從軟件角度來說，總是需要更多的特性和同步原語。例如，有時線程包提供一個調用 mutex_trylock，這個調用嘗試獲取鎖或者返回錯誤碼，但是不會進行加鎖操作。這就給了調用線程一個靈活性，以決定下一步做什麼，是使用替代方法還是等候下去。

Futexes

隨著並行的增加，有效的同步(synchronization)和鎖定(locking) 對於性能來說是非常重要的。如果進程等待時間很短，那麼自旋鎖(Spin lock) 是非常有效；但是如果等待時間比較長，那麼這會浪費 CPU 週期。如果進程很多，那麼阻塞此進程，並僅當鎖被釋放的時候讓內核解除阻塞是更有效的方式。不幸的是，這種方式也會導致另外的問題：它可以在進程競爭頻繁的時候運行良好，但是在競爭不是很激烈的情況下內核切換的消耗會非常大，而且更困難的是，預測鎖的競爭數量更不容易。

有一種有趣的解決方案是把兩者的優點結合起來，提出一種新的思想，稱為 futex，或者是 快速用戶空間互斥(fast user space mutex)，是不是聽起來很有意思？

futex 是 Linux 中的特性實現了基本的鎖定（很像是互斥鎖）而且避免了陷入內核中，因為內核的切換的開銷非常大，這樣做可以大大提高性能。futex 由兩部分組成：內核服務和用戶庫。內核服務提供了了一個 等待隊列(wait queue) 允許多個進程在鎖上排隊等待。除非內核明確的對他們解除阻塞，否則它們不會運行。

對於一個進程來說，把它放到等待隊列需要昂貴的系統調用，這種方式應該被避免。在沒有競爭的情況下，futex 可以直接在用戶空間中工作。這些進程共享一個 32 位整數(integer) 作為公共鎖變量。假設鎖的初始化為 1，我們認為這時鎖已經被釋放了。線程通過執行原子性的操作減少並測試(decrement and test) 來搶佔鎖。decrement and set 是 Linux 中的原子功能，由包裹在 C 函數中的內聯彙編組成，並在頭文件中進行定義。下一步，線程會檢查結果來查看鎖是否已經被釋放。如果鎖現在不是鎖定狀態，那麼剛好我們的線程可以成功搶佔該鎖。然而，如果鎖被其他線程持有，搶佔鎖的線程不得不等待。在這種情況下，futex 庫不會自旋，但是會使用一個系統調用來把線程放在內核中的等待隊列中。這樣一來，切換到內核的開銷已經是合情合理的了，因為線程可以在任何時候阻塞。當線程完成了鎖的工作時，它會使用原子性的 增加並測試(increment and test) 釋放鎖，並檢查結果以查看內核等待隊列上是否仍阻止任何進程。如果有的話，它會通知內核可以對等待隊列中的一個或多個進程解除阻塞。如果沒有鎖競爭，內核則不需要參與競爭。

Pthreads 中的互斥量

Pthreads 提供了一些功能用來同步線程。最基本的機制是使用互斥量變量，可以鎖定和解鎖，用來保護每個關鍵區域。希望進入關鍵區域的線程首先要嘗試獲取 mutex。如果 mutex 沒有加鎖，線程能夠馬上進入並且互斥量能夠自動鎖定，從而阻止其他線程進入。如果 mutex 已經加鎖，調用線程會阻塞，直到 mutex 解鎖。如果多個線程在相同的互斥量上等待，當互斥量解鎖時，只有一個線程能夠進入並且重新加鎖。這些鎖並不是必須的，程序員需要正確使用它們。

下面是與互斥量有關的函數調用

向我們想象中的一樣，mutex 能夠被創建和銷燬，扮演這兩個角色的分別是 Phread_mutex_init 和 Pthread_mutex_destroy。mutex 也可以通過 Pthread_mutex_lock 來進行加鎖，如果互斥量已經加鎖，則會阻塞調用者。還有一個調用Pthread_mutex_trylock 用來嘗試對線程加鎖，當 mutex 已經被加鎖時，會返回一個錯誤代碼而不是阻塞調用者。這個調用允許線程有效的進行忙等。最後，Pthread_mutex_unlock 會對 mutex 解鎖並且釋放一個正在等待的線程。

除了互斥量以外，Pthreads 還提供了第二種同步機制： 條件變量(condition variables) 。mutex 可以很好的允許或阻止對關鍵區域的訪問。條件變量允許線程由於未滿足某些條件而阻塞。絕大多數情況下這兩種方法是一起使用的。下面我們進一步來研究線程、互斥量、條件變量之間的關聯。

下面再來重新認識一下生產者和消費者問題：一個線程將東西放在一個緩衝區內，由另一個線程將它們取出。如果生產者發現緩衝區沒有空槽可以使用了，生產者線程會阻塞起來直到有一個線程可以使用。生產者使用 mutex 來進行原子性檢查從而不受其他線程干擾。但是當發現緩衝區已經滿了以後，生產者需要一種方法來阻塞自己並在以後被喚醒。這便是條件變量做的工作。

下面是一些與條件變量有關的最重要的 pthread 調用

上表中給出了一些調用用來創建和銷燬條件變量。條件變量上的主要屬性是 Pthread_cond_wait 和 Pthread_cond_signal。前者阻塞調用線程，直到其他線程發出信號為止（使用後者調用）。阻塞的線程通常需要等待喚醒的信號以此來釋放資源或者執行某些其他活動。只有這樣阻塞的線程才能繼續工作。條件變量允許等待與阻塞原子性的進程。Pthread_cond_broadcast 用來喚醒多個阻塞的、需要等待信號喚醒的線程。

需要注意的是，條件變量（不像是信號量）不會存在於內存中。如果將一個信號量傳遞給一個沒有線程等待的條件變量，那麼這個信號就會丟失，這個需要注意

下面是一個使用互斥量和條件變量的例子

<code>

管程

為了能夠編寫更加準確無誤的程序，Brinch Hansen 和 Hoare 提出了一個更高級的同步原語叫做 管程(monitor)。他們兩個人的提案略有不同，通過下面的描述你就可以知道。管程是程序、變量和數據結構等組成的一個集合，它們組成一個特殊的模塊或者包。進程可以在任何需要的時候調用管程中的程序，但是它們不能從管程外部訪問數據結構和程序。下面展示了一種抽象的，類似 Pascal 語言展示的簡潔的管程。不能用 C 語言進行描述，因為管程是語言概念而 C 語言並不支持管程。

<code>monitor exampleinteger i;condition c;procedure producer();...end;procedure consumer();.end;end monitor;/<code>

管程有一個很重要的特性，即在任何時候管程中只能有一個活躍的進程，這一特性使管程能夠很方便的實現互斥操作。管程是編程語言的特性，所以編譯器知道它們的特殊性，因此可以採用與其他過程調用不同的方法來處理對管程的調用。通常情況下，當進程調用管程中的程序時，該程序的前幾條指令會檢查管程中是否有其他活躍的進程。如果有的話，調用進程將被掛起，直到另一個進程離開管程才將其喚醒。如果沒有活躍進程在使用管程，那麼該調用進程才可以進入。

進入管程中的互斥由編譯器負責，但是一種通用做法是使用 互斥量(mutex) 和 二進制信號量(binary semaphore)。由於編譯器而不是程序員在操作，因此出錯的幾率會大大降低。在任何時候，編寫管程的程序員都無需關心編譯器是如何處理的。他只需要知道將所有的臨界區轉換成為管程過程即可。絕不會有兩個進程同時執行臨界區中的代碼。

即使管程提供了一種簡單的方式來實現互斥，但在我們看來，這還不夠。因為我們還需要一種在進程無法執行被阻塞。在生產者-消費者問題中，很容易將針對緩衝區滿和緩衝區空的測試放在管程程序中，但是生產者在發現緩衝區滿的時候該如何阻塞呢？

解決的辦法是引入條件變量(condition variables) 以及相關的兩個操作 wait 和 signal。當一個管程程序發現它不能運行時（例如，生產者發現緩衝區已滿），它會在某個條件變量（如 full）上執行 wait 操作。這個操作造成調用進程阻塞，並且還將另一個以前等在管程之外的進程調入管程。在前面的 pthread 中我們已經探討過條件變量的實現細節了。另一個進程，比如消費者可以通過執行 signal 來喚醒阻塞的調用進程。

Brinch Hansen 和 Hoare 在對進程喚醒上有所不同，Hoare 建議讓新喚醒的進程繼續運行；而掛起另外的進程。而 Brinch Hansen 建議讓執行 signal 的進程必須退出管程，這裡我們採用 Brinch Hansen 的建議，因為它在概念上更簡單，並且更容易實現。

如果在一個條件變量上有若干進程都在等待，則在對該條件執行 signal 操作後，系統調度程序只能選擇其中一個進程恢復運行。

順便提一下，這裡還有上面兩位教授沒有提出的第三種方式，它的理論是讓執行 signal 的進程繼續運行，等待這個進程退出管程時，其他進程才能進入管程。

條件變量不是計數器。條件變量也不能像信號量那樣積累信號以便以後使用。所以，如果向一個條件變量發送信號，但是該條件變量上沒有等待進程，那麼信號將會丟失。也就是說，

下面是一個使用 Pascal 語言通過管程實現的生產者-消費者問題的解法

<code>monitor ProducerConsumercondition full,empty;integer count;procedure insert(item:integer);beginif count = N then wait(full);insert_item(item);count := count + 1;if count = 1 then signal(empty);end;function remove:integer;beginif count = 0 then wait(empty);remove = remove_item;count := count - 1;if count = N - 1 then signal(full);end;count := 0;end monitor;procedure producer;beginwhile true do      begin       item = produce_item;      ProducerConsumer.insert(item);      endend;procedure consumer;begin while true dobeginitem = ProducerConsumer.remove;consume_item(item);endend;/<code>

讀者可能覺得 wait 和 signal 操作看起來像是前面提到的 sleep 和 wakeup ，而且後者存在嚴重的競爭條件。它們確實很像，但是有個關鍵的區別：sleep 和 wakeup 之所以會失敗是因為當一個進程想睡眠時，另一個進程試圖去喚醒它。使用管程則不會發生這種情況。管程程序的自動互斥保證了這一點，如果管程過程中的生產者發現緩衝區已滿，它將能夠完成 wait 操作而不用擔心調度程序可能會在 wait 完成之前切換到消費者。甚至，在 wait 執行完成並且把生產者標誌為不可運行之前，是不會允許消費者進入管程的。

儘管類 Pascal 是一種想象的語言，但還是有一些真正的編程語言支持，比如 Java （終於輪到大 Java 出場了），Java 是能夠支持管程的，它是一種 面向對象的語言，支持用戶級線程，還允許將方法劃分為類。只要將關鍵字 synchronized 關鍵字加到方法中即可。Java 能夠保證一旦某個線程執行該方法，就不允許其他線程執行該對象中的任何 synchronized 方法。沒有關鍵字 synchronized ，就不能保證沒有交叉執行。

下面是 Java 使用管程解決的生產者-消費者問題

<code>public class ProducerConsumer {  static final int N = 100;// 定義緩衝區大小的長度  static Producer p = new Producer();// 初始化一個新的生產者線程  static Consumer c = new Consumer();// 初始化一個新的消費者線程  static Our_monitor mon = new Our_monitor();       // 初始化一個管程    static class Producer extends Thread{    public void run(){// run 包含了線程代碼      int item;      while(true){// 生產者循環        item = produce_item();        mon.insert(item);      }    }    private int produce_item(){...}// 生產代碼  }    static class consumer extends Thread {    public void run( ) {// run 包含了線程代碼   int item;      while(true){        item = mon.remove();consume_item(item);      }    }    private int produce_item(){...}// 消費代碼  }    static class Our_monitor {// 這是管程    private int buffer[] = new int[N];    private int count = 0,lo = 0,hi = 0;                                        // 計數器和索引        private synchronized void insert(int val){      if(count == N){        go_to_sleep();// 如果緩衝區是滿的，則進入休眠      }buffer[hi] = val;// 向緩衝區插入內容      hi = (hi + 1) % N; // 找到下一個槽的為止      count = count + 1;// 緩衝區中的數目自增 1       if(count == 1){        notify();// 如果消費者睡眠，則喚醒      }    }        private synchronized void remove(int val){      int val;      if(count == 0){        go_to_sleep();// 緩衝區是空的，進入休眠      }      val = buffer[lo];// 從緩衝區取出數據      lo = (lo + 1) % N;// 設置待取出數據項的槽      count = count - 1;// 緩衝區中的數據項數目減 1       if(count = N - 1){        notify();// 如果生產者睡眠，喚醒它      }      return val;    }        private void go_to_sleep() {      try{        wait( );      }catch(Interr uptedExceptionexc) {};    }  }      }/<code>

上面的代碼中主要設計四個類，外部類(outer class) ProducerConsumer 創建並啟動兩個線程，p 和 c。第二個類和第三個類 Producer 和 Consumer 分別包含生產者和消費者代碼。最後，Our_monitor 是管程，它有兩個同步線程，用於在共享緩衝區中插入和取出數據。

在前面的所有例子中，生產者和消費者線程在功能上與它們是相同的。生產者有一個無限循環，該無限循環產生數據並將數據放入公共緩衝區中；消費者也有一個等價的無限循環，該無限循環用於從緩衝區取出數據並完成一系列工作。

程序中比較耐人尋味的就是 Our_monitor 了，它包含緩衝區、管理變量以及兩個同步方法。當生產者在 insert 內活動時，它保證消費者不能在 remove 方法中運行，從而保證更新變量以及緩衝區的安全性，並且不用擔心競爭條件。變量 count 記錄在緩衝區中數據的數量。變量 lo 是緩衝區槽的序號，指出將要取出的下一個數據項。類似地，hi 是緩衝區中下一個要放入的數據項序號。允許 lo = hi，含義是在緩衝區中有 0 個或 N 個數據。

Java 中的同步方法與其他經典管程有本質差別：Java 沒有內嵌的條件變量。然而，Java 提供了 wait 和 notify 分別與 sleep 和 wakeup 等價。

通過臨界區自動的互斥，管程比信號量更容易保證並行編程的正確性。但是管程也有缺點，我們前面說到過管程是一個編程語言的概念，編譯器必須要識別管程並用某種方式對其互斥作出保證。C、Pascal 以及大多數其他編程語言都沒有管程，所以不能依靠編譯器來遵守互斥規則。

與管程和信號量有關的另一個問題是，這些機制都是設計用來解決訪問共享內存的一個或多個 CPU 上的互斥問題的。通過將信號量放在共享內存中並用 TSL 或 XCHG 指令來保護它們，可以避免競爭。但是如果是在分佈式系統中，可能同時具有多個 CPU 的情況，並且每個 CPU 都有自己的私有內存呢，它們通過網絡相連，那麼這些原語將會失效。因為信號量太低級了，而管程在少數幾種編程語言之外無法使用，所以還需要其他方法。

消息傳遞

上面提到的其他方法就是 消息傳遞(messaage passing)。這種進程間通信的方法使用兩個原語 send 和 receive ，它們像信號量而不像管程，是系統調用而不是語言級別。示例如下

<code>send(destination, &message);receive(source, &message);/<code>

send 方法用於向一個給定的目標發送一條消息，receive 從一個給定的源接受一條消息。如果沒有消息，接受者可能被阻塞，直到接受一條消息或者帶著錯誤碼返回。

消息傳遞系統的設計要點

消息傳遞系統現在面臨著許多信號量和管程所未涉及的問題和設計難點，尤其對那些在網絡中不同機器上的通信狀況。例如，消息有可能被網絡丟失。為了防止消息丟失，發送方和接收方可以達成一致：一旦接受到消息後，接收方馬上回送一條特殊的 確認(acknowledgement) 消息。如果發送方在一段時間間隔內未收到確認，則重發消息。

現在考慮消息本身被正確接收，而返回給發送著的確認消息丟失的情況。發送者將重發消息，這樣接受者將收到兩次相同的消息。

對於接收者來說，如何區分新的消息和一條重發的老消息是非常重要的。通常採用在每條原始消息中嵌入一個連續的序號來解決此問題。如果接受者收到一條消息，它具有與前面某一條消息一樣的序號，就知道這條消息是重複的，可以忽略。

消息系統還必須處理如何命名進程的問題，以便在發送或接收調用中清晰的指明進程。身份驗證(authentication) 也是一個問題，比如客戶端怎麼知道它是在與一個真正的文件服務器通信，從發送方到接收方的信息有可能被中間人所篡改。

用消息傳遞解決生產者-消費者問題

現在我們考慮如何使用消息傳遞來解決生產者-消費者問題，而不是共享緩存。下面是一種解決方式

<code>#define N 100/* buffer 中槽的數量 */void producer(void){    int item;  message m;/* buffer 中槽的數量 */    while(TRUE){    item = produce_item();/* 生成放入緩衝區的數據 */    receive(consumer,&m);/* 等待消費者發送空緩衝區 */    build_message(&m,item);/* 建立一個待發送的消息 */    send(consumer,&m);/* 發送給消費者 */  }  }void consumer(void){    int item,i;  message m;    for(int i = 0;i < N;i++){/* 循環N次 */    send(producer,&m);/* 發送N個緩衝區 */  }  while(TRUE){    receive(producer,&m);/* 接受包含數據的消息 */  item = extract_item(&m);/* 將數據從消息中提取出來 */    send(producer,&m);/* 將空緩衝區發送回生產者 */    consume_item(item);/* 處理數據 */  }  }/<code>

假設所有的消息都有相同的大小，並且在尚未接受到發出的消息時，由操作系統自動進行緩衝。在該解決方案中共使用 N 條消息，這就類似於一塊共享內存緩衝區的 N 個槽。消費者首先將 N 條空消息發送給生產者。當生產者向消費者傳遞一個數據項時，它取走一條空消息並返回一條填充了內容的消息。通過這種方式，系統中總的消息數量保持不變，所以消息都可以存放在事先確定數量的內存中。

如果生產者的速度要比消費者快，則所有的消息最終都將被填滿，等待消費者，生產者將被阻塞，等待返回一條空消息。如果消費者速度快，那麼情況將正相反：所有的消息均為空，等待生產者來填充，消費者將被阻塞，以等待一條填充過的消息。

消息傳遞的方式有許多變體，下面先介紹如何對消息進行 編址。

• 一種方法是為每個進程分配一個唯一的地址，讓消息按進程的地址編址。
• 另一種方式是引入一個新的數據結構，稱為 信箱(mailbox)，信箱是一個用來對一定的數據進行緩衝的數據結構，信箱中消息的設置方法也有多種，典型的方法是在信箱創建時確定消息的數量。在使用信箱時，在 send 和 receive 調用的地址參數就是信箱的地址，而不是進程的地址。當一個進程試圖向一個滿的信箱發送消息時，它將被掛起，直到信箱中有消息被取走，從而為新的消息騰出地址空間。

屏障

最後一個同步機制是準備用於進程組而不是進程間的生產者-消費者情況的。在某些應用中劃分了若干階段，並且規定，除非所有的進程都就緒準備著手下一個階段，否則任何進程都不能進入下一個階段，可以通過在每個階段的結尾安裝一個 屏障(barrier) 來實現這種行為。當一個進程到達屏障時，它會被屏障所攔截，直到所有的屏障都到達為止。屏障可用於一組進程同步，如下圖所示

在上圖中我們可以看到，有四個進程接近屏障，這意味著每個進程都在進行運算，但是還沒有到達每個階段的結尾。過了一段時間後，A、B、D 三個進程都到達了屏障，各自的進程被掛起，但此時還不能進入下一個階段呢，因為進程 B 還沒有執行完畢。結果，當最後一個 C 到達屏障後，這個進程組才能夠進入下一個階段。

避免鎖：讀-複製-更新

最快的鎖是根本沒有鎖。問題在於沒有鎖的情況下，我們是否允許對共享數據結構的併發讀寫進行訪問。答案當然是不可以。假設進程 A 正在對一個數字數組進行排序，而進程 B 正在計算其平均值，而此時你進行 A 的移動，會導致 B 會多次讀到重複值，而某些值根本沒有遇到過。

然而，在某些情況下，我們可以允許寫操作來更新數據結構，即便還有其他的進程正在使用。竅門在於確保每個讀操作要麼讀取舊的版本，要麼讀取新的版本，例如下面的樹

上面的樹中，讀操作從根部到葉子遍歷整個樹。加入一個新節點 X 後，為了實現這一操作，我們要讓這個節點在樹中可見之前使它"恰好正確"：我們對節點 X 中的所有值進行初始化，包括它的子節點指針。然後通過原子寫操作，使 X 稱為 A 的子節點。所有的讀操作都不會讀到前後不一致的版本

在上面的圖中，我們接著移除 B 和 D。首先，將 A 的左子節點指針指向 C 。所有原本在 A 中的讀操作將會後續讀到節點 C ，而永遠不會讀到 B 和 D。也就是說，它們將只會讀取到新版數據。同樣，所有當前在 B 和 D 中的讀操作將繼續按照原始的數據結構指針並且讀取舊版數據。所有操作均能正確運行，我們不需要鎖住任何東西。而不需要鎖住數據就能夠移除 B 和 D 的主要原因就是 讀-複製-更新(Ready-Copy-Update,RCU)，將更新過程中的移除和再分配過程分離開。

本文內容來源於作者“cxuan”，只作為技術分享學習，歡迎大家在評論區評論。

文獻參考：

《現代操作系統》

《Modern Operating System》forth edition

https://www.encyclopedia.com/computing/news-wires-white-papers-and-books/interactive-systems

https://juejin.im/post/5e4b5c786fb9a07ca80aa2b9?utm_source=gold_browser_extension

https://j00ru.vexillium.org/syscalls/nt/32/

https://www.bottomupcs.com/process_hierarchy.xhtml

https://en.wikipedia.org/wiki/Runtime_system

https://en.wikipedia.org/wiki/Execution_model

閱讀更多 湛神帶你寫代碼 的文章

關鍵字: 中央處理器 線程 進程