MPI分布内存并行程序开发-2

MPI分布内存并行

程序开发1

第四章点对点通信函数

传送机制（两种）：阻塞方式，它必须等到消息从本地送出之后才可以执行后续的语句，保证了缓冲区等资源的可再用性；非阻塞方式，它不须等到消息从本地送出就可以执行后续的语句，但非阻塞调用的返回并不保证资源的可再用性。2阻塞通信正确返回后，其后果是：-该调用要求的通信操作已正确完成-该调用的缓冲区可用消息信封要匹配接收到的消息是最早发送的非阻塞通信主要用于计算和通信的重叠，从而提高整个程序执行的效率。3MPI点对点通信函数的参数格式一般如下所示：MPI消息传递函数参数4请求（request）这个参数用于非阻塞发送和非阻塞接收操作。由于非阻塞操作返回后，数据可能继续存在缓冲中，由此需要一种机制来检测资源是否可用。根据该变量调用其它函数完成消息的实际发送和接收。在C程序中，这个参数是指向MPI_Request结构的指针。5通讯模式（4种）：标准通信模式(MPI_SEND)缓存通信模式(MPI_BSEND)同步通信模式(MPI_SSEND)就绪通信模式(MPI_RSEND)6标准（standard）模式：对数据的缓冲由具体MPI实现决定，与用户程序无关；发送操作的正确返回而不要求接收操作收到发送的数据。

SR17缓冲区（buffered）模式：用户定义，使用和回收缓冲区，不管接收操作是否启动，发送操作都可以执行，但是必须保证缓冲区可用。

SR1缓冲区28同步(synchronous)模式：开始不依赖于接收进程相应的操作是否启动，但必须等到接受开始启动发送才可以返回

SR1239就绪(ready)模式：只有当接收操作已经启动时，才可以在发送进程启动发送操作，否则发送将出错。

SR1210

例3、死锁的发送接收序列

CALLMPI_COMM_RANK(comm,rank,ierr)IF(rank.EQ.0)THENCALLMPI_RECV(recvbuf,count,MPI_REAL,1,tag,comm,status,ierr)CALLMPI_SEND(sendbuf,count,MPI_REAL,1,tag,comm,ierr)ELSEIF(rank.EQ.1)CALLMPI_RECV(recvbuf,count,MPI_REAL,0,tag,comm,status,ierr)CALLMPI_SEND(sendbuf,count,MPI_REAL,0,tag,comm,ierr)ENDIF11进程0进程1从进程1接收消息A向进程1发送消息C从进程0接收消息B向进程0发送消息DABCD12例4、不安全的发送接收序列CALLMPI_COMM_RANK(comm,rank,ierr)IF(rank.EQ.0)THENCALLMPI_SEND(sendbuf,count,MPI_REAL,1,tag,comm,ierr)CALLMPI_RECV(recvbuf,count,MPI_REAL,1,tag,comm,status,ierr)ELSEIF(rank.EQ.1)CALLMPI_SEND(sendbuf,count,MPI_REAL,0,tag,comm,ierr)CALLMPI_RECV(recvbuf,count,MPI_REAL,0,tag,comm,status,ierr)ENDIF13进程0进程1从进程1发送消息A向进程1接收消息C从进程0发送消息B向进程0接收消息DABCD系统缓冲区14程序5、安全的发送接收序列CALLMPI_COMM_RANK(comm,rank,ierr)IF(rank.EQ.0)THENCALLMPI_SEND(sendbuf,count,MPI_REAL,1,tag,comm,ierr)CALLMPI_RECV(recvbuf,count,MPI_REAL,1,tag,comm,status,ierr)ELSEIF(rank.EQ.1)CALLMPI_RECV(recvbuf,count,MPI_REAL,0,tag,comm,status,ierr)CALLMPI_SEND(sendbuf,count,MPI_REAL,0,tag,comm,ierr)ENDIF15进程0进程1从进程1发送消息A向进程1接收消息C从进程0接收消息B向进程0发送消息DABCD16例子6clock=(myrank+1)%groupsize;anticlock=(myrank+groupsize-1)%groupsize;MPI_Send(buf1,LENGTH,MPI_CHAR,clock,tag,MPI_COMM_WORLD);MPI_Recv(buf2,LENGTH,MPI_CHAR,anticlock,tag,MPI_COMM_WORLD,&status);01217改进：MPI_Isend(buf1,LENGTH,MPI_CHAR,clock,tag,MPI_COMM_WORLD,&request);MPI_Recv(buf2,LENGTH,MPI_CHAR,anticlock,tag,MPI_COMM_WORLD,&status);MPI_Wait(&request,&status);－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－MPI_Irecv(buf2,LENGTH,MPI_CHAR,anticlock,tag,MPI_COMM_WORLD,&request);MPI_Send(buf2,LENGTH,MPI_CHAR,clock,tag,MPI_COMM_WORLD);MPI_Wait(&request,&status);18第五章集合通信函数集合通信是包含在通信因子中的所有进程都参加操作。集合通信一般实现三个功能通信：组内数据的传输同步：组内所有进程在特定的地点在执行进度上取得一致计算：对给定的数据完成一定的操作19集合操作的三种类型：同步(barrier)：集合中所有进程都到达后，每个进程再接着运行；数据传递：广播(broadcast)、分散(scatter)、收集(gather)、全部到全部(alltoall)；规约(reduction)：集合中的其中一个进程收集所有进程的数据并计算（如：求最大值、求最小值、加、乘等）；20集合通信函数MPI_BarrierMPI_BcastMPI_ScatterMPI_GatherMPI_ScanMPI_Reduce21MPI_Barrier()在组中建立一个同步栅栏。当每个进程都到达MPI_Barrier调用后，程序才接着往下执行：MPI_Barrier(comm)22程序7、同步示例#include“mpi.h”#include“test.h”#include<stdlib.h>#include<stdio.h>intmain(intargc,char**argv){intrank,size,I;int*table;interrors=0;MPI_Aintaddress;MPI_Datatypetype,newtype;intlens;MPI_Init(&argc,&argv);MPI_Comm_rank(MPI_COMM_WORLD,&rank);MPI_Comm_size(MPI_COMM_WORLD,&size);

