實驗3-讀者-寫者問題與進程同步

實驗3讀者/寫者問題與進程同步3.1實驗?zāi)康睦斫馀R界區(qū)和進程互斥的概念，掌握用信號量和PV操作實現(xiàn)進程互斥的方法。3.2實驗要求在linux環(huán)境下編寫一個控制臺應(yīng)用程序，該程序運行時能創(chuàng)立N個線程〔或者進程〕，其中既有讀者線程又有寫者線程，它們按照事先設(shè)計好的測試數(shù)據(jù)進行讀寫操作。請用信號量和PV操作實現(xiàn)讀者/寫者問題。讀者/寫者問題的描述如下：有一個被許多進程共享的數(shù)據(jù)區(qū)，這個數(shù)據(jù)區(qū)可以是一個文件，或者主存的一塊空間〔比方一個數(shù)組或一個變量〕，甚至可以是一組處理器存放器。有一些只讀取這個數(shù)據(jù)區(qū)的進程〔reader〕和一些只往數(shù)據(jù)區(qū)中寫數(shù)據(jù)的進程〔writer〕。以下假設(shè)共享數(shù)據(jù)區(qū)是文件。這些讀者和寫者對數(shù)據(jù)區(qū)的操作必須滿足以下條件：讀—讀允許；讀—寫互斥；寫—寫互斥。這些條件具體來說就是：〔1〕任意多的讀進程可以同時讀這個文件；〔2〕一次只允許一個寫進程往文件中寫；〔3〕如果一個寫進程正在往文件中寫，禁止任何讀進程或?qū)戇M程訪問文件；〔4〕寫進程執(zhí)行寫操作前，應(yīng)讓已有的寫者或讀者全部退出。這說明當(dāng)有讀者在讀文件時不允許寫者寫文件。對于讀者-寫者問題，有三種解決方法：1、讀者優(yōu)先除了上述四個規(guī)那么外，還增加讀者優(yōu)先的規(guī)定，當(dāng)有讀者在讀文件時，對隨后到達的讀者和寫者，要首先滿足讀者，阻塞寫者。這說明只要有一個讀者活潑，那么隨后而來的讀者都將被允許訪問文件，從而導(dǎo)致寫者長時間等待，甚至有可能出現(xiàn)寫者被餓死的情況。2、寫者優(yōu)先除了上述四個規(guī)那么外，還增加寫者優(yōu)先的規(guī)定，即當(dāng)有讀者和寫者同時等待時，首先滿足寫者。當(dāng)一個寫者聲明想寫文件時，不允許新的讀者再訪問文件。3、無優(yōu)先除了上述四個規(guī)那么外，不再規(guī)定讀寫的優(yōu)先權(quán)，誰先等待誰就先使用文件。3.3實驗步驟算法分析1、錯誤的解法圖3-1錯誤的解法semaphorer_w_w=1;reader(){P(r_w_w)；讀文件；V(r_w_w)；}writer(){P(r_w_w)；寫文件；V(r_w_w)；}有同學(xué)認為，可以將文件視為臨界資源，使用臨界資源的代碼就構(gòu)成臨界區(qū)，為了對臨界區(qū)進行管理，只需設(shè)置一個互斥信號量r_w_w，讀或者寫之前執(zhí)行P(r_w_w)，之后執(zhí)行V(r_w_w)即可，從而得到圖3-1所示的算法描述。該方法雖然能滿足讀—寫互斥和寫—寫互斥，但是不滿足讀—讀允許，只要有一個讀者在讀文件，其他的讀者都被阻塞了?？梢姡瑔渭兪褂没コ庑盘柫坎荒芙鉀Q讀者/寫者問題，還需要引入計數(shù)器對讀者進行記數(shù)。2、讀者優(yōu)先如何糾正上述解法中存在的錯誤呢？其實，對于相繼到達的一批讀者，并不是每個讀者都需要執(zhí)行P(r_w_w)和V(r_w_w)。在這批讀者中，只有最先到達的讀者才需要執(zhí)行P(r_w_w)，與寫者競爭對文件的訪問權(quán)，假設(shè)執(zhí)行P(r_w_w)成功那么獲得了文件的訪問權(quán)，其他的讀者可直接訪問文件；同理，只有最后退出臨界區(qū)的讀者需要執(zhí)行V(r_w_w)來歸還文件訪問權(quán)。為了記錄正在讀文件的一批讀者的數(shù)量，需要設(shè)置一個整型變量readercount，每一個讀者到達時都要將readercount加1，退出時都要將readercount減1。由于只要有一個讀者在讀文件，便不允許寫者寫文件，所以，僅當(dāng)readercount=0時，即尚無讀者在讀文件時，讀者才需要執(zhí)行P(r_w_w)操作。假設(shè)P(r_w_w)操作成功，讀者便可去讀文件，相應(yīng)地，readercount+1。同理，僅當(dāng)在執(zhí)行了readercount減1操作后其值為0時，才需要執(zhí)行V(r_w_w)操作，以便讓寫者寫文件。又因為readercount是一個可被多個讀者訪問的臨界資源，所以應(yīng)該為它設(shè)置一個互斥信號量readercount_mutex.。每個讀者在訪問readercount之前執(zhí)行P(readercount_mutex)，之后執(zhí)行V(readercount_mutex)。通過上述分析得到圖3-2所示的算法描述，其中的數(shù)字表示語句對應(yīng)的行號。圖3-2讀者優(yōu)先算法01semaphorer_w_w=1;

