Java建設(shè)者的67篇原創(chuàng)文章

上一句我們剖析了過程和線程的本質(zhì)，過程和線程的實現(xiàn)方式。這篇文章，我們探討它們是如何通信的，過程告訴我，線程不想活了。我不管它是死是死。我只想知道我是誰。那個過程是怎么被我知道的？過程的出現(xiàn)和線程的死亡與我有必然的聯(lián)系嗎？句子給你。上一句我們剖析了過程和線程的本質(zhì)，過程和線程的實現(xiàn)方式。這篇文章我們探討它們是如何通信的，過程說線程不想活了。我不在乎它是死是死。我是誰？那個過程是怎么被我知道的？過程的出現(xiàn)和線程的死亡與我有必然的聯(lián)系嗎？文章，揭露吧。

進程間通信

過程需要經(jīng)常與其他過程交流。 例如，在一個 shell 管道中，第一個進程的輸出必須傳遞給第二個進程，這樣沿著管道進行下去。因此，進程之間如果需要通信的話，必須要使用一種良好的數(shù)據(jù)結(jié)構(gòu)以至于不能被中斷。下面我們會一起討論有關(guān) 進程間通信(Inter Process Communication, IPC) 的問題。

關(guān)于進程間的通信，這里有三個問題

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

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

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

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

競態(tài)條件

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

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

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

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

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

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

臨界區(qū)

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

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

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

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

從抽象的角度來看，我們通常希望進程的行為如上圖所示，在 t1 時刻，進程 A 進入臨界區(qū)，在 t2 的時刻，進程 B 嘗試進入臨界區(qū)，因為此時進程 A 正在處于臨界區(qū)中，所以進程 B 會阻塞直到 t3 時刻進程 A 離開臨界區(qū)，此時進程 B 能夠允許進入臨界區(qū)。最后，在 t4 時刻，進程 B 離開臨界區(qū)，系統(tǒng)恢復到?jīng)]有進程的原始狀態(tài)。

忙等互斥

下面我們會繼續(xù)探討實現(xiàn)互斥的各種設(shè)計，在這些方案中，當一個進程正忙于更新其關(guān)鍵區(qū)域的共享內(nèi)存時，沒有其他進程會進入其關(guān)鍵區(qū)域，也不會造成影響。

屏蔽中斷

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

這個方案可行嗎？進程進入關(guān)鍵區(qū)域是由誰決定的呢？不是用戶進程嗎？當進程進入關(guān)鍵區(qū)域后，用戶進程關(guān)閉中斷，如果經(jīng)過一段較長時間后進程沒有離開，那么中斷不就一直啟用不了，結(jié)果會如何？可能會造成整個系統(tǒng)的終止。而且如果是多處理器的話，屏蔽中斷僅僅對執(zhí)行 disable 指令的 CPU 有效。其他 CPU 仍將繼續(xù)運行，并可以訪問共享內(nèi)存。

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

鎖變量

作為第二種嘗試，可以尋找一種軟件層面解決方案?？紤]有單個共享的（鎖）變量，初始為值為 0 。當一個線程想要進入關(guān)鍵區(qū)域時，它首先會查看鎖的值是否為 0 ，如果鎖的值是 0 ，進程會把它設(shè)置為 1 并讓進程進入關(guān)鍵區(qū)域。如果鎖的狀態(tài)是 1，進程會等待直到鎖變量的值變?yōu)?0 。因此，鎖變量的值是 0 則意味著沒有線程進入關(guān)鍵區(qū)域。如果是 1 則意味著有進程在關(guān)鍵區(qū)域內(nèi)。我們對上圖修改后，如下所示

這種設(shè)計方式是否正確呢？是否存在紕漏呢？假設(shè)一個進程讀出鎖變量的值并發(fā)現(xiàn)它為 0 ，而恰好在它將其設(shè)置為 1 之前，另一個進程調(diào)度運行，讀出鎖的變量為0 ，并將鎖的變量設(shè)置為 1 。然后第一個線程運行，把鎖變量的值再次設(shè)置為 1，此時，關(guān)鍵區(qū)域就會有兩個進程在同時運行。

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

嚴格輪詢法

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

進程 0 的代碼

while(TRUE){ while(turn != 0){ /* 進入關(guān)鍵區(qū)域 */ critical_region(); turn = 1; /* 離開關(guān)鍵區(qū)域 */ noncritical_region(); } }

進程 1 的代碼

while(TRUE){ while(turn != 1){ critical_region(); turn = 0; noncritical_region(); } }

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

