并行程序設(shè)計(jì) 中文課件 05 Linux多線程程序設(shè)計(jì)

ParallelProgrammingInstructor:ZhangWeizhe(張偉哲)ComputerNetworkandInformationSecurityTechniqueResearchCenter,SchoolofComputerScienceandTechnology,HarbinInstituteofTechnologyProgrammingSharedAddressSpacePlatforms3ProgrammingwithpthreadsOutline4WhatarePthreads?5CreatingThreads6WaitingFortheTerminationofThreads7DestroyingThreads8ThreadBasics:CreationandTermination(Example)#include<pthread.h>#include<stdlib.h>#defineMAX_THREADS512void*compute_pi(void*);....main(){...pthread_tp_threads[MAX_THREADS];pthread_attr_tattr;pthread_attr_init(&attr);for(i=0;i<num_threads;i++){hits[i]=i;pthread_create(&p_threads[i],&attr,compute_pi,(void*)&hits[i]);}for(i=0;i<num_threads;i++){pthread_join(p_threads[i],NULL);total_hits+=hits[i];}...}9ThreadBasics:CreationandTermination(Example)void*compute_pi(void*s){intseed,i,*hit_pointer;doublerand_no_x,rand_no_y;intlocal_hits;hit_pointer=(int*)s;seed=*hit_pointer;local_hits=0;for(i=0;i<sample_points_per_thread;i++){rand_no_x=(double)(rand_r(&seed))/(double)((2<<14)-1);rand_no_y=(double)(rand_r(&seed))/(double)((2<<14)-1);if(((rand_no_x-0.5)*(rand_no_x-0.5)+(rand_no_y-0.5)*(rand_no_y-0.5))<0.25)local_hits++;seed*=i;}*hit_pointer=local_hits;pthread_exit(0);}10Phread_detach11pthread_self#include<pthread.h>pthread_tpthread_self(void);returnidentifierofcurrentthread12Example:phreadechocli#include "unpthread.h"void *copyto(void*);staticint sockfd; /*globalforboththreadstoaccess*/staticFILE *fp;voidstr_cli(FILE*fp_arg,intsockfd_arg){ char recvline[MAXLINE]; pthread_t tid; sockfd=sockfd_arg; /*copyargumentstoexternals*/ fp=fp_arg; Pthread_create(&tid,NULL,copyto,NULL); while(Readline(sockfd,recvline,MAXLINE)>0) Fputs(recvline,stdout);}13Example:phreadechoclivoid*copyto(void*arg){ char sendline[MAXLINE]; while(Fgets(sendline,MAXLINE,fp)!=NULL) Writen(sockfd,sendline,strlen(sendline)); Shutdown(sockfd,SHUT_WR); /*EOFonstdin,sendFIN*/ return(NULL); /*return(i.e.,threadterminates)whenend-of-fileonstdin*/}14Example:phreadechoser#include "unpthread.h"staticvoid *doit(void*); /*eachthreadexecutesthisfunction*/intmain(intargc,char**argv){ int listenfd,connfd; socklen_t addrlen,len; structsockaddr *cliaddr; if(argc==2) listenfd=Tcp_listen(NULL,argv[1],&addrlen); elseif(argc==3) listenfd=Tcp_listen(argv[1],argv[2],&addrlen); else err_quit("usage:tcpserv01[<host>]<serviceorport>"); cliaddr=Malloc(addrlen); for(;;){ len=addrlen; connfd=Accept(listenfd,cliaddr,&len); Pthread_create(NULL,NULL,&doit,(void*)connfd); }}15Example:phreadechoserstaticvoid*doit(void*arg){ Pthread_detach(pthread_self()); str_echo((int)arg); /*samefunctionasbefore*/ Close((int)arg); /*wearedonewithconnectedsocket*/ return(NULL);}16SynchronizationPrimitivesinPthreadsWhenmultiplethreadsattempttomanipulatethesamedataitem,theresultscanoftenbeincoherentifpropercareisnottakentosynchronizethem.當(dāng)多個(gè)線程嘗試操作相同的數(shù)據(jù)項(xiàng)時(shí)，如果不采取適當(dāng)?shù)淖⒁鈦?lái)同步它們，則結(jié)果通常可能是不相干的。Consider:/*eachthreadtriestoupdatevariablebest_costasfollows*/if(my_cost<best_cost)best_cost=my_cost;Assumethattherearetwothreads,theinitialvalueofbest_costis100,andthevaluesofmy_costare50and75atthreadst1andt2.Dependingonthescheduleofthethreads,thevalueofbest_costcouldbe50or75!Thevalue75doesnotcorrespondtoanyserializationofthethreads.17Mutex---protectingyourcriticalsectionsAmutexwillbeownedbyonlyonethreadatonetime互斥在同一時(shí)間將僅由一個(gè)線程所擁有Ifanotherthreadcallpthread_mutex_lock()onaunlockedmutex,anotherthreadwhichcallspthread_mutex_locktooissuspendedandwaitsfortheowneroflockedmutexthreadtounlockthemutexfirst.如果另一個(gè)線程在解鎖的互斥體上調(diào)用pthread_mutex_lock（），那么調(diào)用pthread_mutex_lock的另一個(gè)線程也被掛起，并等待鎖定的互斥線程的所有者首先解鎖互斥鎖。18AvoidingDeadlocks19AvoidingDeadlocksCont20OperationsonMutexes#include<pthread.h>