02semaphorereadercount_mutex=1;03intreadercount=0;04reader(){05P(readercount_mutex);06if(readercount==0)P(r_w_w);07readercount++;08V(readercount_mutex);09讀文件;10P(readercount_mutex);11readercount--;12if(readercount==0)V(r_w_w);13V(readercount_mutex);14}1516writer(){17P(r_w_w);18寫文件;19V(r_w_w);20}3、寫者優(yōu)先通過增加信號量并修改上述程序可以得到寫者優(yōu)先算法。為了實現(xiàn)寫者優(yōu)先算法，需要將寫者和讀者分開排隊，并且第一個讀者和其它讀者也要分開排隊。這樣就需要三個隊列，一個是寫者排隊的地方，另一個是第一個讀者排隊的地方，第三個是其它讀者排隊的地方。相應(yīng)地需要設(shè)置三個信號量，r_w_w、first_reader_wait和reader_wait。當(dāng)一個寫者聲明想寫文件時，可以讓新的讀者中的第一個到first_reader_wait上排隊等待；當(dāng)有讀者阻塞在first_reader_wait上時，讓其它讀者阻塞在reader_wait上；當(dāng)有一個寫者在寫文件時，其它寫者到r_w_w上排隊。只要有活潑的寫者或者寫者隊列不為空，那么阻塞新到達的讀者。為了記錄已經(jīng)發(fā)出聲明的寫者數(shù)量，需要設(shè)置一個整數(shù)writercount，以表示聲明要寫文件的寫者數(shù)目。由于只要有一個寫者到達，就不允許讀者去讀，因此僅當(dāng)writercount=0，表示無寫者聲明寫時，寫者才需要執(zhí)行P(first_reader_wait)操作，假設(shè)操作成功，寫者便可以執(zhí)行P(r_w_w)去競爭寫文件權(quán)利。其它寫者不需要再向讀者聲明，可以直接執(zhí)行P(r_w_w)去競爭寫文件權(quán)利。同理僅當(dāng)寫者在執(zhí)行writercount減1操作后其值為0時，才需要執(zhí)行V(first_reader_wait)操作，以便喚醒第一個被阻塞的讀者去讀文件。又因為writercount是一個可被多個寫者訪問的臨界資源，所以，應(yīng)該為它設(shè)置一個互斥信號量writer_mutex。4、無優(yōu)先除了在讀者優(yōu)先時需要的信號量r_w_w和readercount_mutex之外，還需要設(shè)置一個信號量wait供讀者和寫者排隊。讀者和寫者都排在wait隊列上。假設(shè)有讀者在讀文件，那么第一個寫者阻塞在r_w_w上，其它的寫者和讀者阻塞在wait上；假設(shè)有一個寫者在寫文件，那么其它寫者和讀者都阻塞在wait上。無優(yōu)先的算法描述如圖3-3所示。圖3-3無優(yōu)先算法01semaphorer_w_w=1;02semaphorewait=1;

