進(jìn)程間的通信

進(jìn)程間的通信（Inter Process Communication, IPC）問題主要有3個：
(1) 一個進(jìn)程如何把信息傳遞給另一個進(jìn)程；
(2) 確保兩個或更多進(jìn)程在關(guān)鍵活動中不會出現(xiàn)交叉；
(3) 有協(xié)作關(guān)系的進(jìn)程的時序問題。

兩個或多個進(jìn)程讀寫某些共享數(shù)據(jù)，而最后的結(jié)果取決于進(jìn)程運(yùn)行的精確時序，稱為競爭條件（race condition）。我們把對共享內(nèi)存進(jìn)行訪問的程序片段稱作臨界區(qū)域（critical region）或臨界區(qū)（critical section），如果我們能夠保證兩個進(jìn)程不可能同時處于臨界區(qū)中，就能避免競爭條件。為了避免競爭條件，以某種手段確保當(dāng)前一個進(jìn)程在使用一個共享變量或文件時，其他進(jìn)程不能做同樣的操作，稱為互斥（mutual exclusion）。
一個好的互斥方案需要滿足下面4個條件：
(1) 任何兩個進(jìn)程不能同時處于其臨界區(qū)；
(2) 臨界區(qū)外運(yùn)行的進(jìn)程不得阻塞其他進(jìn)程；
(3) 不得使進(jìn)程無限期等待進(jìn)入臨界區(qū)；
(4) 不應(yīng)對CPU的速度和數(shù)量做任何假設(shè)。
下面將討論幾種實(shí)現(xiàn)互斥的方案。

1、忙等等（busy waiting）

1.1 屏蔽中斷

在單處理器系統(tǒng)中，最簡單的辦法是使每個進(jìn)程在剛剛進(jìn)入臨界區(qū)域后立即屏蔽所有中斷，并在就要離開之前打開中斷。在屏蔽中斷之后，CPU將不會切換到其他進(jìn)程。

但這個方案并不好，把屏蔽中斷的權(quán)利交給用戶進(jìn)程會引發(fā)風(fēng)險，如果進(jìn)程中斷屏蔽后不再打開，將使得整個系統(tǒng)終止。此外，在多CPU系統(tǒng)中，屏蔽中斷僅僅對執(zhí)行disable指令的那個CPU有效，其他CPU仍將繼續(xù)運(yùn)行，并可以訪問共享內(nèi)存。

1.2 鎖變量

設(shè)想有一個共享鎖變量，其初始值為0。0表示臨界區(qū)沒有進(jìn)程，1表示已經(jīng)有某個進(jìn)程進(jìn)入臨界區(qū)。
當(dāng)一個進(jìn)程要進(jìn)入臨界區(qū)時，先測試這把鎖，如果該鎖的值為0，則該進(jìn)程將其設(shè)置為1并進(jìn)入臨界區(qū)，若這把鎖的值為1，則該進(jìn)程將等待直到其值變?yōu)?。
鎖變量的缺陷：如果一個進(jìn)程讀出鎖變量的值為0，但在將它設(shè)置為1之前，另一個進(jìn)程被調(diào)度運(yùn)行，將該鎖變量設(shè)置為1。當(dāng)?shù)谝粋€進(jìn)程再次能運(yùn)行時，它同樣也將該鎖設(shè)置為1，則此時有兩個進(jìn)程進(jìn)入臨界區(qū)中。

1.3 嚴(yán)格輪轉(zhuǎn)法

[cpp] view plain copy

//進(jìn)程0
while(True){
while(turn != 0); //等待turn等于0
critical_region();
turn = 1; //離開臨界區(qū)
noncritical_region();
}
//進(jìn)程1
while(True){
while(turn != 1); //等待turn等于1
critical_region();
turn = 0; //離開臨界區(qū)
noncritical_region();
}