pthread_mutex_tfastmutex=PTHREAD_MUTEX_INITIALIZER;

intpthread_mutex_init(pthread_mutex_t*mutex,constpthread_mutexattr_t*mutexattr);

intpthread_mutex_lock(pthread_mutex_t*mutex); intpthread_mutex_trylock(pthread_mutex_t*mutex); intpthread_mutex_unlock(pthread_mutex_t*mutex); intpthread_mutex_destroy(pthread_mutex_t*mutex);IntheLinuxThreadsimplementation,noresourcesareassociatedwithmutexobjects,thuspthread_mutex_destroyactuallydoesnothingexceptcheckingthatthemutexisunlocked.在LinuxThreads實(shí)現(xiàn)中，沒(méi)有任何資源與互斥體對(duì)象相關(guān)聯(lián)，因此pthread_mutex_destroy除了檢查互斥鎖是否被解鎖外，實(shí)際上什么都不做。21MutualExclusionWecannowwriteourpreviouslyincorrectcodesegmentas:pthread_mutex_tminimum_value_lock;...main(){....pthread_mutex_init(&minimum_value_lock,NULL);....}void*find_min(void*list_ptr){....pthread_mutex_lock(&minimum_value_lock);if(my_min<minimum_value)minimum_value=my_min;/*andunlockthemutex*/pthread_mutex_unlock(&minimum_value_lock);}

22Itistimetowriteasimpleprogram23Cont24ConditionVariablesforSynchronizationAconditionvariableallowsathreadtoblockitselfuntilspecifieddatareachesapredefinedstate.條件變量允許線程阻塞自身，直到指定的數(shù)據(jù)達(dá)到預(yù)定義的狀態(tài)。Aconditionvariableisassociatedwiththispredicate.Whenthepredicatebecomestrue,theconditionvariableisusedtosignaloneormorethreadswaitingonthecondition.條件變量與此謂詞相關(guān)聯(lián)。當(dāng)謂詞變?yōu)閠rue時(shí)，條件變量用于發(fā)送一個(gè)或多個(gè)等待條件的線程。Asingleconditionvariablemaybeassociatedwithmorethanonepredicate.單個(gè)條件變量可能與多個(gè)謂詞相關(guān)聯(lián)。Aconditionvariablealwayshasamutexassociatedwithit.Athreadlocksthismutexandteststhepredicatedefinedonthesharedvariable.條件變量始終具有與其相關(guān)聯(lián)的互斥體。線程鎖定該互斥體并測(cè)試共享變量上定義的謂詞。Ifthepredicateisnottrue,thethreadwaitsontheconditionvariableassociatedwiththepredicateusingthefunctionpthread_cond_wait.如果謂詞為假，線程將使用函數(shù)pthread_cond_wait等待與謂詞關(guān)聯(lián)的條件變量。25ConditionVariablesforSynchronization

