【MPI】使用非阻塞Send/Recv实现并行折叠整数和

1. 实验要求

本次实验需采用非阻塞式编程实现Master和Worker之间的通信。要求如下：

⑴．Master进程被动等待Worker进程发来数据量请求和上次计算结果。

⑵. Master进程将结果叠加至Result中，并根据数据请求量进行数据分段：

a.当有数据可分时且数据量满足Worker请求时，将请求量大小的数据段分给该请求进程Worker；

b. 当有数据可分但数据量不足够大不满足Worker请求,Master向该Worker发送退出信号；

c.当无数据可分，向该Worker发送退出信号；

⑶.Worker主动向Master请求数据，随机生成一个满足要求的随机数(例如大小在2~20之间)作为下次向Master请求的数据量，并将其与上次计算结果发送给Master；接受到Master发送来的数据进行叠加和或者退出。

2. 算法设计及代码实现

⑴.a.程序流程图

 

b.实验代码框架图 

① 随机生成数RANNOM，生成满足区间(a，b)的整数值；

② buf_read()函数接受动态从键盘输入数据；

③ master()函数负责主进程Master的与Worker进程的通信，接受Worker的数据请求、任务分配和答案总和；

④ worker()函数负责进程Worker与主进程Master的通信，子任务的完成；

⑤ sum()函数为一次叠加和，将buf中长度为len的数据段进行叠加和。

⑵．实验主要代码

从键盘接受输入数据函数buf_read()和一次叠加和代码函数sum()与实验一代码一样，此处不再列出

①.当进程只有一个时，为串行计算，代码如下：

 

1. if(numprocs==1) 
2. { 
3.        result = sum(buffer,totalnum); 
4.        etime = MPI_Wtime();     
5.        printf("The result is  %d.\n",result); 
6.        printf("The sequential collapse takes %f seconds.\n",etime-stime); 
7. } 

②.当进程数大于2时，主进程Master被动接受Worker的发送请求量和上次数据的叠加和等，处理过程如下：

 

1. MPI_Request* reqs = (MPI_Request*)calloc(numprocs - 1, sizeof(MPI_Request)); 
2. MPI_Status* stats = (MPI_Status*)calloc(numprocs - 1, sizeof(MPI_Status)); 
3. int* indices = (int*)calloc((numprocs - 1), sizeof(int)); 
4. int** lsum = (int **)calloc(numprocs - 1, sizeof(int *));   
5. int sendlen =0,hassend=0; 
6. for(i = 0; i < numprocs - 1; i++) 
7. {//初始化 
8.     lsum[i] = (int *)calloc(2, sizeof(int));  
9. }  
10. count = numprocs - 1; 
11. allocatedJobs = numprocs - 1; 
12. for(i = 0; i < count; i++) 
13. {//初始化 
14.     indices[i] = i; 
15. }  
16. flag = 1; 
17. for(i = 0; i < count; i++){//初始化 
18.     MPI_Irecv(lsum[indices[i]], 2, MPI_INT, indices[i] + 1, 100, MPI_COMM_WORLD, reqs + indices[i]); 
19. }   
20. while(allocatedJobs||flag) 
21. { 
22.     MPI_Waitsome(numprocs - 1,reqs, &count, indices, stats);  
23.     allocatedJobs -=count;  
24.     for(i = 0; i < count; i++) 
25.     { 
26.         sendlen = (lsum[indices[i]])[1];  
27.         temp = (lsum[indices[i]])[0]; 
28.         result = (result + temp-1)%9+1; 
29.         printf("Master--worker %d require %d datas and answer %d.\n",indices[i] + 1,sendlen,temp);  
30.         if(read>=sendlen) 
31.         { 
32.                       
33.             printf("Master--allocate %d datas to worker %d.\n",sendlen,indices[i] + 1);  
34.             MPI_Isend(buffer+hassend,sendlen, MPI_CHAR, indices[i] + 1, 99, MPI_COMM_WORLD, &request);                           
35.             hassend += sendlen; 
36.             read -=sendlen;   
37.             MPI_Wait(&request, &status); 
38.             MPI_Irecv(lsum[indices[i]], 2, MPI_INT, indices[i] + 1, 100, MPI_COMM_WORLD, reqs + indices[i]); 
39.             allocatedJobs++;  
40.         } 
41.         else 
42.         {//向后来的Worker发送退出信号 
43.             MPI_Isend(buffer,0, MPI_CHAR, indices[i] + 1, 99, MPI_COMM_WORLD, &request);  
44.             printf("Master--kill worker %d.\n",0,indices[i] + 1);  
45.             MPI_Wait(&request, &status); 
46.             flag = 0;                    
47.         }  
48.     }  
49. } 
50. if(read!=0) 
51. {//剩余数据量不足以分配时，则由Master计算 
52.     temp = sum(buffer+hassend,read); 
53.     result = (result+temp-1)%9+1; 
54. } 

