source: CIVL/src/include/civl/civl-mpi.cvl@ 01188a2

1.23 2.0 acw/focus-triggers main test-branch
Last change on this file since 01188a2 was bf584ca, checked in by Manchun Zheng <zmanchun@…>, 11 years ago

cleaned up #ifdef ... #else of headers using #ifndef.

git-svn-id: ​svn://vsl.cis.udel.edu/civl/trunk@2292 fb995dde-84ed-4084-dfe6-e5aef3e2452c
  • Property mode set to 100644
File size: 16.7 KB
Line 
1#ifndef __CIVL_CIVLMPI__
2#define __CIVL_CIVLMPI__
3
4#include <civlc.cvh>
5#include <concurrency.cvh>
6#include <comm.cvh>
7#include <bundle.cvh>
8#include <mpi.h>
9#include <civl-mpi.cvh>
10#include <string.h>
11#include <pointer.cvh>
12
13/**************************** Duplicated Part *************************************/
14/* Duplicated definition with the same struct in mpi.h.
15 The reason of this duplication is to make civlmpi.cvl
16 independent with mpi.cvl. */
17typedef struct MPI_Comm {
18 $comm p2p; // point-to-point communication
19 $comm col; // collective communication
20 $collect_checker collect_checker;
21 $barrier barrier;
22 int gcommIndex; //the index of the corresponding global communicator.
23}MPI_Comm;
24
25/* Definition of CMPI_Gcomm (CMPI_Gcomm has a type of __CMPI_Gcomm)
26 and MPI_Comm */
27struct CMPI_Gcomm {
28 $gcomm p2p; // point-to-point communication
29 $gcomm col; // collective communication
30 $gcollect_checker collect_checker;
31 $gbarrier gbarrier;
32};
33
34/****************************** Helper Functions **********************************/
35int sizeofDatatype(MPI_Datatype datatype) {
36 switch (datatype) {
37 case MPI_INT:
38 return sizeof(int);
39 case MPI_2INT:
40 return (sizeof(int)*2);
41 case MPI_FLOAT:
42 return sizeof(float);
43 case MPI_DOUBLE:
44 return sizeof(double);
45 case MPI_CHAR:
46 return sizeof(char);
47 case MPI_BYTE:
48 return sizeof(char); // char is always one byte ?
49 case MPI_SHORT:
50 return sizeof(short);
51 case MPI_LONG:
52 return sizeof(long);
53 case MPI_LONG_DOUBLE:
54 return sizeof(long double);
55 case MPI_LONG_LONG_INT:
56 return sizeof(long long int);
57 case MPI_LONG_LONG:
58 return sizeof(long long);
59 case MPI_UNSIGNED_LONG_LONG:
60 return sizeof(unsigned long long);
61 default:
62 $assert(0, "Unreachable");
63 }
64}
65
66/************************** MPI LIB Implementations *******************************/
67CMPI_Gcomm CMPI_Gcomm_create($scope scope, int size) {
68 CMPI_Gcomm result;
69
70 result.p2p = $gcomm_create(scope, size);
71 result.col = $gcomm_create(scope, size);
72 result.collect_checker = $gcollect_checker_create(scope);
73 result.gbarrier = $gbarrier_create(scope, size);
74 return result;
75}
76
77void CMPI_Gcomm_destroy(CMPI_Gcomm gc) {
78 $gcomm_destroy(gc.p2p);
79 $gcomm_destroy(gc.col);
80 $gcollect_checker_destroy(gc.collect_checker);
81 $gbarrier_destroy(gc.gbarrier);
82}
83
84MPI_Comm CMPI_Comm_create($scope scope, CMPI_Gcomm gc, int rank) {
85 MPI_Comm result;
86
87 result.p2p = $comm_create(scope, gc.p2p, rank);
88 result.col = $comm_create(scope, gc.col, rank);
89 result.collect_checker = $collect_checker_create(scope, gc.collect_checker);
90 result.barrier = $barrier_create(scope, gc.gbarrier, rank);
91 result.gcommIndex = 0;
92 return result;
93}
94
95void CMPI_Comm_destroy(MPI_Comm comm) {
96 __MPI_Sys_status__ curr_status;
97
98 curr_status = CMPI_Get_status();
99 if(comm.gcommIndex == 0)
100 $assert(curr_status == __FINALIZED, "Process terminates without "
101 "calling MPI_Finalize() first.");
102 $comm_destroy(comm.p2p);
103 $comm_destroy(comm.col);
104 $collect_checker_destroy(comm.collect_checker);
105 $barrier_destroy(comm.barrier);
106}
107
108int _MPI_Init(void) {
109 CMPI_Set_status(__INIT);
110 return 0;
111}
112
113int _MPI_Finalize(void) {
114 CMPI_Set_status(__FINALIZED);
115 return 0;
116}
117
118void * CMPI_PointerAdd(const void * ptr, int offset, MPI_Datatype datatype) {
119 int type_size = sizeofDatatype(datatype);
120
121 return $pointer_add(ptr, offset, type_size);
122}
123
124/********************* Lower level MPI routines *********************/
125/* CMPI_Send and CMPI_Recv are a pair of send receives functions that
126 help implementing MPI routines. They should never be block which
127 means no potential deadlocks related to these functions */
128int CMPI_Send(void *buf, int count, MPI_Datatype datatype, int dest,
129 int tag, $comm comm) {
130 if (dest >= 0) {
131 int size = count*sizeofDatatype(datatype);
132 int place = $comm_place(comm);
133 $message out = $message_pack(place, dest, tag, buf, size);
134 $comm_enqueue(comm, out);
135 }
136 return 0;
137}
138
139int CMPI_Recv(void *buf, int count, MPI_Datatype datatype, int source,
140 int tag, $comm comm, MPI_Status *status) {
141 if (source >= 0 || source == MPI_ANY_SOURCE) {
142 $message in = $comm_dequeue(comm, source, tag);
143 int size = count*sizeofDatatype(datatype);
144
145 $message_unpack(in, buf, size);
146 if (status != MPI_STATUS_IGNORE) {
147 status->size = $message_size(in);
148 status->MPI_SOURCE = $message_source(in);
149 status->MPI_TAG = $message_tag(in);
150 status->MPI_ERROR = 0;
151 }
152 }
153 return 0;
154}
155
156int CMPI_Sendrecv(const void *sendbuf, int sendcount, MPI_Datatype sendtype,
157 int dest, int sendtag, void *recvbuf, int recvcount,
158 MPI_Datatype recvtype, int source, int recvtag,
159 $comm comm, MPI_Status *status) {
160 //send and receive triggering flags
161 if((dest >= 0) && ((source >= 0 || source == MPI_ANY_SOURCE))) {
162 $message out, in;
163 int size = sendcount*sizeofDatatype(sendtype);
164 int place = $comm_place(comm);
165
166 out = $message_pack(place, dest, sendtag, sendbuf, size);
167 $choose {
168 $when (1){
169 $comm_enqueue(comm, out);
170 in = $comm_dequeue(comm, source, recvtag);
171 }
172 $when (1){
173 in = $comm_dequeue(comm, source, recvtag);
174 $comm_enqueue(comm, out);
175 }
176 }
177 size = recvcount*sizeofDatatype(recvtype);
178 $message_unpack(in, recvbuf, size);
179 if (status != MPI_STATUS_IGNORE) {
180 status->size = $message_size(in);
181 status->MPI_SOURCE = $message_source(in);
182 status->MPI_TAG = $message_tag(in);
183 status->MPI_ERROR = 0;
184 }
185 }
186 else if (dest >= 0) {
187 CMPI_Send(sendbuf, sendcount, sendtype, dest, sendtag, comm);
188 }
189 else if (source >= 0 || source == MPI_ANY_SOURCE) {
190 CMPI_Recv(recvbuf, recvcount, recvtype, source, recvtag, comm, status);
191 }
192 return 0;
193}
194
195/********************* Collective helper functions ********************/
196/* Note: collective helpers functions are functions have same
197 behaviors as MPI collective functions, it can be re-used as a part
198 of implementation by different MPI routines. For example,
199 MPI_Allreduce will call CMPI_Reduce and CMPI_Bcast, both of them
200 should throw errors (if encounters any) as if errors are thrown
201 from MPI_Allreduce.
202*/
203int CMPI_Collective_recv(void *buf, int count, MPI_Datatype datatype,
204 int source, int tag, $comm comm,
205 MPI_Status * status, char * routName) {
206 if(source >= 0 || source == MPI_ANY_SOURCE) {
207 $message in = $comm_dequeue(comm, source, MPI_ANY_TAG);
208 int size = count*sizeofDatatype(datatype);
209 int recvTag;
210
211 recvTag = $message_tag(in);
212 $assert (recvTag == tag , "Collective routine %s receives a "
213 "message with a mismatched tag\n", routName);
214 $message_unpack(in, buf, size);
215 if (status != MPI_STATUS_IGNORE) {
216 status->size = $message_size(in);
217 status->MPI_SOURCE = $message_source(in);
218 status->MPI_TAG = recvTag;
219 status->MPI_ERROR = 0;
220 }
221 }
222 return 0;
223}
224
225/* Broadcast helper function that uses any specified message tag */
226int CMPI_Bcast(void *buf, int count, MPI_Datatype datatype, int root, int tag,
227 MPI_Comm comm, char * routName) {
228 if ($comm_place(comm.col) == root) {
229 int nprocs = $comm_size(comm.col);
230
231 for (int i=0; i<nprocs; i++)
232 if (i != root)
233 CMPI_Send(buf, count, datatype, i, tag, comm.col);
234 } else
235 CMPI_Collective_recv(buf, count, datatype, root, tag, comm.col,
236 MPI_STATUS_IGNORE, routName);
237 return 0;
238}
239
240/* Reduction helper function that uses any specified message tag */
241int CMPI_Reduce(const void* sendbuf, void* recvbuf, int count,
242 MPI_Datatype datatype, MPI_Op op, int root, int tag,
243 MPI_Comm comm, char * routName) {
244 int rank;
245
246 rank = $comm_place(comm.col);
247 if (rank != root)
248 CMPI_Send(sendbuf, count, datatype, root, tag, comm.col);
249 else {
250 int nprocs = $comm_size(comm.col);
251 int size;
252
253 size = count * sizeofDatatype(datatype);
254 memcpy(recvbuf, sendbuf, size);
255 for (int i = 0; i<nprocs; i++) {
256 if(i != root){
257 int colTag;
258 $message in = $comm_dequeue(comm.col, i, MPI_ANY_TAG);
259
260 colTag = $message_tag(in);
261 $assert (colTag == tag , "Collective routine %s receives a "
262 "message with a mismatched tag\n", routName);
263 /* the third argument "count" indicates the number of cells needs doing the
264 operation. */
265 $bundle_unpack_apply(in.data, recvbuf, count, op);
266 $assert (in.size <= size ,
267 "Message of size %d exceeds the specified size %d.", in.size, size);
268 }
269 }
270 }
271 return 0;
272}
273
274/* Gathering helper function that uses any specified message tag */
275int CMPI_Gather(const void* sendbuf, int sendcount, MPI_Datatype sendtype,
276 void* recvbuf, int recvcount, MPI_Datatype recvtype,
277 int root, int tag, MPI_Comm comm, char * routName){
278 int rank, nprocs;
279 MPI_Status status;
280
281 rank = $comm_place(comm.col);
282 nprocs = $comm_size(comm.col);
283 /* MPI standard requirement:
284 * For root process, sendtype must be equal to
285 * recvtype. */
286 if(rank == root)
287 $assert (sendtype == recvtype,
288 "%s asks for equality "
289 "between 'sendtype' and 'recvtype'.", routName);
290 /* MPI_standard requirement:
291 * Only root process can use MPI_IN_PLACE*/
292 if(sendbuf == MPI_IN_PLACE){
293 $assert (root == rank,
294 "Only root can replace 'sendbuf' with 'MPI_IN_PLACE'.");
295 } else if(root == rank) {
296 void * ptr;
297
298 $assert(sendcount == recvcount, "Root process of routine %d without using"
299 " MPI_IN_PLACE should give the same value for recvcount and sendcount",
300 routName);
301 ptr = CMPI_PointerAdd(recvbuf, root * recvcount, recvtype);
302 memcpy(ptr, sendbuf, recvcount * sizeofDatatype(recvtype));
303 } else
304 CMPI_Send(sendbuf, sendcount, sendtype, root, tag, comm.col);
305 /* Root process receives messages and put them in right places */
306 if(rank == root){
307 int real_recvcount;
308 int offset;
309
310 for(int i=0; i<nprocs; i++){
311 if(i != root) {
312 void * ptr;
313
314 offset = i * recvcount;
315 ptr = CMPI_PointerAdd(recvbuf, offset, recvtype);
316 CMPI_Collective_recv(ptr, recvcount, recvtype,
317 i, tag, comm.col, &status, routName);
318 real_recvcount = status.size/sizeofDatatype(recvtype);
319 $assert(real_recvcount == recvcount,
320 "%s asks for equality between"
321 " the amount of data sent and the "
322 "amount of data received.", routName);
323 }
324 }
325 }
326 return 0;
327}
328
329int CMPI_Gatherv(const void* sendbuf, int sendcount, MPI_Datatype sendtype,
330 void* recvbuf, const int recvcounts[], const int displs[],
331 MPI_Datatype recvtype, int root, int tag,
332 MPI_Comm comm, char * routName){
333 int rank, nprocs;
334
335 rank = $comm_place(comm.col);
336 nprocs = $comm_size(comm.col);
337 /* MPI standard requirement:
338 * For root process, sendtype must be equal to
339 * recvtype. */
340 if(rank == root)
341 $assert(sendtype == recvtype, "%s asks for equality "
342 "between 'sendtype' and 'recvtype'.", routName);
343 /* MPI_standard requirement:
344 * Only root process can use MPI_IN_PLACE*/
345 if(sendbuf == MPI_IN_PLACE){
346 $assert(root == rank, "Only root can replace 'sendbuf' with 'MPI_IN_PLACE'.");
347 }else if(root == rank) {
348 void * ptr;
349
350 $assert(sendcount == recvcounts[root], "For routine %s, recvcounts[%d] "
351 "should be same as the sendcount of the process with rank %d.\n",
352 routName, root, root);
353 ptr = CMPI_PointerAdd(recvbuf, displs[rank], recvtype);
354 memcpy(ptr, sendbuf, sendcount * sizeofDatatype(recvtype));
355 }else{
356 CMPI_Send(sendbuf, sendcount, sendtype, root, tag, comm.col);
357 }
358 /* Root process receives messages and put them in right places */
359 if(rank == root){
360 int real_recvcount;
361 MPI_Status status;
362
363 for(int i=0; i<nprocs; i++){
364 if(i != root){
365 void * ptr = CMPI_PointerAdd(recvbuf, displs[i], recvtype);
366
367 CMPI_Collective_recv(ptr, recvcounts[i],
368 recvtype, i, tag, comm.col, &status, routName);
369 real_recvcount = status.size/sizeofDatatype(recvtype);
370 $assert(real_recvcount == recvcounts[i], "%s asks for equality between"
371 " the amount of data sent and the "
372 "amount of data received.", routName);
373 }
374 }
375 }
376 return 0;
377}
378
379/* Scatter helper function that uses any specified message tag */
380int CMPI_Scatter(const void* sendbuf, int sendcount, MPI_Datatype sendtype,
381 void* recvbuf, int recvcount, MPI_Datatype recvtype, int root,
382 int tag, MPI_Comm comm, char * routName){
383 int rank, nprocs;
384
385 rank = $comm_place(comm.col);
386 nprocs = $comm_size(comm.col);
387 /* MPI standard requirement:
388 * For root process, sendtype must be equal to
389 * recvtype. */
390 if(rank == root)
391 $assert(sendtype == recvtype, "MPI_Scatter() asks for equality "
392 "between 'sendtype' and 'recvtype'.");
393 /* MPI_standard requirement:
394 * Only root process can use MPI_IN_PLACE */
395 if(recvbuf == MPI_IN_PLACE){
396 $assert(root == rank, "Only root can replace 'recvbuf' with 'MPI_IN_PLACE'.");
397 }else if(rank == root) {
398 void * ptr;
399
400 $assert(sendcount == recvcount, "Root process of routine %d without using"
401 " MPI_IN_PLACE should give the same value for recvcount and sendcount",
402 routName);
403 ptr = CMPI_PointerAdd(sendbuf, root*recvcount, sendtype);
404 memcpy(recvbuf, ptr, sizeofDatatype(recvtype)*recvcount);
405 }
406 /* Root process scatters data to other processes */
407 if(rank == root){
408 int offset;
409
410 for(int i=0; i<nprocs; i++){
411 if(i != root) {
412 void * ptr;
413
414 offset = i * sendcount;
415 ptr = CMPI_PointerAdd(sendbuf, offset, sendtype);
416 CMPI_Send(ptr, sendcount, sendtype, i, tag, comm.col);
417 }
418 }
419 }
420 /* Non-root processes receive data */
421 if(!(root == rank)){
422 int real_recvcount;
423 MPI_Status status;
424
425 CMPI_Collective_recv(recvbuf, recvcount, recvtype,
426 root, tag, comm.col, &status, routName);
427 real_recvcount = status.size/sizeofDatatype(recvtype);
428 $assert(real_recvcount == recvcount,
429 "%s asks for equality between"
430 " the amount of data sent and the "
431 "amount of data received.", routName);
432 }
433 return 0;
434}
435
436/* Scatterv helper function that uses any specified message tag */
437int CMPI_Scatterv(const void* sendbuf, const int sendcounts[], const
438 int displs[], MPI_Datatype sendtype, void* recvbuf,
439 int recvcount, MPI_Datatype recvtype, int root, int tag,
440 MPI_Comm comm, char * routName){
441 int rank, nprocs;
442
443 rank = $comm_place(comm.col);
444 nprocs = $comm_size(comm.col);
445 /* MPI standard requirement:
446 * For root process, sendtype must be equal to
447 * recvtype. */
448 if(rank == root)
449 $assert(sendtype == recvtype, "%s asks for equality "
450 "between 'sendtype' and 'recvtype'.", routName);
451 /* MPI_standard requirement:
452 * Only root process can use MPI_IN_PLACE */
453 if(recvbuf == MPI_IN_PLACE){
454 $assert(root == rank, "Only root can replace 'recvbuf' with 'MPI_IN_PLACE'.");
455 } else if(rank == root) {
456 void * ptr;
457
458 $assert(sendcounts[root] == recvcount, "For routine %s, sendcounts[%d] "
459 "should be same as the recvcount of the process with rank %d.\n",
460 routName, root, root);
461 ptr = CMPI_PointerAdd(sendbuf, displs[root], sendtype);
462 memcpy(recvbuf, ptr, recvcount*sizeofDatatype(recvtype));
463 }
464 /* Root process scatters data to other processes */
465 if(rank == root){
466 for(int i=0; i<nprocs; i++){
467 if(i != root) {
468 void * ptr = CMPI_PointerAdd(sendbuf, displs[i], sendtype);
469
470 CMPI_Send(ptr, sendcounts[i], sendtype, i,
471 tag, comm.col);
472 }
473 }
474 }
475 if(!(root == rank)){
476 MPI_Status status;
477 int real_recvcount;
478
479 CMPI_Collective_recv(recvbuf, recvcount, recvtype,
480 root, tag, comm.col, &status, routName);
481 real_recvcount = status.size/sizeofDatatype(recvtype);
482 $assert(real_recvcount == recvcount, "Process rank:%d\n%s asks for equality between"
483 " the amount of data sent (%d) and the "
484 "amount of data received (%d).", rank, routName, real_recvcount, recvcount);
485 }
486 return 0;
487}
488
489int CMPI_Comm_dup($scope scope, MPI_Comm comm, MPI_Comm * newcomm, char * routName) {
490 int place = $comm_place(comm.col);
491 CMPI_Gcomm newgcomm;
492 int idx;
493 $scope CMPI_ROOT_SCOPE = CMPI_Root_scope(comm.col);
494
495 if(place == 0) {
496 int size = $comm_size(comm.col);
497
498 newgcomm = CMPI_Gcomm_create(CMPI_ROOT_SCOPE, size);
499 idx = CMPI_NewGcomm(CMPI_ROOT_SCOPE, newgcomm);
500 }
501 CMPI_Bcast(&idx, 1, MPI_INT, 0, COMMDUP_TAG,
502 comm, routName);
503 newgcomm = CMPI_GetGcomm(CMPI_ROOT_SCOPE, idx);
504 (*newcomm) = CMPI_Comm_create(scope, newgcomm, place);
505 newcomm->gcommIndex = idx;
506 $barrier_call(comm.barrier);
507 $gcomm_dup(comm.p2p, newcomm->p2p);
508 $gcomm_dup(comm.col, newcomm->col);
509 $barrier_call(comm.barrier);
510 return 0;
511}
512
513int CMPI_Comm_free(MPI_Comm * comm) {
514 int place = $comm_place(comm->col);
515 int size = $comm_size(comm->col);
516 int buf[size];
517 int gcommIndex = comm->gcommIndex;
518 $scope CMPI_ROOT_SCOPE = CMPI_Root_scope(comm->col);
519
520 //TODO: CMPI_Gather here is just a ugly synchronization
521 CMPI_Gather(&place, 1, MPI_INT, buf, 1, MPI_INT, 0,
522 COMMFREE_TAG, (*comm), "MPI_Comm_free synchronization.");
523 CMPI_Comm_destroy(*comm);
524 if(place == 0) {
525 CMPI_Gcomm temp = CMPI_GetGcomm(CMPI_ROOT_SCOPE, gcommIndex);
526
527 CMPI_Gcomm_destroy(temp);
528 }
529 return 0;
530}
531
532#endif
533
Note: See TracBrowser for help on using the repository browser.