Pthreadsprovidesthefollowingfunctionsforconditionvariables:intpthread_cond_wait(pthread_cond_t*cond,pthread_mutex_t*mutex);intpthread_cond_signal(pthread_cond_t*cond);intpthread_cond_broadcast(pthread_cond_t*cond);intpthread_cond_init(pthread_cond_t*cond,constpthread_condattr_t*attr);intpthread_cond_destroy(pthread_cond_t*cond);26pthread_cond_tAconditionisasynchronizationdevicethatallowsthreadstosuspendexecutionandrelinquishtheprocessorsuntilsomepredicateonshareddataissatisfied.條件是允許線程暫停執(zhí)行并放棄處理器的同步設(shè)備，直到滿足共享數(shù)據(jù)的某些謂詞為止。Thebasicoperationsonconditionsare:條件的基本操作是：signalthecondition(whenthepredicatebecomestrue),andwaitforthecondition,信號(hào)的條件（當(dāng)謂詞為真），并且等待狀態(tài)，suspendingthethreadexecutionuntilanotherthreadsignalsthecondition.暫停線程執(zhí)行，直到另一個(gè)線程發(fā)出信號(hào)。Aconditionvariablemustalwaysbeassociatedwithamutex,toavoidtheraceconditionwhereathreadpreparestowaitonaconditionvariableandanotherthreadsignalstheconditionjustbeforethefirstthreadactuallywaitsonit.條件變量必須始終與互斥量相關(guān)聯(lián)，以避免條件變量的競(jìng)爭(zhēng)，另一個(gè)線程在第一個(gè)線程實(shí)際等待之前就會(huì)通知該條件。27pthread_cond_init28Example29pthread_cond_signal30Example31pthread_cond_broadcast32Example33pthread_cond_wait34Cont35Cont.36Example37pthread_cond_timedwait38pthread_cond_destroy39AnExamplefromLinuxManual40WorkshopModifytheprogramcounter.cthenletathreadwaitasignalwhenthesumisupto10,andprintfthemessage“Getasignalthatthesumhasbeenupto10!”.41Solution#include<stdio.h>#include<pthread.h>#defineTHREAD_NUMBER10pthread_mutex_tmutex=PTHREAD_MUTEX_INITIALIZER;pthread_cond_tcond=PTHREAD_COND_INITIALIZER;intsum=0;void*th_counter(void*argc){inti;i=*(int*)argc;sleep(1);