嚴(yán)格輪轉(zhuǎn)法采用忙等待，即連續(xù)測試一個變量直到某個值出現(xiàn)為止，用于忙等待的鎖稱為自旋鎖（spin lock），這種方式比較浪費(fèi)CPU時間，通常應(yīng)該避免。
代碼說明：進(jìn)程0離開臨界區(qū)，將turn設(shè)置為1，以便允許進(jìn)程1進(jìn)入其臨界區(qū)。假設(shè)進(jìn)程1很快便離開臨界區(qū)，則此時兩個進(jìn)程都處于臨界區(qū)之外，turn的值又被設(shè)置為0。如果此時進(jìn)程1突然結(jié)束了非臨界區(qū)并且返回循環(huán)的開始，但是，這時它不能進(jìn)入臨界區(qū)，因?yàn)閠urn的值為0，而此時進(jìn)程0還在忙于非臨界區(qū)的操作，進(jìn)程1只有繼續(xù)while循環(huán)，直到進(jìn)程0把turn的值改為1。這實(shí)際上違反了前面敘述的互斥條件(2)，即臨界區(qū)外運(yùn)行的進(jìn)程不得阻塞其他進(jìn)程。

1.4 Peterson解法

[cpp] view plain copy

#define N 2 //進(jìn)程數(shù)量
int turn; //鎖變量
int interested[N];
void enter_region(int process){
int other;
other = 1 - process; //其他進(jìn)程
interested[process] = True;
turn = process;
while(turn==process && interested[other]==True); //等待other離開臨界區(qū)
}
void leave_region(int process){
interested[process] = False;
}

代碼說明：一開始，沒有任何進(jìn)程處于臨界區(qū)，現(xiàn)在進(jìn)程0調(diào)用enter_region，它通過設(shè)置數(shù)組元素和將turn置為0來表示它希望進(jìn)入臨界區(qū)。由于進(jìn)程1并不處于臨界區(qū)，enter_region很快便返回。如果此時進(jìn)程1調(diào)用enter_region，進(jìn)程1將在此處掛起直到interested[0]變成False，該事件只有在進(jìn)程0調(diào)用leave_region退出臨界區(qū)時才會發(fā)生。

1.5 TSL指令/XCHG指令

在某些計算機(jī)上，有下面這樣的指令：
TSL RX, LOCK
TSL（Test and Set Lock）指令將一個內(nèi)存字lock讀入寄存器RX中，然后將該內(nèi)存地址上存一個非零值。讀字和寫字操作保證是不可分割的，即該指令結(jié)束之前其他處理器均不允許訪問該內(nèi)存字。執(zhí)行TSL指令的CPU將鎖住內(nèi)存總線，以禁止其他CPU在本指令結(jié)束之前訪問內(nèi)存。

[plain] view plain copy

//用TSL指令進(jìn)入和離開臨界區(qū)
enter_region:
TSL REGISTER, LOCK ;復(fù)制鎖變量到寄存器，并將鎖設(shè)為1
CMP REGISTER, #0
JNE enter_region ;若鎖不等于0，繼續(xù)循環(huán)等待
RET
leave_region:
MOVE LOCK, #0
RET

從代碼可以看出，進(jìn)程在進(jìn)入臨界區(qū)之前先調(diào)用enter_region，這將導(dǎo)致忙等待，直到鎖空閑為止，隨和它獲得該鎖并返回。在進(jìn)程從臨界區(qū)返回時它調(diào)用leave_region，這將把鎖設(shè)置為0。
一個可替代TSL指令的是XCHG，它原子性地交換兩個位置的內(nèi)容。

[plain] view plain copy

//用XCHG指令進(jìn)入和離開臨界區(qū)
enter_region:
MOVE REGISTER, #1
XCHG REGISTER, LOCK ;交換寄存器與鎖變量的內(nèi)容
CMP REGISTER, #0 ;測試鎖
JNE enter_region ;若鎖不等于0，繼續(xù)循環(huán)等待
RET
leave_region:
MOVE LOCK, #0
RET

2、睡眠和喚醒（sleep & wake up）

忙等待的缺點(diǎn)：當(dāng)一個進(jìn)程要進(jìn)入臨界區(qū)時，先檢查是否允許進(jìn)入，若不允許，則該進(jìn)程將原地等待，直到允許為止。
考慮一種情況：現(xiàn)在有兩個進(jìn)程H和L，H的優(yōu)先級更高，L處于臨界區(qū)，H處于就緒態(tài)。由于H的優(yōu)先級高于L，L將不會被調(diào)度，因此也無法離開臨界區(qū)，這時H將始終處于忙等待，這種情況稱為優(yōu)先級反轉(zhuǎn)問題（priority inversion problem）。
睡眠是將一個無法進(jìn)入臨界區(qū)的進(jìn)程阻塞，而不是忙等待，該進(jìn)程被掛起，直到另外一個進(jìn)程將其喚醒。

