并行程序设计 中文课件 05 Linux多线程程序设计

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.当多个线程尝试操作相同的数据项时，如果不采取适当的注意来同步它们，则结果通常可能是不相干的。Consider:/*eachthreadtriestoupdatevariablebest_costasfollows*/if(my_cost<best_cost)best_cost=my_cost;Assumethattherearetwothreads,theinitialvalueofbest_costis100,andthevaluesofmy_costare50and75atthreadst1andt2.Dependingonthescheduleofthethreads,thevalueofbest_costcouldbe50or75!Thevalue75doesnotcorrespondtoanyserializationofthethreads.17Mutex---protectingyourcriticalsectionsAmutexwillbeownedbyonlyonethreadatonetime互斥在同一时间将仅由一个线程所拥有Ifanotherthreadcallpthread_mutex_lock()onaunlockedmutex,anotherthreadwhichcallspthread_mutex_locktooissuspendedandwaitsfortheowneroflockedmutexthreadtounlockthemutexfirst.如果另一个线程在解锁的互斥体上调用pthread_mutex_lock（），那么调用pthread_mutex_lock的另一个线程也被挂起，并等待锁定的互斥线程的所有者首先解锁互斥锁。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实现中，没有任何资源与互斥体对象相关联，因此pthread_mutex_destroy除了检查互斥锁是否被解锁外，实际上什么都不做。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.条件变量允许线程阻塞自身，直到指定的数据达到预定义的状态。Aconditionvariableisassociatedwiththispredicate.Whenthepredicatebecomestrue,theconditionvariableisusedtosignaloneormorethreadswaitingonthecondition.条件变量与此谓词相关联。当谓词变为true时，条件变量用于发送一个或多个等待条件的线程。Asingleconditionvariablemaybeassociatedwithmorethanonepredicate.单个条件变量可能与多个谓词相关联。Aconditionvariablealwayshasamutexassociatedwithit.Athreadlocksthismutexandteststhepredicatedefinedonthesharedvariable.条件变量始终具有与其相关联的互斥体。线程锁定该互斥体并测试共享变量上定义的谓词。Ifthepredicateisnottrue,thethreadwaitsontheconditionvariableassociatedwiththepredicateusingthefunctionpthread_cond_wait.如果谓词为假，线程将使用函数pthread_cond_wait等待与谓词关联的条件变量。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.条件是允许线程暂停执行并放弃处理器的同步设备，直到满足共享数据的某些谓词为止。Thebasicoperationsonconditionsare:条件的基本操作是：signalthecondition(whenthepredicatebecomestrue),andwaitforthecondition,信号的条件（当谓词为真），并且等待状态，suspendingthethreadexecutionuntilanotherthreadsignalsthecondition.暂停线程执行，直到另一个线程发出信号。Aconditionvariablemustalwaysbeassociatedwithamutex,toavoidtheraceconditionwhereathreadpreparestowaitonaconditionvariableandanotherthreadsignalstheconditionjustbeforethefirstthreadactuallywaitsonit.条件变量必须始终与互斥量相关联，以避免条件变量的竞争，另一个线程在第一个线程实际等待之前就会通知该条件。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.通过设计，Pthreads可以为基本操作提供支持。Higherlevelconstructscanbebuiltusingbasicsynchronizationconstructs.可以使用基本的同步构造构建更高级别的构造。Wediscusstwosuchconstructs-read-writelocksandbarriers.我们讨论两个这样的构造-读写锁和障碍。48Read-WriteLocks

Inmanyapplications,adatastructureisreadfrequentlybutwritteninfrequently.Forsuchapplications,weshoulduseread-writelocks.在许多应用中，经常读取数据结构，但不经常写入。对于这样的应用程序，我们应该使用读写锁。Areadlockisgrantedwhenthereareotherthreadsthatmayalreadyhavereadlocks.当有其他线程可能已经具有读取锁定时，将会授予读取锁定。Ifthereisawritelockonthedata(oriftherearequeuedwritelocks),thethreadperformsaconditionwait.如果对数据有写锁定（或者如果有排队的写锁），则线程执行条件等待。Iftherearemultiplethreadsrequestingawritelock,theymustperformaconditionwait.如果有多个线程请求写锁定，则它们必须执行条件等待。Withthisdescription,wecandesignfunctionsforreadlocksmylib_rwlock_rlock,writelocksmylib_rwlock_wlock,andunlockingmylib_rwlock_unlock.有了这个描述，我们可以设计读锁的功能mylib_rwlock_rlock，写锁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->