實(shí)驗(yàn)3--讀者-寫者問題與進(jìn)程同步

1、實(shí)驗(yàn)3 讀者/寫者問題與進(jìn)程同步3.1 實(shí)驗(yàn)?zāi)康睦斫馀R界區(qū)和進(jìn)程互斥的概念，掌握用信號量和PV操作實(shí)現(xiàn)進(jìn)程互斥的方法。3.2 實(shí)驗(yàn)要求在linux環(huán)境下編寫一個(gè)控制臺應(yīng)用程序，該程序運(yùn)行時(shí)能創(chuàng)建N個(gè)線程（或者進(jìn)程），其中既有讀者線程又有寫者線程，它們按照事先設(shè)計(jì)好的測試數(shù)據(jù)進(jìn)行讀寫操作。請用信號量和PV操作實(shí)現(xiàn)讀者/寫者問題。讀者/寫者問題的描述如下：有一個(gè)被許多進(jìn)程共享的數(shù)據(jù)區(qū)，這個(gè)數(shù)據(jù)區(qū)可以是一個(gè)文件，或者主存的一塊空間（比如一個(gè)數(shù)組或一個(gè)變量），甚至可以是一組處理器寄存器。有一些只讀取這個(gè)數(shù)據(jù)區(qū)的進(jìn)程（reader）和一些只往數(shù)據(jù)區(qū)中寫數(shù)據(jù)的進(jìn)程（writer）。以下假設(shè)共享數(shù)據(jù)區(qū)是文件

2、。這些讀者和寫者對數(shù)據(jù)區(qū)的操作必須滿足以下條件：讀讀允許；讀寫互斥；寫寫互斥。這些條件具體來說就是：（1）任意多的讀進(jìn)程可以同時(shí)讀這個(gè)文件；（2）一次只允許一個(gè)寫進(jìn)程往文件中寫；（3）如果一個(gè)寫進(jìn)程正在往文件中寫，禁止任何讀進(jìn)程或?qū)戇M(jìn)程訪問文件；（4）寫進(jìn)程執(zhí)行寫操作前，應(yīng)讓已有的寫者或讀者全部退出。這說明當(dāng)有讀者在讀文件時(shí)不允許寫者寫文件。對于讀者-寫者問題，有三種解決方法：1、讀者優(yōu)先除了上述四個(gè)規(guī)則外，還增加讀者優(yōu)先的規(guī)定，當(dāng)有讀者在讀文件時(shí)，對隨后到達(dá)的讀者和寫者，要首先滿足讀者，阻塞寫者。這說明只要有一個(gè)讀者活躍，那么隨后而來的讀者都將被允許訪問文件，從而導(dǎo)致寫者長時(shí)間等待，甚至有可

3、能出現(xiàn)寫者被餓死的情況。2、寫者優(yōu)先除了上述四個(gè)規(guī)則外，還增加寫者優(yōu)先的規(guī)定，即當(dāng)有讀者和寫者同時(shí)等待時(shí)，首先滿足寫者。當(dāng)一個(gè)寫者聲明想寫文件時(shí)，不允許新的讀者再訪問文件。3、無優(yōu)先除了上述四個(gè)規(guī)則外，不再規(guī)定讀寫的優(yōu)先權(quán)，誰先等待誰就先使用文件。3.3 實(shí)驗(yàn)步驟 算法分析1、錯(cuò)誤的解法圖3-1 錯(cuò)誤的解法semaphore r_w_w=1;reader()P(r_w_w)；讀文件；V(r_w_w)；writer()P(r_w_w)；寫文件；V(r_w_w)；有同學(xué)認(rèn)為，可以將文件視為臨界資源，使用臨界資源的代碼就構(gòu)成臨界區(qū)，為了對臨界區(qū)進(jìn)行管理，只需設(shè)置一個(gè)互斥信號量r_w_w，讀或者寫之前