23

/*Makedatatable*/table=(int*)calloc(size,sizeof(int));table[rank]=rank+1;/*准备要广播的数据*/MPI_Barrier(MPI_COMM_WORLD);/*将数据广播出去*/

for(i=0;i<size,i++)MPI_Bcast(&table[i],1,MPI_INT,i,MPI_COMM_WORLD);/*检查接收到的数据的正确性*/

for(i=0;i<size,i++)if(table[i]!=i+1)errors++;MPI_Barrier(MPI_COMM_WORLD);/*检查完毕后执行一次同步*/……/*其他的计算*/

MPI_Finalize();}24MPI_Bcast()从指定的一个根进程中把数据广播发送给组中的所有其它进程：MPI_Bcast(*buffer,count,datatype,root,comm)对于root进程：buffer既是接收缓冲又是发送缓冲；对于其他进程：buffer就是接收缓冲。2526程序8、广播程序示例#include<stdio.h>#include“mpi.h”intmain(argc,argv)intargc;Char**argv;{intrank,value;MPI_Init(&argc,&argv);MPI_Comm_rank(MPI_COMM_WORLD,&rank);27do{if(rank==0)/*进程0读入需要广播的数据*/

scanf(“%d”,&value);/*将该数据广播出去*/MPI_Bcast(&value,1,MPI_INT,0,MPI_COMM_WORLD);/*各进程打印收到的数据*/printf(“Process%dgot%d\n”,rank,value);}while(value>=0);MPI_Finalize();return0;}28MPI_Scatter()把根进程中的数据分散发送给组中的所有进程（包括自己）：MPI_Scatter(*sendbuf,sendcnt,sendtype,*recvbuf,recvcnt,recvtype,root,comm)root用MPI_Send(sendbuf,sendcount·n,sendtype,…)发送一个消息。这个消息分成n个相等的段，第i个段发送到进程组的第i个进程，sendcnt必须要和recvcnt相同。2930MPI_Gather()在组中指定一个进程收集组中所有进程发送来的消息，这个函数操作与MPI_Scatter函数操作相反：MPI_Gather(*sendbuf,sendcnt,sendtype,*recvbuf,ecvcount,recvtype,root,comm)3132MPI_Reduce()在组内所有的进程中，执行一个规约操作，并把结果存放在指定的一个进程中：MPI_Reduce(*sendbuf,*recvbuf,count,datatype,op,root,comm)MPI缺省定义了如下的规约操作，用户可根据自己的需要用MPI_Op_create函数创建新的规约操作：33程序9、规约示例 #include“mpi.h”#include<stdio.h>#include<math.h>doublef(doublex);/*定义函数f(x)*/{return(4.0/(1.0+x*x));}intmain(intargc,char*argv[]){intdone=0,n,myid,numprocs,i;doublePI25DT=3.1493238462643;doublemypi,pi,h,sum,x;

