MPI 堆排序

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

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

1. #include "mpi.h"   
2. #include <stdio.h>   
3. #include <stdlib.h>   
4.   
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;  
13.   
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. }