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

1、生產(chǎn)者-消費者(producer-consume r)問題，也稱作有界緩沖區(qū)(bounded-buffe r)問題，兩 個進程共享一個公共的固定大小的緩沖區(qū)。其中一個是生產(chǎn)者，用于將消息放入緩沖區(qū)；另 外一個是消費者，用于從緩沖區(qū)中取出消息。問題出現(xiàn)在當(dāng)緩沖區(qū)已經(jīng)滿了，而此時生產(chǎn)者 還想向其中放入一個新的數(shù)據(jù)項的情形，其解決方法是讓生產(chǎn)者此時進行休眠，等待消費者 從緩沖區(qū)中取走了一個或者多個數(shù)據(jù)后再去喚醒它。同樣地，當(dāng)緩沖區(qū)已經(jīng)空了，而消費者 還想去取消息，此時也可以讓消費者進行休眠，等待生產(chǎn)者放入一個或者多個數(shù)據(jù)時再喚醒 它。聽起來好像蠻對的，無懈可擊似的，但其實在實現(xiàn)時會有一個競爭條件存在

2、的。為了跟蹤緩 沖區(qū)中的消息數(shù)目，需要一個變量count。如果緩沖區(qū)最多存放N個消息，則生產(chǎn)者的代 碼會首先檢查count是否達到N，如果是，則生產(chǎn)者休眠；否則，生產(chǎn)者向緩沖區(qū)中放入 一個消息，并增加count的值。消費者的代碼也與此類似，首先檢測count是否為0，如果是，則休眠；否則，從緩沖區(qū) 中取出消息并遞減count的值。同時，每個進程也需要檢查是否需要喚醒另一個進程。代 碼可能如下：/緩沖區(qū)大小#define N 100int count = 0;/跟蹤緩沖區(qū)的記錄數(shù)/ 緩沖區(qū)中的數(shù)據(jù)項/*生產(chǎn)者進程*/ void procedure(void) int item;while(tru

3、e) /無限循環(huán)/ 產(chǎn)生下一個數(shù)據(jù)項/ 如果緩沖區(qū)滿了，/ 將新數(shù)據(jù)項放入緩沖區(qū)/計數(shù)器加1/表明插入之前為空，/ 消費/ 喚醒消費者item = produce_item();if (count = N)進行休眠sleep();insert_item(item);count = count + 1;if (count = 1)者等待wakeup(consumer);while(true)/ 無限循環(huán)if (count = 0)入休眠sleep();item = remove_item();count = count - 1;if (count = N -1)生產(chǎn)者wakeup(produce

4、r);consume_item(item);如果緩沖區(qū)為空，進/從緩沖區(qū)中取出一個數(shù)據(jù)項/計數(shù)器減1/緩沖區(qū)有空槽/喚醒/打印出數(shù)據(jù)項/*消費者進程*/ void consumer(void) /緩沖區(qū)中的數(shù)據(jù)項int item;看上去很美，哪里出了問題，這里對count的訪問是有可能出現(xiàn)競爭條件的：緩沖區(qū)為空， 消費者剛剛讀取count的值為0，而此時調(diào)度程序決定暫停消費者并啟動執(zhí)行生產(chǎn)者。生 產(chǎn)者向緩沖區(qū)中加入一個數(shù)據(jù)項，count加1。現(xiàn)在count的值變成了 1.它推斷剛才count 為0，所以此時消費者一定在休眠，于是生產(chǎn)者開始調(diào)用wakeup（consumer）來喚醒消費者。 但是

5、，此時消費者在邏輯上并沒有休眠，所以wakeup信號就丟失了。當(dāng)消費者下次運行 時，它將測試先前讀到的count值，發(fā)現(xiàn)為0（注意，其實這個時刻count已經(jīng)為1 了）， 于是開始休眠（邏輯上）。而生產(chǎn)者下次運行的時候，count會繼續(xù)遞增，并且不會喚醒 consumer 了，所以遲早會填滿緩沖區(qū)的，然后生產(chǎn)者也休眠，這樣兩個進程就都永遠(yuǎn)的休 眠下去了。1，使用信號量解決生產(chǎn)者-消費者問題首先了解一下信號量吧，信號量是E.W.Dijkstra在1965年提出的一種方法，它是使用一 個整型變量來累計喚醒的次數(shù)，供以后使用。在他的建議中，引入了一個新的變量類型，稱 為信號量（semaphore），

6、一個信號量的取值可以為0 （表示沒有保存下來的喚醒操作）或者為 正值（表示有一個或多個喚醒操作）。并且設(shè)立了兩種操作：down和up（分別為一般化后的sleep和wakeup，其實也是一般教 科書上說的P/V向量）。對一個信號量執(zhí)行down操作，表示檢查其值是否大于0，如果 該值大于0，則將其值減1（即用掉一個保存的喚醒信號）并繼續(xù)；如果為0，則進程休眠， 而且此時down操作并未結(jié)束。另外，就是檢查數(shù)值，修改變量值以及可能發(fā)生的休眠操作都作為單一的，不可分割的原子操作來完成。下面開始考慮用信號量來解決生產(chǎn)者-消費者問題了，不過在此之前，再次分析一下這個問 題的本質(zhì)會更清晰點：問題的實質(zhì)在于發(fā)

