pthread計(jì)算劃分和任務(wù)分配

pthread編程:任務(wù)粒度和分配策略(schedulingstrategy)、優(yōu)化處理器間由規(guī)約操作導(dǎo)致的W-W。統(tǒng)自動(dòng)調(diào)度到具體的處理器上執(zhí)行。同時(shí)，pthread為線程之間的進(jìn)度同步提供了一組同步pthread提供的線程管理相關(guān)的主要函數(shù)有：pthread_create-createanewthread -terminatecallingthreadpthread_join-waitforthreadterminationpthread_timedjoin_np-trytojoinwithaterminatedthreadpthread_tryjoin_np-trytojoinwithaterminatedthreadpthread_detach-detachathreadpthread_equal-comparethreadIDspthread_kill-sendasignaltoathreadpthread_kill_other_threads_np-terminateallotherthreadsinprocesspthread_self-getthecallingthreadIDpthread提供了三種線程同步機(jī)制：鎖機(jī)制、信號(hào)量機(jī)制(semaphore)、條件信號(hào)機(jī)制(condition)。其中，鎖機(jī)制可用來實(shí)現(xiàn)線程的互斥操作，pthread提供了互斥鎖(mutex)、自旋鎖(spinlock)(read-writelock)。線程都通過“加鎖”操作進(jìn)入互斥操作區(qū)，通過pthread_mutex_init-destroyandinitializeamutexpthread_mutex_destroy-destroyandinitializeamutexpthread_mutex_lock-lockandunlockamutexpthread_mutex_trylock-lockandunlockamutexpthread_mutex_timedlock-lockamutexpthread_mutex_unlock-lockandunlockapthread_spin_init-destroyorinitializeaspinlockobjectpthread_spin_destroy-destroyorinitializeaspinlockobjectpthread_spin_lock-lockaspinlockobjectpthread_spin_trylock(3p)-lockaspinlockobjectpthread_spin_unlock(3p)-unlockaspinlockobjectpthread_rwlock_init-destroyandinitializearead-writelockobjectpthread_rwlock_destroy-destroyandinitializearead-writelockobjectpthread_rwlock_rdlock-lockaread-writelockobjectforreadingpthread_rwlock_tryrdlock-lockaread-writelockobjectforreadingpthread_rwlock_timedrdlock-lockaread-writelockforreadingpthread_rwlock_wrlock-lockaread-writelockobjectforwritingpthread_rwlock_trywrlock-lockaread-writelockobjectforwritingpthread_rwlock_timedwrlock-lockaread-writelockforwritingsem_init-initializeanunnamedsemaphoresem_destroy-destroyanunnamedsemaphoresem_getvalue-getthevalueofasemaphoresem_wait-lockasemaphoresem_trywait-lockasemaphoresem_timedwait-lockasemaphoresem_post-unlockapthread_cond_init-destroyandinitializeconditionvariablespthread_cond_destroy-destroyandinitializeconditionvariablespthread_cond_wait-waitonaconditionpthread_cond_timedwait-waitonaconditionpthread_cond_signal-signalaconditionpthread_cond_broadcast-broadcastacondition并行的任務(wù)執(zhí)行規(guī)約操作時(shí)，需要進(jìn)入互斥操作區(qū)，以避免寫寫。采用pthread的unsignedintunsignedlonglonglongtypesync_fetch_and_add(type*ptr,typevalue);typesync_fetch_and_sub(type*ptr,typevalue);typesync_fetch_and_or(type*ptr,typevalue);typesync_fetch_and_and(type*ptr,typevalue);typesync_fetch_and_xor(type*ptr,typevalue);typesync_fetch_and_nandtype*ptrtypevalue);typesync_add_and_fetch(type*ptr,typevalue);typesync_sub_and_fetch(type*ptr,typevalue);typesync_or_and_fetch(type*ptr,typevalue);typesync_and_and_fetch(type*ptr,typevalue);typesync_xor_and_fetch(type*ptr,typevalue);typesync_nand_and_fetch(type*ptr,typevalue); pare_and_swap(type*ptr,typeoldval,type pare_and_swap(type*ptr,typeoldval,type==sync_synchronizefull sync_lock_test_and_set(type*ptr,type sync_lock_releasetype*ptr)將*ptr0BSPP(N0)的求解，K0是已發(fā)現(xiàn)的最大素?cái)?shù)，M=min(N+1,??2)，則當(dāng)前超級(jí)計(jì)算步要搜索的自然數(shù)區(qū)間為(N0M)。當(dāng)前超級(jí)計(jì)算步完成P(Ｍ)的求解。根據(jù)BSP系統(tǒng)的結(jié)構(gòu)和訪存機(jī)制，在每顆處理器的P(N0)的解的副本r_P(N0)l_P(Ｍ)應(yīng)緩存在寫緩沖區(qū)。在當(dāng)前超級(jí)計(jì)算步的數(shù)據(jù)交換階段，每個(gè)處理因?qū)⒈镜貙懢彌_區(qū)中的l_P(Ｍ)發(fā)送給其他每顆處理器；將其他各個(gè)處理器發(fā)送來的素?cái)?shù)集合與本地的l_P(Ｍ)、r_P(N0)合并，作為r_P(Ｍ)BSP處理器執(zhí)行0r_P(N0){2，0While(??2<N)00Mmin(N+1,0采用動(dòng)態(tài)計(jì)算劃分模式并行計(jì)算：根據(jù)r_P(N0)計(jì)算l_P(Ｍ)l_P(Ｍ)，將其他各個(gè)處理器發(fā)送來的素?cái)?shù)集合與本地的l_P(Ｍ)一起并入集合r_P(N0);K0max{k:kr_}空間組成，每個(gè)線程負(fù)責(zé)本地局部空間與全局空間的數(shù)據(jù)交換。采用pthread編程實(shí)現(xiàn)問題P(N)的并行算法，需解決三個(gè)問題：P(N0)、l_P(Ｍ)的管理和；M的動(dòng)態(tài)P(N0)、l_P(Ｍ)的管理和。P(N0)和l_P(Ｍ)中元素的數(shù)量都是未知的。為提高程序的算步上，各個(gè)線程開始搜索素?cái)?shù)之前，都需要P(N0)；完成素?cái)?shù)搜后，要分別將自己找到的素?cái)?shù)l_P(Ｍ)并入到P(N0)，這中間有可能需要為P(N0)重新分配更大的空MM中搜索素?cái)?shù)時(shí)，各個(gè)自然數(shù)涉及的整數(shù)除法運(yùn)算量不可預(yù)估，適各自搜索到的素?cái)?shù)集合l_P(Ｍ)并入到P(N0，才能夠更新K0、并開始下一個(gè)超級(jí)計(jì)算步。pthreadmasterthreadK0的計(jì)算，然后與其他線實(shí)現(xiàn)方式1：mtxvecPrime:當(dāng)前找到的全部素?cái)?shù)地程度的任務(wù)后再合并到全局的vecPrime中。實(shí)現(xiàn)方式2：atomic全局的vecPrime中的素?cái)?shù)到各個(gè)線程本地的vecPrime中；而各線程在超級(jí)計(jì)#include<stdio.h>#include<string.h>#include<time.h>//#include<math.h>#include<stdlib.h>#defineNANO #defineBLOCK_SIZE longintn longint*vecPrime,nPrime, longint int cMaster,cWorker; longintlbound,ubound,{longinti,j,k,um=vecPrime[0]=2;vecPrime[1]=j=nPrime=lbound=ubound=vecPrime[nPrime]*vecPrime[nPrime];if(ubound<0||ubound>arg)ubound=arg;for(i=lbound;i<ubound;i+=2)for(k=1;k<j;k++)if(i%vecPrime[k]==0)break;if(k<j)continue;if(nPrime==um{um+=temp=vecPrime=(longint*)malloc(um*sizeof(longint));memcpy(vecPrime,temp,(um-BLOCK_SIZE)*sizeof(longint));}}lbound=ubound+}}intcmpLongInt(constvoid*p1,constvoid*p2)longintval1=*((longint*)p1),val2=*((longintif(val1<val2)return-1;if(val1==val2)return0;return1;}{longinti,j,k,longintloc_lbound,loc_ubound; while(pMyStatus->id!=pthread_self())pMyStatus++;pMyStatus->um=BLOCK_SIZE;j=2; {pthread_cond_wait(&cMaster,&mtx);if(lbound== UM)pMyStatus->nPrime=0;while(true){lbound+=task_size;loc_lbound=lbound;if(loc_lbound>=ubound)break;loc_ubound=loc_lbound+task_size;if(loc_ubound>ubound)loc_ubound=for(i=loc_lbound;i<loc_ubound;i+=2)for(k=1;k<j;k++)if(i%vecPrime[k]==0)break;if(k<j)continue;if(pMyStatus->nPrime==pMyStatus->um{temp=pMyStatus->vecPrime;pMyStatus->um+=BLOCK_SIZE; um*sizeof(longint)); pMyStatus->nPrime*sizeof(longint));}}}}return}longint i,j,k,*temp;um=vecPrime[0]=2;vecPrime[1]=j=nPrime=lbound=totalThread=0;pthread_mutex_init(&mtx,NULL);pthread_cond_init(&cMaster,NULL);for(i=0;i<thread_num;i++)pthread_create(&(threads[i].id),NULL,mtx_worker,ubound=vecPrime[nPrime-1]*vecPrime[nPrime-1];if(ubound<0||ubound>arg)ubound=arg;task_size=(ubound-lbound)/(10*thread_num);if(task_size<10)task_size=(ubound-lbound+thread_num-1)/thread_num;if(task_size%2==1)task_size++;totalThread=0;while(totalThread!=thread_num)pthread_cond_wait(&cWorker,&mtx);for(i=0;i<thread_num;i++){if(threads[i].nPrime==0)if( um{temp= }