4、執(zhí)行P(r_w_w)，之后執(zhí)行V(r_w_w)即可，從而得到圖3-1所示的算法描述。該方法雖然能滿足讀寫互斥和寫寫互斥，但是不滿足讀讀允許，只要有一個(gè)讀者在讀文件，其他的讀者都被阻塞了。可見，單純使用互斥信號量不能解決讀者/寫者問題，還需要引入計(jì)數(shù)器對讀者進(jìn)行記數(shù)。2、讀者優(yōu)先如何糾正上述解法中存在的錯(cuò)誤呢？其實(shí)，對于相繼到達(dá)的一批讀者，并不是每個(gè)讀者都需要執(zhí)行P(r_w_w)和V(r_w_w)。在這批讀者中，只有最先到達(dá)的讀者才需要執(zhí)行P(r_w_w)，與寫者競爭對文件的訪問權(quán)，若執(zhí)行P(r_w_w)成功則獲得了文件的訪問權(quán)，其他的讀者可直接訪問文件；同理，只有最后退出臨界區(qū)的讀者需要執(zhí)行V

5、(r_w_w)來歸還文件訪問權(quán)。為了記錄正在讀文件的一批讀者的數(shù)量，需要設(shè)置一個(gè)整型變量readercount，每一個(gè)讀者到達(dá)時(shí)都要將readercount加1，退出時(shí)都要將readercount減1。由于只要有一個(gè)讀者在讀文件，便不允許寫者寫文件，所以，僅當(dāng)readercount=0時(shí)，即尚無讀者在讀文件時(shí)，讀者才需要執(zhí)行P(r_w_w)操作。若P(r_w_w)操作成功，讀者便可去讀文件，相應(yīng)地，readercount+1。同理，僅當(dāng)在執(zhí)行了readercount減1操作后其值為0時(shí)，才需要執(zhí)行V(r_w_w)操作，以便讓寫者寫文件。又因?yàn)閞eadercount是一個(gè)可被多個(gè)讀者訪問的臨界資

6、源，所以應(yīng)該為它設(shè)置一個(gè)互斥信號量readercount_mutex.。每個(gè)讀者在訪問readercount之前執(zhí)行P(readercount_mutex)，之后執(zhí)行V(readercount_mutex)。通過上述分析得到圖3-2所示的算法描述，其中的數(shù)字表示語句對應(yīng)的行號。圖3-2 讀者優(yōu)先算法01 semaphore r_w_w=1;02 semaphore readercount_mutex=1;03 int readercount=0; 04 reader()05 P(readercount_mutex);06 if(readercount=0) P(r_w_w);07 reader

7、count+;08 V(readercount_mutex);09 讀文件;10 P(readercount_mutex);11 readercount-;12 if(readercount=0) V(r_w_w);13 V(readercount_mutex);14 1516 writer()17 P(r_w_w);18 寫文件;19 V(r_w_w);20 3、寫者優(yōu)先通過增加信號量并修改上述程序可以得到寫者優(yōu)先算法。為了實(shí)現(xiàn)寫者優(yōu)先算法，需要將寫者和讀者分開排隊(duì)，并且第一個(gè)讀者和其它讀者也要分開排隊(duì)。這樣就需要三個(gè)隊(duì)列，一個(gè)是寫者排隊(duì)的地方，另一個(gè)是第一個(gè)讀者排隊(duì)的地方，第三個(gè)是其它讀者

8、排隊(duì)的地方。相應(yīng)地需要設(shè)置三個(gè)信號量，r_w_w、first_reader_wait和reader_wait。當(dāng)一個(gè)寫者聲明想寫文件時(shí)，可以讓新的讀者中的第一個(gè)到first_reader_wait上排隊(duì)等待；當(dāng)有讀者阻塞在first_reader_wait上時(shí)，讓其它讀者阻塞在reader_wait上；當(dāng)有一個(gè)寫者在寫文件時(shí)，其它寫者到r_w_w上排隊(duì)。只要有活躍的寫者或者寫者隊(duì)列不為空，則阻塞新到達(dá)的讀者。為了記錄已經(jīng)發(fā)出聲明的寫者數(shù)量，需要設(shè)置一個(gè)整數(shù)writercount，以表示聲明要寫文件的寫者數(shù)目。由于只要有一個(gè)寫者到達(dá)，就不允許讀者去讀，因此僅當(dāng)writercount=0，表示無寫