進程 0 離開臨界區(qū)時，它將 turn 的值設(shè)置為 1，以便允許進程 1 進入其臨界區(qū)。假設(shè)進程 1 很快便離開了臨界區(qū)，則此時兩個進程都處于臨界區(qū)之外，turn 的值又被設(shè)置為 0 。現(xiàn)在進程 0 很快就執(zhí)行完了整個循環(huán)，它退出臨界區(qū)，并將 turn 的值設(shè)置為 1。此時，turn 的值為 1，兩個進程都在其臨界區(qū)外執(zhí)行。

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

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

Peterson 解法

荷蘭數(shù)學家 T.Dekker 通過將鎖變量與警告變量相結(jié)合，最早提出了一個不需要嚴格輪換的軟件互斥算法

后來， G.L.Peterson 發(fā)現(xiàn)了一種簡單很多的互斥算法，它的算法如下

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

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

現(xiàn)在來看看這個辦法是如何工作的。一開始，沒有任何進程處于臨界區(qū)中，現(xiàn)在進程 0 調(diào)用 enter_region。它通過設(shè)置數(shù)組元素和將 turn 置為 0 來表示它希望進入臨界區(qū)。由于進程 1 并不想進入臨界區(qū)，所以 enter_region 很快便返回。如果進程現(xiàn)在調(diào)用 enter_region，進程 1 將在此處掛起直到 interested[0] 變?yōu)?FALSE，這種情況只有在進程 0 調(diào)用 leave_region 退出臨界區(qū)時才會發(fā)生。

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

TSL 指令

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

TSL RX,lock

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

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

為了使用 TSL 指令，要使用一個共享變量 lock 來協(xié)調(diào)對共享內(nèi)存的訪問。當 lock 為 0 時，任何進程都可以使用 TSL 指令將其設(shè)置為 1，并讀寫共享內(nèi)存。當操作結(jié)束時，進程使用 move 指令將 lock 的值重新設(shè)置為 0 。

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

enter_region: TSL REGISTER,LOCK | 復制鎖到寄存器并將鎖設(shè)為1 CMP REGISTER,#0 | 鎖是 0 嗎？ JNE enter_region | 若不是零，說明鎖已被設(shè)置，所以循環(huán) RET | 返回調(diào)用者，進入臨界區(qū) leave_region: MOVE LOCK,#0 | 在鎖中存入 0 RET | 返回調(diào)用者

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

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

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

enter_region: MOVE REGISTER,#1 | 把 1 放在內(nèi)存器中 XCHG REGISTER,LOCK | 交換寄存器和鎖變量的內(nèi)容 CMP REGISTER,#0 | 鎖是 0 嗎？ JNE enter_region | 若不是 0 ，鎖已被設(shè)置，進行循環(huán) RET | 返回調(diào)用者，進入臨界區(qū) leave_region: MOVE LOCK,#0 | 在鎖中存入 0 RET | 返回調(diào)用者

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

睡眠與喚醒

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

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

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

生產(chǎn)者-消費者問題

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

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

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

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

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

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

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

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

信號量

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

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

“

原子性操作指的是在計算機科學的許多其他領(lǐng)域中，一組相關(guān)操作全部執(zhí)行而沒有中斷或根本不執(zhí)行。

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

用信號量解決生產(chǎn)者 - 消費者問題

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

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

為了確保信號量能正確工作，最重要的是要采用一種不可分割的方式來實現(xiàn)它。通常是將 up 和 down 作為系統(tǒng)調(diào)用來實現(xiàn)。而且操作系統(tǒng)只需在執(zhí)行以下操作時暫時屏蔽全部中斷：檢查信號量、更新、必要時使進程睡眠。由于這些操作僅需要非常少的指令，因此中斷不會造成影響。如果使用多個 CPU，那么信號量應該被鎖進行保護。使用 TSL 或者 XCHG 指令用來確保同一時刻只有一個 CPU 對信號量進行操作。

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

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

現(xiàn)在我們有了一個好的進程間原語的保證。然后我們再來看一下中斷的順序保證

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

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

上面的代碼實際上是通過兩種不同的方式來使用信號量的，而這兩種信號量之間的區(qū)別也是很重要的。mutex 信號量用于互斥。它用于確保任意時刻只有一個進程能夠?qū)彌_區(qū)和相關(guān)變量進行讀寫?；コ馐怯糜诒苊膺M程混亂所必須的一種操作。

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

互斥量

如果不需要信號量的計數(shù)能力時，可以使用信號量的一個簡單版本，稱為 mutex(互斥量)?；コ饬康膬?yōu)勢就在于在一些共享資源和一段代碼中保持互斥。由于互斥的實現(xiàn)既簡單又有效，這使得互斥量在實現(xiàn)用戶空間線程包時非常有用。

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