03semaphorereadercount_mutex=1;04intreadercount=0;05reader(){06P(wait);07P(readercount_mutex);08if(readercount==0)P(r_w_w);09readercount++;10V(readercount_mutex);11V(wait);12讀文件;13P(readercount_mutex);14readercount--;15if(readercount==0)V(r_w_w);16V(readercount_mutex);17}18writer(){19P(wait);20P(r_w_w);21寫文件;22V(r_w_w);23V(wait);24}3.3.2該程序采用簡單的控制臺應(yīng)用程序界面，在主界面上顯示程序的功能。該程序的功能如下：演示讀者優(yōu)先算法；演示寫者優(yōu)先算法；演示無優(yōu)先算法；退出。3.3.3實現(xiàn)讀者/寫者問題的源程序名稱是reader_and_writer.cpp。該程序共包括10個函數(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個線程并等待它們結(jié)束三WF_reader_thread()WF_writer_thread()writer_first()寫者優(yōu)先算法的讀者線程函數(shù)寫者優(yōu)先算法的寫者線程函數(shù)寫者優(yōu)先算法的初始化函數(shù)：創(chuàng)立10個線程并等待它們結(jié)束四FIFO_reader_thread()FIFO_writer_thread()first_come_first_serverd()無優(yōu)先算法的讀者線程函數(shù)無者優(yōu)先算法的寫者線程函數(shù)無者優(yōu)先算法的初始化函數(shù)：創(chuàng)立10個線程并等待它們結(jié)束圖3-4函數(shù)功能簡述程序開始局部定義了宏MAX_THREAD，表示程序中創(chuàng)立的線程數(shù)。還定義了測試數(shù)據(jù)的結(jié)構(gòu)體TEST_INFO，該結(jié)構(gòu)體包含三個數(shù)據(jù)項：線程名稱；提出請求的時刻；操作持續(xù)時間。接著定義了全局變量，這些全局變量的作用如下：數(shù)組test_data保存了10個線程的測試數(shù)據(jù)；整數(shù)read_count記錄一段時間內(nèi)同時對文件進行讀操作的線程數(shù)；整數(shù)write_count記錄一段時間內(nèi)提出寫操作請求的線程數(shù)，該整數(shù)只在寫者優(yōu)先算法中使用；CS_DATA是臨界區(qū)變量，用來保護文件，實現(xiàn)對文件的讀—寫互斥和寫—寫互斥〔相當(dāng)于算法描述中的r_w_w〕；互斥體h_mutex_read_count用來保護整數(shù)read_count，以保證多個讀者對read_count的互斥訪問；互斥體h_mutex_write_count用來保護整數(shù)write_count，以保證多個寫者對write_count的互斥訪問，該互斥體只在寫者優(yōu)先算法中使用；互斥體h_mutex_first_reader_wait和h_mutex_reader_wait只在寫者優(yōu)先算法中使用，當(dāng)有寫者在寫文件時，提出讀請求的第一個讀者阻塞在h_mutex_first_reader_wait上，其余的讀者阻塞在h_mutex_reader_wait上；互斥體h_mutex_wait只在無優(yōu)先算法中使用，當(dāng)文件被使用時，后繼的讀請求和寫請求依次阻塞在h_mutex_wait上。參考源程序3.3.4.1eq\o\ac(○,1)編譯命令gccreader_and_writer.cpp–oreader_and_writer.o–lcurses–lpthreadeq\o\ac(○,2)程序清單#include<unistd.h>#include<pthread.h>#include<curses.h>#include<stdlib.h>#include<string.h>#defineMAX_THREAD10typedefstruct{ charthread_name[3]; unsignedintrequire_moment; unsignedintpersist_time;}TEST_INFO;TEST_INFOtest_data[MAX_THREAD]={ {"r1",0,15}, {"r2",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}};intread_count=0;intwrite_count=0;pthread_mutex_tCS_DATA;pthread_mutex_th_mutex_read_count;pthread_mutex_th_mutex_write_count;pthread_mutex_th_mutex_reader_wait;pthread_mutex_th_mutex_first_reader_wait;pthread_mutex_th_mutex_wait;void*RF_reader_thread(void*data){ charthread_name[3]; strcpy(thread_name,((TEST_INFO*)data)->thread_name); sleep(((TEST_INFO*)data)->require_moment); 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); 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); pthread_mutex_unlock(&h_mutex_read_count); return0;}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(&CS_DATA); return0;}voidreader_first(){ inti=0; pthread_th_thread[MAX_THREAD]; printw("readerfirstrequiresequence:"); for(i=0;i<MAX_THREAD;i++){ printw("%s",test_data[i].thread_name); }; printw("\n"); printw("readerfirstoperationsequence:"); refresh(); pthread_mutex_init(&CS_DATA,NULL); for(i=0;i<MAX_THREAD;i++){ if(test_data[i].thread_name[0]=='r') pthread_create(&h_thread[i],NULL,RF_reader_thread,&test_data[i]); else pthread_create(&h_thread[i],NULL,RF_writer_thread,&test_data[i]); } for(i=0;i<MAX_THREAD;i++){ pthread_join(h_thread[i],NULL); }printw("\n"); refresh();}void*FIFO_reader_thread(void*data){ charthread_name[3]; 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_lock(&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); pthread_mutex_unlock(&h_mutex_read_count); return0;}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)->persist_time); pthread_mutex_unlock(&CS_DATA); pthread_mutex_unlock(&h_mutex_wait); return0;}voidfirst_come_first_served(){ inti=0; pthread_th_thread[MAX_THREAD]; printw("FCFSrequiresequence:"); for(i=0;i<MAX_THREAD;i++){ printw("%s",test_data[i].thread_name); }; printw("\n"); printw("FCFS:operationsequence:"); refresh(); pthread_mutex_init(&CS_DATA,NULL); for(i=0;i<MAX_THREAD;i++){ if(test_data[i].thread_name[0]=='r') pthread_create(&h_thread[i],NULL,FIFO_reader_thread,&test_data[i]); else pthread_create(&h_thread[i],NULL,FIFO_writer_thread,&test_data[i]); } for(i=0;i<MAX_THREAD;i++){ pthread_join(h_thread[i],NULL); }printw("\n"); refresh();}void*WF_reader_thread(void*data){ charthread_name[3]; strcpy(thread_name,((TEST_INFO*)data)->thread_name); sleep(((TEST_INFO*)data)->require_moment); pthread_mutex_lock(&h_mutex_reader_wait); pthread_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); 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); pthread_mutex_unlock(&h_mutex_read_count); return0;}void*WF_writer_thread(void*data){ sleep(((TEST_INFO*)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(); 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); return0;}voidwriter_first(){ inti=0; pthread_th_thread[MAX_THREAD]; printw("writerfirstrequiresequence:"); for(i=0;i<MAX_THREAD;i++){ printw("%s",test_data[i].thread_name); } printw("\n"); printw("writerfirstoperationsequence:"); refresh(); pthread_mutex_init(&CS_DATA,NULL); for(i=0;i<MAX_THREAD;i++){ if(test_data[i].thread_name[0]=='r') pthread_create(&h_thread[i],NULL,WF_reader_thread,&test_data[i]); else pthread_create(&h_thread[i],NULL,WF_writer_thread,&test_data[i]); } for(i=0;i<MAX_THREAD;i++){ pthread_join(h_thread[i],NULL); }printw("\n"); refresh();}intmain(intargc,char*argv[]){ charselect; boolend=false; initscr(); while(!end){ clear(); refresh(); printw("|-----------------------------------|\n"); printw("|1:readerfirst