9、者聲明寫時(shí)，寫者才需要執(zhí)行P(first_reader_wait)操作，若操作成功，寫者便可以執(zhí)行P(r_w_w)去競爭寫文件權(quán)利。其它寫者不需要再向讀者聲明，可以直接執(zhí)行P(r_w_w)去競爭寫文件權(quán)利。同理僅當(dāng)寫者在執(zhí)行writercount減1操作后其值為0時(shí)，才需要執(zhí)行V(first_reader_wait)操作，以便喚醒第一個(gè)被阻塞的讀者去讀文件。又因?yàn)閣ritercount是一個(gè)可被多個(gè)寫者訪問的臨界資源，所以，應(yīng)該為它設(shè)置一個(gè)互斥信號量writer_mutex。4、無優(yōu)先除了在讀者優(yōu)先時(shí)需要的信號量r_w_w和readercount_mutex之外，還需要設(shè)置一個(gè)信號量wait供

10、讀者和寫者排隊(duì)。讀者和寫者都排在wait隊(duì)列上。若有讀者在讀文件，則第一個(gè)寫者阻塞在r_w_w上，其它的寫者和讀者阻塞在wait上；若有一個(gè)寫者在寫文件，則其它寫者和讀者都阻塞在wait上。無優(yōu)先的算法描述如圖3-3所示。圖3-3 無優(yōu)先算法01 semaphore r_w_w=1;02 semaphore wait=1;03 semaphore readercount_mutex=1;04 int readercount=0; 05 reader()06 P(wait);07 P(readercount_mutex);08 if(readercount=0) P(r_w_w);09 read

11、ercount+;10 V(readercount_mutex);11 V(wait);12 讀文件;13 P(readercount_mutex);14 readercount-;15 if(readercount=0) V(r_w_w);16 V(readercount_mutex);17 18 writer()19 P(wait);20 P(r_w_w);21 寫文件;22 V(r_w_w);23 V(wait);24 3.3.2 程序功能及界面設(shè)計(jì)該程序采用簡單的控制臺應(yīng)用程序界面，在主界面上顯示程序的功能。該程序的功能如下：1. 演示讀者優(yōu)先算法；2. 演示寫者優(yōu)先算法；3. 演示無

12、優(yōu)先算法；4. 退出。3.3.3 函數(shù)設(shè)計(jì)實(shí)現(xiàn)讀者/寫者問題的源程序名稱是reader_and_writer.cpp。該程序共包括10個(gè)函數(shù)。這些函數(shù)可以分成4組。各組包含的函數(shù)及其功能如圖3-4。組別包括函數(shù)函數(shù)功能一main()顯示主菜單，接收用戶的選擇并執(zhí)行相應(yīng)的功能。二RF_reader_thread()RF_writer_thread()reader_first()讀者優(yōu)先算法的讀者線程函數(shù)讀者優(yōu)先算法的寫者線程函數(shù)讀者優(yōu)先算法的初始化函數(shù)：創(chuàng)建10個(gè)線程并等待它們結(jié)束三WF_reader_thread()WF_writer_thread()writer_first()寫者優(yōu)先算法的

13、讀者線程函數(shù)寫者優(yōu)先算法的寫者線程函數(shù)寫者優(yōu)先算法的初始化函數(shù)：創(chuàng)建10個(gè)線程并等待它們結(jié)束四FIFO_reader_thread()FIFO_writer_thread()first_come_first_serverd()無優(yōu)先算法的讀者線程函數(shù)無者優(yōu)先算法的寫者線程函數(shù)無者優(yōu)先算法的初始化函數(shù)：創(chuàng)建10個(gè)線程并等待它們結(jié)束圖3-4 函數(shù)功能簡述程序開始部分定義了宏MAX_THREAD，表示程序中創(chuàng)建的線程數(shù)。還定義了測試數(shù)據(jù)的結(jié)構(gòu)體TEST_INFO，該結(jié)構(gòu)體包含三個(gè)數(shù)據(jù)項(xiàng)：線程名稱；提出請求的時(shí)刻；操作持續(xù)時(shí)間。接著定義了全局變量，這些全局變量的作用如下：數(shù)組test_data保存了1