doublestartwtime=0.0,endwtime;intnamelen;charprocessor_name[MPI_MAXPROCESSOR_NAME];34MPI_Init(&argc,&argv);MPI_Comm_size(MPI_COMM_WORLD,&numprocs);MPI_Comm_rank(MPI_COMM_WORLD,&myid);MPI_Get_processor_name(processor_name,&namelen);fprint(stdout,”Process%dof%don%s\n”,myid,numprocs,processor_name);n=0;if(myid==0){printf(“PleasegiveN=”);scanf(&n);startwtime=MPI_Wtime();}/*将n值广播出去*/MPI_Bcast(&n,1,MPI_INT,0,MPI_COMM_WORLD);35

h=1.0/(double)n;sum=0.0;for(i=myid+1;i<=n;i+=numprocs)/*每一个进程计算一部分矩形的面积，若进程总数numprocs为4，将0-1区间划分为100个矩形，则各个进程分别计算矩形块0进程1，5，9，13，……，971进程2，6，10，14，……，982进程3，7，11，15，……，993进程4，8，12，16，……，100*/{

x=h*((double)i-0.5);sum+=f(x);}

mypi=h*sum;/*各进程并行计算得到的部分和*/36/*将部分和累加得到所有矩形的面积，该面积和即为近似PI值*/MPI_Reduce(&mypi,&pi,1,MPI_DOUBLE,MPI_SUM,0,MPI_COMM_WORLD);if(myid==0){printf(“piisapproximately%.16f,Erroris%.16f\n”,pi,fabs(pi-PI25DT));endwtime=MPI_Wtime();printf(“wallclocktime=%f\n”,endwtime-startwtime);fflush(stdout);}MPI_Finalize();}37MPI_Scan()用来对分布在进程组上的数据执行前缀归约：MPI_Scan(*sendbuf,*recvbuf,count,datatype,op,comm)3839进程数据缓冲区的变化情况40群集函数的特点：通讯因子中所有进程都要调用除了MPI_Barrier(),其他函数使用类似标准阻塞的通信模式。一个进程一旦结束了它所参与的群集操作就从群集例程中返回，并不保证其他进程执行该群集例程已经完成。一个群集例程是不是同步操作取决于实现。41MPI并行程序的两种基本模式对等模式的MPI程序设计主从模式的MPI程序设计42一.对等模式的MPI程序设计1.问题描述——Jacobi迭代

Jacobi迭代是一种比较常见的迭代方法，其核心部分可以用程序1来表示。简单的说，Jacobi迭代得到的新值是原来旧值点相邻数值点的平均。

Jacobi迭代的局部性很好，可以取得很高的并行性。将参加迭代的数据按块分割后，各块之间除了相邻的元素需要通信外，在各块的内部可以完全独立的并行计算。43程序10串行表示的Jacobi迭代……REALA(N+1,N+1),B(N+1,N+1)……DOK=1,STEPDOJ=1,NDOI=1,NB(I,J)=0.25*(A(I-1,J)+A(I+1,J)+A(I,J+1)+A(I,J-1))ENDDOENDDODOJ=1,NDOI=1,NA(I,J)=B(I,J)ENDDOENDDOENDDO44

2.用MPI程序实现Jacobi迭代

为了并行求解，这里将参加迭代的数据按列进行分割，假设有4个进程同时并行计算，数据的分割结果如图：45假设需要迭代的数据是M*M的二维数组A(M,M),令M=4*N,按上图进行数据划分，则分布在4个不同进程上的数据分别是：进程0：A(M,1:N);

进程1：A(M,N+1:2*N);

进程2：A(M,2*N+1:3*N);

进程3：A(M,3*N+1:4*N).