mutex 使用兩個過程，當一個線程（或者進程）需要訪問關(guān)鍵區(qū)域時，會調(diào)用 mutex_lock 進行加鎖。如果互斥鎖當前處于解鎖狀態(tài)（表示關(guān)鍵區(qū)域可用），則調(diào)用成功，并且調(diào)用線程可以自由進入關(guān)鍵區(qū)域。

另一方面，如果 mutex 互斥量已經(jīng)鎖定的話，調(diào)用線程會阻塞直到關(guān)鍵區(qū)域內(nèi)的線程執(zhí)行完畢并且調(diào)用了 mutex_unlock 。如果多個線程在 mutex 互斥量上阻塞，將隨機選擇一個線程并允許它獲得鎖。

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

mutex_lock: TSL REGISTER,MUTEX | 將互斥信號量復制到寄存器，并將互斥信號量置為1 CMP REGISTER,#0 | 互斥信號量是 0 嗎？ JZE ok | 如果互斥信號量為0，它被解鎖，所以返回 CALL thread_yield | 互斥信號正在使用；調(diào)度其他線程 JMP mutex_lock | 再試一次 ok: RET | 返回調(diào)用者，進入臨界區(qū) mutex_unlcok: MOVE MUTEX,#0 | 將 mutex 置為 0 RET | 返回調(diào)用者

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

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

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

上面就是 enter_region 和 mutex_lock 的差別所在。由于 thread_yield 僅僅是一個用戶空間的線程調(diào)度，所以它的運行非?？旖?。這樣，mutex_lock 和 mutex_unlock 都不需要任何內(nèi)核調(diào)用。通過使用這些過程，用戶線程完全可以實現(xiàn)在用戶空間中的同步，這個過程僅僅需要少量的同步。

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

Futexes

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

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

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

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

Pthreads 中的互斥量

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

下面是與互斥量有關(guān)的函數(shù)調(diào)用

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

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

下面再來重新認識一下生產(chǎn)者和消費者問題：一個線程將東西放在一個緩沖區(qū)內(nèi)，由另一個線程將它們?nèi)〕觥Ｈ绻a(chǎn)者發(fā)現(xiàn)緩沖區(qū)沒有空槽可以使用了，生產(chǎn)者線程會阻塞起來直到有一個線程可以使用。生產(chǎn)者使用 mutex 來進行原子性檢查從而不受其他線程干擾。但是當發(fā)現(xiàn)緩沖區(qū)已經(jīng)滿了以后，生產(chǎn)者需要一種方法來阻塞自己并在以后被喚醒。這便是條件變量做的工作。

下面是一些與條件變量有關(guān)的最重要的 pthread 調(diào)用

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

“

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

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

#include <; #include <; #define MAX 1000000000 /* 需要生產(chǎn)的數(shù)量 */ pthread_mutex_t the_mutex; pthread_cond_t condc,condp; /* 使用信號量 */ int buffer = 0; void *producer(void *ptr){ /* 生產(chǎn)數(shù)據(jù) */ int i; for(int i = 0;i <= MAX;i++){ pthread_mutex_lock(&the_mutex); /* 緩沖區(qū)獨占訪問，也就是使用 mutex 獲取鎖 */ while(buffer != 0){ pthread_cond_wait(&condp,&the_mutex); } buffer = i; /* 把他們放在緩沖區(qū)中 */ pthread_cond_signal(&condc); /* 喚醒消費者 */ pthread_mutex_unlock(&the_mutex); /* 釋放緩沖區(qū) */ } pthread_exit(0); } void *consumer(void *ptr){ /* 消費數(shù)據(jù) */ int i; for(int i = 0;i <= MAX;i++){ pthread_mutex_lock(&the_mutex); /* 緩沖區(qū)獨占訪問，也就是使用 mutex 獲取鎖 */ while(buffer == 0){ pthread_cond_wait(&condc,&the_mutex); } buffer = 0; /* 把他們從緩沖區(qū)中取出 */ pthread_cond_signal(&condp); /* 喚醒生產(chǎn)者 */ pthread_mutex_unlock(&the_mutex); /* 釋放緩沖區(qū) */ } pthread_exit(0); }

管程

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

monitor example integer i; condition c; procedure producer(); . . . end; procedure consumer(); . end; end monitor;

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

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

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

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

“