[cpp] view plain copy

#define N 100 //緩沖區(qū)大小
int count = 0;
//數(shù)據(jù)生產(chǎn)者
void producer(void){
int item;
while(True){
item = produce_item(); //產(chǎn)生下一新數(shù)據(jù)項(xiàng)
if(count==N)sleep(); //如果緩沖區(qū)滿，就進(jìn)入休眠狀態(tài)
insert_item(item); //將新數(shù)據(jù)項(xiàng)放入緩沖區(qū)
count++;
if(count==1)wakeup(consumer);//喚醒消費(fèi)者
}
}
//數(shù)據(jù)消費(fèi)者
void consumer(void){
int item;
while(True){
if(count==0)sleep(); //緩沖區(qū)為空，就進(jìn)入休眠狀態(tài)
item = remove_item(); //從緩沖區(qū)取走一個數(shù)據(jù)項(xiàng)
count--;
if(count==N-1)wakeup(producer);//喚醒生產(chǎn)者
consume_item(item);
}
}

代碼說明：在生產(chǎn)者-消費(fèi)者（producer-consumer）問題中，兩個進(jìn)程共享一個公共的固定大小的緩沖區(qū)（bounded-buffer），其中一個是生產(chǎn)者，將信息放入緩沖區(qū)；另一個是消費(fèi)者，從緩沖區(qū)取走信息。當(dāng)緩沖區(qū)滿時，讓生產(chǎn)者睡眠，待消費(fèi)者從緩沖區(qū)取出一個或多個數(shù)據(jù)項(xiàng)時再喚醒它。同樣地，當(dāng)消費(fèi)者試圖從空緩沖區(qū)取數(shù)據(jù)時，消費(fèi)者就睡眠，直到生產(chǎn)者向其中放入一些數(shù)據(jù)項(xiàng)時再喚醒它。
不過上面的代碼仍存在一個問題，其原因是對count的訪問未加限制。這種情況是：當(dāng)緩沖區(qū)為空時，消費(fèi)者剛剛讀取count的值為0，而此時調(diào)度程序恰好將消費(fèi)者掛起，并啟動生產(chǎn)者，生產(chǎn)者向緩沖區(qū)加入一個數(shù)據(jù)項(xiàng)，count加1。現(xiàn)在count的值為1，它推斷認(rèn)為由于count剛才為0，所以消費(fèi)者一定在睡眠，于是生產(chǎn)者調(diào)用wakeup來喚醒消費(fèi)者。但是，消費(fèi)者此時在邏輯上并未睡眠，所以wakeup信號丟失。當(dāng)消費(fèi)者再次運(yùn)行時，它將測試先前讀到的count值，發(fā)現(xiàn)它為0，于是睡眠。這樣生產(chǎn)者遲早會填滿整個緩沖區(qū)，兩個進(jìn)程都將永遠(yuǎn)睡眠下去。
上面這個問題的實(shí)質(zhì)在于給一個清醒的進(jìn)程發(fā)送的wakeup信號被丟失了。我們可以設(shè)置一個喚醒等待位，當(dāng)一個清醒的進(jìn)程收到wakeup信號時，將喚醒等待為置為1，隨后，如果該進(jìn)程收到sleep信號，先檢測喚醒等待位，如果喚醒等待位為1，則不睡眠，而只是將喚醒等待位清0。

3、信號量（semaphore）

信號量是設(shè)置一個整型變量來累計喚醒次數(shù)。對信號量有兩種操作：down和up（一般化后的sleep和wankeup）。
對信號量執(zhí)行down操作，則是先檢查其值是否大于0，若該值大于0，則將其值減1并繼續(xù)，若該值為0，則進(jìn)程將睡眠。這里，檢查數(shù)值、修改變量值以及可能發(fā)生的睡眠操作是一個原子操作（不會被中間打斷）。
up操作對信號量加1，信號量的增值1和喚醒操作同樣是不可分割的。