14、0個(gè)線程的測試數(shù)據(jù)；整數(shù)read_count記錄一段時(shí)間內(nèi)同時(shí)對文件進(jìn)行讀操作的線程數(shù)；整數(shù)write_count記錄一段時(shí)間內(nèi)提出寫操作請求的線程數(shù)，該整數(shù)只在寫者優(yōu)先算法中使用；CS_DATA是臨界區(qū)變量，用來保護(hù)文件，實(shí)現(xiàn)對文件的讀寫互斥和寫寫互斥（相當(dāng)于算法描述中的r_w_w）；互斥體h_mutex_read_count用來保護(hù)整數(shù)read_count，以保證多個(gè)讀者對read_count的互斥訪問；互斥體h_mutex_write_count用來保護(hù)整數(shù)write_count，以保證多個(gè)寫者對write_count的互斥訪問，該互斥體只在寫者優(yōu)先算法中使用；互斥體h_mutex_fi

15、rst_reader_wait和h_mutex_reader_wait只在寫者優(yōu)先算法中使用，當(dāng)有寫者在寫文件時(shí)，提出讀請求的第一個(gè)讀者阻塞在h_mutex_first_reader_wait上，其余的讀者阻塞在h_mutex_reader_wait上；互斥體h_mutex_wait只在無優(yōu)先算法中使用，當(dāng)文件被使用時(shí)，后繼的讀請求和寫請求依次阻塞在h_mutex_wait上。3.3.4 參考源程序3.3.4.1 Linux下的參考源程序編譯命令gcc reader_and_writer .cpp o reader_and_writer.o lcurses lpthread程序清單#inclu

16、de <unistd.h>#include <pthread.h>#include <curses.h>#include <stdlib.h>#include <string.h>#define MAX_THREAD 10 typedef structchar thread_name3;unsigned int require_moment;unsigned int persist_time;TEST_INFO;TEST_INFO test_dataMAX_THREAD="r1",0,15,"r2&quo

17、t;,1, 15,"w1",3,3,"r3",4, 2,"w2",5,6,"w3",6,10,"r4",7,8,"r5",9,2,"w4",10,18,"w5",12,2;int read_count=0;int write_count=0;pthread_mutex_t CS_DATA;pthread_mutex_t h_mutex_read_count;pthread_mutex_t h_mutex_write_count;pthr

18、ead_mutex_t h_mutex_reader_wait;pthread_mutex_t h_mutex_first_reader_wait;pthread_mutex_t h_mutex_wait;void* RF_reader_thread(void *data)char thread_name3;strcpy(thread_name,(TEST_INFO *)data)->thread_name);sleep(TEST_INFO *)data)->require_moment);pthread_mutex_lock(&h_mutex_read_count);re

19、ad_count+;if(read_count=1)pthread_mutex_lock(&CS_DATA);pthread_mutex_unlock(&h_mutex_read_count);printw("%s ",thread_name);refresh();sleep(TEST_INFO *)data)->persist_time);pthread_mutex_lock(&h_mutex_read_count);read_count-;if(read_count=0)pthread_mutex_unlock(&CS_DATA);

20、pthread_mutex_unlock(&h_mutex_read_count);return 0;void* RF_writer_thread(void *data)sleep(TEST_INFO *)data)->require_moment);pthread_mutex_lock(&CS_DATA);printw("%s ",(TEST_INFO *)data)->thread_name);refresh();sleep(TEST_INFO *)data)->persist_time);pthread_mutex_unlock(&a

21、mp;CS_DATA);return 0;void reader_first()int i=0;pthread_t h_threadMAX_THREAD;printw("reader first require sequence:");for(i=0;i<MAX_THREAD;i+)printw("%s ",test_datai.thread_name);printw("n");printw("reader first operation sequence:");refresh();pthread_mutex

22、_init(&CS_DATA,NULL);for(i=0;i<MAX_THREAD;i+)if(test_datai.thread_name0='r')pthread_create(&h_threadi,NULL,RF_reader_thread,&test_datai);elsepthread_create(&h_threadi,NULL,RF_writer_thread,&test_datai);for(i=0;i<MAX_THREAD;i+)pthread_join(h_threadi,NULL); printw(&qu

23、ot;n");refresh();void* FIFO_reader_thread(void *data)char thread_name3;strcpy(thread_name,(TEST_INFO *)data)->thread_name);sleep(TEST_INFO *)data)->require_moment);pthread_mutex_lock(&h_mutex_wait);pthread_mutex_lock(&h_mutex_read_count);read_count+;if(read_count=1)pthread_mutex_l

24、ock(&CS_DATA);pthread_mutex_unlock(&h_mutex_read_count);pthread_mutex_unlock(&h_mutex_wait);printw("%s ",thread_name);refresh();sleep(TEST_INFO *)data)->persist_time);pthread_mutex_lock(&h_mutex_read_count);read_count-;if(read_count=0)pthread_mutex_unlock(&CS_DATA);p