Brinch Hansen 和 Hoare 在對進程喚醒上有所不同，Hoare 建議讓新喚醒的進程繼續(xù)運行；而掛起另外的進程。而 Brinch Hansen 建議讓執(zhí)行 signal 的進程必須退出管程，這里我們采用 Brinch Hansen 的建議，因為它在概念上更簡單，并且更容易實現(xiàn)。

如果在一個條件變量上有若干進程都在等待，則在對該條件執(zhí)行 signal 操作后，系統(tǒng)調(diào)度程序只能選擇其中一個進程恢復運行。

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

條件變量不是計數(shù)器。條件變量也不能像信號量那樣積累信號以便以后使用。所以，如果向一個條件變量發(fā)送信號，但是該條件變量上沒有等待進程，那么信號將會丟失。也就是說，wait 操作必須在 signal 之前執(zhí)行。

下面是一個使用 Pascal 語言通過管程實現(xiàn)的生產(chǎn)者-消費者問題的解法

monitor ProducerConsumer condition full,empty; integer count; procedure insert(item:integer); begin if count = N then wait(full); insert_item(item); count := count + 1; if count = 1 then signal(empty); end; function remove:integer; begin if 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; begin while true do begin item = produce_item; ProducerCon(item); end end; procedure consumer; begin while true do begin item = ProducerCon; consume_item(item); end end;

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

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

下面是 Java 使用管程解決的生產(chǎn)者-消費者問題

public class ProducerConsumer { static final int N = 100; // 定義緩沖區(qū)大小的長度 static Producer p = new Producer(); // 初始化一個新的生產(chǎn)者線程 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){ // 生產(chǎn)者循環(huán) item = produce_item(); mon.insert(item); } } private int produce_item(){...} // 生產(chǎn)代碼 } 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; // 計數(shù)器和索引 private synchronized void insert(int val){ if(count == N){ go_to_sleep(); // 如果緩沖區(qū)是滿的，則進入休眠 } buffer[hi] = val; // 向緩沖區(qū)插入內(nèi)容 hi = (hi + 1) % N; // 找到下一個槽的為止 count = count + 1; // 緩沖區(qū)中的數(shù)目自增 1 if(count == 1){ notify(); // 如果消費者睡眠，則喚醒 } } private synchronized void remove(int val){ int val; if(count == 0){ go_to_sleep(); // 緩沖區(qū)是空的，進入休眠 } val = buffer[lo]; // 從緩沖區(qū)取出數(shù)據(jù) lo = (lo + 1) % N; // 設(shè)置待取出數(shù)據(jù)項的槽 count = count - 1; // 緩沖區(qū)中的數(shù)據(jù)項數(shù)目減 1 if(count = N - 1){ notify(); // 如果生產(chǎn)者睡眠，喚醒它 } return val; } private void go_to_sleep() { try{ wait( ); }catch(Interr uptedExceptionexc) {}; } } }

上面的代碼中主要設(shè)計四個類，外部類(outer class) ProducerConsumer 創(chuàng)建并啟動兩個線程，p 和 c。第二個類和第三個類 Producer 和 Consumer 分別包含生產(chǎn)者和消費者代碼。最后，Our_monitor 是管程，它有兩個同步線程，用于在共享緩沖區(qū)中插入和取出數(shù)據(jù)。

在前面的所有例子中，生產(chǎn)者和消費者線程在功能上與它們是相同的。生產(chǎn)者有一個無限循環(huán)，該無限循環(huán)產(chǎn)生數(shù)據(jù)并將數(shù)據(jù)放入公共緩沖區(qū)中；消費者也有一個等價的無限循環(huán)，該無限循環(huán)用于從緩沖區(qū)取出數(shù)據(jù)并完成一系列工作。

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

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

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

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

消息傳遞

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

send(destination, &message); receive(source, &message);

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

消息傳遞系統(tǒng)的設(shè)計要點

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

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

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

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

用消息傳遞解決生產(chǎn)者-消費者問題

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

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

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

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

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

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

屏障

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

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

避免鎖：讀-復制-更新

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

然而，在某些情況下，我們可以允許寫操作來更新數(shù)據(jù)結(jié)構(gòu)，即便還有其他的進程正在使用。竅門在于確保每個讀操作要么讀取舊的版本，要么讀取新的版本，例如下面的樹

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

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

往期精選

萬字長文帶你還原進程和線程

這些操作系統(tǒng)的概念，保你沒聽過！

什么叫操作系統(tǒng)啊 | 戰(zhàn)術(shù)后仰

你要問我應用層？我就和你扯扯扯

文章參考：

《現(xiàn)代操作系統(tǒng)》

《Modern Operating System》forth edition