MPI 堆排序

堆排是串行排序中最適合並行化的排序之一

1/分為主線程和從線程
2/主線程分發數據，使用大小與從線程個數相同的堆作為私有堆，進行最後整理用
3/從線程維護一個堆，每次返回給主線程堆頂元素
4/主線程 提取堆頂元素 通知相應從線程提交新的堆頂元素
5/主從線程並行進行重建堆（heapify）的動作

1. #include "mpi.h"
2. #include <stdio.h>
3. #include <stdlib.h>
5. #define ROOT 0
6. #define TAG 0
7. #define NEXT 1
8. #define END -1
9. typedef struct node {
10. int v;
11. int r;
12. } Node;
14. static inline int left(int i) {
15. return i*2+1;
16. }
17. static inline int right(int i) {
18. return left(i)+1;
19. }
20. void defineIntVector(MPI_Datatype* type,int length) {
21. MPI_Type_vector(length,1,1,MPI_INT,type);
22. MPI_Type_commit(type);
23. }
24. //for slaves
25. void heapifyI(int *a,int root,int size) {
26. int l=left(root);
27. int r=right(root);
28. int m=root;
29. if(l<size&&a[l]<a[m]) {
30. m=l;
31. }
32. if(r<size&&a[r]<a[m]) {
33. m=r;
34. }
35. if(m!=root) {
36. int t=a[root];
37. a[root]=a[m];
38. a[m]=t;
39. heapifyI(a,m,size);
40. }
41. }
42. void buildHeapI(int *a,int size) {
43. for(int i=(size-1)/2;i>=0;--i) {
44. heapifyI(a,i,size);
45. }
46. }
47. //for root
48. void heapifyN(Node *a,int root,int size) {
49. int l=left(root);
50. int r=right(root);
51. int m=root;
52. if(l<size&&a[l].v<a[m].v) {
53. m=l;
54. }
55. if(r<size&&a[r].v<a[m].v) {
56. m=r;
57. }
58. if(m!=root) {
59. Node t=a[root];
60. a[root]=a[m];
61. a[m]=t;
62. heapifyN(a,m,size);
63. }
64. }
65. void buildHeapN(Node *a,int size) {
66. for(int i=(size-1)/2;i>=0;--i) {
67. heapifyN(a,i,size);
68. }
69. }
70. int main(int argc,char *argv[]) {
71. int self,size;
72. int part;//threshold
73. MPI_Init(&argc,&argv);
74. MPI_Comm_rank(MPI_COMM_WORLD,&self);
75. MPI_Comm_size(MPI_COMM_WORLD,&size);
76. MPI_Status status;
77. MPI_Datatype MPI_VEC;
78. if(ROOT==self) {
79. //all data
80. int values[]={1,5,3,8,4,2,9,6,7,54,34,6,13,63,32,14,53,21,65,33,24,213,23,14,21,65,32};
81. int length=27;
82. //for each process
83. part=length/(size-1);
84. MPI_Bcast(&part,1,MPI_INT,ROOT,MPI_COMM_WORLD);
85. defineIntVector(&MPI_VEC,part);
86. //heap for min value from each process
87. Node *nodes=(Node*)malloc((size-1)*sizeof(Node));
88. for(int i=0;i<size-1;++i) {
89. nodes[i].r=i+1;
90. }
91. //send out all the data
92. for(int i=1;i<size;++i) {
93. MPI_Ssend(&values[(i-1)*part],1,MPI_VEC,i,TAG,MPI_COMM_WORLD);
94. }
95. for(int i=1;i<size;++i) {
96. MPI_Recv(&(nodes[i-1].v),1,MPI_INT,i,TAG,MPI_COMM_WORLD,&status);
97. }
98. //end flag
99. part=size-1;
100. //build a heap representing process
101. buildHeapN(nodes,part);
102. int p=0;
103. while(part>0) {
104. values[p++]=nodes[0].v;
105. //similar to the message passing system of select sort
106. MPI_Ssend(&p,1,MPI_INT,nodes[0].r,NEXT,MPI_COMM_WORLD);
107. MPI_Recv(&(nodes[0].v),1,MPI_INT,nodes[0].r,TAG,MPI_COMM_WORLD,&status);
108. if(END==nodes[0].v) {
109. nodes[0]=nodes[--part];
110. }
111. heapifyN(nodes,0,part);
112. }
113. for(int i=0;i<length;++i)
114. printf("%d ",values[i]);
115. free(nodes);
116. } else {
117. MPI_Bcast(&part,1,MPI_INT,ROOT,MPI_COMM_WORLD);
118. defineIntVector(&MPI_VEC,part);
119. int *values=(int*)malloc(part*sizeof(int));
120. MPI_Recv(values,1,MPI_VEC,ROOT,TAG,MPI_COMM_WORLD,&status);
121. buildHeapI(values,part);
122. //send the smallest
123. MPI_Ssend(&values[0],1,MPI_INT,ROOT,TAG,MPI_COMM_WORLD);
124. //pop out one
125. values[0]=values[--part];
126. //rebuild heap
127. heapifyI(values,0,part);
128. while(part>0) {
129. MPI_Recv(&values[part],1,MPI_INT,ROOT,NEXT,MPI_COMM_WORLD,&status);
130. MPI_Ssend(&values[0],1,MPI_INT,ROOT,TAG,MPI_COMM_WORLD);
131. values[0]=values[--part];
132. heapifyI(values,0,part);
133. }
134. MPI_Recv(&values[part],1,MPI_INT,ROOT,NEXT,MPI_COMM_WORLD,&status);
135. values[0]=END;
136. MPI_Ssend(&values[0],1,MPI_INT,ROOT,TAG,MPI_COMM_WORLD);
137. free(values);
138. }
139. MPI_Finalize();
140. return 0;
141. }