③.Worker进程随机生成请求量的大小，主动向Master请求任务，主要代码实现 ：

//初始化变量

 

1. //初始化变量 
2. char  buf[Block_SIZE];   
3. int result[2],recvlen,myid; 
4. MPI_Status status; 
5. MPI_Request request; 
6. MPI_Comm_rank(MPI_COMM_WORLD, &myid);   
7. result[0] = 0;//第一次计算结果设置为0 
8. result[1] = RANDOM(2,Block_SIZE); //生成随机请求量 
9. MPI_Isend(result, 2, MPI_INT, Master_Id, 100, MPI_COMM_WORLD, &request); 
10. MPI_Wait(&request, &status); 
11. while(1) 
12. { 
13.     memset(buf, '\0', Block_SIZE); 
14.     MPI_Irecv(buf, result[1], MPI_CHAR, 0, 99, MPI_COMM_WORLD, &request); 
15.     MPI_Wait(&request, &status); 
16.     MPI_Get_count(&status, MPI_CHAR, &recvlen);   
17.     if(recvlen<=0){ //退出信号 
18.         break; 
19.     } 
20.     printf("worker--%d receive %s from master.\n",myid,buf); 
21.     result[0] = sum(buf,recvlen); 
22.     result[1] = RANDOM(2,Block_SIZE); //生成随机请求量 
23.     MPI_Isend(result, 2, MPI_INT,Master_Id, 100, MPI_COMM_WORLD, &request); 
24.     MPI_Wait(&request, &status); 
25. }  

3. 实验测试结果

①.当只有一个进程时，为串行计算，使用测试数据集100.tst和999.tst结果如下：

100.tst 

999.tst

 

②. 设置最大请求量MaxSize=20bit:

a.当有两个进程时，一个为Master，一个为Worker，即只有一个计算进程，从结果中与①对比可以看出通信量给总体任务执行完的影响，使用测试数据集100.tst和999.tst结果如下：

100.tst

999.tst

b.输入进程数为5，查看Worker进程数量对计算时间的影响

100.tst

999.tst

③设置MaxSize = 50bit时，设置进程数为5时结果如下：

100.tst

999.tst

4. 性能分析

当设定每次Master向Worker发送的最大数据量为20bit时，改变进程数结果对比如下：

                           表格 1 进程数对性能的影响

固定总进程数为5时，Worker请求发送的最大数据量MaxSize对实验的影响，实验数据为执行时间为秒（s）：

                           表格 2 请求量阈值的影响

实验分析：从表1中非阻塞式编程对试验数据进行测试时，当有1个进程时为串行计算，有2个进程时一个为Master，一个为Worker，但串行执行反而更快，执行时间更短，分析源于并行执行的通信时间过大，且每次Worker不是一次性请求数据，分为多次请求、多次通信，时间大部分都浪费在通信上。当为并行执行时，增加进程数并不一定会减少执行时间，表中对测试数据100.tst，从进程10增加到进程数为20个时，执行时间反而增大，100.tst数据量小，对于一次开启20个进程，即19个Worker时，都请求数据时，虽然请求数据量不同，但进程过多，将导致有部分Worker连一次数据都没有请求到，而Master却要为没有数据可分的Worker维护通信，故使执行时间增大。对于大数据量测试集999.tst，增加进程数从5到20，执行时间一直呈下降趋势，进程数多，并发度就越高，总体执行时间更小，故效率更高。对于表2，固定进程数，增大Worker每次能向Master请求数据量的上限，虽然是随机请求数据量，但时间一直呈下降趋势。故增大最大请求量阈值可提高计算性能。总体来看需要权衡需要开启的进程数，对于一定的数据处理量，恰当设置Worker每次请求的数据量的最大阈值对于并行度影响也比较大，最大程度的利用每个Worker且尽量增大每次Master向Worker发送的数据量可尽可能的利用最大资源，减少通信浪费的时间，总体提高效率。