由于在迭代过程中，边界点新值的计算需要相邻边界其他块的数据，因此在每一个数据块的两侧各增加1列的数据空间，用于存放从相邻数据块通信得到的数据。每个数据块的大小就从M*N扩大到M*(N+2)。计算和通信过程是这样的：首先对数组赋初值，边界赋为8，内部为0。然后开始迭代，迭代之前，每个进程都需要从相邻的进程得到数据块，同时也向相邻的进程提供数据块(FORTRAN数组在内存中是按列优先排列的)。46进程0进程1进程2进程3发送发送发送发送发送发送接收接收接收接收接收接收47程序11、并行的Jacobi迭代programmaininclude‘mpif.h’integertotalsize,mysize,stepsParameter(totalsize=16)(定义全局数组的规模）parameter(mysize=totalsize/4,steps=10)integern,myid,numprocs,i,j,rcReala(totalsize,mysize+2),b(totalsize,mysize+2)Integerbegin_col,end_col,ierrIntegerstatus(MPI_STATUS_SIZE)48callMPI_INIT(ierr)callMPI_COMM_RANK(MPI_COMM_WORLD,myid,ierr)callMPI_COMM_SIZE(MPI_COMM_WORLD,numprocs,ierr)print*,”Process”,myid,”of”,numprocs,”isalive”(数组初始化)doj=1,mysize+2doi=1,totalsizea(i,j)=0.0enddoenddoIf(myid.eq.0)thendoi=1,totalsizea(i,2)=8.0enddoendif49If(myid.eq.3)thendoi=1,totalsizea(i,mysize+1)=8.0enddoendifIf(myid.eq.3)thendoi=1,totalsizea(i,mysize+1)=8.0enddoendifdoi=1,mysize+2a(1,i)=8.0a(totalsize,i)=8.0enddo50(Jacobi迭代部分)don=1,steps(从右侧的邻居得到数据)if(myid.lt.3)thencallMPI_RECV(a(1,mysize+2),totalsize,MPI_REAL,myid+1,10,MPI_COMM_WORLD,status,ierr)endif(向左侧的邻居发送数据)if(myid.gt.0)thencallMPI_SEND(a(1,2),totalsize,MPI_REAL,myid-1,10,MPI_COMM_WORLD,ierr)endif51//向右侧的邻居发送数据if(myid.lt.3)thencallMPI_SEND(a(1,mysize+1),totalsize,MPI_REAL,myid+1,10,MPI_COMM_WORLD,ierr)endif//从左侧的邻居接收数据if(myid.gt.0)thencallMPI_RECV(a(1,1),totalsize,MPI_REAL,myid-1,10,MPI_COMM_WORLD,status,ierr)endifbegin_col=2end_col=mysize+152if(myid.eq.0)thenbegin_col=3endifif(myid.eq.3)thenend_col=mysizeendifdoj=begin_col,end_coldoi=2,totalsize-1b(i,j)=0.25*(a(i,j+1)+a(i,j-1)+a(i+1,j)+a(i-1,j))enddoenddo53doj=begin_col,end_coldoi=2,totalsize-1a(i,j)=b(i,j)enddoenddoenddodoi=2,totalsize-1print*,myid,(a(i,j),j=begin_col,end_col)enddocallMPI_FINALIZE(rc)end54二.主从模式的MPI程序设计1.问题描述——矩阵向量乘

实现矩阵C=AxB。具体实现方法是：主进程将向量B广播给所有的从进程，然后将矩阵A的各行依次发送给从进程，从进程计算一行和B相乘的结果，然后将结果发送给主进程。主进程循环向各个从进程发送一行的数据，直到将A各行的数据发送完毕。一旦主进程将A的各行发送完毕，则每收到一个结果，就向相应的从进程发送结束标志，从进程接收到结束标志后退出执行。主进程收集完所有的结果后也结束。

55发送矩阵A的各行数据回收各行与B相乘的结果计算计算计算计算主进程从进程送回结果56程序12、矩阵向量乘programmaininclude“mpif.h”integerMAX_ROWS,MAX_COLS,rows,colsparameter(MAX_ROWS=1000,MAX_COLS=1000)doubleprecisiona(MAX_ROWS,MAX_COLS),b(MAX_COLS),c(MAX_COLS)doublepresicionbuffer(MAX_COLS),ansintegermyid,master,numprocs,ierr,status(MPI_STATUS_SIZE)integeri,j,numsent,numrcvd,senderintegeranstype,row57callMPI_INIT(ierr)callMPI_COMM_RANK(MPI_COMM_WORLD,myid,ierr)callMPI_COMM_SIZE(MPI_COMM_WORLD,numprocs,ierr)master=0rows=100cols=100If(myid.eq.master)then(主进程对矩阵A和B赋初值)doi=1,colsb(i)=1doj=1,rowsa(I,j)=1enddoenddo58

numsent=0numrcvd=0(将矩阵B发送给所有其他的从进程，通过下面的广播语句实现)callMPI_BCAST(b,cols,MPI_DOUBLE_PRECISION,master,*MPI_COMM_WORLD,ierr)(依次将矩阵A的各行发送给其他的numprocs-1个从进程)doi=1,min(numprocs-1,rows)doj=1,cols(将一行的数据取出来依次放到缓冲区中)

buffer(j)=a(i,j)enddo(将准备好的一行数据发送出去)

callMPI_SEND(buffer,cols,MPI_DOUBLE_PRECISION,i,i,*MPI_COMM_WORLD,ierr)numsent=numsent+1enddo59

(对所有的行，依次接收从进程对一行数据的计算结果)doi=1,rowcallMPI_RECV(ans,1,MPI_DOUBLE_PRECISION,MPI_ANY_SOURCE,*MPI_ANY_TAG,MPI_COMM_WORLD,status,ierr)sender=status(MPI_SOURCE)

anstype=status(MPI_TAG)(将该行数据