25、thread_mutex_unlock(&h_mutex_read_count);return 0;void* FIFO_writer_thread(void *data)sleep(TEST_INFO *)data)->require_moment);pthread_mutex_lock(&h_mutex_wait);pthread_mutex_lock(&CS_DATA);printw("%s ",(TEST_INFO *)data)->thread_name);refresh();sleep(TEST_INFO *)data)-&g

26、t;persist_time);pthread_mutex_unlock(&CS_DATA);pthread_mutex_unlock(&h_mutex_wait);return 0;void first_come_first_served()int i=0;pthread_t h_threadMAX_THREAD;printw("FCFS require sequence:");for(i=0;i<MAX_THREAD;i+)printw("%s ",test_datai.thread_name);printw("n&q

27、uot;);printw("FCFS:operation sequence:");refresh();pthread_mutex_init(&CS_DATA,NULL);for(i=0;i<MAX_THREAD;i+)if(test_datai.thread_name0='r')pthread_create(&h_threadi,NULL,FIFO_reader_thread,&test_datai);elsepthread_create(&h_threadi,NULL,FIFO_writer_thread,&t

28、est_datai);for(i=0;i<MAX_THREAD;i+)pthread_join(h_threadi,NULL); printw("n");refresh();void* WF_reader_thread(void *data)char thread_name3;strcpy(thread_name,(TEST_INFO *)data)->thread_name);sleep(TEST_INFO *)data)->require_moment);pthread_mutex_lock(&h_mutex_reader_wait);pthr

29、ead_mutex_lock(&h_mutex_first_reader_wait);pthread_mutex_lock(&h_mutex_read_count);read_count+;if(read_count=1)pthread_mutex_lock(&CS_DATA);pthread_mutex_unlock(&h_mutex_read_count);pthread_mutex_unlock(&h_mutex_first_reader_wait);pthread_mutex_unlock(&h_mutex_reader_wait);pr

30、intw("%s ",thread_name);refresh();sleep(TEST_INFO *)data)->persist_time);pthread_mutex_lock(&h_mutex_read_count);read_count-;if(read_count=0)pthread_mutex_unlock(&CS_DATA);pthread_mutex_unlock(&h_mutex_read_count);return 0;void* WF_writer_thread(void *data)sleep(TEST_INFO *)

31、data)->require_moment);pthread_mutex_lock(&h_mutex_write_count);if(write_count=0)pthread_mutex_lock(&h_mutex_first_reader_wait);write_count+;pthread_mutex_unlock(&h_mutex_write_count);pthread_mutex_lock(&CS_DATA);printw("%s ",(TEST_INFO *)data)->thread_name);refresh()

32、;sleep(TEST_INFO *)data)->persist_time);pthread_mutex_unlock(&CS_DATA);pthread_mutex_lock(&h_mutex_write_count);write_count-;if(write_count=0)pthread_mutex_unlock(&h_mutex_first_reader_wait);pthread_mutex_unlock(&h_mutex_write_count);return 0;void writer_first()int i=0;pthread_t h

33、_threadMAX_THREAD;printw("writer first require sequence:");for(i=0;i<MAX_THREAD;i+)printw("%s ",test_datai.thread_name);printw("n");printw("writer first operation sequence:");refresh();pthread_mutex_init(&CS_DATA,NULL);for(i=0;i<MAX_THREAD;i+)if(test

34、_datai.thread_name0='r')pthread_create(&h_threadi,NULL,WF_reader_thread,&test_datai);elsepthread_create(&h_threadi,NULL,WF_writer_thread,&test_datai);for(i=0;i<MAX_THREAD;i+)pthread_join(h_threadi,NULL); printw("n");refresh();int main(int argc,char *argv)char select;bool end=false;initscr();while(!end)clear();refresh();printw("|-|n");printw("|