vecPrime=(longint*)malloc(um*sizeof(longint));memcpy(vecPrime,temp,nPrime*sizeof(longint)); threads[i].nPrime*sizeof(longint));nPrime+=}qsort(vecPrime,nPrime,sizeof(longint),cmpLongInt);lbound=ubound+2;}lbound= for(i=0;i<thread_num;i++)pthread_join(threads[i].id,NULL);}{longinti,j,k,longintloc_lbound,loc_ubound; while(pMyStatus->id!=pthread_self())pMyStatus++; um=BLOCK_SIZE;j=2; {pthread_cond_wait(&cMaster,&mtx);if(lbound== UM)pMyStatus->nPrime=0;while(true){loc_lbound= if(loc_lbound>=ubound)break;loc_ubound=loc_lbound+if(loc_ubound>ubound)loc_ubound=for(i=loc_lbound;i<loc_ubound;i+=2)for(k=1;k<j;k++)if(i%vecPrime[k]==0)break;if(k<j)continue;if(pMyStatus->nPrime==pMyStatus->um{temp=pMyStatus->vecPrime;pMyStatus->um+=BLOCK_SIZE; malloc(pMyStatus->um*sizeof(longint)); pMyStatus->nPrime*sizeof(longint));}}}}return}longint i,j,k,*temp;um=vecPrime=(longint*)malloc( um*sizeof(longint));vecPrime[0]=2;vecPrime[1]=j=nPrime=lbound=totalThread=0;pthread_mutex_init(&mtx,NULL);pthread_cond_init(&cMaster,NULL);for(i=0;i<thread_num;i++)pthread_create(&(threads[i].id),NULL,atomic_worker,ubound=vecPrime[nPrime-1]*vecPrime[nPrime-1];if(ubound<0||ubound>arg)ubound=arg;task_size=(ubound-lbound)/(10*thread_num);if(task_size<10)task_size=(ubound-lbound+thread_num-1)/thread_num;if(task_size%2==1)task_size++;totalThread=0;while(totalThread!=thread_num)pthread_cond_wait(&cWorker,&mtx);for(i=0;i<thread_num;i++){if(threads[i].nPrime==0)if(nPrime+threads[i].nPrime>um{temp= }

vecPrime=(longint*)malloc(um*sizeof(longint));memcpy(vecPrime,temp,nPrime*sizeof(longint)); threads[i].nPrime*sizeof(longint));nPrime+=}qsort(vecPrime,nPrime,sizeof(longint),cmpLongInt);lbound=ubound+2;}lbound= for(i=0;i<thread_num;i++)pthread_join(threads[i].id,NULL);}longinti,j,k,*temp,loc_vecPrime[BLOCK_SIZE];longintloc_lbound,loc_ubound,loc_nPrime; while(pMyStatus->id!=pthread_self())pMyStatus++;j=2; {pthread_cond_wait(&cMaster,&mtx);if(lbound== UM)loc_nPrime=0;while(true){loc_lbound= if(loc_lbound>=ubound)break;loc_ubound=loc_lbound+if(loc_ubound>ubound)loc_ubound=for(i=loc_lbound;i<loc_ubound;i+=2)while(i>pMyStatus->vecPrime[j-1]*pMyStatus->vecPrime[j-1])j++;for(k=1;k<j;k++)if(i%pMyStatus->vecPrime[k]==0)break;if(k<j)continue;if(loc_nPrime==BLOCK_SIZE)temp=vecPrime;um+=vecPrime=(longint*)malloc( um*sizeof(longint));memcpy(vecPrime,temp,nPrime*sizeof(longint)); nPrime+=BLOCK_SIZE;loc_nPrime=0;}}}if(loc_nPrime>0){if(nPrime+loc_nPrime um{temp=um+=vecPrime=(longint*)malloc( um*sizeof(longint));memcpy(vecPrime,temp,nPrime*sizeof(longint));}memcpy(vecPrime+nPrime,loc_vecPrime,loc_nPrime*sizeof(longint));nPrime+=loc_nPrime;}}return(void*)0;}longint i,j,k,*temp;um=vecPrime=(longint*)malloc(um*sizeof(longint));vecPrime[0]=2;vecPrime[1]=j=nPrime=lbound=totalThread=0;pthread_mutex_init(&mtx,NULL);pthread_cond_init(&cMaster,NULL);if(thread_num>Max_Thread_Num)thread_num=Max_Thread_Num;for(i=0;i<thread_num;i++){threads[i].vecPrime=NULL;pthread_create(&(threads[i].id),NULL,dup_worker,NULL);}ubound=vecPrime[nPrime-1]*vecPrime[nPrime-1];if(ubound<0||ubound>arg)ubound=arg;task_size=(ubound-lbound)/(10*thread_num);if(task_size<10)task_size=(u