[cpp] view plain copy

#define N 100 //緩沖區(qū)大小
typedef int semaphore;
semaphore full = 0; //緩沖區(qū)已用數(shù)目
semaphore empty = N; //緩沖區(qū)剩余數(shù)目
semaphore mutex = 1; //控制對臨界區(qū)的訪問
//數(shù)據(jù)生產(chǎn)者
void producer(void){
int item;
while(True){
item = produce_item(); //產(chǎn)生下一新數(shù)據(jù)項(xiàng)
down(&empty);
down(&mutex); //進(jìn)入臨界區(qū)
insert_item(item);
up(&mutex); //離開臨界區(qū)
up(&full);
}
}
//數(shù)據(jù)消費(fèi)者
void consumer(void){
int item;
while(True){
down(&full);
down(&mutex);
item = remove_item();
up(&mutex);
up(&empty);
consume_item(item);
}
}

代碼說明：mutex是一個二元信號量，每個進(jìn)程在進(jìn)入臨界區(qū)前都對它執(zhí)行一個down操作，離開臨界區(qū)后執(zhí)行一個up操作，就能夠?qū)崿F(xiàn)互斥。信號量的另一個作用是實(shí)現(xiàn)同步（synchronization），信號量full和empty用來保證當(dāng)緩沖區(qū)滿地時候生產(chǎn)者停止運(yùn)行，以及當(dāng)緩沖區(qū)空的時候消費(fèi)者停止運(yùn)行。

4、互斥量（mutex）

互斥量是一種退化的信號量，它只有兩種狀態(tài)：加鎖和解鎖，這樣只要一個二進(jìn)制位即可表示。
互斥量在實(shí)現(xiàn)用戶級線程包時非常有用。當(dāng)一個線程需要訪問臨界區(qū)時，調(diào)用mutex_lock，如果該互斥量當(dāng)前是解鎖的，此調(diào)用成功，調(diào)用線程可以自由進(jìn)入臨界區(qū)。如果該互斥量已經(jīng)加鎖，調(diào)用線程被阻塞，直到在臨界區(qū)中的線程完成并調(diào)用mutex_unlock。如果多個線程被阻塞在該互斥量上，將隨機(jī)選擇一個線程并允許它獲得鎖。

[cpp] view plain copy

mutex_lock:
TSL REGISTER, MUTEX
CMP REGISTER, #0 ;測試互斥量
JZE ok ;解鎖
CALL thread_yield ;互斥量忙，調(diào)度另一個線程
JMP mutex_lock ;稍后再試
ok: RET
mutex_unlock:
MOVE MUTEX, #0
RET

可以看出，線程互斥量與1.5節(jié)中的TSL指令實(shí)現(xiàn)的進(jìn)程互斥類似，但有一個關(guān)鍵區(qū)別：
當(dāng)enter_region進(jìn)入臨界區(qū)失敗時，它始終重復(fù)測試鎖（忙等待）。實(shí)際上，由于時鐘超時作用，會調(diào)度其他進(jìn)程運(yùn)行，這樣遲早擁有鎖的進(jìn)程會進(jìn)入運(yùn)行并釋放鎖。
當(dāng)mutex_lock進(jìn)入臨界區(qū)失敗時，它調(diào)用thread_yield主動釋放CPU給另外一個線程，這樣就沒有忙等待。這是因?yàn)榫€程沒有時鐘中斷，通過忙等待的方式來獲取鎖的線程將永遠(yuǎn)循環(huán)下去，絕不會得到鎖，而其他線程也沒有機(jī)會運(yùn)行了。

下面列出了幾個與線程互斥量相關(guān)的調(diào)用

pthread_mutex_init	創(chuàng)建一個互斥量
pthread_mutex_destroy	撤銷一個已經(jīng)存在的互斥量
pthread_mutex_lock	獲得一個鎖或阻塞
pthread_mutex_unlock	解鎖
pthread_mutex_trylock	獲得一個鎖或失敗

