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

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

Add implementation of MPI_Scan and MPI_Exscan

git-svn-id: ​svn://vsl.cis.udel.edu/civl/trunk@4522 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#ifdef _MPI_CONTRACT
318 $elaborate(count);
319#else
320 $comm_enqueue(comm.col, out);
321#endif
322 }
323 return 0;
324}
325
326int $mpi_collective_recv(void *buf, int count, MPI_Datatype datatype,
327 int source, int tag, MPI_Comm comm,
328 MPI_Status * status, char * routName) {
329 if(source >= 0 || source == MPI_ANY_SOURCE) {
330 $elaborate(source);
331 $message in = $comm_dequeue(comm.col, source, MPI_ANY_TAG);
332 int size = count*sizeofDatatype(datatype);
333 int recvTag;
334
335 /* This routine should only be used by collective routines, there
336 is no non-deterministic tags for collective routines.*/
337 recvTag = $message_tag(in);
338 $assert (recvTag == tag, "Collective routine %s receives a "
339 "message with a mismatched tag\n", routName);
340 $message_unpack(in, buf, size);
341 if (status != MPI_STATUS_IGNORE) {
342 status->size = $message_size(in);
343 status->MPI_SOURCE = $message_source(in);
344 status->MPI_TAG = recvTag;
345 status->MPI_ERROR = 0;
346 }
347 }
348 return 0;
349}
350
351/* Broadcast helper function that uses any specified message tag */
352int $mpi_bcast(void *buf, int count, MPI_Datatype datatype, int root, int tag,
353 MPI_Comm comm, char * routName) {
354 if ($comm_place(comm.col) == root) {
355 int nprocs = $comm_size(comm.col);
356
357 for (int i=0; i<nprocs; i++)
358 if (i != root)
359 $mpi_collective_send(buf, count, datatype, i, tag, comm);
360 } else
361 $mpi_collective_recv(buf, count, datatype, root, tag, comm,
362 MPI_STATUS_IGNORE, routName);
363 return 0;
364}
365
366/* Reduction helper function that uses any specified message tag */
367int $mpi_reduce(const void* sendbuf, void* recvbuf, int count,
368 MPI_Datatype datatype, MPI_Op op, int root, int tag,
369 MPI_Comm comm, char * routName) {
370 int rank;
371
372 rank = $comm_place(comm.col);
373 if (rank != root)
374 $mpi_collective_send(sendbuf, count, datatype, root, tag, comm);
375 else {
376 int nprocs = $comm_size(comm.col);
377 int size;
378
379 size = count * sizeofDatatype(datatype);
380 memcpy(recvbuf, sendbuf, size);
381 for (int i = 0; i<nprocs; i++) {
382 if(i != root){
383 int colTag;
384 void * applybuf;
385 $message in = $comm_dequeue(comm.col, i, MPI_ANY_TAG);
386
387 /* Collective routines have no non-deterministic tags.*/
388 colTag = $message_tag(in);
389 $assert (colTag == tag , "Collective routine %s receives a "
390 "message with a mismatched tag\n", routName);
391 /* the third argument "count" indicates the number of cells needs doing the
392 operation. */
393 applybuf = $mpi_malloc(count, datatype);
394 $bundle_unpack_apply(in.data, recvbuf, op, count, applybuf);
395 memcpy(recvbuf, applybuf, size);
396 free(applybuf);
397 $assert (in.size <= size ,
398 "Message of size %d exceeds the specified size %d.", in.size, size);
399 }
400 }
401 }
402 return 0;
403}
404
405/* Gathering helper function that uses any specified message tag */
406int $mpi_gather(const void* sendbuf, int sendcount, MPI_Datatype sendtype,
407 void* recvbuf, int recvcount, MPI_Datatype recvtype,
408 int root, int tag, MPI_Comm comm, char * routName){
409 int rank, nprocs;
410 MPI_Status status;
411
412 rank = $comm_place(comm.col);
413 nprocs = $comm_size(comm.col);
414 /* MPI standard requirement:
415 * For root process, sendtype must be equal to
416 * recvtype. */
417 if(rank == root)
418 $assert (sendtype == recvtype,
419 "%s asks for equality "
420 "between 'sendtype' and 'recvtype'.", routName);
421 /* MPI_standard requirement:
422 * Only root process can use MPI_IN_PLACE*/
423 if(sendbuf == MPI_IN_PLACE){
424 $assert (root == rank,
425 "Only root can replace 'sendbuf' with 'MPI_IN_PLACE'.");
426 } else if(root == rank) {
427 void * ptr;
428
429 $assert(sendcount == recvcount, "Root process of routine %d without using"
430 " MPI_IN_PLACE should give the same value for recvcount and sendcount",
431 routName);
432 ptr = $mpi_pointer_add(recvbuf, root * recvcount, recvtype);
433 memcpy(ptr, sendbuf, recvcount * sizeofDatatype(recvtype));
434 } else
435 $mpi_collective_send(sendbuf, sendcount, sendtype, root, tag, comm);
436 /* Root process receives messages and put them in right places */
437 if(rank == root){
438 int real_recvcount;
439 int offset;
440
441 for(int i=0; i<nprocs; i++){
442 if(i != root) {
443 void * ptr;
444
445 offset = i * recvcount;
446 ptr = $mpi_pointer_add(recvbuf, offset, recvtype);
447 $mpi_collective_recv(ptr, recvcount, recvtype,
448 i, tag, comm, &status, routName);
449 real_recvcount = status.size/sizeofDatatype(recvtype);
450 $assert(real_recvcount == recvcount,
451 "%s asks for equality between"
452 " the amount of data sent and the "
453 "amount of data received.", routName);
454 }
455 }
456 }
457 return 0;
458}
459
460int $mpi_gatherv(const void* sendbuf, int sendcount, MPI_Datatype sendtype,
461 void* recvbuf, const int recvcounts[], const int displs[],
462 MPI_Datatype recvtype, int root, int tag,
463 MPI_Comm comm, char * routName){
464 int rank, nprocs;
465
466 rank = $comm_place(comm.col);
467 nprocs = $comm_size(comm.col);
468 /* MPI standard requirement:
469 * For root process, sendtype must be equal to
470 * recvtype. */
471 if(rank == root)
472 $assert(sendtype == recvtype, "%s asks for equality "
473 "between 'sendtype' and 'recvtype'.", routName);
474 /* MPI_standard requirement:
475 * Only root process can use MPI_IN_PLACE*/
476 if(sendbuf == MPI_IN_PLACE){
477 $assert(root == rank, "Only root can replace 'sendbuf' with 'MPI_IN_PLACE'.");
478 }else if(root == rank) {
479 void * ptr;
480
481 $assert(sendcount == recvcounts[root], "For routine %s, recvcounts[%d] "
482 "should be same as the sendcount of the process with rank %d.\n",
483 routName, root, root);
484 ptr = $mpi_pointer_add(recvbuf, displs[rank], recvtype);
485 memcpy(ptr, sendbuf, sendcount * sizeofDatatype(recvtype));
486 }else{
487 $mpi_collective_send(sendbuf, sendcount, sendtype, root, tag, comm);
488 }
489 /* Root process receives messages and put them in right places */
490 if(rank == root){
491 int real_recvcount;
492 MPI_Status status;
493
494 for(int i=0; i<nprocs; i++){
495 if(i != root){
496 void * ptr = $mpi_pointer_add(recvbuf, displs[i], recvtype);
497
498 $mpi_collective_recv(ptr, recvcounts[i],
499 recvtype, i, tag, comm, &status, routName);
500 real_recvcount = status.size/sizeofDatatype(recvtype);
501 $assert(real_recvcount == recvcounts[i], "%s asks for equality between"
502 " the amount of data sent and the "
503 "amount of data received.", routName);
504 }
505 }
506 }
507 return 0;
508}
509
510/* Scatter helper function that uses any specified message tag */
511int $mpi_scatter(const void* sendbuf, int sendcount, MPI_Datatype sendtype,
512 void* recvbuf, int recvcount, MPI_Datatype recvtype, int root,
513 int tag, MPI_Comm comm, char * routName){
514 int rank, nprocs;
515
516 rank = $comm_place(comm.col);
517 nprocs = $comm_size(comm.col);
518 /* MPI standard requirement:
519 * For root process, sendtype must be equal to
520 * recvtype. */
521 if(rank == root)
522 $assert(sendtype == recvtype, "MPI_Scatter() asks for equality "
523 "between 'sendtype' and 'recvtype'.");
524 /* MPI_standard requirement:
525 * Only root process can use MPI_IN_PLACE */
526 if(recvbuf == MPI_IN_PLACE){
527 $assert(root == rank, "Only root can replace 'recvbuf' with 'MPI_IN_PLACE'.");
528 }else if(rank == root) {
529 void * ptr;
530
531 $assert(sendcount == recvcount, "Root process of routine %d without using"
532 " MPI_IN_PLACE should give the same value for recvcount and sendcount",
533 routName);
534 ptr = $mpi_pointer_add(sendbuf, root*recvcount, sendtype);
535 memcpy(recvbuf, ptr, sizeofDatatype(recvtype)*recvcount);
536 }
537 /* Root process scatters data to other processes */
538 if(rank == root){
539 int offset;
540
541 for(int i=0; i<nprocs; i++){
542 if(i != root) {
543 void * ptr;
544
545 offset = i * sendcount;
546 ptr = $mpi_pointer_add(sendbuf, offset, sendtype);
547 $mpi_collective_send(ptr, sendcount, sendtype, i, tag, comm);
548 }
549 }
550 }
551 /* Non-root processes receive data */
552 if(!(root == rank)){
553 int real_recvcount;
554 MPI_Status status;
555
556 $mpi_collective_recv(recvbuf, recvcount, recvtype,
557 root, tag, comm, &status, routName);
558 real_recvcount = status.size/sizeofDatatype(recvtype);
559 $assert(real_recvcount == recvcount,
560 "%s asks for equality between"
561 " the amount of data sent and the "
562 "amount of data received.", routName);
563 }
564 return 0;
565}
566
567/* Scatterv helper function that uses any specified message tag */
568int $mpi_scatterv(const void* sendbuf, const int sendcounts[], const
569 int displs[], MPI_Datatype sendtype, void* recvbuf,
570 int recvcount, MPI_Datatype recvtype, int root, int tag,
571 MPI_Comm comm, char * routName){
572 int rank, nprocs;
573
574 rank = $comm_place(comm.col);
575 nprocs = $comm_size(comm.col);
576 /* MPI standard requirement:
577 * For root process, sendtype must be equal to
578 * recvtype. */
579 if(rank == root)
580 $assert(sendtype == recvtype, "%s asks for equality "
581 "between 'sendtype' and 'recvtype'.", routName);
582 /* MPI_standard requirement:
583 * Only root process can use MPI_IN_PLACE */
584 if(recvbuf == MPI_IN_PLACE){
585 $assert(root == rank, "Only root can replace 'recvbuf' with 'MPI_IN_PLACE'.");
586 } else if(rank == root) {
587 void * ptr;
588
589 $assert(sendcounts[root] == recvcount, "For routine %s, sendcounts[%d] "
590 "should be same as the recvcount of the process with rank %d.\n",
591 routName, root, root);
592 ptr = $mpi_pointer_add(sendbuf, displs[root], sendtype);
593 memcpy(recvbuf, ptr, recvcount*sizeofDatatype(recvtype));
594 }
595 /* Root process scatters data to other processes */
596 if(rank == root){
597 for(int i=0; i<nprocs; i++){
598 if(i != root) {
599 void * ptr = $mpi_pointer_add(sendbuf, displs[i], sendtype);
600
601 $mpi_collective_send(ptr, sendcounts[i], sendtype, i,
602 tag, comm);
603 }
604 }
605 }
606 if(!(root == rank)){
607 MPI_Status status;
608 int real_recvcount;
609
610 $mpi_collective_recv(recvbuf, recvcount, recvtype,
611 root, tag, comm, &status, routName);
612 real_recvcount = status.size/sizeofDatatype(recvtype);
613 $assert(real_recvcount == recvcount, "Process rank:%d\n%s asks for equality between"
614 " the amount of data sent (%d) and the "
615 "amount of data received (%d).", rank, routName, real_recvcount, recvcount);
616 }
617 return 0;
618}
619
620/* The helper function for (inclusive) MPI_Scan */
621int $mpi_scan(const void* sendbuf, void* recvbuf, int count,
622 MPI_Datatype datatype, MPI_Op op, MPI_Comm comm) {
623 int place, nprocs;
624 int size = count * sizeofDatatype(datatype);
625 void * tmp_space = $mpi_malloc(count, datatype);
626 void * tmp_ptr, * tmp_space_const = tmp_space;
627
628 place = $comm_place(comm.col);
629 nprocs = $comm_size(comm.col);
630 /* Each process do reduction from 0 .. rank (inclusive):
631 * Send to process rank + 1 ... nprocs-1 (inclusive)
632 * Recv from process 0 .. rank (exclusive)
633 */
634 for (int i = place + 1; i < nprocs; i++)
635 $mpi_collective_send(sendbuf, count, datatype, i, SCAN_TAG, comm);
636 if (sendbuf != MPI_IN_PLACE)
637 memcpy(recvbuf, sendbuf, size);
638 for (int i = 0; i < place; i++) {
639 $message in = $comm_dequeue(comm.col, i, SCAN_TAG);
640
641 // so far, unpack_apply requires that 'recvbuf' is not aliasing
642 // 'tmp_space' (can be improved in the future):
643 $bundle_unpack_apply(in.data, recvbuf, op, count, tmp_space);
644 // swap tmp_space with recvbuf
645 tmp_ptr = tmp_space;
646 tmp_space = recvbuf;
647 recvbuf = tmp_ptr;
648 $assert (in.size <= size ,
649 "Message of size %d exceeds the specified size %d.", in.size, size);
650 }
651 if (recvbuf == tmp_space_const)
652 memcpy(tmp_space, recvbuf, size);
653 free(tmp_space_const);
654 return 0;
655}
656
657/* The helper function for (the exclusive) MPI_Exscan */
658int $mpi_exscan(const void* sendbuf, void* recvbuf, int count,
659 MPI_Datatype datatype, MPI_Op op, MPI_Comm comm) {
660 int place, nprocs;
661 int size = count * sizeofDatatype(datatype);
662 void * tmp_space = $mpi_malloc(count, datatype);
663 void * tmp_ptr, * tmp_space_const = tmp_space;
664
665 place = $comm_place(comm.col);
666 nprocs = $comm_size(comm.col);
667 /* The “in place” option for intracommunicators is specified by
668 * passing MPI_IN_PLACE in the sendbuf argument. In this case, the
669 * input data is taken from the receive buffer, and replaced by the
670 * output data. The receive buffer on rank 0 is not changed by this
671 * operation. */
672 if (sendbuf == MPI_IN_PLACE)
673 sendbuf = recvbuf;
674 /* Each process do reduction from 0 .. rank (inclusive):
675 * Send to process rank + 1 ... nprocs-1 (inclusive)
676 * Recv from process 0 .. rank (exclusive)
677 */
678 for (int i = place + 1; i < nprocs; i++)
679 $mpi_collective_send(sendbuf, count, datatype, i, EXSCAN_TAG, comm);
680 // no-op for rank 0
681 if (place != 0) {
682 $message in = $comm_dequeue(comm.col, 0, EXSCAN_TAG);
683 $bundle_unpack(in.data, recvbuf);
684 for (int i = 1; i < place; i++) {
685 in = $comm_dequeue(comm.col, i, EXSCAN_TAG);
686 // so far, unpack_apply requires that 'recvbuf' is not aliasing
687 // 'tmp_space' (can be improved in the future):
688 $bundle_unpack_apply(in.data, recvbuf, op, count, tmp_space);
689 // swap tmp_space with recvbuf
690 tmp_ptr = tmp_space;
691 tmp_space = recvbuf;
692 recvbuf = tmp_ptr;
693 $assert (in.size <= size ,
694 "Message of size %d exceeds the specified size %d.", in.size, size);
695 }
696 }
697 if (recvbuf == tmp_space_const)
698 memcpy(tmp_space, recvbuf, size);
699 free(tmp_space_const);
700 return 0;
701}
702
703
704/* ******************** End of collective routines ********************* */
705
706int $mpi_comm_dup($scope scope, MPI_Comm comm, MPI_Comm * newcomm, char * routName) {
707 int place = $comm_place(comm.col);
708 $mpi_gcomm newgcomm;
709 int idx;
710 $scope CMPI_ROOT_SCOPE = $mpi_root_scope(comm.col);
711
712 if(place == 0) {
713 int size = $comm_size(comm.col);
714
715 newgcomm = $mpi_gcomm_create(CMPI_ROOT_SCOPE, size);
716 idx = $mpi_new_gcomm(CMPI_ROOT_SCOPE, newgcomm);
717 }
718 $mpi_bcast(&idx, 1, MPI_INT, 0, COMMDUP_TAG,
719 comm, routName);
720 newgcomm = $mpi_get_gcomm(CMPI_ROOT_SCOPE, idx);
721 (*newcomm) = $mpi_comm_create(scope, newgcomm, place);
722 newcomm->gcommIndex = idx;
723 $barrier_call(comm.barrier);
724 $gcomm_dup(comm.p2p, newcomm->p2p);
725 $gcomm_dup(comm.col, newcomm->col);
726 $barrier_call(comm.barrier);
727 return 0;
728}
729
730int $mpi_comm_free(MPI_Comm *comm, $mpi_state mpi_state) {
731 int place = $comm_place(comm->col);
732 int size = $comm_size(comm->col);
733 int buf[size];
734 int gcommIndex = comm->gcommIndex;
735 $scope CMPI_ROOT_SCOPE = $mpi_root_scope(comm->col);
736
737 //TODO: $mpi_gather here is just a ugly synchronization
738 $mpi_gather(&place, 1, MPI_INT, buf, 1, MPI_INT, 0,
739 COMMFREE_TAG, (*comm), "MPI_Comm_free synchronization.");
740 $mpi_comm_destroy(*comm, mpi_state);
741 if(place == 0) {
742 $mpi_gcomm temp = $mpi_get_gcomm(CMPI_ROOT_SCOPE, gcommIndex);
743
744 $mpi_gcomm_destroy(temp);
745 }
746 return 0;
747}
748
749$bundle $mpi_create_coroutine_entry(int routineTag, int root,
750 int op, int numDatatypes, int * datatypes) {
751 int zero = 0;
752 $bundle bundledEntry;
753 struct Entry {
754 int routine_tag;
755 int root;
756 int op;
757 int numTypes;
758 int datatypes[];
759 }entry;
760
761 entry.routine_tag = routineTag;
762 entry.root = root;
763 entry.op = op;
764 entry.numTypes = numDatatypes;
765 $seq_init(&entry.datatypes, numDatatypes, &zero);
766 for(int i = 0; i < numDatatypes; i++)
767 entry.datatypes[i] = datatypes[i];
768 bundledEntry = $bundle_pack(&entry, sizeof(struct Entry));
769 return bundledEntry;
770}
771
772void $mpi_diff_coroutine_entries($bundle specEntry, $bundle mineEntry, int rank) {
773 struct Entry {
774 int routine_tag;
775 int root;
776 int op;
777 int numTypes;
778 int datatypes[];
779 }spec, mine;
780 char * routine;
781 int numTypes;
782
783 $bundle_unpack(specEntry, &spec);
784 $bundle_unpack(mineEntry, &mine);
785 routine = $mpi_coroutine_name(spec.routine_tag);
786 if(spec.routine_tag != mine.routine_tag) {
787 char * mineRoutine = $mpi_coroutine_name(mine.routine_tag);
788
789 $assert($false, "Process with rank %d reaches an MPI collective routine "
790 "%s while at least one of others are collectively reaching %s.",
791 rank, mineRoutine, routine);
792 }
793 else if(spec.root != mine.root) {
794 $assert($false, "Process with rank %d reaches an MPI collective routine "
795 "%s which has a different root with at least one of others.", rank, routine);
796 } else if(spec.op != mine.op) {
797 $assert($false, "Process with rank %d reaches an MPI collective routine "
798 "%s which has a different MPI_Op with at least one of others", rank, routine);
799 } else if(spec.numTypes != mine.numTypes) {
800 $assert($false, "Process with rank %d reaches an MPI collective routine "
801 "%s which has an inconsistent datatype specification with at least"
802 " one of others",
803 rank, routine);
804 }
805 numTypes = spec.numTypes;
806 for(int i = 0; i < numTypes; i++)
807 if(spec.datatypes[i] != mine.datatypes[i]) {
808 $assert($false, "Process with rank %d reaches an MPI collective routine "
809 "%s which has an inconsistent datatype specification with at "
810 "least one of others",
811 rank, routine);
812 break;
813 }
814}
815
816#ifdef _MPI_CONTRACT
817
818$collate_state $mpi_snapshot(MPI_Comm comm, $scope scope) {
819 return $collate_arrives(comm.collator, scope);
820}
821
822void $mpi_unsnapshot(MPI_Comm comm, $collate_state cs) {
823 $collate_departs(comm.collator, cs);
824}
825
826void $mpi_assigns(void * buf, int count, MPI_Datatype datatype) {
827 if ($is_concrete_int(datatype)) {
828 size_t size = sizeofDatatype(datatype);
829 int _int[count];
830 int _2int[count * 2];
831 char _char[count];
832 $real _real[count];
833
834 switch (datatype) {
835 case MPI_INT:
836 case MPI_SHORT:
837 case MPI_LONG:
838 case MPI_LONG_LONG_INT:
839 case MPI_LONG_LONG:
840 case MPI_UNSIGNED_LONG_LONG:
841 memcpy(buf, _int, count * size);
842 break;
843 case MPI_2INT:
844 memcpy(buf, _2int, 2 * count * size);
845 break;
846 case MPI_FLOAT:
847 case MPI_DOUBLE:
848 case MPI_LONG_DOUBLE:
849 memcpy(buf, _real, count * size);
850 break;
851 case MPI_CHAR:
852 case MPI_BYTE:
853 memcpy(buf, _char, count * size);
854 break;
855 default:
856 $assert(0, "Unreachable");
857 }
858 } else {
859 size_t realCount = count * $mpi_extentof(datatype);
860 char newValues[realCount];
861
862 memcpy(buf, newValues, count * sizeofDatatype(datatype));
863 }
864}
865
866$atomic_f void $mpi_comm_empty(MPI_Comm comm, MPI_COMM_MODE mode) {
867 _Bool empty;
868
869 if (mode == P2P) {
870 empty = $comm_empty_in(comm.p2p);
871 empty &= $comm_empty_out(comm.p2p);
872 } else {
873 empty = $comm_empty_in(comm.col);
874 empty &= $comm_empty_out(comm.col);
875 }
876 $assert(empty, "Messages are remaining in MPI communicator\n");
877}
878
879#endif
880
881/********************* Private helper functions *********************/
882/* Returns the string literal of MPI collective routine names by
883 * giving the unique message tag. */
884char * $mpi_coroutine_name(int tag) {
885 switch(tag) {
886 case 9999: return "MPI_Bcast";
887 case 9998: return "MPI_Reduce";
888 case 9997: return "MPI_Allreduce";
889 case 9996: return "MPI_Gather";
890 case 9995: return "MPI_Scatter";
891 case 9994: return "MPI_Gatherv";
892 case 9993: return "MPI_Scatterv";
893 case 9992: return "MPI_Allgather";
894 case 9991: return "MPI_Reduce_scatter";
895 case 9990: return "MPI_Alltoall";
896 case 9989: return "MPI_Alltoallv";
897 case 9988: return "MPI_Alltoallw";
898 case 9987: return "MPI_Barrier";
899 case 9986: return "MPI_Commdup";
900 case 9985: return "MPI_Commfree";
901 case 9984: return "MPI_Scan";
902 case 9983: return "MPI_Exscan";
903 default: $assert($false, "Internal Error: Unexpected MPI routine tag:%d.\n", tag);
904 }
905}
906#endif
907
Note: See TracBrowser for help on using the repository browser.