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

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

make the collective_checker much more general

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