除互斥量外，pthread提供了另外一種同步機(jī)制——條件變量（condition variable）?；コ饬吭试S或阻塞對臨界區(qū)的訪問，條件變量則允許線程由于一些未達(dá)到的條件而阻塞。條件變量和互斥量經(jīng)常一起使用，這種模式用于讓一個線程鎖住一個互斥量，然后當(dāng)它不能獲得它期待的結(jié)果時等待一個條件變量。最后另一個線程會向他發(fā)信號，使它可以繼續(xù)執(zhí)行。注意：條件變量不會存在內(nèi)存中，如果將一個信號量傳遞給一個沒有線程在等待的條件變量，那么這個信號就會丟失。

下面列出了幾個與條件變量相關(guān)的調(diào)用

pthread_cond_init	創(chuàng)建一個條件變量
pthread_cond_destroy	撤銷一個條件變量
pthread_cond_wait	阻塞調(diào)用線程直到另一個線程給它發(fā)信號
pthread_cond_signal	向另一個線程發(fā)信號來喚醒它
pthread_cond_broadcast	向多個線程發(fā)信號來讓它們?nèi)繂拘?/span>

5、管程（monitor）

6、消息傳遞（message passing）

消息傳遞使用的兩條原語：send和receive，前一個調(diào)用向目標(biāo)發(fā)送一條消息，后一個調(diào)用從一個給定的源接收一條消息。如果沒有消息可用，則接收者可能被阻塞，直到下一條消息到達(dá)，或者帶著一個錯誤碼立即返回。
消息傳遞過程中可能發(fā)生消息丟失的現(xiàn)象，因此，一旦接收到消息，接收方應(yīng)該回送一條確認(rèn)消息（acknowledge），如果發(fā)送方在一段時間間隔內(nèi)沒有收到確認(rèn)，則重發(fā)消息。而如果消息本身被正確接收，但返回給發(fā)送方的確認(rèn)消息丟失，發(fā)送者將重發(fā)消息，這樣將導(dǎo)致接收者接收到兩次相同的消息。通常采用在每條原始消息中嵌入一個連續(xù)的序號來解決此問題，如果接收者接收到一條消息，并且它具有與前面某條消息一樣的序號，就知道這條消息是重復(fù)的。

[cpp] view plain copy

#define N 100
void producer(void){
int item;
message m;
while(True){
item = produce_item();
receive(consumer, &m); //等待消費(fèi)者發(fā)送空緩沖區(qū)
build_message(&m, item);//建立一個待發(fā)送的信號
send(consumer, &m); //發(fā)送數(shù)據(jù)項(xiàng)給消費(fèi)者
}
}
void consumer(void){
int item;
message m;
for(int i=0; i<N; ++i)
send(producer, &m); //發(fā)送N個空緩沖區(qū)
while(True){
receive(produce, &m); //接收包含數(shù)據(jù)項(xiàng)的消息
item = extract_item(&m);//將數(shù)據(jù)項(xiàng)從消息中提取出來
send(producer, &m); //將空緩沖區(qū)發(fā)送回生產(chǎn)者
consume_item(item);
}
}

代碼說明：消費(fèi)者先將N條空消息發(fā)送給生產(chǎn)者，當(dāng)生產(chǎn)者向消費(fèi)者傳遞一個數(shù)據(jù)項(xiàng)時，它取走一條空消息并送回一條填充了內(nèi)容的消息。通過這種方式，系統(tǒng)中的消息總數(shù)保持不變。如果生產(chǎn)者的速度比消費(fèi)者快，所有消息都將被填滿，等待消費(fèi)者；相反，如果消費(fèi)者速度比生產(chǎn)者快，所有消息均為空，等待生產(chǎn)者來填充它們，消費(fèi)者被阻塞，以等待一條填充過的消息。

7、屏障（barrier）

屏障機(jī)制適用于進(jìn)程組。在有些應(yīng)用中劃分了若干階段，并且規(guī)定，除非所有進(jìn)程都就緒準(zhǔn)備著手下一個階段，否則任何進(jìn)程都不能進(jìn)入下一個階段。可以通過在每個階段的結(jié)尾設(shè)置屏障來實(shí)現(xiàn)這種行為。如下圖所示，在所有進(jìn)程到達(dá)屏障前，先到達(dá)的進(jìn)程會被掛起，只有當(dāng)所有進(jìn)程都就緒后，所有進(jìn)程一起被釋放。