source: CIVL/src/include/civl/civl-mpi.cvl@ 7d546e9

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

re-naming shapshot functions and add comments

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