MPI 实现并行直接插入排序
基本思路是
1/主线程控制数据读取 分发和汇总
2/从线程中数据按线程号非降有序 每次有新数据 主线程广播 从线程按照数据大小判断是否在自己处理范围内 如果在的话 在自己保存的数组中进行直接插入
3/最后 主线程按序收集显示所有数据
#include "mpi.h"#include <unistd.h>#include <fcntl.h>#include <ctype.h>#include <stdio.h>#include <stdlib.h>#define END 0#define INIT 1#define DATA 2#define ROOT 0int readValue(char* file,int* values){int fd,size,count=0;char buffer[80],num[11];fd=open(file,O_RDONLY);do{size=read(fd,buffer,sizeof(buffer));int j=0;for(int i=0;i<size;++i){if(buffer[i]<'9'&&buffer[i]>'0'){num[j++]=buffer[i];}else{num[j]='\0';values[count++]=atoi(num);j=0;}}}while(size!=0);close(fd);return count;}void insert(int* arr,int pos,int temp){while(pos>=0&&temp<arr[pos]){arr[pos+1]=arr[pos];--pos;}arr[pos+1]=temp;}void serial(int* arr,int size){for(int i=1;i<size;++i){insert(arr,i-1,arr[i]);}}int main(int argc,char *argv[]){int* values;int self,size,length,temp;MPI_Init(&argc,&argv);MPI_Comm_size(MPI_COMM_WORLD,&size);MPI_Comm_rank(MPI_COMM_WORLD,&self);MPI_Status status;if(self==0){//read value from filevalues=(int*)malloc(100*sizeof(int));length=readValue("/home/gt/parellel/sort/data.in",values);//initial root value of each processserial(values,size);//broadcast the max length of each processMPI_Bcast(&length,1,MPI_INT,ROOT,MPI_COMM_WORLD);//the boundaryfor(int i=1;i<size;++i){MPI_Ssend(&values[i-1],1,MPI_INT,i,INIT,MPI_COMM_WORLD);}MPI_Recv(&temp,1,MPI_INT,self+1,DATA,MPI_COMM_WORLD,&status);//broadcast each value and do insert in each processfor(int i=size-1;i<length;++i){temp=values[i];MPI_Bcast(&temp,1,MPI_INT,ROOT,MPI_COMM_WORLD);printf("broadcast %d \n",values[i]);//MPI_Barrier(MPI_COMM_WORLD);}//endsprintf("ends \n");temp=END;MPI_Bcast(&temp,1,MPI_INT,ROOT,MPI_COMM_WORLD);//recieve from each process in orderint pos=0;int len=0;for(int i=1;i<size;++i){MPI_Recv(&len,1,MPI_INT,i,DATA,MPI_COMM_WORLD,&status);for(int j=0;j<len;++j){MPI_Recv(&temp,1,MPI_INT,i,DATA,MPI_COMM_WORLD,&status);values[pos++]=temp;}}for(int i=0;i<length;++i){printf("%d \n",values[i]);}free(values);}else{MPI_Request request;//recieve max lengthMPI_Bcast(&length,1,MPI_INT,ROOT,MPI_COMM_WORLD);//positionint pos=0;//post irecvMPI_Irecv(&temp,1,MPI_INT,ROOT,INIT,MPI_COMM_WORLD,&request);//data on this processint* local=(int*) malloc(length*sizeof(int)+1);//wait for the recieveMPI_Wait(&request,&status);//exchange upper limitint upper=999999;//max of intif(self!=size-1)MPI_Irecv(&upper,1,MPI_INT,self+1,DATA,MPI_COMM_WORLD,&request);MPI_Ssend(&temp,1,MPI_INT,self-1,DATA,MPI_COMM_WORLD);if(self!=size-1)MPI_Wait(&request,&status);local[pos++]=temp;MPI_Bcast(&temp,1,MPI_INT,ROOT,MPI_COMM_WORLD);//MPI_Barrier(MPI_COMM_WORLD);while(temp!=END){printf("%d recieve %d \n",self,temp);if(temp<=upper&&(self==1||local[0]<temp)){printf("%d insert %d \n",self,temp);local[pos++]=temp;insert(local,pos-2,temp);}MPI_Bcast(&temp,1,MPI_INT,ROOT,MPI_COMM_WORLD);//MPI_Barrier(MPI_COMM_WORLD);}//ends//sends the data in order to root processMPI_Ssend(&pos,1,MPI_INT,ROOT,DATA,MPI_COMM_WORLD);for(int i=0;i<pos;++i){MPI_Ssend(&local[i],1,MPI_INT,ROOT,DATA,MPI_COMM_WORLD);printf("% d send back %d \n",self,local[i]);}free(local);}MPI_Finalize();return 0;}