source: CIVL/src/include/civl/civl-mpi.cvl@ 15298b6

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

merged in the contract branch

git-svn-id: ​svn://vsl.cis.udel.edu/civl/trunk@5159 fb995dde-84ed-4084-dfe6-e5aef3e2452c
  • Property mode set to 100644
File size: 29.0 KB
Line 
1#ifndef __CIVL_CIVLMPI__
2#define __CIVL_CIVLMPI__
3
4#include <civlc.cvh>
5#include <concurrency.cvh>
6#include <collate.cvh>
7#include <comm.cvh>
8#include <bundle.cvh>
9#include <mpi.h>
10#include <civl-mpi.cvh>
11#include <string.h>
12#include <pointer.cvh>
13#include <seq.cvh>
14#include <stdlib.h>
15#include <stdio.h>
16
17
18/* Library private helper function declaration */
19char * $mpi_coroutine_name(int tag);
20
21/**************************** Duplicated Part *************************************/
22/* Duplicated definition with the same struct in mpi.h.
23 The reason of this duplication is to make civlmpi.cvl
24 independent with mpi.cvl. */
25typedef struct MPI_Comm {
26 $comm p2p; // point-to-point communication
27 $comm col; // collective communication
28 $collator collator;
29 $barrier barrier;
30 int gcommIndex; //the index of the corresponding global communicator.
31}MPI_Comm;
32
33/* Definition of CMPI_Gcomm (CMPI_Gcomm has a type of __CMPI_Gcomm)
34 and MPI_Comm */
35struct $mpi_gcomm {
36 $gcomm p2p; // point-to-point communication
37 $gcomm col; // collective communication
38 $gcollator gcollator;
39 $gbarrier gbarrier;
40};
41
42/****************************** Helper Functions **********************************/
43$state_f size_t $mpi_extentof(MPI_Datatype datatype) {
44 $abstract size_t $mpi_sizeof(MPI_Datatype datatype);
45
46 return $mpi_sizeof(datatype);
47}
48
49int sizeofDatatype(MPI_Datatype datatype) {
50 size_t result;
51
52#ifdef _MPI_CONTRACT
53 /* In MPI contract mode, it's possible that datatype is
54 non-concrete. Then use an abstract function to represent the
55 extent of the datatype. */
56 if (!$is_concrete_int(datatype))
57 return $mpi_extentof(datatype);
58#endif
59
60 switch (datatype) {
61 case MPI_INT:
62 result = sizeof(int); break;
63 case MPI_2INT:
64 result = (sizeof(int)*2); break;
65 case MPI_FLOAT:
66 result = sizeof(float); break;
67 case MPI_DOUBLE:
68 result = sizeof(double); break;
69 case MPI_CHAR:
70 result = sizeof(char); break;
71 case MPI_BYTE:
72 result = sizeof(char); break; // char is always one byte ?
73 case MPI_SHORT:
74 result = sizeof(short); break;
75 case MPI_LONG:
76 result = sizeof(long); break;
77 case MPI_LONG_DOUBLE:
78 result = sizeof(long double); break;
79 case MPI_LONG_LONG_INT:
80 result = sizeof(long long int); break;
81 case MPI_LONG_LONG:
82 result = sizeof(long long); break;
83 case MPI_UNSIGNED_LONG_LONG:
84 result = sizeof(unsigned long long); break;
85 default:
86 $assert(0, "Unreachable");
87 }
88#ifdef _MPI_CONTRACT
89 /* In the case that a datatype X which is previously non-concrete
90 and whose extent $mpi_sizeof(X) was used. Later X is simplified
91 to a concrete number representing 'int', we need somehow
92 associate SIZEOF_INT to $mpi_sizeof(X) by assume the following
93 predicate. */
94 $assume($mpi_extentof(datatype) == result);
95#endif
96 return result;
97}
98
99void * $mpi_malloc(int count, MPI_Datatype datatype) {
100 switch (datatype) {
101 case MPI_INT:
102 case MPI_SHORT:
103 case MPI_LONG:
104 case MPI_LONG_LONG_INT:
105 case MPI_LONG_LONG:
106 case MPI_UNSIGNED_LONG_LONG:
107 return (int *)malloc(sizeof(int) * count);
108 case MPI_2INT:
109 return (int *)malloc(2 * count * sizeof(int));
110 case MPI_FLOAT:
111 case MPI_DOUBLE:
112 case MPI_LONG_DOUBLE:
113 return ($real *)malloc(count * sizeof($real));
114 case MPI_CHAR:
115 case MPI_BYTE:
116 return (char *)malloc(count * sizeof(char));
117 default:
118 $assert(0, "Unreachable");
119 }
120 return (void *)0;
121}
122
123/************************** MPI LIB Implementations *******************************/
124$mpi_gcomm $mpi_gcomm_create($scope scope, int size) {
125 $mpi_gcomm result;
126
127 result.p2p = $gcomm_create(scope, size);
128 result.col = $gcomm_create(scope, size);
129 result.gcollator = $gcollator_create(scope, size);
130 result.gbarrier = $gbarrier_create(scope, size);
131 return result;
132}
133
134void $mpi_gcomm_destroy($mpi_gcomm gc) {
135 /* This function will report errors for any messages remaining the
136 $mpi_gcomm. Those messages are junk messages. */
137 int numJunkRecord;
138 int numJunkMsg;
139 $message junkMsgs[]; // A CIVL-C sequence for junk messages.
140
141 $seq_init(&junkMsgs, 0, NULL);
142 numJunkMsg = $gcomm_destroy(gc.p2p, &junkMsgs);
143 /* Informations of reporting junk messages in p2p communicator and
144 collective communicator are different: */
145 for(int i = 0; i < numJunkMsg; i++) {
146 int src, dest, tag;
147
148 src = $message_source(junkMsgs[i]);
149 dest = $message_dest(junkMsgs[i]);
150 tag = $message_tag(junkMsgs[i]);
151 $assert($false, "MPI message leak: There is a message from rank %d to rank %d with tag %d "
152 "has been sent but is never received in point-to-point communication.",
153 src, dest, tag);
154 }
155 numJunkMsg = $gcomm_destroy(gc.col, &junkMsgs);
156 for(int i = 0; i < numJunkMsg; i++) {
157 int src, tag;
158 char * routine;
159
160 src = $message_source(junkMsgs[i]);
161 tag = $message_tag(junkMsgs[i]);
162 routine = $mpi_coroutine_name(tag);
163 $assert($false, "MPI message leak: There is a message sent by rank %d for collective routine %s"
164 " that is never received.",
165 src, routine);
166 }
167 $gcollator_destroy(gc.gcollator);
168 $gbarrier_destroy(gc.gbarrier);
169}
170
171MPI_Comm $mpi_comm_create($scope scope, $mpi_gcomm gc, int rank) {
172 MPI_Comm result;
173
174 result.p2p = $comm_create(scope, gc.p2p, rank);
175 result.col = $comm_create(scope, gc.col, rank);
176 result.collator = $collator_create(gc.gcollator, scope, rank);
177 result.barrier = $barrier_create(scope, gc.gbarrier, rank);
178 result.gcommIndex = 0;
179 return result;
180}
181
182void $mpi_comm_destroy(MPI_Comm comm, $mpi_state mpi_state) {
183#ifndef _MPI_CONTRACT
184 if(comm.gcommIndex == 0)
185 $assert(mpi_state == _MPI_FINALIZED, "Process terminates without "
186 "calling MPI_Finalize() first.");
187#endif
188 $comm_destroy(comm.p2p);
189 $comm_destroy(comm.col);
190 $free(comm.collator);
191 $barrier_destroy(comm.barrier);
192}
193
194void * $mpi_pointer_add(const void * ptr, int offset, MPI_Datatype datatype) {
195#ifdef _MPI_CONTRACT
196 if (!$is_concrete_int(datatype)) {
197 int datatypeExtent = $mpi_extentof(datatype);
198
199 return $pointer_add(ptr, offset * datatypeExtent, sizeof(char));
200 }
201#endif
202
203 int type_size = sizeofDatatype(datatype);
204
205 return $pointer_add(ptr, offset, type_size);
206}
207
208/********************* Lower level MPI routines *********************/
209/* CMPI_Send and CMPI_Recv are a pair of send receives functions that
210 help implementing MPI routines. They should never be block which
211 means no potential deadlocks related to these functions */
212int $mpi_send(const void *buf, int count, MPI_Datatype datatype, int dest,
213 int tag, MPI_Comm comm) {
214 if (dest >= 0) {
215 int size = count*sizeofDatatype(datatype);
216 int place = $comm_place(comm.p2p);
217 $message out;
218
219 $elaborate(dest);
220 out = $message_pack(place, dest, tag, buf, size);
221 $comm_enqueue(comm.p2p, out);
222 }
223 return 0;
224}
225
226int $mpi_recv(void *buf, int count, MPI_Datatype datatype, int source,
227 int tag, MPI_Comm comm, MPI_Status *status) {
228 if ((source >= 0 || source == MPI_ANY_SOURCE)) {
229 $message in;
230 int place = $comm_place(comm.p2p);
231 int deterministicTag;
232
233 $assert(tag == -2 || tag >= 0, "Illegal MPI message receive tag %d.\n", tag);
234 deterministicTag = tag < 0 ? -2 : tag;
235 $elaborate(source);
236 in = $comm_dequeue(comm.p2p, source, deterministicTag);
237
238 int size = count*sizeofDatatype(datatype);
239
240 $message_unpack(in, buf, size);
241 if (status != MPI_STATUS_IGNORE) {
242 status->size = $message_size(in);
243 status->MPI_SOURCE = $message_source(in);
244 status->MPI_TAG = $message_tag(in);
245 status->MPI_ERROR = 0;
246 }
247 }
248 return 0;
249}
250
251int $mpi_sendrecv(const void *sendbuf, int sendcount, MPI_Datatype sendtype,
252 int dest, int sendtag, void *recvbuf, int recvcount,
253 MPI_Datatype recvtype, int source, int recvtag,
254 MPI_Comm comm, MPI_Status *status) {
255 int deterministicRecvTag;
256
257 $assert(sendtag >= 0, "MPI sendtag should be greater than or equal to zero");
258 $assert(recvtag == -2 || recvtag >= 0, "Illegal MPI message receive tag %d.\n", recvtag);
259
260 deterministicRecvTag = recvtag < 0 ? -2 : recvtag;
261 if((dest >= 0) && ((source >= 0 || source == MPI_ANY_SOURCE))) {
262 $message out, in;
263 int size = sendcount*sizeofDatatype(sendtype);
264 int place = $comm_place(comm.p2p);
265
266 out = $message_pack(place, dest, sendtag, sendbuf, size);
267 $elaborate(source);
268 $elaborate(dest);
269 $choose {
270 $when($true){
271 $comm_enqueue(comm.p2p, out);
272 in = $comm_dequeue(comm.p2p, source, deterministicRecvTag);
273 }
274 $when($false){
275 /* This $choose branch plays a trick which correctly
276 implements the sendrecv() semantically. Such a branch
277 ensures that there is no chance of potential deadlocks when
278 all processes do send then recv collectively. However,
279 effectively, this branch is no need and never will be
280 executed.*/
281 in = $comm_dequeue(comm.p2p, source, deterministicRecvTag);
282 $comm_enqueue(comm.p2p, out);
283 }
284 }
285 size = recvcount*sizeofDatatype(recvtype);
286 $message_unpack(in, recvbuf, size);
287 if (status != MPI_STATUS_IGNORE) {
288 status->size = $message_size(in);
289 status->MPI_SOURCE = $message_source(in);
290 status->MPI_TAG = $message_tag(in);
291 status->MPI_ERROR = 0;
292 }
293 }
294 else if (dest >= 0)
295 $mpi_send(sendbuf, sendcount, sendtype, dest, sendtag, comm);
296 else if (source >= 0 || source == MPI_ANY_SOURCE)
297 $mpi_recv(recvbuf, recvcount, recvtype, source, deterministicRecvTag, comm, status);
298 return 0;
299}
300
301/********************* Collective helper functions ********************/
302/* Note: collective helpers functions are functions have same
303 behaviors as MPI collective functions, it can be re-used as a part
304 of implementation by different MPI routines. For example,
305 MPI_Allreduce will call CMPI_Reduce and CMPI_Bcast, both of them
306 should throw errors (if encounters any) as if errors are thrown
307 from MPI_Allreduce.
308*/
309int $mpi_collective_send(const void *buf, int count, MPI_Datatype datatype, int dest,
310 int tag, MPI_Comm comm) {
311 if (dest >= 0) {
312 int size = count*sizeofDatatype(datatype);
313 int place = $comm_place(comm.col);
314 $message out = $message_pack(place, dest, tag, buf, size);
315
316 $elaborate(dest);
317 $comm_enqueue(comm.col, out);
318 }
319 return 0;
320}
321
322int $mpi_collective_recv(void *buf, int count, MPI_Datatype datatype,
323 int source, int tag, MPI_Comm comm,
324 MPI_Status * status, char * routName) {
325 if(source >= 0 || source == MPI_ANY_SOURCE) {
326 $elaborate(source);
327 $message in = $comm_dequeue(comm.col, source, MPI_ANY_TAG);
328 int size = count*sizeofDatatype(datatype);
329 int recvTag;
330
331 /* This routine should only be used by collective routines, there
332 is no non-deterministic tags for collective routines.*/
333 recvTag = $message_tag(in);
334 $assert (recvTag == tag, "Collective routine %s receives a "
335 "message with a mismatched tag\n", routName);
336 $message_unpack(in, buf, size);
337 if (status != MPI_STATUS_IGNORE) {
338 status->size = $message_size(in);
339 status->MPI_SOURCE = $message_source(in);
340 status->MPI_TAG = recvTag;
341 status->MPI_ERROR = 0;
342 }
343 }
344 return 0;
345}
346
347/* Broadcast helper function that uses any specified message tag */
348int $mpi_bcast(void *buf, int count, MPI_Datatype datatype, int root, int tag,
349 MPI_Comm comm, char * routName) {
350 if ($comm_place(comm.col) == root) {
351 int nprocs = $comm_size(comm.col);
352
353 for (int i=0; i<nprocs; i++)
354 if (i != root)
355 $mpi_collective_send(buf, count, datatype, i, tag, comm);
356 } else
357 $mpi_collective_recv(buf, count, datatype, root, tag, comm,
358 MPI_STATUS_IGNORE, routName);
359 return 0;
360}
361
362/* Reduction helper function that uses any specified message tag */
363int $mpi_reduce(const void* sendbuf, void* recvbuf, int count,
364 MPI_Datatype datatype, MPI_Op op, int root, int tag,
365 MPI_Comm comm, char * routName) {
366 int rank;
367
368 rank = $comm_place(comm.col);
369 if (rank != root)
370 $mpi_collective_send(sendbuf, count, datatype, root, tag, comm);
371 else {
372 int nprocs = $comm_size(comm.col);
373 int size;
374
375 size = count * sizeofDatatype(datatype);
376 memcpy(recvbuf, sendbuf, size);
377 for (int i = 0; i<nprocs; i++) {
378 if(i != root){
379 int colTag;
380 void * applybuf;
381 $message in = $comm_dequeue(comm.col, i, MPI_ANY_TAG);
382
383 /* Collective routines have no non-deterministic tags.*/
384 colTag = $message_tag(in);
385 $assert (colTag == tag , "Collective routine %s receives a "
386 "message with a mismatched tag\n", routName);
387 /* the third argument "count" indicates the number of cells needs doing the
388 operation. */
389 applybuf = $mpi_malloc(count, datatype);
390 $bundle_unpack_apply(in.data, recvbuf, op, count, applybuf);
391 memcpy(recvbuf, applybuf, size);
392 free(applybuf);
393 $assert (in.size <= size ,
394 "Message of size %d exceeds the specified size %d.", in.size, size);
395 }
396 }
397 }
398 return 0;
399}
400
401/* Gathering helper function that uses any specified message tag */
402int $mpi_gather(const void* sendbuf, int sendcount, MPI_Datatype sendtype,
403 void* recvbuf, int recvcount, MPI_Datatype recvtype,
404 int root, int tag, MPI_Comm comm, char * routName){
405 int rank, nprocs;
406 MPI_Status status;
407
408 rank = $comm_place(comm.col);
409 nprocs = $comm_size(comm.col);
410 /* MPI standard requirement:
411 * For root process, sendtype must be equal to
412 * recvtype. */
413 if(rank == root)
414 $assert (sendtype == recvtype,
415 "%s asks for equality "
416 "between 'sendtype' and 'recvtype'.", routName);
417 /* MPI_standard requirement:
418 * Only root process can use MPI_IN_PLACE*/
419 if(sendbuf == MPI_IN_PLACE){
420 $assert (root == rank,
421 "Only root can replace 'sendbuf' with 'MPI_IN_PLACE'.");
422 } else if(root == rank) {
423 void * ptr;
424
425 $assert(sendcount == recvcount, "Root process of routine %d without using"
426 " MPI_IN_PLACE should give the same value for recvcount and sendcount",
427 routName);
428 ptr = $mpi_pointer_add(recvbuf, root * recvcount, recvtype);
429 memcpy(ptr, sendbuf, recvcount * sizeofDatatype(recvtype));
430 } else
431 $mpi_collective_send(sendbuf, sendcount, sendtype, root, tag, comm);
432 /* Root process receives messages and put them in right places */
433 if(rank == root){
434 int real_recvcount;
435 int offset;
436
437 for(int i=0; i<nprocs; i++){
438 if(i != root) {
439 void * ptr;
440
441 offset = i * recvcount;
442 ptr = $mpi_pointer_add(recvbuf, offset, recvtype);
443 $mpi_collective_recv(ptr, recvcount, recvtype,
444 i, tag, comm, &status, routName);
445 real_recvcount = status.size/sizeofDatatype(recvtype);
446 $assert(real_recvcount == recvcount,
447 "%s asks for equality between"
448 " the amount of data sent and the "
449 "amount of data received.", routName);
450 }
451 }
452 }
453 return 0;
454}
455
456int $mpi_gatherv(const void* sendbuf, int sendcount, MPI_Datatype sendtype,
457 void* recvbuf, const int recvcounts[], const int displs[],
458 MPI_Datatype recvtype, int root, int tag,
459 MPI_Comm comm, char * routName){
460 int rank, nprocs;
461
462 rank = $comm_place(comm.col);
463 nprocs = $comm_size(comm.col);
464 /* MPI standard requirement:
465 * For root process, sendtype must be equal to
466 * recvtype. */
467 if(rank == root)
468 $assert(sendtype == recvtype, "%s asks for equality "
469 "between 'sendtype' and 'recvtype'.", routName);
470 /* MPI_standard requirement:
471 * Only root process can use MPI_IN_PLACE*/
472 if(sendbuf == MPI_IN_PLACE){
473 $assert(root == rank, "Only root can replace 'sendbuf' with 'MPI_IN_PLACE'.");
474 }else if(root == rank) {
475 void * ptr;
476
477 $assert(sendcount == recvcounts[root], "For routine %s, recvcounts[%d] "
478 "should be same as the sendcount of the process with rank %d.\n",
479 routName, root, root);
480 ptr = $mpi_pointer_add(recvbuf, displs[rank], recvtype);
481 memcpy(ptr, sendbuf, sendcount * sizeofDatatype(recvtype));
482 }else{
483 $mpi_collective_send(sendbuf, sendcount, sendtype, root, tag, comm);
484 }
485 /* Root process receives messages and put them in right places */
486 if(rank == root){
487 int real_recvcount;
488 MPI_Status status;
489
490 for(int i=0; i<nprocs; i++){
491 if(i != root){
492 void * ptr = $mpi_pointer_add(recvbuf, displs[i], recvtype);
493
494 $mpi_collective_recv(ptr, recvcounts[i],
495 recvtype, i, tag, comm, &status, routName);
496 real_recvcount = status.size/sizeofDatatype(recvtype);
497 $assert(real_recvcount == recvcounts[i], "%s asks for equality between"
498 " the amount of data sent and the "
499 "amount of data received.", routName);
500 }
501 }
502 }
503 return 0;
504}
505
506/* Scatter helper function that uses any specified message tag */
507int $mpi_scatter(const void* sendbuf, int sendcount, MPI_Datatype sendtype,
508 void* recvbuf, int recvcount, MPI_Datatype recvtype, int root,
509 int tag, MPI_Comm comm, char * routName){
510 int rank, nprocs;
511
512 rank = $comm_place(comm.col);
513 nprocs = $comm_size(comm.col);
514 /* MPI standard requirement:
515 * For root process, sendtype must be equal to
516 * recvtype. */
517 if(rank == root)
518 $assert(sendtype == recvtype, "MPI_Scatter() asks for equality "
519 "between 'sendtype' and 'recvtype'.");
520 /* MPI_standard requirement:
521 * Only root process can use MPI_IN_PLACE */
522 if(recvbuf == MPI_IN_PLACE){
523 $assert(root == rank, "Only root can replace 'recvbuf' with 'MPI_IN_PLACE'.");
524 }else if(rank == root) {
525 void * ptr;
526
527 $assert(sendcount == recvcount, "Root process of routine %d without using"
528 " MPI_IN_PLACE should give the same value for recvcount and sendcount",
529 routName);
530 ptr = $mpi_pointer_add(sendbuf, root*recvcount, sendtype);
531 memcpy(recvbuf, ptr, sizeofDatatype(recvtype)*recvcount);
532 }
533 /* Root process scatters data to other processes */
534 if(rank == root){
535 int offset;
536
537 for(int i=0; i<nprocs; i++){
538 if(i != root) {
539 void * ptr;
540
541 offset = i * sendcount;
542 ptr = $mpi_pointer_add(sendbuf, offset, sendtype);
543 $mpi_collective_send(ptr, sendcount, sendtype, i, tag, comm);
544 }
545 }
546 }
547 /* Non-root processes receive data */
548 if(!(root == rank)){
549 int real_recvcount;
550 MPI_Status status;
551
552 $mpi_collective_recv(recvbuf, recvcount, recvtype,
553 root, tag, comm, &status, routName);
554 real_recvcount = status.size/sizeofDatatype(recvtype);
555 $assert(real_recvcount == recvcount,
556 "%s asks for equality between"
557 " the amount of data sent and the "
558 "amount of data received.", routName);
559 }
560 return 0;
561}
562
563/* Scatterv helper function that uses any specified message tag */
564int $mpi_scatterv(const void* sendbuf, const int sendcounts[], const
565 int displs[], MPI_Datatype sendtype, void* recvbuf,
566 int recvcount, MPI_Datatype recvtype, int root, int tag,
567 MPI_Comm comm, char * routName){
568 int rank, nprocs;
569
570 rank = $comm_place(comm.col);
571 nprocs = $comm_size(comm.col);
572 /* MPI standard requirement:
573 * For root process, sendtype must be equal to
574 * recvtype. */
575 if(rank == root)
576 $assert(sendtype == recvtype, "%s asks for equality "
577 "between 'sendtype' and 'recvtype'.", routName);
578 /* MPI_standard requirement:
579 * Only root process can use MPI_IN_PLACE */
580 if(recvbuf == MPI_IN_PLACE){
581 $assert(root == rank, "Only root can replace 'recvbuf' with 'MPI_IN_PLACE'.");
582 } else if(rank == root) {
583 void * ptr;
584
585 $assert(sendcounts[root] == recvcount, "For routine %s, sendcounts[%d] "
586 "should be same as the recvcount of the process with rank %d.\n",
587 routName, root, root);
588 ptr = $mpi_pointer_add(sendbuf, displs[root], sendtype);
589 memcpy(recvbuf, ptr, recvcount*sizeofDatatype(recvtype));
590 }
591 /* Root process scatters data to other processes */
592 if(rank == root){
593 for(int i=0; i<nprocs; i++){
594 if(i != root) {
595 void * ptr = $mpi_pointer_add(sendbuf, displs[i], sendtype);
596
597 $mpi_collective_send(ptr, sendcounts[i], sendtype, i,
598 tag, comm);
599 }
600 }
601 }
602 if(!(root == rank)){
603 MPI_Status status;
604 int real_recvcount;
605
606 $mpi_collective_recv(recvbuf, recvcount, recvtype,
607 root, tag, comm, &status, routName);
608 real_recvcount = status.size/sizeofDatatype(recvtype);
609 $assert(real_recvcount == recvcount, "Process rank:%d\n%s asks for equality between"
610 " the amount of data sent (%d) and the "
611 "amount of data received (%d).", rank, routName, real_recvcount, recvcount);
612 }
613 return 0;
614}
615
616/* The helper function for (inclusive) MPI_Scan */
617int $mpi_scan(const void* sendbuf, void* recvbuf, int count,
618 MPI_Datatype datatype, MPI_Op op, MPI_Comm comm) {
619 int place, nprocs;
620 int size = count * sizeofDatatype(datatype);
621 void * tmp_space = $mpi_malloc(count, datatype);
622 void * tmp_ptr, * tmp_space_const = tmp_space;
623
624 place = $comm_place(comm.col);
625 nprocs = $comm_size(comm.col);
626 /* Each process do reduction from 0 .. rank (inclusive):
627 * Send to process rank + 1 ... nprocs-1 (inclusive)
628 * Recv from process 0 .. rank (exclusive)
629 */
630 for (int i = place + 1; i < nprocs; i++)
631 $mpi_collective_send(sendbuf, count, datatype, i, SCAN_TAG, comm);
632 if (sendbuf != MPI_IN_PLACE)
633 memcpy(recvbuf, sendbuf, size);
634 for (int i = 0; i < place; i++) {
635 $message in = $comm_dequeue(comm.col, i, SCAN_TAG);
636
637 // so far, unpack_apply requires that 'recvbuf' is not aliasing
638 // 'tmp_space' (can be improved in the future):
639 $bundle_unpack_apply(in.data, recvbuf, op, count, tmp_space);
640 // swap tmp_space with recvbuf
641 tmp_ptr = tmp_space;
642 tmp_space = recvbuf;
643 recvbuf = tmp_ptr;
644 $assert (in.size <= size ,
645 "Message of size %d exceeds the specified size %d.", in.size, size);
646 }
647 if (recvbuf == tmp_space_const)
648 memcpy(tmp_space, recvbuf, size);
649 free(tmp_space_const);
650 return 0;
651}
652
653/* The helper function for (the exclusive) MPI_Exscan */
654int $mpi_exscan(const void* sendbuf, void* recvbuf, int count,
655 MPI_Datatype datatype, MPI_Op op, MPI_Comm comm) {
656 int place, nprocs;
657 int size = count * sizeofDatatype(datatype);
658 void * tmp_space = $mpi_malloc(count, datatype);
659 void * tmp_ptr, * tmp_space_const = tmp_space;
660
661 place = $comm_place(comm.col);
662 nprocs = $comm_size(comm.col);
663 /* The “in place” option for intracommunicators is specified by
664 * passing MPI_IN_PLACE in the sendbuf argument. In this case, the
665 * input data is taken from the receive buffer, and replaced by the
666 * output data. The receive buffer on rank 0 is not changed by this
667 * operation. */
668 if (sendbuf == MPI_IN_PLACE)
669 sendbuf = recvbuf;
670 /* Each process do reduction from 0 .. rank (inclusive):
671 * Send to process rank + 1 ... nprocs-1 (inclusive)
672 * Recv from process 0 .. rank (exclusive)
673 */
674 for (int i = place + 1; i < nprocs; i++)
675 $mpi_collective_send(sendbuf, count, datatype, i, EXSCAN_TAG, comm);
676 // no-op for rank 0
677 if (place != 0) {
678 $message in = $comm_dequeue(comm.col, 0, EXSCAN_TAG);
679 $bundle_unpack(in.data, recvbuf);
680 for (int i = 1; i < place; i++) {
681 in = $comm_dequeue(comm.col, i, EXSCAN_TAG);
682 // so far, unpack_apply requires that 'recvbuf' is not aliasing
683 // 'tmp_space' (can be improved in the future):
684 $bundle_unpack_apply(in.data, recvbuf, op, count, tmp_space);
685 // swap tmp_space with recvbuf
686 tmp_ptr = tmp_space;
687 tmp_space = recvbuf;
688 recvbuf = tmp_ptr;
689 $assert (in.size <= size ,
690 "Message of size %d exceeds the specified size %d.", in.size, size);
691 }
692 }
693 if (recvbuf == tmp_space_const)
694 memcpy(tmp_space, recvbuf, size);
695 free(tmp_space_const);
696 return 0;
697}
698
699
700/* ******************** End of collective routines ********************* */
701
702int $mpi_comm_dup($scope scope, MPI_Comm comm, MPI_Comm * newcomm, char * routName) {
703 int place = $comm_place(comm.col);
704 $mpi_gcomm newgcomm;
705 int idx;
706 $scope CMPI_ROOT_SCOPE = $mpi_root_scope(comm.col);
707
708 if(place == 0) {
709 int size = $comm_size(comm.col);
710
711 newgcomm = $mpi_gcomm_create(CMPI_ROOT_SCOPE, size);
712 idx = $mpi_new_gcomm(CMPI_ROOT_SCOPE, newgcomm);
713 }
714 $mpi_bcast(&idx, 1, MPI_INT, 0, COMMDUP_TAG,
715 comm, routName);
716 newgcomm = $mpi_get_gcomm(CMPI_ROOT_SCOPE, idx);
717 (*newcomm) = $mpi_comm_create(scope, newgcomm, place);
718 newcomm->gcommIndex = idx;
719 $barrier_call(comm.barrier);
720 $gcomm_dup(comm.p2p, newcomm->p2p);
721 $gcomm_dup(comm.col, newcomm->col);
722 $barrier_call(comm.barrier);
723 return 0;
724}
725
726int $mpi_comm_free(MPI_Comm *comm, $mpi_state mpi_state) {
727 int place = $comm_place(comm->col);
728 int size = $comm_size(comm->col);
729 int buf[size];
730 int gcommIndex = comm->gcommIndex;
731 $scope CMPI_ROOT_SCOPE = $mpi_root_scope(comm->col);
732
733 //TODO: $mpi_gather here is just a ugly synchronization
734 $mpi_gather(&place, 1, MPI_INT, buf, 1, MPI_INT, 0,
735 COMMFREE_TAG, (*comm), "MPI_Comm_free synchronization.");
736 $mpi_comm_destroy(*comm, mpi_state);
737 if(place == 0) {
738 $mpi_gcomm temp = $mpi_get_gcomm(CMPI_ROOT_SCOPE, gcommIndex);
739
740 $mpi_gcomm_destroy(temp);
741 }
742 return 0;
743}
744
745$bundle $mpi_create_coroutine_entry(int routineTag, int root,
746 int op, int numDatatypes, int * datatypes) {
747 int zero = 0;
748 $bundle bundledEntry;
749 struct Entry {
750 int routine_tag;
751 int root;
752 int op;
753 int numTypes;
754 int datatypes[];
755 }entry;
756
757 entry.routine_tag = routineTag;
758 entry.root = root;
759 entry.op = op;
760 entry.numTypes = numDatatypes;
761 $seq_init(&entry.datatypes, numDatatypes, &zero);
762 for(int i = 0; i < numDatatypes; i++)
763 entry.datatypes[i] = datatypes[i];
764 bundledEntry = $bundle_pack(&entry, sizeof(struct Entry));
765 return bundledEntry;
766}
767
768void $mpi_diff_coroutine_entries($bundle specEntry, $bundle mineEntry, int rank) {
769 struct Entry {
770 int routine_tag;
771 int root;
772 int op;
773 int numTypes;
774 int datatypes[];
775 }spec, mine;
776 char * routine;
777 int numTypes;
778
779 $bundle_unpack(specEntry, &spec);
780 $bundle_unpack(mineEntry, &mine);
781 routine = $mpi_coroutine_name(spec.routine_tag);
782 if(spec.routine_tag != mine.routine_tag) {
783 char * mineRoutine = $mpi_coroutine_name(mine.routine_tag);
784
785 $assert($false, "Process with rank %d reaches an MPI collective routine "
786 "%s while at least one of others are collectively reaching %s.",
787 rank, mineRoutine, routine);
788 }
789 else if(spec.root != mine.root) {
790 $assert($false, "Process with rank %d reaches an MPI collective routine "
791 "%s which has a different root with at least one of others.", rank, routine);
792 } else if(spec.op != mine.op) {
793 $assert($false, "Process with rank %d reaches an MPI collective routine "
794 "%s which has a different MPI_Op with at least one of others", rank, routine);
795 } else if(spec.numTypes != mine.numTypes) {
796 $assert($false, "Process with rank %d reaches an MPI collective routine "
797 "%s which has an inconsistent datatype specification with at least"
798 " one of others",
799 rank, routine);
800 }
801 numTypes = spec.numTypes;
802 for(int i = 0; i < numTypes; i++)
803 if(spec.datatypes[i] != mine.datatypes[i]) {
804 $assert($false, "Process with rank %d reaches an MPI collective routine "
805 "%s which has an inconsistent datatype specification with at "
806 "least one of others",
807 rank, routine);
808 break;
809 }
810}
811
812#ifdef _MPI_CONTRACT
813
814$collate_state $mpi_snapshot(MPI_Comm comm, $scope scope) {
815 return $collate_arrives(comm.collator, scope);
816}
817
818void $mpi_unsnapshot(MPI_Comm comm, $collate_state cs) {
819 $collate_departs(comm.collator, cs);
820}
821
822void $mpi_assigns(void * buf, int count, MPI_Datatype datatype) {
823 if ($is_concrete_int(datatype)) {
824 size_t size = sizeofDatatype(datatype);
825 int _int[count];
826 int _2int[count * 2];
827 char _char[count];
828 $real _real[count];
829
830 switch (datatype) {
831 case MPI_INT:
832 case MPI_SHORT:
833 case MPI_LONG:
834 case MPI_LONG_LONG_INT:
835 case MPI_LONG_LONG:
836 case MPI_UNSIGNED_LONG_LONG:
837 memcpy(buf, _int, count * size);
838 break;
839 case MPI_2INT:
840 memcpy(buf, _2int, 2 * count * size);
841 break;
842 case MPI_FLOAT:
843 case MPI_DOUBLE:
844 case MPI_LONG_DOUBLE:
845 memcpy(buf, _real, count * size);
846 break;
847 case MPI_CHAR:
848 case MPI_BYTE:
849 memcpy(buf, _char, count * size);
850 break;
851 default:
852 $assert(0, "Unreachable");
853 }
854 } else {
855 size_t realCount = count * $mpi_extentof(datatype);
856 char newValues[realCount];
857
858 memcpy(buf, newValues, count * sizeofDatatype(datatype));
859 }
860}
861
862$atomic_f void $mpi_comm_empty(MPI_Comm comm, MPI_COMM_MODE mode) {
863 _Bool empty;
864
865 if (mode == P2P) {
866 empty = $comm_empty_in(comm.p2p);
867 empty &= $comm_empty_out(comm.p2p);
868 } else {
869 empty = $comm_empty_in(comm.col);
870 empty &= $comm_empty_out(comm.col);
871 }
872 $assert(empty, "Messages are remaining in MPI communicator\n");
873}
874
875#endif
876
877/********************* Private helper functions *********************/
878/* Returns the string literal of MPI collective routine names by
879 * giving the unique message tag. */
880char * $mpi_coroutine_name(int tag) {
881 switch(tag) {
882 case 9999: return "MPI_Bcast";
883 case 9998: return "MPI_Reduce";
884 case 9997: return "MPI_Allreduce";
885 case 9996: return "MPI_Gather";
886 case 9995: return "MPI_Scatter";
887 case 9994: return "MPI_Gatherv";
888 case 9993: return "MPI_Scatterv";
889 case 9992: return "MPI_Allgather";
890 case 9991: return "MPI_Reduce_scatter";
891 case 9990: return "MPI_Alltoall";
892 case 9989: return "MPI_Alltoallv";
893 case 9988: return "MPI_Alltoallw";
894 case 9987: return "MPI_Barrier";
895 case 9986: return "MPI_Commdup";
896 case 9985: return "MPI_Commfree";
897 case 9984: return "MPI_Scan";
898 case 9983: return "MPI_Exscan";
899 default: $assert($false, "Internal Error: Unexpected MPI routine tag:%d.\n", tag);
900 }
901}
902#endif
903
Note: See TracBrowser for help on using the repository browser.