source: CIVL/text/include/mpi.cvl@ d6f8945

1.23 2.0 acw/focus-triggers main test-branch
Last change on this file since d6f8945 was e15d0a3, checked in by Ziqing Luo <ziqing@…>, 12 years ago

Added Implementation of MPI_Reduce and MPI_Allreduce
Move definitions of CMPI_Gcomm and MPI_Comm into mpi-common.h

git-svn-id: ​svn://vsl.cis.udel.edu/civl/trunk@944 fb995dde-84ed-4084-dfe6-e5aef3e2452c
  • Property mode set to 100644
File size: 12.6 KB
Line 
1#ifdef __MPI__
2#else
3 #include <mpi-common.h>
4 #ifdef __CIVLC__
5 #else
6 #include <civlc.h>
7 #endif
8#endif
9
10/****************************** Helper Functions **********************************/
11int sizeofDatatype(MPI_Datatype datatype) {
12 switch (datatype) {
13 case MPI_INT:
14 return sizeof(int);
15 case MPI_FLOAT:
16 return sizeof(float);
17 case MPI_DOUBLE:
18 return sizeof(double);
19 case MPI_CHAR:
20 return sizeof(char);
21 default:
22 $assert(0, "Unreachable");
23 }
24}
25
26/* Helpers for MPI_Reduce and MPI_Allreduce */
27void sumInt(int result[], int buf[], int real_count, int nprocs){
28 //init result array
29 $atomic{
30 for(int i=0; i<real_count; i++){
31 result[i] = 0;
32 }
33 //sum up
34 for(int i=0; i<nprocs; i++)
35 for(int j=0; j<real_count; j++)
36 result[j] = result[j] + buf[i * real_count + j];
37 }
38}
39
40void sumFloat(float result[], float buf[], int real_count, int nprocs){
41 $atomic{
42 for(int i=0; i<real_count; i++){
43 result[i] = 0;
44 }
45 //sum up
46 for(int i=0; i<nprocs; i++)
47 for(int j=0; j<real_count; j++)
48 result[j] = result[j] + buf[i * real_count + j];
49 }
50}
51
52void sumDouble(double result[], double buf[], int real_count, int nprocs){
53 $atomic{
54 for(int i=0; i<real_count; i++){
55 result[i] = 0;
56 }
57 //sum up
58 for(int i=0; i<nprocs; i++)
59 for(int j=0; j<real_count; j++)
60 result[j] = result[j] + buf[i * real_count + j];
61 }
62}
63
64
65void maxInt(int result[], int buf[], int real_count, int nprocs){
66 //init result arra
67 $atomic{
68 for(int i=0; i<real_count; i++){
69 result[i] = buf[i * real_count];
70 }
71
72 for(int i=0; i<nprocs; i++)
73 for(int j=0; j<real_count; j++){
74 if(buf[i * real_count + j] > result[j])
75 result[j] = buf[i * real_count + j];
76 }
77 }
78}
79
80void maxFloat(float result[], float buf[], int real_count, int nprocs){
81 //init result array
82 $atom{
83 for(int i=0; i<real_count; i++){
84 result[i] = buf[i * real_count];
85 }
86
87 for(int i=0; i<nprocs; i++)
88 for(int j=0; j<real_count; j++){
89 if(buf[i * real_count + j] > result[j])
90 result[j] = buf[i * real_count + j];
91 }
92 }
93}
94
95void maxDouble(double result[], double buf[], int real_count, int nprocs){
96 //init result array
97 $atom{
98 for(int i=0; i<real_count; i++){
99 result[i] = buf[i * real_count];
100 }
101
102 for(int i=0; i<nprocs; i++)
103 for(int j=0; j<real_count; j++){
104 if(buf[i * real_count + j] > result[j])
105 result[j] = buf[i * real_count + j];
106 }
107 }
108}
109
110void minInt(int result[], int buf[], int real_count, int nprocs){
111 //init result array
112 $atom{
113 for(int i=0; i<real_count; i++){
114 result[i] = buf[i * real_count];
115 }
116
117 for(int i=0; i<nprocs; i++)
118 for(int j=0; j<real_count; j++){
119 if(buf[i * real_count + j] < result[j])
120 result[j] = buf[i * real_count + j];
121 }
122 }
123}
124
125void minFloat(float result[], float buf[], int real_count, int nprocs){
126 //init result array
127 $atom{
128 for(int i=0; i<real_count; i++){
129 result[i] = buf[i * real_count];
130 }
131
132 for(int i=0; i<nprocs; i++)
133 for(int j=0; j<real_count; j++){
134 if(buf[i * real_count + j] < result[j])
135 result[j] = buf[i * real_count + j];
136 }
137 }
138}
139
140void minDouble(double result[], double buf[], int real_count, int nprocs){
141 //init result array
142 $atom{
143 for(int i=0; i<real_count; i++){
144 result[i] = buf[i * real_count];
145 }
146
147 for(int i=0; i<nprocs; i++)
148 for(int j=0; j<real_count; j++){
149 if(buf[i * real_count + j] < result[j])
150 result[j] = buf[i * real_count + j];
151 }
152 }
153}
154
155/************************** MPI LIB Implementations *******************************/
156CMPI_Gcomm CMPI_Gcomm_create($scope scope, int size) {
157 CMPI_Gcomm result;
158
159 result.p2p = $gcomm_create(scope, size);
160 result.col = $gcomm_create(scope, size);
161 return result;
162}
163
164void CMPI_Gcomm_destroy(CMPI_Gcomm gc) {
165 $gcomm_destroy(gc.p2p);
166 $gcomm_destroy(gc.col);
167}
168
169MPI_Comm MPI_Comm_create($scope scope, CMPI_Gcomm gc, int rank) {
170 MPI_Comm result;
171
172 result.p2p = $comm_create(scope, gc.p2p, rank);
173 result.col = $comm_create(scope, gc.col, rank);
174 return result;
175}
176
177void MPI_Comm_destroy(MPI_Comm comm) {
178 $comm_destroy(comm.p2p);
179 $comm_destroy(comm.col);
180}
181
182int MPI_Init(int *argc, char *** argv){
183 return 0;
184}
185
186int __MPI_Init() {
187 return 0;
188}
189
190int MPI_Finalize() {
191 return 0;
192}
193
194int MPI_Comm_size(MPI_Comm comm, int *size) {
195 *size = $comm_size(comm.p2p);
196 return 0;
197}
198
199int MPI_Comm_rank(MPI_Comm comm, int *rank) {
200 *rank = $comm_place(comm.p2p);
201 return 0;
202}
203
204
205int CMPI_Send(void *buf, int count, MPI_Datatype datatype, int dest,
206 int tag, $comm comm) {
207 if (dest >= 0) {
208 int size = count*sizeofDatatype(datatype);
209 int place = $comm_place(comm);
210 $message out = $message_pack(place, dest, tag, buf, size);
211
212 $comm_enqueue(comm, out);
213 }
214 return 0;
215}
216
217int MPI_Send(void *buf, int count, MPI_Datatype datatype, int dest,
218 int tag, MPI_Comm comm) {
219 return CMPI_Send(buf, count, datatype, dest, tag, comm.p2p);
220}
221
222
223int CMPI_Recv(void *buf, int count, MPI_Datatype datatype, int source,
224 int tag, $comm comm, MPI_Status *status) {
225 if (source >= 0) {
226 $message in = $comm_dequeue(comm, source, tag);
227 int size = count*sizeofDatatype(datatype);
228
229 $message_unpack(in, buf, size);
230 if (status != MPI_STATUS_IGNORE) {
231 status->size = $message_size(in);
232 status->MPI_SOURCE = $message_source(in);
233 status->MPI_TAG = $message_tag(in);
234 status->MPI_ERROR = 0;
235 }
236 }
237 return 0;
238}
239
240int MPI_Recv(void *buf, int count, MPI_Datatype datatype, int source,
241 int tag, MPI_Comm comm, MPI_Status *status) {
242 return CMPI_Recv(buf, count, datatype, source, tag, comm.p2p, status);
243}
244
245int MPI_Get_count(MPI_Status *status, MPI_Datatype datatype, int *count) {
246 *count = status->size/sizeofDatatype(datatype);
247 return 0;
248}
249
250int MPI_Sendrecv(void *sendbuf, int sendcount, MPI_Datatype sendtype,
251 int dest, int sendtag,
252 void *recvbuf, int recvcount, MPI_Datatype recvtype,
253 int source, int recvtag,
254 MPI_Comm comm, MPI_Status *status) {
255 MPI_Send(sendbuf, sendcount, sendtype, dest, sendtag, comm);
256 MPI_Recv(recvbuf, recvcount, recvtype, source, recvtag, comm, status);
257 return 0;
258}
259
260/* Broadcasts a message from root to everyone else.
261 * Need to use a differnt comm.
262 */
263int MPI_Bcast(void *buf, int count, MPI_Datatype datatype, int root,
264 MPI_Comm comm) {
265 int place = $comm_place(comm.col);
266
267 if (place == root) {
268 int nprocs = $comm_size(comm.col);
269
270 for (int i=0; i<nprocs; i++)
271 if (i != root)
272 CMPI_Send(buf, count, datatype, i, BCAST_TAG, comm.col);
273 } else {
274 CMPI_Recv(buf, count, datatype, root, BCAST_TAG, comm.col,
275 MPI_STATUS_IGNORE);
276 }
277 return 0;
278}
279
280/* Reduces values on all processes to a single value */
281int MPI_Reduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype,
282 MPI_Op op, int root, MPI_Comm comm){
283 int place = $comm_place(comm.col);
284 MPI_Status status;
285
286 //non-root
287 CMPI_Send(sendbuf, count, datatype, root, REDUCE_TAG, comm.col);
288 if(place != root)
289 return 0;
290 else{
291 //root
292 int nprocs = $comm_size(comm.col);
293 int real_count = -1; //the real count gotton from MPI_Status
294 //void ** buf; //Buffer stores data from other processes
295 void * recv; //Buffer used in MPI_Recv()
296 void * results; //Array of results
297 $scope here = $root;
298 struct Bundle_buf{
299 int I[nprocs * count];
300 float F[nprocs * count];
301 double D[nprocs * count];
302 } buf;
303
304 switch(datatype){
305 case MPI_INT :
306 recv = (int *)$malloc(here, count * sizeof(int));
307 results = (int *)$malloc(here, count * sizeof(int));
308 break;
309 case MPI_FLOAT :
310 recv = (float *)$malloc(here, count * sizeof(float));
311 results = (float *)$malloc(here, count * sizeof(float));
312 break;
313 case MPI_DOUBLE :
314 recv = (double *)$malloc(here, count * sizeof(double));
315 results = (double *)$malloc(here, count * sizeof(double));
316 break;
317 }
318 for(int i=0; i<nprocs; i++){
319 CMPI_Recv(recv, count, datatype, i, REDUCE_TAG, comm.col,
320 &status);
321 MPI_Get_count(&status, datatype, &real_count);
322 $assert(real_count == count);
323
324 switch(datatype){
325 case MPI_INT :
326 for(int j=0; j<real_count; j++)
327 buf.I[i * real_count + j] = ((int *)recv)[j];
328 break;
329 case MPI_FLOAT :
330 for(int j=0; j<real_count; j++)
331 buf.F[i * real_count + j] = ((float *)recv)[j];
332 break;
333 case MPI_DOUBLE :
334 for(int j=0; j<real_count; j++)
335 buf.D[i * real_count + j] = ((double *)recv)[j];
336 break;
337 }
338 }
339
340 //operations
341 switch(datatype){
342 case MPI_INT:
343 if(op == MPI_SUM)
344 sumInt(results, buf.I, real_count, nprocs);
345 if(op == MPI_MAX)
346 maxInt(results, buf.I, real_count, nprocs);
347 if(op == MPI_MIN)
348 minInt(results, buf.I, real_count, nprocs);
349 break;
350 case MPI_FLOAT:
351 if(op == MPI_SUM)
352 sumFloat(results, buf.F, real_count, nprocs);
353 if(op == MPI_MAX)
354 maxFloat(results, buf.F, real_count, nprocs);
355 if(op == MPI_MIN)
356 minFloat(results, buf.F, real_count, nprocs);
357 break;
358 case MPI_DOUBLE:
359 if(op == MPI_SUM)
360 sumDouble(results, buf.D, real_count, nprocs);
361 if(op == MPI_MAX)
362 maxDouble(results, buf.D, real_count, nprocs);
363 if(op == MPI_MIN)
364 minDouble(results, buf.D, real_count, nprocs);
365 break;
366 }
367 MPI_Sendrecv(results, real_count, MPI_INT,
368 place, 0,
369 recvbuf, real_count, MPI_INT,
370 place, 0,
371 comm, &status);
372 $free(recv);
373 $free(results);
374 }
375 return 0;
376}
377
378/* Combines values from all processes and distributes the result back to all processes */
379/* default root is 0 */
380int MPI_Allreduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype,
381 MPI_Op op, MPI_Comm comm){
382 int place = $comm_place(comm.col);
383 int root = 0;
384 MPI_Status status;
385
386 //non-root
387 CMPI_Send(sendbuf, count, datatype, root, REDUCE_TAG, comm.col);
388 if(place != root){
389 MPI_Bcast(recvbuf, count, datatype, root,
390 comm);
391 return 0;
392 }
393 else{
394 //root
395 int nprocs = $comm_size(comm.col);
396 int real_count = -1; //the real count gotton from MPI_Status
397 //void ** buf; //Buffer stores data from other processes
398 void * recv; //Buffer used in MPI_Recv()
399 void * results; //Array of results
400 $scope here = $root;
401 struct Bundle_buf{
402 int I[nprocs * count];
403 float F[nprocs * count];
404 double D[nprocs * count];
405 } buf;
406
407 switch(datatype){
408 case MPI_INT :
409 recv = (int *)$malloc(here, count * sizeof(int));
410 results = (int *)$malloc(here, count * sizeof(int));
411 break;
412 case MPI_FLOAT :
413 recv = (float *)$malloc(here, count * sizeof(float));
414 results = (float *)$malloc(here, count * sizeof(float));
415 break;
416 case MPI_DOUBLE :
417 recv = (double *)$malloc(here, count * sizeof(double));
418 results = (double *)$malloc(here, count * sizeof(double));
419 break;
420 }
421 //recv = (void *)$malloc(here, count * sizeofDatatype(datatype));
422 //results = (void *)$malloc(here, count * sizeofDatatype(datatype));
423
424 for(int i=0; i<nprocs; i++){
425 CMPI_Recv(recv, count, datatype, i, REDUCE_TAG, comm.col,
426 &status);
427 MPI_Get_count(&status, datatype, &real_count);
428 $assert(real_count == count);
429
430 switch(datatype){
431 case MPI_INT :
432 for(int j=0; j<real_count; j++)
433 buf.I[i * real_count + j] = ((int *)recv)[j];
434 break;
435 case MPI_FLOAT :
436 for(int j=0; j<real_count; j++)
437 buf.F[i * real_count + j] = ((float *)recv)[j];
438 break;
439 case MPI_DOUBLE :
440 for(int j=0; j<real_count; j++)
441 buf.D[i * real_count + j] = ((double *)recv)[j];
442 break;
443 }
444 }
445
446 //operations
447 switch(datatype){
448 case MPI_INT:
449 if(op == MPI_SUM)
450 sumInt(results, buf.I, real_count, nprocs);
451 if(op == MPI_MAX)
452 maxInt(results, buf.I, real_count, nprocs);
453 if(op == MPI_MIN)
454 minInt(results, buf.I, real_count, nprocs);
455 break;
456 case MPI_FLOAT:
457 if(op == MPI_SUM)
458 sumFloat(results, buf.F, real_count, nprocs);
459 if(op == MPI_MAX)
460 maxFloat(results, buf.F, real_count, nprocs);
461 if(op == MPI_MIN)
462 minFloat(results, buf.F, real_count, nprocs);
463 break;
464 case MPI_DOUBLE:
465 if(op == MPI_SUM)
466 sumDouble(results, buf.D, real_count, nprocs);
467 if(op == MPI_MAX)
468 maxDouble(results, buf.D, real_count, nprocs);
469 if(op == MPI_MIN)
470 minDouble(results, buf.D, real_count, nprocs);
471 break;
472 }
473 MPI_Sendrecv(results, real_count, MPI_INT,
474 place, 0,
475 recvbuf, real_count, MPI_INT,
476 place, 0,
477 comm, &status);
478 MPI_Bcast(results, real_count, datatype, root,
479 comm);
480 $free(recv);
481 $free(results);
482 }
483 return 0;
484}
Note: See TracBrowser for help on using the repository browser.