source: ThirdParty/mpqc_open/src/lib/util/group/messmpi.cc@ 894040

Candidate_v1.7.0 stable
Last change on this file since 894040 was fc38cb, checked in by Frederik Heber <frederik.heber@…>, 3 years ago

Fixing MPI calls for 3.0.

  • MPI_ErrorHandler_set has been deprecated.
  • Property mode set to 100644
File size: 15.9 KB
Line 
1//
2// messmpi.cc
3//
4// Copyright (C) 1996 Limit Point Systems, Inc.
5//
6// Author: Curtis Janssen <cljanss@limitpt.com>
7// Maintainer: LPS
8//
9// This file is part of the SC Toolkit.
10//
11// The SC Toolkit is free software; you can redistribute it and/or modify
12// it under the terms of the GNU Library General Public License as published by
13// the Free Software Foundation; either version 2, or (at your option)
14// any later version.
15//
16// The SC Toolkit is distributed in the hope that it will be useful,
17// but WITHOUT ANY WARRANTY; without even the implied warranty of
18// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19// GNU Library General Public License for more details.
20//
21// You should have received a copy of the GNU Library General Public License
22// along with the SC Toolkit; see the file COPYING.LIB. If not, write to
23// the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
24//
25// The U.S. Government is granted a limited license as per AL 91-7.
26//
27
28#include <stdio.h> // for sprintf
29#include <unistd.h> // for fchdir etc.
30#include <fcntl.h> // for open on AIX
31
32#define MPICH_SKIP_MPICXX
33#include <mpi.h>
34extern int MPI_Initialized(int *); // missing in mpi.h
35
36#include <util/keyval/keyval.h>
37#include <util/group/messmpi.h>
38#include <util/misc/formio.h>
39#include <util/misc/newstring.h>
40
41MPI_Comm global_commgrp;
42
43using namespace std;
44using namespace sc;
45
46// Define this to use immediate mode. This was added added to work
47// around bugs in non-immediate mode optimizations in an MPI impl.
48#undef USE_IMMEDIATE_MODE
49
50// OP_COMMUTES is zero to work around a bug in MPI/Pro 1.5b5 and earlier
51#define OP_COMMUTES 1
52
53///////////////////////////////////////////////////////////////////////
54
55int MPIMessageGrp::nmpi_grps=0;
56Ref<ThreadLock> MPIMessageGrp::grplock;
57
58static
59void
60print_error_and_abort(int me, int mpierror)
61{
62 char msg[MPI_MAX_ERROR_STRING+1];
63 int size;
64 MPI_Error_string(mpierror, msg, &size);
65 msg[size] = '\0';
66 ExEnv::outn() << me << ": " << msg << endl;
67 ExEnv::outn().flush();
68 //MPI_Abort(MPI_COMM_WORLD, mpierror);
69}
70
71static
72const char *
73mpi_thread_string(int level)
74{
75 switch (level) {
76#ifdef HAVE_MPI_INIT_THREAD
77 case MPI_THREAD_SINGLE:
78 return "MPI_THREAD_SINGLE";
79 case MPI_THREAD_FUNNELED:
80 return "MPI_THREAD_FUNNELED";
81 case MPI_THREAD_SERIALIZED:
82 return "MPI_THREAD_SERIALIZED";
83 case MPI_THREAD_MULTIPLE:
84 return "MPI_THREAD_MULTIPLE";
85#endif
86 default:
87 return "unknown";
88 }
89}
90
91///////////////////////////////////////////////////////////////////////
92// The MPIMessageGrp class
93
94static ClassDesc MPIMessageGrp_cd(
95 typeid(MPIMessageGrp),"MPIMessageGrp",1,"public MessageGrp",
96 create<MPIMessageGrp>, create<MPIMessageGrp>, 0);
97
98MPIMessageGrp::MPIMessageGrp()
99{
100 init(MPI_COMM_WORLD);
101}
102
103MPIMessageGrp::MPIMessageGrp(MPI_Comm comm)
104{
105 init(comm);
106}
107
108MPIMessageGrp::MPIMessageGrp(int *argc, char ***argv)
109{
110 init(MPI_COMM_WORLD,argc,argv);
111}
112
113MPIMessageGrp::MPIMessageGrp(const Ref<KeyVal>& keyval):
114 MessageGrp(keyval)
115{
116 if (keyval->exists("argv")) {
117 int argc = keyval->count("argv");
118 char **argv = new char*[argc+1];
119 argv[argc] = 0;
120 for (int arg=0; arg<argc; arg++) {
121 argv[arg] = keyval->pcharvalue("argv",arg);
122 }
123 init(MPI_COMM_WORLD, &argc, &argv);
124 }
125 else {
126 init(MPI_COMM_WORLD);
127 }
128
129 if (keyval->booleanvalue("errors_return")) {
130 if (me()==0)
131 ExEnv::outn() << indent << "MPIMessageGrp: errors_return is true" << endl;
132 MPI_Comm_set_errhandler(commgrp, MPI_ERRORS_RETURN);
133 }
134
135 if (debug_) {
136 ExEnv::outn() << indent << "MPIMessageGrp: KeyVal CTOR: done" << endl;
137 }
138}
139
140void
141MPIMessageGrp::init(MPI_Comm comm, int *argc, char ***argv)
142{
143 int me, nproc;
144
145 if (debug_) {
146 ExEnv::outn() << "MPIMessageGrp::init: entered" << endl;
147 }
148
149 int flag;
150 MPI_Initialized(&flag);
151 if (!flag) {
152 int tmp_argc;
153 char **tmp_argv;
154 int *inits_argc;
155 char ***inits_argv;
156 if (argc && argv) {
157 inits_argc = argc;
158 inits_argv = argv;
159 }
160 else {
161 tmp_argc = 0;
162 tmp_argv = new char*[tmp_argc+1];
163 tmp_argv[tmp_argc] = 0;
164 inits_argc = &tmp_argc;
165 inits_argv = &tmp_argv;
166 }
167 // This dot business is to work around problems with some MPI
168 // implementations.
169 int dot = open(".",O_RDONLY);
170 if (debug_) {
171 ExEnv::outn() << indent
172 << "Calling MPI_Init with";
173 for (int i=0; i<*argc; i++) {
174 ExEnv::outn() << " " << *argv[i];
175 }
176 ExEnv::outn() << endl;
177 }
178#ifdef HAVE_MPI_INIT_THREAD
179 int provided, desired = SC_MPI_THREAD_LEVEL;
180 MPI_Init_thread(inits_argc, inits_argv, desired, &provided);
181 int me;
182 MPI_Comm_rank(MPI_COMM_WORLD, &me);
183 if (provided != desired && me == 0) {
184 ExEnv::outn() << indent
185 << "WARNING: desired "
186 << mpi_thread_string(desired)
187 << " MPI threading support but got "
188 << mpi_thread_string(provided)
189 << endl;
190 }
191#else
192 MPI_Init(inits_argc, inits_argv);
193#endif
194#ifdef HAVE_FCHDIR
195 fchdir(dot);
196#endif
197 close(dot);
198 }
199
200 MPI_Comm_dup(comm, &commgrp);
201 global_commgrp = commgrp;
202
203 MPI_Comm_set_errhandler(commgrp, MPI_ERRORS_ARE_FATAL);
204
205 if (!nmpi_grps) {
206 threadgrp = ThreadGrp::get_default_threadgrp();
207 grplock = threadgrp->new_lock();
208 }
209 grplock->lock();
210 nmpi_grps++;
211 grplock->unlock();
212
213 MPI_Comm_rank(commgrp,&me);
214 MPI_Comm_size(commgrp, &nproc);
215 bufsize = 4000000;
216 buf = 0;
217 //buf = (void*) new char[bufsize];
218 //MPI_Buffer_attach(buf,bufsize);
219
220 if (getenv("MPIMESSAGEGRP_MESSAGEGRP_COLLECTIVES"))
221 use_messagegrp_collectives_ = true;
222 else
223 use_messagegrp_collectives_ = false;
224
225 initialize(me, nproc);
226
227 //MPIL_Trace_on();
228
229 if (debug_) {
230 ExEnv::outn() << me << ": MPIMessageGrp::init: done" << endl;
231 }
232
233 SCFormIO::init_mp(me);
234}
235
236MPIMessageGrp::~MPIMessageGrp()
237{
238 //MPIL_Trace_off();
239 //MPI_Buffer_detach(&buf, &bufsize);
240 delete[] (char*) buf;
241
242 grplock->lock();
243 nmpi_grps--;
244 if (!nmpi_grps) MPI_Finalize();
245 grplock->unlock();
246
247}
248
249Ref<MessageGrp> MPIMessageGrp::clone(void)
250{
251 Ref<MessageGrp> mgrp = new MPIMessageGrp;
252 return mgrp;
253}
254
255void
256MPIMessageGrp::raw_send(int target, const void* data, int nbyte)
257{
258 if (debug_) {
259 ExEnv::outn() << scprintf("%3d: MPI_Send"
260 "(0x%08x, %5d, MPI_BYTE, %3d, 0, commgrp)",
261 me(), data, nbyte, target)
262 << endl;
263 }
264 int ret;
265#ifndef USE_IMMEDIATE_MODE
266 ret = MPI_Send(const_cast<void*>(data),nbyte,MPI_BYTE,target,0,commgrp);
267#else
268 MPI_Request mpireq;
269 MPI_Status status;
270 ret = MPI_Isend(data,nbyte,MPI_BYTE,target,0,commgrp,&mpireq);
271 if (ret == MPI_SUCCESS) ret = MPI_Wait(&mpireq,&status);
272#endif // USE_IMMEDIATE_MODE
273 if (ret != MPI_SUCCESS) {
274 ExEnv::outn() << me() << ": MPIMessageGrp::raw_send("
275 << target << ",," << nbyte << "): mpi error:" << endl;
276 print_error_and_abort(me(), ret);
277 }
278 if (debug_) ExEnv::outn() << scprintf("%3d: sent\n", me()) << endl;
279}
280
281void
282MPIMessageGrp::raw_recv(int sender, void* data, int nbyte)
283{
284 MPI_Status status;
285 if (sender == -1) sender = MPI_ANY_SOURCE;
286 if (debug_) {
287 ExEnv::outn() << scprintf("%3d: MPI_Recv"
288 "(0x%08x, %5d, MPI_BYTE, %3d, 0, commgrp,)",
289 me(), data, nbyte, sender)
290 << endl;
291 }
292 int ret;
293#ifndef USE_IMMEDIATE_MODE
294 ret = MPI_Recv(data,nbyte,MPI_BYTE,sender,0,commgrp,&status);
295#else
296 MPI_Request mpireq;
297 ret = MPI_Irecv(data,nbyte,MPI_BYTE,sender,0,commgrp,&mpireq);
298 if (ret == MPI_SUCCESS) ret = MPI_Wait(&mpireq,&status);
299#endif // USE_IMMEDIATE_MODE
300 if (ret != MPI_SUCCESS) {
301 ExEnv::outn() << me() << ": MPIMessageGrp::raw_recv("
302 << sender << ",," << nbyte << "): mpi error:" << endl;
303 print_error_and_abort(me(), ret);
304 }
305 rnode = status.MPI_SOURCE;
306 rtag = status.MPI_TAG;
307 rlen = nbyte;
308 if (debug_) ExEnv::outn() << scprintf("%3d: recvd %d bytes\n", me(), rlen) << endl;
309}
310
311void
312MPIMessageGrp::raw_sendt(int target, int type, const void* data, int nbyte)
313{
314 type = (type<<1) + 1;
315 if (debug_) {
316 ExEnv::outn() << scprintf("%3d: MPI_Send"
317 "(0x%08x, %5d, MPI_BYTE, %3d, %5d, commgrp)",
318 me(), data, nbyte, target, type)
319 << endl;
320 }
321 int ret;
322#ifndef USE_IMMEDIATE_MODE
323 ret = MPI_Send(const_cast<void*>(data),nbyte,MPI_BYTE,target,type,commgrp);
324#else
325 MPI_Request mpireq;
326 MPI_Status status;
327 ret = MPI_Isend(data,nbyte,MPI_BYTE,target,type,commgrp,&mpireq);
328 if (ret == MPI_SUCCESS) ret = MPI_Wait(&mpireq,&status);
329#endif
330 if (ret != MPI_SUCCESS) {
331 ExEnv::outn() << me() << ": MPIMessageGrp::raw_sendt("
332 << target << "," << type << ",," << nbyte << "): mpi error:" << endl;
333 print_error_and_abort(me(), ret);
334 }
335 if (debug_) ExEnv::outn() << scprintf("%3d: sent\n", me()) << endl;
336}
337
338void
339MPIMessageGrp::raw_recvt(int type, void* data, int nbyte)
340{
341 MPI_Status status;
342 if (type == -1) type = MPI_ANY_TAG;
343 else type = (type<<1) + 1;
344 if (debug_) {
345 ExEnv::outn() << scprintf("%3d: MPI_Recv(0x%08x, %5d, MPI_BYTE, "
346 "MPI_ANY_SOURCE, %5d, commgrp,)",
347 me(), data, nbyte, type)
348 << endl;
349 }
350 int ret;
351#ifndef USE_IMMEDIATE_MODE
352 ret = MPI_Recv(data,nbyte,MPI_BYTE,MPI_ANY_SOURCE,type,commgrp,&status);
353#else
354 MPI_Request mpireq;
355 ret = MPI_Irecv(data,nbyte,MPI_BYTE,MPI_ANY_SOURCE,type,commgrp,&mpireq);
356 if (ret == MPI_SUCCESS) ret = MPI_Wait(&mpireq,&status);
357#endif // USE_IMMEDIATE_MODE
358 if (ret != MPI_SUCCESS) {
359 ExEnv::outn() << me() << ": MPIMessageGrp::raw_recvt("
360 << type << ",," << nbyte << "): mpi error:" << endl;
361 print_error_and_abort(me(), ret);
362 }
363 rnode = status.MPI_SOURCE;
364 rtag = status.MPI_TAG;
365 rlen = nbyte;
366 if (debug_) {
367 ExEnv::outn() << scprintf("%3d: recvd %d bytes from %d with tag %d\n",
368 me(), rlen, rnode, rtag) << endl;
369 }
370}
371
372int
373MPIMessageGrp::probet(int type)
374{
375 int flag;
376 MPI_Status status;
377
378 if (type == -1) type = MPI_ANY_TAG;
379 else type = (type<<1) + 1;
380 int ret;
381 if (debug_) {
382 ExEnv::outn() << scprintf("%3d: MPI_Iprobe(MPI_ANY_SOURCE, %5d, commgrp, "
383 "&flag, &status)", me(), type)
384 << endl;
385 }
386 if ((ret = MPI_Iprobe(MPI_ANY_SOURCE,type,commgrp,&flag,&status))
387 != MPI_SUCCESS ) {
388 ExEnv::outn() << me() << ": MPIMessageGrp::probet("
389 << type << "): mpi error:" << endl;
390 print_error_and_abort(me(), ret);
391 }
392 if (flag) {
393 rnode = status.MPI_SOURCE;
394 rtag = status.MPI_TAG;
395 MPI_Get_count(&status, MPI_BYTE, &rlen);
396 return 1;
397 }
398 else {
399 rnode = rtag = rlen = 0;
400 }
401
402 return 0;
403}
404
405void
406MPIMessageGrp::sync()
407{
408 int ret;
409 if (debug_) {
410 ExEnv::outn() << scprintf("%3d: MPI_Barrier(commgrp)", me()) << endl;
411 }
412 if ((ret = MPI_Barrier(commgrp)) != MPI_SUCCESS) {
413 ExEnv::outn() << me() << ": MPIMessageGrp::sync(): mpi error:" << endl;
414 print_error_and_abort(me(), ret);
415 }
416}
417
418#define REDUCEMEMBER(name, type, mpitype) \
419static GrpReduce<type>* name ## reduceobject; \
420extern "C" void \
421name ## reduce(void*b, void*a, int*len, MPI_Datatype*datatype) \
422{ \
423 name ## reduceobject->reduce((type*)a, (type*)b, *len); \
424} \
425void \
426MPIMessageGrp::reduce(type*d, int n, GrpReduce<type>&r, \
427 type*scratch, int target) \
428{ \
429 if (use_messagegrp_collectives_) { \
430 MessageGrp::reduce(d,n,r,scratch,target); \
431 return; \
432 } \
433 \
434 name ## reduceobject = &r; \
435 \
436 MPI_Op op; \
437 MPI_Op_create(name ## reduce, OP_COMMUTES, &op); \
438 \
439 type *work; \
440 if (!scratch) work = new type[n]; \
441 else work = scratch; \
442 \
443 int ret; \
444 \
445 if (target == -1) { \
446 if (debug_) { \
447 ExEnv::outn() << scprintf("%3d: MPI_Allreduce" \
448 "(0x%08x, 0x%08x, %5d, %3d, op, commgrp)", \
449 me(), d, work, n, mpitype) \
450 << endl; \
451 } \
452 ret = MPI_Allreduce(d, work, n, mpitype, op, commgrp); \
453 if (debug_) \
454 ExEnv::outn() << scprintf("%3d: done with Allreduce", me()) << endl; \
455 } \
456 else { \
457 if (debug_) { \
458 ExEnv::outn() << scprintf("%3d: MPI_Reduce" \
459 "(0x%08x, 0x%08x, %5d, %3d, op, %3d, commgrp)", \
460 me(), d, work, n, mpitype, target) \
461 << endl; \
462 } \
463 ret = MPI_Reduce(d, work, n, mpitype, op, target, commgrp); \
464 if (debug_) \
465 ExEnv::outn() << scprintf("%3d: done with Reduce", me()) << endl; \
466 } \
467 \
468 if (ret != MPI_SUCCESS) { \
469 ExEnv::outn() << me() << ": MPIMessageGrp::reduce(," \
470 << n << ",,," << target << "): mpi error:" << endl; \
471 print_error_and_abort(me(), ret); \
472 } \
473 \
474 if (target == -1 || target == me()) { \
475 for (int i=0; i<n; i++) d[i] = work[i]; \
476 } \
477 \
478 MPI_Op_free(&op); \
479 \
480 if (!scratch) delete[] work; \
481}
482
483REDUCEMEMBER(double, double, MPI_DOUBLE)
484REDUCEMEMBER(float, float, MPI_FLOAT)
485REDUCEMEMBER(uint, unsigned int, MPI_INT)
486REDUCEMEMBER(int, int, MPI_INT)
487REDUCEMEMBER(short, short, MPI_SHORT)
488REDUCEMEMBER(long, long, MPI_LONG)
489REDUCEMEMBER(char, char, MPI_CHAR)
490REDUCEMEMBER(uchar, unsigned char, MPI_UNSIGNED_CHAR)
491#ifdef MPI_SIGNED_CHAR
492REDUCEMEMBER(schar, signed char, MPI_SIGNED_CHAR)
493#else
494void
495MPIMessageGrp::reduce(signed char* d, int n, GrpReduce<signed char>& r,
496 signed char*scratch, int target)
497{
498 MessageGrp::reduce(d,n,r,scratch,target);
499}
500#endif
501
502#define SUMMEMBER(name, type, mpitype) \
503void \
504MPIMessageGrp::sum(type*d, int n, type*scratch, int target) \
505{ \
506 if (use_messagegrp_collectives_) { \
507 MessageGrp::sum(d,n,scratch,target); \
508 return; \
509 } \
510 \
511 type *work; \
512 if (!scratch) work = new type[n]; \
513 else work = scratch; \
514 \
515 int ret; \
516 \
517 if (target == -1) { \
518 if (debug_) { \
519 ExEnv::outn() << scprintf("%3d: MPI_Allreduce" \
520 "(0x%08x, 0x%08x, %5d, %3d, MPI_SUM, commgrp)", \
521 me(), d, work, n, mpitype) \
522 << endl; \
523 } \
524 ret = MPI_Allreduce(d, work, n, mpitype, MPI_SUM, commgrp); \
525 if (debug_) \
526 ExEnv::outn() << scprintf("%3d: done with Allreduce", me()) << endl; \
527 } \
528 else { \
529 if (debug_) { \
530 ExEnv::outn() << scprintf("%3d: MPI_Reduce" \
531 "(0x%08x, 0x%08x, %5d, %3d, MPI_SUM, %3d, commgrp)", \
532 me(), d, work, n, mpitype, target) \
533 << endl; \
534 } \
535 ret = MPI_Reduce(d, work, n, mpitype, MPI_SUM, target, commgrp); \
536 if (debug_) \
537 ExEnv::outn() << scprintf("%3d: done with Reduce", me()) << endl; \
538 } \
539 \
540 if (ret != MPI_SUCCESS) { \
541 ExEnv::outn() << me() << ": MPIMessageGrp::sum(," \
542 << n << ",,," << target << "): mpi error:" << endl; \
543 print_error_and_abort(me(), ret); \
544 } \
545 \
546 if (target == -1 || target == me()) { \
547 for (int i=0; i<n; i++) d[i] = work[i]; \
548 } \
549 \
550 if (!scratch) delete[] work; \
551}
552SUMMEMBER(double, double, MPI_DOUBLE)
553SUMMEMBER(int, int, MPI_INT)
554
555void
556MPIMessageGrp::raw_bcast(void* data, int nbyte, int from)
557{
558 if (n() == 1) return;
559
560 if (use_messagegrp_collectives_) {
561 MessageGrp::raw_bcast(data,nbyte,from);
562 return;
563 }
564
565 if (debug_) {
566 ExEnv::outn() << scprintf("%3d: MPI_Bcast("
567 "0x%08x, %5d, MPI_BYTE, %3d, commgrp)",
568 me(), data, nbyte, from)
569 << endl;
570 }
571 int ret;
572 if ((ret = MPI_Bcast(data, nbyte, MPI_BYTE, from, commgrp))
573 != MPI_SUCCESS) {
574 ExEnv::outn() << me() << ": MPIMessageGrp::raw_bcast(,"
575 << nbyte << "," << from << "): mpi error:" << endl;
576 print_error_and_abort(me(), ret);
577 }
578 if (debug_) {
579 ExEnv::outn() << scprintf("%3d: done with bcast", me()) << endl;
580 }
581}
582
583/////////////////////////////////////////////////////////////////////////////
584
585// Local Variables:
586// mode: c++
587// c-file-style: "CLJ"
588// End:
Note: See TracBrowser for help on using the repository browser.