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

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

remove "$elaborate(count)"

git-svn-id: ​svn://vsl.cis.udel.edu/civl/trunk@2526 fb995dde-84ed-4084-dfe6-e5aef3e2452c
  • Property mode set to 100644
File size: 17.4 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, $comm comm) {
131 if (dest >= 0) {
132 int size = count*sizeofDatatype(datatype);
133 int place = $comm_place(comm);
134 $message out = $message_pack(place, dest, tag, buf, size);
135 $comm_enqueue(comm, out);
136 }
137 return 0;
138}
139
140int $mpi_recv(void *buf, int count, MPI_Datatype datatype, int source,
141 int tag, $comm comm, MPI_Status *status) {
142 if (source >= 0 || source == MPI_ANY_SOURCE) {
143 $elaborate(source);
144 $message in = $comm_dequeue(comm, source, tag);
145 int size = count*sizeofDatatype(datatype);
146
147 $message_unpack(in, buf, size);
148 if (status != MPI_STATUS_IGNORE) {
149 status->size = $message_size(in);
150 status->MPI_SOURCE = $message_source(in);
151 status->MPI_TAG = $message_tag(in);
152 status->MPI_ERROR = 0;
153 }
154 }
155 return 0;
156}
157
158int $mpi_sendrecv(const void *sendbuf, int sendcount, MPI_Datatype sendtype,
159 int dest, int sendtag, void *recvbuf, int recvcount,
160 MPI_Datatype recvtype, int source, int recvtag,
161 $comm comm, MPI_Status *status) {
162 //send and receive triggering flags
163 if((dest >= 0) && ((source >= 0 || source == MPI_ANY_SOURCE))) {
164 $message out, in;
165 int size = sendcount*sizeofDatatype(sendtype);
166 int place = $comm_place(comm);
167
168 out = $message_pack(place, dest, sendtag, sendbuf, size);
169 $elaborate(source);
170 $choose {
171 $when (1){
172 $comm_enqueue(comm, out);
173 in = $comm_dequeue(comm, source, recvtag);
174 }
175 $when (1){
176 in = $comm_dequeue(comm, source, recvtag);
177 $comm_enqueue(comm, out);
178 }
179 }
180 size = recvcount*sizeofDatatype(recvtype);
181 $message_unpack(in, recvbuf, size);
182 if (status != MPI_STATUS_IGNORE) {
183 status->size = $message_size(in);
184 status->MPI_SOURCE = $message_source(in);
185 status->MPI_TAG = $message_tag(in);
186 status->MPI_ERROR = 0;
187 }
188 }
189 else if (dest >= 0) {
190 $mpi_send(sendbuf, sendcount, sendtype, dest, sendtag, comm);
191 }
192 else if (source >= 0 || source == MPI_ANY_SOURCE) {
193 $mpi_recv(recvbuf, recvcount, recvtype, source, recvtag, comm, status);
194 }
195 return 0;
196}
197
198/********************* Collective helper functions ********************/
199/* Note: collective helpers functions are functions have same
200 behaviors as MPI collective functions, it can be re-used as a part
201 of implementation by different MPI routines. For example,
202 MPI_Allreduce will call CMPI_Reduce and CMPI_Bcast, both of them
203 should throw errors (if encounters any) as if errors are thrown
204 from MPI_Allreduce.
205*/
206int $mpi_collective_recv(void *buf, int count, MPI_Datatype datatype,
207 int source, int tag, $comm comm,
208 MPI_Status * status, char * routName) {
209 if(source >= 0 || source == MPI_ANY_SOURCE) {
210 $elaborate(source);
211 $message in = $comm_dequeue(comm, source, MPI_ANY_TAG);
212 int size = count*sizeofDatatype(datatype);
213 int recvTag;
214
215 recvTag = $message_tag(in);
216 $assert (recvTag == tag , "Collective routine %s receives a "
217 "message with a mismatched tag\n", routName);
218 $message_unpack(in, buf, size);
219 if (status != MPI_STATUS_IGNORE) {
220 status->size = $message_size(in);
221 status->MPI_SOURCE = $message_source(in);
222 status->MPI_TAG = recvTag;
223 status->MPI_ERROR = 0;
224 }
225 }
226 return 0;
227}
228
229/* Broadcast helper function that uses any specified message tag */
230int $mpi_bcast(void *buf, int count, MPI_Datatype datatype, int root, int tag,
231 MPI_Comm comm, char * routName) {
232 if ($comm_place(comm.col) == root) {
233 int nprocs = $comm_size(comm.col);
234
235 for (int i=0; i<nprocs; i++)
236 if (i != root)
237 $mpi_send(buf, count, datatype, i, tag, comm.col);
238 } else
239 $mpi_collective_recv(buf, count, datatype, root, tag, comm.col,
240 MPI_STATUS_IGNORE, routName);
241 return 0;
242}
243
244/* Reduction helper function that uses any specified message tag */
245int $mpi_reduce(const void* sendbuf, void* recvbuf, int count,
246 MPI_Datatype datatype, MPI_Op op, int root, int tag,
247 MPI_Comm comm, char * routName) {
248 int rank;
249
250 rank = $comm_place(comm.col);
251 if (rank != root)
252 $mpi_send(sendbuf, count, datatype, root, tag, comm.col);
253 else {
254 int nprocs = $comm_size(comm.col);
255 int size;
256
257 size = count * sizeofDatatype(datatype);
258 memcpy(recvbuf, sendbuf, size);
259 for (int i = 0; i<nprocs; i++) {
260 if(i != root){
261 int colTag;
262 $message in = $comm_dequeue(comm.col, i, MPI_ANY_TAG);
263
264 colTag = $message_tag(in);
265 $assert (colTag == tag , "Collective routine %s receives a "
266 "message with a mismatched tag\n", routName);
267 /* the third argument "count" indicates the number of cells needs doing the
268 operation. */
269 $bundle_unpack_apply(in.data, recvbuf, count, op);
270 $assert (in.size <= size ,
271 "Message of size %d exceeds the specified size %d.", in.size, size);
272 }
273 }
274 }
275 return 0;
276}
277
278/* Gathering helper function that uses any specified message tag */
279int $mpi_gather(const void* sendbuf, int sendcount, MPI_Datatype sendtype,
280 void* recvbuf, int recvcount, MPI_Datatype recvtype,
281 int root, int tag, MPI_Comm comm, char * routName){
282 int rank, nprocs;
283 MPI_Status status;
284
285 rank = $comm_place(comm.col);
286 nprocs = $comm_size(comm.col);
287 /* MPI standard requirement:
288 * For root process, sendtype must be equal to
289 * recvtype. */
290 if(rank == root)
291 $assert (sendtype == recvtype,
292 "%s asks for equality "
293 "between 'sendtype' and 'recvtype'.", routName);
294 /* MPI_standard requirement:
295 * Only root process can use MPI_IN_PLACE*/
296 if(sendbuf == MPI_IN_PLACE){
297 $assert (root == rank,
298 "Only root can replace 'sendbuf' with 'MPI_IN_PLACE'.");
299 } else if(root == rank) {
300 void * ptr;
301
302 $assert(sendcount == recvcount, "Root process of routine %d without using"
303 " MPI_IN_PLACE should give the same value for recvcount and sendcount",
304 routName);
305 ptr = $mpi_pointerAdd(recvbuf, root * recvcount, recvtype);
306 memcpy(ptr, sendbuf, recvcount * sizeofDatatype(recvtype));
307 } else
308 $mpi_send(sendbuf, sendcount, sendtype, root, tag, comm.col);
309 /* Root process receives messages and put them in right places */
310 if(rank == root){
311 int real_recvcount;
312 int offset;
313
314 for(int i=0; i<nprocs; i++){
315 if(i != root) {
316 void * ptr;
317
318 offset = i * recvcount;
319 ptr = $mpi_pointerAdd(recvbuf, offset, recvtype);
320 $mpi_collective_recv(ptr, recvcount, recvtype,
321 i, tag, comm.col, &status, routName);
322 real_recvcount = status.size/sizeofDatatype(recvtype);
323 $assert(real_recvcount == recvcount,
324 "%s asks for equality between"
325 " the amount of data sent and the "
326 "amount of data received.", routName);
327 }
328 }
329 }
330 return 0;
331}
332
333int $mpi_gatherv(const void* sendbuf, int sendcount, MPI_Datatype sendtype,
334 void* recvbuf, const int recvcounts[], const int displs[],
335 MPI_Datatype recvtype, int root, int tag,
336 MPI_Comm comm, char * routName){
337 int rank, nprocs;
338
339 rank = $comm_place(comm.col);
340 nprocs = $comm_size(comm.col);
341 /* MPI standard requirement:
342 * For root process, sendtype must be equal to
343 * recvtype. */
344 if(rank == root)
345 $assert(sendtype == recvtype, "%s asks for equality "
346 "between 'sendtype' and 'recvtype'.", routName);
347 /* MPI_standard requirement:
348 * Only root process can use MPI_IN_PLACE*/
349 if(sendbuf == MPI_IN_PLACE){
350 $assert(root == rank, "Only root can replace 'sendbuf' with 'MPI_IN_PLACE'.");
351 }else if(root == rank) {
352 void * ptr;
353
354 $assert(sendcount == recvcounts[root], "For routine %s, recvcounts[%d] "
355 "should be same as the sendcount of the process with rank %d.\n",
356 routName, root, root);
357 ptr = $mpi_pointerAdd(recvbuf, displs[rank], recvtype);
358 memcpy(ptr, sendbuf, sendcount * sizeofDatatype(recvtype));
359 }else{
360 $mpi_send(sendbuf, sendcount, sendtype, root, tag, comm.col);
361 }
362 /* Root process receives messages and put them in right places */
363 if(rank == root){
364 int real_recvcount;
365 MPI_Status status;
366
367 for(int i=0; i<nprocs; i++){
368 if(i != root){
369 void * ptr = $mpi_pointerAdd(recvbuf, displs[i], recvtype);
370
371 $mpi_collective_recv(ptr, recvcounts[i],
372 recvtype, i, tag, comm.col, &status, routName);
373 real_recvcount = status.size/sizeofDatatype(recvtype);
374 $assert(real_recvcount == recvcounts[i], "%s asks for equality between"
375 " the amount of data sent and the "
376 "amount of data received.", routName);
377 }
378 }
379 }
380 return 0;
381}
382
383/* Scatter helper function that uses any specified message tag */
384int $mpi_scatter(const void* sendbuf, int sendcount, MPI_Datatype sendtype,
385 void* recvbuf, int recvcount, MPI_Datatype recvtype, int root,
386 int tag, MPI_Comm comm, char * routName){
387 int rank, nprocs;
388
389 rank = $comm_place(comm.col);
390 nprocs = $comm_size(comm.col);
391 /* MPI standard requirement:
392 * For root process, sendtype must be equal to
393 * recvtype. */
394 if(rank == root)
395 $assert(sendtype == recvtype, "MPI_Scatter() asks for equality "
396 "between 'sendtype' and 'recvtype'.");
397 /* MPI_standard requirement:
398 * Only root process can use MPI_IN_PLACE */
399 if(recvbuf == MPI_IN_PLACE){
400 $assert(root == rank, "Only root can replace 'recvbuf' with 'MPI_IN_PLACE'.");
401 }else if(rank == root) {
402 void * ptr;
403
404 $assert(sendcount == recvcount, "Root process of routine %d without using"
405 " MPI_IN_PLACE should give the same value for recvcount and sendcount",
406 routName);
407 ptr = $mpi_pointerAdd(sendbuf, root*recvcount, sendtype);
408 memcpy(recvbuf, ptr, sizeofDatatype(recvtype)*recvcount);
409 }
410 /* Root process scatters data to other processes */
411 if(rank == root){
412 int offset;
413
414 for(int i=0; i<nprocs; i++){
415 if(i != root) {
416 void * ptr;
417
418 offset = i * sendcount;
419 ptr = $mpi_pointerAdd(sendbuf, offset, sendtype);
420 $mpi_send(ptr, sendcount, sendtype, i, tag, comm.col);
421 }
422 }
423 }
424 /* Non-root processes receive data */
425 if(!(root == rank)){
426 int real_recvcount;
427 MPI_Status status;
428
429 $mpi_collective_recv(recvbuf, recvcount, recvtype,
430 root, tag, comm.col, &status, routName);
431 real_recvcount = status.size/sizeofDatatype(recvtype);
432 $assert(real_recvcount == recvcount,
433 "%s asks for equality between"
434 " the amount of data sent and the "
435 "amount of data received.", routName);
436 }
437 return 0;
438}
439
440/* Scatterv helper function that uses any specified message tag */
441int $mpi_scatterv(const void* sendbuf, const int sendcounts[], const
442 int displs[], MPI_Datatype sendtype, void* recvbuf,
443 int recvcount, MPI_Datatype recvtype, int root, int tag,
444 MPI_Comm comm, char * routName){
445 int rank, nprocs;
446
447 rank = $comm_place(comm.col);
448 nprocs = $comm_size(comm.col);
449 /* MPI standard requirement:
450 * For root process, sendtype must be equal to
451 * recvtype. */
452 if(rank == root)
453 $assert(sendtype == recvtype, "%s asks for equality "
454 "between 'sendtype' and 'recvtype'.", routName);
455 /* MPI_standard requirement:
456 * Only root process can use MPI_IN_PLACE */
457 if(recvbuf == MPI_IN_PLACE){
458 $assert(root == rank, "Only root can replace 'recvbuf' with 'MPI_IN_PLACE'.");
459 } else if(rank == root) {
460 void * ptr;
461
462 $assert(sendcounts[root] == recvcount, "For routine %s, sendcounts[%d] "
463 "should be same as the recvcount of the process with rank %d.\n",
464 routName, root, root);
465 ptr = $mpi_pointerAdd(sendbuf, displs[root], sendtype);
466 memcpy(recvbuf, ptr, recvcount*sizeofDatatype(recvtype));
467 }
468 /* Root process scatters data to other processes */
469 if(rank == root){
470 for(int i=0; i<nprocs; i++){
471 if(i != root) {
472 void * ptr = $mpi_pointerAdd(sendbuf, displs[i], sendtype);
473
474 $mpi_send(ptr, sendcounts[i], sendtype, i,
475 tag, comm.col);
476 }
477 }
478 }
479 if(!(root == rank)){
480 MPI_Status status;
481 int real_recvcount;
482
483 $mpi_collective_recv(recvbuf, recvcount, recvtype,
484 root, tag, comm.col, &status, routName);
485 real_recvcount = status.size/sizeofDatatype(recvtype);
486 $assert(real_recvcount == recvcount, "Process rank:%d\n%s asks for equality between"
487 " the amount of data sent (%d) and the "
488 "amount of data received (%d).", rank, routName, real_recvcount, recvcount);
489 }
490 return 0;
491}
492
493int $mpi_comm_dup($scope scope, MPI_Comm comm, MPI_Comm * newcomm, char * routName) {
494 int place = $comm_place(comm.col);
495 $mpi_gcomm newgcomm;
496 int idx;
497 $scope CMPI_ROOT_SCOPE = $mpi_root_scope(comm.col);
498
499 if(place == 0) {
500 int size = $comm_size(comm.col);
501
502 newgcomm = $mpi_gcomm_create(CMPI_ROOT_SCOPE, size);
503 idx = $mpi_newGcomm(CMPI_ROOT_SCOPE, newgcomm);
504 }
505 $mpi_bcast(&idx, 1, MPI_INT, 0, COMMDUP_TAG,
506 comm, routName);
507 newgcomm = $mpi_getGcomm(CMPI_ROOT_SCOPE, idx);
508 (*newcomm) = $mpi_comm_create(scope, newgcomm, place);
509 newcomm->gcommIndex = idx;
510 $barrier_call(comm.barrier);
511 $gcomm_dup(comm.p2p, newcomm->p2p);
512 $gcomm_dup(comm.col, newcomm->col);
513 $barrier_call(comm.barrier);
514 return 0;
515}
516
517int $mpi_comm_free(MPI_Comm * comm) {
518 int place = $comm_place(comm->col);
519 int size = $comm_size(comm->col);
520 int buf[size];
521 int gcommIndex = comm->gcommIndex;
522 $scope CMPI_ROOT_SCOPE = $mpi_root_scope(comm->col);
523
524 //TODO: $mpi_gather here is just a ugly synchronization
525 $mpi_gather(&place, 1, MPI_INT, buf, 1, MPI_INT, 0,
526 COMMFREE_TAG, (*comm), "MPI_Comm_free synchronization.");
527 $mpi_comm_destroy(*comm);
528 if(place == 0) {
529 $mpi_gcomm temp = $mpi_getGcomm(CMPI_ROOT_SCOPE, gcommIndex);
530
531 $mpi_gcomm_destroy(temp);
532 }
533 return 0;
534}
535
536$bundle $mpi_createCoroutineEntries(int routineTag, int root,
537 int op, int numDatatypes, int * datatypes) {
538 int zero = 0;
539 $bundle bundledEntries;
540 struct Entries {
541 int routine_tag;
542 int root;
543 int op;
544 int numTypes;
545 int datatypes[];
546 }entries;
547
548 entries.routine_tag = routineTag;
549 entries.root = root;
550 entries.op = op;
551 entries.numTypes = numDatatypes;
552 $seq_init(&entries.datatypes, numDatatypes, &zero);
553 for(int i = 0; i < numDatatypes; i++)
554 entries.datatypes[i] = datatypes[i];
555 bundledEntries = $bundle_pack(&entries, sizeof(struct Entries));
556 return bundledEntries;
557}
558
559#endif
560
Note: See TracBrowser for help on using the repository browser.