7、給一個（尚）未休眠進程（如上的消費者進程在只 判斷了 count = 0后即被調(diào)度出來，還未休眠）的wakeup信號丟失（如上的生產(chǎn)者進程 在判斷了 count = 1后以為消費者進程休眠，而喚醒它）了。如果它沒有丟失，則一切都 會很好。#define N 100typedef int semaphore;semaphore mutex = 1;semaphore empty = N;semaphore full= 0;/緩沖區(qū)中的槽數(shù)目/信號量一般被定義為特殊的整型數(shù)據(jù)/控制對臨界區(qū)的訪問/計數(shù)緩沖區(qū)中的空槽數(shù)目/計數(shù)緩沖區(qū)中的滿槽數(shù)目/*生產(chǎn)者進程*/void proceducer(voi

8、d)int item;while(1)item = procedure_item();down(&empty);down(&mutex);insert_item(item);up(&mutex);up(&full);加1/*消費者進程*/void consumer(voi)int item;while(1)down(&full);down(&mutex);item = remove_item();up(&mutex);up(&empty);/生成數(shù)據(jù)/將空槽數(shù)目減1/進入臨界區(qū)/將新數(shù)據(jù)放入緩沖區(qū)/離開臨界區(qū)/ 將滿槽的數(shù)目/將滿槽數(shù)目減1/ 進入臨界區(qū)/從緩沖區(qū)中取出數(shù)據(jù)項/ 離開臨界區(qū)/

9、將空槽數(shù)目加1consumer_item(item);/處理數(shù)據(jù)項該解決方案使用了三個信號量：一個為full,用來記錄充滿的緩沖槽的數(shù)目，一個為empty, 記錄空的緩沖槽總數(shù)，一個為mutex，用來確保生產(chǎn)者和消費者不會同時訪問緩沖區(qū)。mutex 的初始值為1，供兩個或者多個進程使用的信號量，保證同一個時刻只有一個進程可以進入 臨界區(qū)，稱為二元信號量(binary semaphore)0如果每一個進程在進入臨界區(qū)前都執(zhí)行一個 down(.)，在剛剛退出臨界區(qū)時執(zhí)行一個up(.)，就能夠?qū)崿F(xiàn)互斥。另外，通常是將down和up操作作為系統(tǒng)調(diào)用來實現(xiàn)，而且OS只需要在執(zhí)行以下操作 時暫時禁止全部中

10、斷：測試信號量，更新信號量以及在需要時使某個進程休眠。這里使用了三個信號量，但是它們的目的卻不相同，其中full和empty用來同步(synchronization)，而 mutex 用來實現(xiàn)互斥。2,使用消息傳遞解決生產(chǎn)者-消費者問題這種IPC方式使用兩條原語send和receive，也是系統(tǒng)調(diào)用。如：send(dest, &msg) /將消息msg發(fā)送到目標(biāo)(進程)dest中receive(src, &msg) /接收由src過來的msg,如果沒有消息可用，則可能阻塞接收者消息傳遞系統(tǒng)會面臨位于網(wǎng)絡(luò)中不同機器上的通信進程的情形，所以會更加的復(fù)雜。如：消 息可能被網(wǎng)絡(luò)丟失，一般使用確認(rèn)(AC

11、K)消息。如果發(fā)送方在一定的時間段內(nèi)沒有收到 確認(rèn)消息，則重發(fā)消息。如果消息本身被正確接收，但是返回的ACK消息丟失，發(fā)送方則重發(fā)消息，這樣接收方 就會收到兩份同樣的消息。一般使用在每條原始消息的頭部嵌入一個連續(xù)的序號來解決這個 問題。另外，消息傳遞系統(tǒng)還需要解決進程命名的問題，在send和receive系統(tǒng)調(diào)用中指定的進 程必須沒有二義性的。還有其他的一些問題，如性能問題，身份認(rèn)證等等，不過那個就會扯 多了，還是看看如果解決這個生產(chǎn)者-消費者的問題吧：#define N 100/緩沖區(qū)中的槽數(shù)目/*生產(chǎn)者進程*/void proceducer(void)int item;message ms

12、g;/ 消息緩沖區(qū)while(1)/生成數(shù)據(jù)/等待消費者發(fā)送空的緩沖區(qū)/創(chuàng)建待發(fā)送消息/發(fā)送數(shù)據(jù)項給消費者item = procedure_item(); receive(consumer, &msg); build_msg(&msg, item);send(consumer, &msg); /*消費者進程*/void consumer(voi)int item, i;message msg;for(i=0; iN; i+)send(producer, &msg);while(1)receive(producer, &msg);item = extract_item(&msg);send(pr

13、oceduer, &msg);產(chǎn)者consumer_item(item);/發(fā)送給生產(chǎn)者N個空緩沖區(qū)/接收包含數(shù)據(jù)項的消息/解析消息，并組裝成數(shù)據(jù)項/然后又將空緩沖區(qū)發(fā)送回生/處理數(shù)據(jù)項在這個解決方案中，共使用了 N條消息，有點類似于上一個的共享內(nèi)存緩沖區(qū)的N個槽， 消費者進程這邊首先通過一個for循環(huán)將N條空消息發(fā)送給生產(chǎn)者。當(dāng)生產(chǎn)者向消費者傳 遞一個數(shù)據(jù)項時，是通過取走每一條接收到的空消息，然后送回填充了內(nèi)容的消息給消費者 的。通過這種方式，整個消息傳遞系統(tǒng)中的總的消息數(shù)（包括空的消息+存了數(shù)據(jù)項的消 息=N）是不變的。如果運行過程中，生產(chǎn)者進程的速度比消費者快，則所有的消息最終都會塞滿，然后生產(chǎn)者 進程就會等待消費者（即使調(diào)用procedure也是阻塞在receive處），直到消費者返回一條 空的消息；反之亦然。下面再來看一下消息傳遞方式的兩種變體。一種是：為每一個進程分配一個唯一的地址，讓 消息按照這個進程的地址進行編址。也就是send和receive調(diào)用的第一個