MPI 實現並行直接插入排序

話說 直插排序串行只需要十幾行 並行就大不一樣。。。

基本思路是

1/主線程控制數據讀取 分發和匯總

2/從線程中數據按線程號非降有序 每次有新數據 主線程廣播 從線程按照數據大小判斷是否在自己處理范圍內 如果在的話 在自己保存的數組中進行直接插入

3/最後 主線程按序收集顯示所有數據

1. #include "mpi.h"
2. #include <unistd.h>
3. #include <fcntl.h>
4. #include <ctype.h>
5. #include <stdio.h>
6. #include <stdlib.h>
8. #define END 0
9. #define INIT 1
10. #define DATA 2
11. #define ROOT 0
13. int readValue(char* file,int* values)
14. {
15. int fd,size,count=0;
16. char buffer[80],num[11];
17. fd=open(file,O_RDONLY);
18. do
19. {
20. size=read(fd,buffer,sizeof(buffer));
21. int j=0;
22. for(int i=0;i<size;++i)
23. {
24. if(buffer[i]<'9'&&buffer[i]>'0')
25. {
26. num[j++]=buffer[i];
27. }else
28. {
29. num[j]='\0';
30. values[count++]=atoi(num);
31. j=0;
32. }
33. }
34. }while(size!=0);
35. close(fd);
36. return count;
37. }
38. void insert(int* arr,int pos,int temp)
39. {
40. while(pos>=0&&temp<arr[pos])
41. {
42. arr[pos+1]=arr[pos];
43. --pos;
44. }
45. arr[pos+1]=temp;
46. }
47. void serial(int* arr,int size)
48. {
49. for(int i=1;i<size;++i)
50. {
51. insert(arr,i-1,arr[i]);
52. }
53. }
55. int main(int argc,char *argv[])
56. {
57. int* values;
58. int self,size,length,temp;
59. MPI_Init(&argc,&argv);
60. MPI_Comm_size(MPI_COMM_WORLD,&size);
61. MPI_Comm_rank(MPI_COMM_WORLD,&self);
62. MPI_Status status;
63. if(self==0)
64. {
65. //read value from file
66. values=(int*)malloc(100*sizeof(int));
67. length=readValue("/home/gt/parellel/sort/data.in",values);
68. //initial root value of each process
69. serial(values,size);
70. //broadcast the max length of each process
71. MPI_Bcast(&length,1,MPI_INT,ROOT,MPI_COMM_WORLD);
72. //the boundary
73. for(int i=1;i<size;++i)
74. {
75. MPI_Ssend(&values[i-1],1,MPI_INT,i,INIT,MPI_COMM_WORLD);
76. }
77. MPI_Recv(&temp,1,MPI_INT,self+1,DATA,MPI_COMM_WORLD,&status);
78. //broadcast each value and do insert in each process
79. for(int i=size-1;i<length;++i)
80. {
81. temp=values[i];
82. MPI_Bcast(&temp,1,MPI_INT,ROOT,MPI_COMM_WORLD);
83. printf("broadcast %d \n",values[i]);
84. //MPI_Barrier(MPI_COMM_WORLD);
85. }
87. //ends
88. printf("ends \n");
89. temp=END;
90. MPI_Bcast(&temp,1,MPI_INT,ROOT,MPI_COMM_WORLD);
91. //recieve from each process in order
92. int pos=0;
93. int len=0;
94. for(int i=1;i<size;++i)
95. {
96. MPI_Recv(&len,1,MPI_INT,i,DATA,MPI_COMM_WORLD,&status);
97. for(int j=0;j<len;++j)
98. {
99. MPI_Recv(&temp,1,MPI_INT,i,DATA,MPI_COMM_WORLD,&status);
100. values[pos++]=temp;
101. }
102. }
103. for(int i=0;i<length;++i)
104. {
105. printf("%d \n",values[i]);
106. }
107. free(values);
108. }
109. else
110. {
111. MPI_Request request;
112. //recieve max length
113. MPI_Bcast(&length,1,MPI_INT,ROOT,MPI_COMM_WORLD);
114. //position
115. int pos=0;
116. //post irecv
117. MPI_Irecv(&temp,1,MPI_INT,ROOT,INIT,MPI_COMM_WORLD,&request);
118. //data on this process
119. int* local=(int*) malloc(length*sizeof(int)+1);
120. //wait for the recieve
121. MPI_Wait(&request,&status);
122. //exchange upper limit
123. int upper=999999;//max of int
124. if(self!=size-1)MPI_Irecv(&upper,1,MPI_INT,self+1,DATA,MPI_COMM_WORLD,&request);
125. MPI_Ssend(&temp,1,MPI_INT,self-1,DATA,MPI_COMM_WORLD);
126. if(self!=size-1)MPI_Wait(&request,&status);
127. local[pos++]=temp;
128. MPI_Bcast(&temp,1,MPI_INT,ROOT,MPI_COMM_WORLD);
129. //MPI_Barrier(MPI_COMM_WORLD);
130. while(temp!=END)
131. {
132. printf("%d recieve %d \n",self,temp);
133. if(temp<=upper&&(self==1||local[0]<temp))
134. {
135. printf("%d insert %d \n",self,temp);
136. local[pos++]=temp;
137. insert(local,pos-2,temp);
138. }
139. MPI_Bcast(&temp,1,MPI_INT,ROOT,MPI_COMM_WORLD);
140. //MPI_Barrier(MPI_COMM_WORLD);
141. }
142. //ends
143. //sends the data in order to root process
144. MPI_Ssend(&pos,1,MPI_INT,ROOT,DATA,MPI_COMM_WORLD);
145. for(int i=0;i<pos;++i)
146. {
147. MPI_Ssend(&local[i],1,MPI_INT,ROOT,DATA,MPI_COMM_WORLD);
148. printf("% d send back %d \n",self,local[i]);
149. }
150. free(local);
151. }
152. MPI_Finalize();
153. return 0;
154. }