pthread_mutex_lock(&mutex);sum=sum+i;if(sum>10)pthread_cond_signal(&cond);pthread_mutex_unlock(&mutex);printf("count%disover\n",i);return;}42void*waitsum(void*argc){

pthread_mutex_lock(&mutex);while(sum<=10)pthread_cond_wait(&cond,&mutex);

printf("Getasignalthatthesumhasbeenupto10!\n");pthread_mutex_unlock(&mutex);}intmain(void){pthread_tpt[THREAD_NUMBER];inti;intarg[THREAD_NUMBER];pthread_create(&pt[THREAD_NUMBER-1],NULL,waitsum,NULL);43for(i=0;i<THREAD_NUMBER-1;i++){arg[i]=i;pthread_create(&pt[i],NULL,th_counter,(void*)&arg[i]);}for(i=0;i<THREAD_NUMBER;i++)pthread_detach(pt[i]);//pthread_join(pt[i],NULL);printf("Themainthreadiswaitingforallthethreadsfinishing...\n");sleep(5);printf("sumis%d\n",sum);pthread_mutex_destroy(&mutex);pthread_cond_destroy(&cond);return0;}44Example:Producer-ConsumerUsingConditionVariablespthread_cond_tcond_queue_empty,cond_queue_full;pthread_mutex_ttask_queue_cond_lock;inttask_available;/*otherdatastructureshere*/main(){/*declarationsandinitializations*/task_available=0;pthread_init();pthread_cond_init(&cond_queue_empty,NULL);pthread_cond_init(&cond_queue_full,NULL);pthread_mutex_init(&task_queue_cond_lock,NULL);/*createandjoinproducerandconsumerthreads*/}45Example:Producer-ConsumerUsingConditionVariables

void*producer(void*producer_thread_data){intinserted;while(!done()){create_task();pthread_mutex_lock(&task_queue_cond_lock);while(task_available==1)pthread_cond_wait(&cond_queue_empty,&task_queue_cond_lock);insert_into_queue();task_available=1;pthread_cond_signal(&cond_queue_full);pthread_mutex_unlock(&task_queue_cond_lock);}}46Example:Producer-ConsumerUsingConditionVariables

void*consumer(void*consumer_thread_data){while(!done()){pthread_mutex_lock(&task_queue_cond_lock);while(task_available==0)pthread_cond_wait(&cond_queue_full,&task_queue_cond_lock);my_task=extract_from_queue();task_available=0;pthread_cond_signal(&cond_queue_empty);pthread_mutex_unlock(&task_queue_cond_lock);process_task(my_task);}}47CompositeSynchronizationConstructsBydesign,Pthreadsprovidesupportforabasicsetofoperations.通過(guò)設(shè)計(jì)，Pthreads可以為基本操作提供支持。Higherlevelconstructscanbebuiltusingbasicsynchronizationconstructs.可以使用基本的同步構(gòu)造構(gòu)建更高級(jí)別的構(gòu)造。Wediscusstwosuchconstructs-read-writelocksandbarriers.我們討論兩個(gè)這樣的構(gòu)造-讀寫(xiě)鎖和障礙。48Read-WriteLocks

Inmanyapplications,adatastructureisreadfrequentlybutwritteninfrequently.Forsuchapplications,weshoulduseread-writelocks.在許多應(yīng)用中，經(jīng)常讀取數(shù)據(jù)結(jié)構(gòu)，但不經(jīng)常寫(xiě)入。對(duì)于這樣的應(yīng)用程序，我們應(yīng)該使用讀寫(xiě)鎖。Areadlockisgrantedwhenthereareotherthreadsthatmayalreadyhavereadlocks.當(dāng)有其他線程可能已經(jīng)具有讀取鎖定時(shí)，將會(huì)授予讀取鎖定。Ifthereisawritelockonthedata(oriftherearequeuedwritelocks),thethreadperformsaconditionwait.如果對(duì)數(shù)據(jù)有寫(xiě)鎖定（或者如果有排隊(duì)的寫(xiě)鎖），則線程執(zhí)行條件等待。Iftherearemultiplethreadsrequestingawritelock,theymustperformaconditionwait.如果有多個(gè)線程請(qǐng)求寫(xiě)鎖定，則它們必須執(zhí)行條件等待。Withthisdescription,wecandesignfunctionsforreadlocksmylib_rwlock_rlock,writelocksmylib_rwlock_wlock,andunlockingmylib_rwlock_unlock.有了這個(gè)描述，我們可以設(shè)計(jì)讀鎖的功能mylib_rwlock_rlock，寫(xiě)鎖mylib_rwlock_wlock，并解鎖mylib_rwlock_unlock。49Read-WriteLocks

Thelockdatatypemylib_rwlock_tholdsthefollowing:acountofthenumberofreaders,thewriter(a0/1integerspecifyingwhetherawriterispresent),aconditionvariablereaders_proceedthatissignaledwhenreaderscanproceed,aconditionvariablewriter_proceedthatissignaledwhenoneofthewriterscanproceed,acountpending_writersofpendingwriters,andamutexread_write_lockassociatedwiththeshareddatastructure50Read-WriteLocks

typedefstruct{intreaders;intwriter;pthread_cond_treaders_proceed;pthread_cond_twriter_proceed;intpending_writers;pthread_mutex_tread_write_lock;}mylib_rwlock_t;voidmylib_rwlock_init(mylib_rwlock_t*l){l->readers=l->writer=l->pending_writers=0;pthread_mutex_init(&(l->read_write_lock),NULL);pthread_cond_init(&(l->readers_proceed),NULL);pthread_cond_init(&(l->writer_proceed),NULL);}51Read-WriteLocks

voidmylib_rwlock_rlock(mylib_rwlock_t*l){/*ifthereisawritelockorpendingwriters,performconditionwait..elseincrementcountofreadersandgrantreadlock*/pthread_mutex_lock(&(l->