source: CIVL/text/include/mpi.cvl@ 277ddb4

1.23 2.0 acw/focus-triggers main test-branch
Last change on this file since 277ddb4 was 49ab71c, checked in by Stephen Siegel <siegel@…>, 12 years ago

Removed extraneous #endif in mpi.cvl.

git-svn-id: ​svn://vsl.cis.udel.edu/civl/trunk@1171 fb995dde-84ed-4084-dfe6-e5aef3e2452c
  • Property mode set to 100644
File size: 14.1 KB
Line 
1#ifdef __MPI_CVL__
2#else
3#define __MPI_CVL__
4#include <mpi-common.h>
5#include <civlc.h>
6
7//TODO make a Datatype struct, which has a field "int size;" Define one of these objects for MPI_INT, MPI_DOUBLE, etc.
8//TODO Then provide methods like MPI provides for creating new ones.
9//TODO then support MPI_Type_contig(datatype, int n).
10
11/* Definition of CMPI_Gcomm and MPI_Comm */
12struct CMPI_Gcomm {
13 $gcomm p2p; // point-to-point communication
14 $gcomm col; // collective communication
15};
16
17struct MPI_Comm {
18 $comm p2p; // point-to-point communication
19 $comm col; // collective communication
20 __MPI_Status__ status;
21};
22
23/****************************** Helper Functions **********************************/
24int sizeofDatatype(MPI_Datatype datatype) {
25 switch (datatype) {
26 case MPI_INT:
27 return sizeof(int);
28 case MPI_FLOAT:
29 return sizeof(float);
30 case MPI_DOUBLE:
31 return sizeof(double);
32 case MPI_CHAR:
33 return sizeof(char);
34 default:
35 $assert(0, "Unreachable");
36 }
37}
38
39/* Helpers for MPI_Reduce and MPI_Allreduce */
40void sumInt(int result[], int buf[], int real_count, int nprocs){
41 //init result array
42 $atomic{
43 for(int i=0; i<real_count; i++){
44 result[i] = 0;
45 }
46 //sum up
47 for(int i=0; i<nprocs; i++)
48 for(int j=0; j<real_count; j++)
49 result[j] = result[j] + buf[i * real_count + j];
50 }
51}
52
53void sumFloat(float result[], float buf[], int real_count, int nprocs){
54 $atomic{
55 for(int i=0; i<real_count; i++){
56 result[i] = 0;
57 }
58 //sum up
59 for(int i=0; i<nprocs; i++)
60 for(int j=0; j<real_count; j++)
61 result[j] = result[j] + buf[i * real_count + j];
62 }
63}
64
65void sumDouble(double result[], double buf[], int real_count, int nprocs){
66 $atomic{
67 for(int i=0; i<real_count; i++){
68 result[i] = 0;
69 }
70 //sum up
71 for(int i=0; i<nprocs; i++)
72 for(int j=0; j<real_count; j++)
73 result[j] = result[j] + buf[i * real_count + j];
74 }
75}
76
77
78void maxInt(int result[], int buf[], int real_count, int nprocs){
79 //init result arra
80 $atomic{
81 for(int i=0; i<real_count; i++){
82 result[i] = buf[i * real_count];
83 }
84
85 for(int i=0; i<nprocs; i++)
86 for(int j=0; j<real_count; j++){
87 if(buf[i * real_count + j] > result[j])
88 result[j] = buf[i * real_count + j];
89 }
90 }
91}
92
93void maxFloat(float result[], float buf[], int real_count, int nprocs){
94 //init result array
95 $atom{
96 for(int i=0; i<real_count; i++){
97 result[i] = buf[i * real_count];
98 }
99
100 for(int i=0; i<nprocs; i++)
101 for(int j=0; j<real_count; j++){
102 if(buf[i * real_count + j] > result[j])
103 result[j] = buf[i * real_count + j];
104 }
105 }
106}
107
108void maxDouble(double result[], double buf[], int real_count, int nprocs){
109 //init result array
110 $atom{
111 for(int i=0; i<real_count; i++){
112 result[i] = buf[i * real_count];
113 }
114
115 for(int i=0; i<nprocs; i++)
116 for(int j=0; j<real_count; j++){
117 if(buf[i * real_count + j] > result[j])
118 result[j] = buf[i * real_count + j];
119 }
120 }
121}
122
123void minInt(int result[], int buf[], int real_count, int nprocs){
124 //init result array
125 $atom{
126 for(int i=0; i<real_count; i++){
127 result[i] = buf[i * real_count];
128 }
129
130 for(int i=0; i<nprocs; i++)
131 for(int j=0; j<real_count; j++){
132 if(buf[i * real_count + j] < result[j])
133 result[j] = buf[i * real_count + j];
134 }
135 }
136}
137
138void minFloat(float result[], float buf[], int real_count, int nprocs){
139 //init result array
140 $atom{
141 for(int i=0; i<real_count; i++){
142 result[i] = buf[i * real_count];
143 }
144
145 for(int i=0; i<nprocs; i++)
146 for(int j=0; j<real_count; j++){
147 if(buf[i * real_count + j] < result[j])
148 result[j] = buf[i * real_count + j];
149 }
150 }
151}
152
153void minDouble(double result[], double buf[], int real_count, int nprocs){
154 //init result array
155 $atom{
156 for(int i=0; i<real_count; i++){
157 result[i] = buf[i * real_count];
158 }
159
160 for(int i=0; i<nprocs; i++)
161 for(int j=0; j<real_count; j++){
162 if(buf[i * real_count + j] < result[j])
163 result[j] = buf[i * real_count + j];
164 }
165 }
166}
167
168/************************** MPI LIB Implementations *******************************/
169CMPI_Gcomm CMPI_Gcomm_create($scope scope, int size) {
170 CMPI_Gcomm result;
171
172 result.p2p = $gcomm_create(scope, size);
173 result.col = $gcomm_create(scope, size);
174 return result;
175}
176
177void CMPI_Gcomm_destroy(CMPI_Gcomm gc) {
178 $gcomm_destroy(gc.p2p);
179 $gcomm_destroy(gc.col);
180}
181
182MPI_Comm MPI_Comm_create($scope scope, CMPI_Gcomm gc, int rank) {
183 MPI_Comm result;
184
185 result.p2p = $comm_create(scope, gc.p2p, rank);
186 result.col = $comm_create(scope, gc.col, rank);
187 result.status = __UNINIT;
188 return result;
189}
190
191void MPI_Comm_destroy(MPI_Comm comm) {
192 $comm_destroy(comm.p2p);
193 $comm_destroy(comm.col);
194}
195
196int __MPI_Init(MPI_Comm *comm) {
197 comm->status = __INIT;
198 return 0;
199}
200
201int __MPI_Finalize(MPI_Comm *comm) {
202 comm->status = __FINALIZED;
203 return 0;
204}
205
206int MPI_Comm_size(MPI_Comm comm, int *size) {
207 $assert(comm.status == __INIT, "MPI_Comm_size() cannot be invoked without MPI_Init() being called before.\n");
208 *size = $comm_size(comm.p2p);
209 return 0;
210}
211
212int MPI_Comm_rank(MPI_Comm comm, int *rank) {
213 $assert(comm.status == __INIT, "MPI_Comm_rank() cannot be invoked without MPI_Init() being called before.\n");
214 *rank = $comm_place(comm.p2p);
215 return 0;
216}
217
218
219int CMPI_Send(void *buf, int count, MPI_Datatype datatype, int dest,
220 int tag, $comm comm) {
221 if (dest >= 0) {
222 int size = count*sizeofDatatype(datatype);
223 int place = $comm_place(comm);
224 $message out = $message_pack(place, dest, tag, buf, size);
225
226 $comm_enqueue(comm, out);
227 }
228 return 0;
229}
230
231int MPI_Send(void *buf, int count, MPI_Datatype datatype, int dest,
232 int tag, MPI_Comm comm) {
233 $assert(comm.status == __INIT, "MPI_Send() cannot be invoked without MPI_Init() being called before.\n");
234 return CMPI_Send(buf, count, datatype, dest, tag, comm.p2p);
235}
236
237
238int CMPI_Recv(void *buf, int count, MPI_Datatype datatype, int source,
239 int tag, $comm comm, MPI_Status *status) {
240 if (source >= 0) {
241 $message in = $comm_dequeue(comm, source, tag);
242 int size = count*sizeofDatatype(datatype);
243
244 $message_unpack(in, buf, size);
245 if (status != MPI_STATUS_IGNORE) {
246 status->size = $message_size(in);
247 status->MPI_SOURCE = $message_source(in);
248 status->MPI_TAG = $message_tag(in);
249 status->MPI_ERROR = 0;
250 }
251 }
252 return 0;
253}
254
255int MPI_Recv(void *buf, int count, MPI_Datatype datatype, int source,
256 int tag, MPI_Comm comm, MPI_Status *status) {
257 $assert(comm.status == __INIT, "MPI_Recv() cannot be invoked without MPI_Init() being called before.\n");
258 return CMPI_Recv(buf, count, datatype, source, tag, comm.p2p, status);
259}
260
261int MPI_Get_count(MPI_Status *status, MPI_Datatype datatype, int *count) {
262 //$assert(__my_status == __INIT, "MPI status is not INIT.\n");
263 *count = status->size/sizeofDatatype(datatype);
264 return 0;
265}
266
267int MPI_Sendrecv(void *sendbuf, int sendcount, MPI_Datatype sendtype,
268 int dest, int sendtag,
269 void *recvbuf, int recvcount, MPI_Datatype recvtype,
270 int source, int recvtag,
271 MPI_Comm comm, MPI_Status *status) {
272 $assert(comm.status == __INIT, "MPI_Sendrecv() cannot be invoked without MPI_Init() being called before.\n");
273 MPI_Send(sendbuf, sendcount, sendtype, dest, sendtag, comm);
274 MPI_Recv(recvbuf, recvcount, recvtype, source, recvtag, comm, status);
275 return 0;
276}
277
278/* Broadcasts a message from root to everyone else.
279 * Need to use a differnt comm.
280 */
281int MPI_Bcast(void *buf, int count, MPI_Datatype datatype, int root,
282 MPI_Comm comm) {
283 int place = $comm_place(comm.col);
284
285 $assert(comm.status == __INIT, "MPI_Bcast() cannot be invoked without MPI_Init() being called before.\n");
286 place = $comm_place(comm.col);
287 if (place == root) {
288 int nprocs = $comm_size(comm.col);
289
290 for (int i=0; i<nprocs; i++)
291 if (i != root)
292 CMPI_Send(buf, count, datatype, i, BCAST_TAG, comm.col);
293 } else {
294 CMPI_Recv(buf, count, datatype, root, BCAST_TAG, comm.col,
295 MPI_STATUS_IGNORE);
296 }
297 return 0;
298}
299
300/* Reduces values on all processes to a single value */
301int MPI_Reduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype,
302 MPI_Op op, int root, MPI_Comm comm){
303 int place;
304 MPI_Status status;
305
306 $assert(comm.status == __INIT, "MPI_Reduce() cannot be invoked without MPI_Init() being called before.\n");
307 place = $comm_place(comm.col);
308 //non-root
309 CMPI_Send(sendbuf, count, datatype, root, REDUCE_TAG, comm.col);
310 if(place != root)
311 return 0;
312 else{
313 //root
314 int nprocs = $comm_size(comm.col);
315 int real_count = -1; //the real count gotton from MPI_Status
316 //void ** buf; //Buffer stores data from other processes
317 void * recv; //Buffer used in MPI_Recv()
318 void * results; //Array of results
319 $scope here = $root;
320 struct Bundle_buf{
321 int I[nprocs * count];
322 float F[nprocs * count];
323 double D[nprocs * count];
324 } buf;
325
326 switch(datatype){
327 case MPI_INT :
328 recv = (int *)$malloc(here, count * sizeof(int));
329 results = (int *)$malloc(here, count * sizeof(int));
330 break;
331 case MPI_FLOAT :
332 recv = (float *)$malloc(here, count * sizeof(float));
333 results = (float *)$malloc(here, count * sizeof(float));
334 break;
335 case MPI_DOUBLE :
336 recv = (double *)$malloc(here, count * sizeof(double));
337 results = (double *)$malloc(here, count * sizeof(double));
338 break;
339 }
340 for(int i=0; i<nprocs; i++){
341 CMPI_Recv(recv, count, datatype, i, REDUCE_TAG, comm.col,
342 &status);
343 MPI_Get_count(&status, datatype, &real_count);
344 $assert(real_count == count);
345
346 switch(datatype){
347 case MPI_INT :
348 for(int j=0; j<real_count; j++)
349 buf.I[i * real_count + j] = ((int *)recv)[j];
350 break;
351 case MPI_FLOAT :
352 for(int j=0; j<real_count; j++)
353 buf.F[i * real_count + j] = ((float *)recv)[j];
354 break;
355 case MPI_DOUBLE :
356 for(int j=0; j<real_count; j++)
357 buf.D[i * real_count + j] = ((double *)recv)[j];
358 break;
359 }
360 }
361
362 //operations
363 switch(datatype){
364 case MPI_INT:
365 if(op == MPI_SUM)
366 sumInt(results, buf.I, real_count, nprocs);
367 if(op == MPI_MAX)
368 maxInt(results, buf.I, real_count, nprocs);
369 if(op == MPI_MIN)
370 minInt(results, buf.I, real_count, nprocs);
371 break;
372 case MPI_FLOAT:
373 if(op == MPI_SUM)
374 sumFloat(results, buf.F, real_count, nprocs);
375 if(op == MPI_MAX)
376 maxFloat(results, buf.F, real_count, nprocs);
377 if(op == MPI_MIN)
378 minFloat(results, buf.F, real_count, nprocs);
379 break;
380 case MPI_DOUBLE:
381 if(op == MPI_SUM)
382 sumDouble(results, buf.D, real_count, nprocs);
383 if(op == MPI_MAX)
384 maxDouble(results, buf.D, real_count, nprocs);
385 if(op == MPI_MIN)
386 minDouble(results, buf.D, real_count, nprocs);
387 break;
388 }
389 MPI_Sendrecv(results, real_count, datatype,
390 place, 0,
391 recvbuf, real_count, datatype,
392 place, 0,
393 comm, &status);
394 $free(recv);
395 $free(results);
396 }
397 return 0;
398}
399
400/* Combines values from all processes and distributes the result back to all processes */
401/* default root is 0 */
402int MPI_Allreduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype,
403 MPI_Op op, MPI_Comm comm){
404 int place;
405 int root;
406 MPI_Status status;
407
408 $assert(comm.status == __INIT, "MPI_Allreduce() cannot be invoked without MPI_Init() being called before.\n");
409 place = $comm_place(comm.col);
410 root = 0;
411 //non-root
412 CMPI_Send(sendbuf, count, datatype, root, REDUCE_TAG, comm.col);
413 if(place != root){
414 MPI_Bcast(recvbuf, count, datatype, root,
415 comm);
416 return 0;
417 }
418 else{
419 //root
420 int nprocs = $comm_size(comm.col);
421 int real_count = -1; //the real count gotton from MPI_Status
422 //void ** buf; //Buffer stores data from other processes
423 void * recv; //Buffer used in MPI_Recv()
424 void * results; //Array of results
425 $scope here = $root;
426 struct Bundle_buf{
427 int I[nprocs * count];
428 float F[nprocs * count];
429 double D[nprocs * count];
430 } buf;
431
432 switch(datatype){
433 case MPI_INT :
434 recv = (int *)$malloc(here, count * sizeof(int));
435 results = (int *)$malloc(here, count * sizeof(int));
436 break;
437 case MPI_FLOAT :
438 recv = (float *)$malloc(here, count * sizeof(float));
439 results = (float *)$malloc(here, count * sizeof(float));
440 break;
441 case MPI_DOUBLE :
442 recv = (double *)$malloc(here, count * sizeof(double));
443 results = (double *)$malloc(here, count * sizeof(double));
444 break;
445 }
446 //recv = (void *)$malloc(here, count * sizeofDatatype(datatype));
447 //results = (void *)$malloc(here, count * sizeofDatatype(datatype));
448
449 for(int i=0; i<nprocs; i++){
450 CMPI_Recv(recv, count, datatype, i, REDUCE_TAG, comm.col,
451 &status);
452 MPI_Get_count(&status, datatype, &real_count);
453 $assert(real_count == count);
454
455 switch(datatype){
456 case MPI_INT :
457 for(int j=0; j<real_count; j++)
458 buf.I[i * real_count + j] = ((int *)recv)[j];
459 break;
460 case MPI_FLOAT :
461 for(int j=0; j<real_count; j++)
462 buf.F[i * real_count + j] = ((float *)recv)[j];
463 break;
464 case MPI_DOUBLE :
465 for(int j=0; j<real_count; j++)
466 buf.D[i * real_count + j] = ((double *)recv)[j];
467 break;
468 }
469 }
470
471 //operations
472 switch(datatype){
473 case MPI_INT:
474 if(op == MPI_SUM)
475 sumInt(results, buf.I, real_count, nprocs);
476 if(op == MPI_MAX)
477 maxInt(results, buf.I, real_count, nprocs);
478 if(op == MPI_MIN)
479 minInt(results, buf.I, real_count, nprocs);
480 break;
481 case MPI_FLOAT:
482 if(op == MPI_SUM)
483 sumFloat(results, buf.F, real_count, nprocs);
484 if(op == MPI_MAX)
485 maxFloat(results, buf.F, real_count, nprocs);
486 if(op == MPI_MIN)
487 minFloat(results, buf.F, real_count, nprocs);
488 break;
489 case MPI_DOUBLE:
490 if(op == MPI_SUM)
491 sumDouble(results, buf.D, real_count, nprocs);
492 if(op == MPI_MAX)
493 maxDouble(results, buf.D, real_count, nprocs);
494 if(op == MPI_MIN)
495 minDouble(results, buf.D, real_count, nprocs);
496 break;
497 }
498 MPI_Sendrecv(results, real_count, datatype,
499 place, 0,
500 recvbuf, real_count, datatype,
501 place, 0,
502 comm, &status);
503 MPI_Bcast(results, real_count, datatype, root,
504 comm);
505 $free(recv);
506 $free(results);
507 }
508 return 0;
509}
510
511#endif
Note: See TracBrowser for help on using the repository browser.