src/ParaMEDMEM/MPIAccess.cxx

   1 // Copyright (C) 2007-2016  CEA/DEN, EDF R&D
   2 //
   3 // This library is free software; you can redistribute it and/or
   4 // modify it under the terms of the GNU Lesser General Public
   5 // License as published by the Free Software Foundation; either
   6 // version 2.1 of the License, or (at your option) any later version.
   7 //
   8 // This library is distributed in the hope that it will be useful,
   9 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  11 // Lesser General Public License for more details.
  12 //
  13 // You should have received a copy of the GNU Lesser General Public
  14 // License along with this library; if not, write to the Free Software
  15 // Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  16 //
  17 // See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
  18 //
  19
  20 #include "MPIAccess.hxx"
  21 #include "InterpolationUtils.hxx"
  22
  23 #include <iostream>
  24
  25 using namespace std;
  26
  27 namespace MEDCoupling
  28 {
  29   /**!
  30     \anchor MPIAccess-det
  31     \class MPIAccess
  32
  33     The class \a MPIAccess is the gateway to the MPI library.
  34     It is a helper class that gathers the calls to the MPI
  35     library that are made in the ParaMEDMEM library. This gathering
  36     allows easier gathering of information about the communication
  37     in the library. With MPIAccess, tags are managed automatically
  38     and asynchronous operations are easier.
  39
  40     It is typically called after the MPI_Init() call in a program. It is afterwards passed as a parameter to the constructors of ParaMEDMEM objects so that they access the MPI library via the MPIAccess.
  41
  42     As an example, the following code initializes a processor group made of the zero processor.
  43
  44     \verbatim
  45     #include "MPIAccess.hxx"
  46     #include "ProcessorGroup.hxx"
  47
  48     int main(int argc, char** argv)
  49     {
  50     //initialization
  51     MPI_Init(&argc, &argv);
  52     MEDCoupling::CommInterface comm_interface;
  53
  54     //setting up a processor group with proc 0
  55     set<int> procs;
  56     procs.insert(0);
  57     MEDCoupling::ProcessorGroup group(procs, comm_interface);
  58
  59     MEDCoupling::MPIAccess mpi_access(group);
  60
  61     //cleanup
  62     MPI_Finalize();
  63     }
  64     \endverbatim
  65   */
  66
  67
  68   /*! Creates a MPIAccess that is based on the processors included in \a ProcessorGroup.
  69     This class may be called for easier use of MPI API.
  70
  71     \param ProcessorGroup MPIProcessorGroup object giving access to group management
  72     \param BaseTag and MaxTag define the range of tags to be used.
  73     Tags are managed by MPIAccess. They are cyclically incremented.
  74     When there is a Send or a Receive operation there is a new RequestId tag returned
  75     to the caller. That RequestId may be used to manage the operation Wait, Check of
  76     status etc... The MPITag internally managed by MPIAccess is used as "tag" argument
  77     in MPI call.
  78   */
  79
  80   MPIAccess::MPIAccess(MPIProcessorGroup * ProcessorGroup, int BaseTag, int MaxTag) :
  81     _comm_interface( ProcessorGroup->getCommInterface() ) ,
  82     _intra_communicator( ProcessorGroup->getComm() )
  83   {
  84     void *v ;
  85     int mpitagub ;
  86     int flag ;
  87     //MPI_Comm_get_attr does not run with _IntraCommunicator ???
  88     //MPI_Comm_get_attr(*_IntraCommunicator,MPID_TAG_UB,&mpitagub,&flag) ;
  89     MPI_Comm_get_attr(MPI_COMM_WORLD,MPI_TAG_UB,&v,&flag) ;
  90     mpitagub=*(reinterpret_cast<int*>(v));
  91     if ( BaseTag != 0 )
  92       BaseTag = (BaseTag/MODULO_TAG)*MODULO_TAG ;
  93     if ( MaxTag == 0 )
  94       MaxTag = (mpitagub/MODULO_TAG-1)*MODULO_TAG ;
  95     MPI_Comm_rank( *_intra_communicator, &_my_rank ) ;
  96     if ( !flag | (BaseTag < 0) | (BaseTag >= MaxTag) | (MaxTag > mpitagub) )
  97       throw INTERP_KERNEL::Exception("wrong call to MPIAccess constructor");
  98
  99     _processor_group = ProcessorGroup ;
 100     _processor_group_size = _processor_group->size() ;
 101     _trace = false ;
 102
 103     _base_request = -1 ;
 104     _max_request = std::numeric_limits<int>::max() ;
 105     _request = _base_request ;
 106
 107     _base_MPI_tag = BaseTag ;
 108     _max_MPI_tag = MaxTag ;
 109
 110     _send_request = new int[ _processor_group_size ] ;
 111     _recv_request = new int[ _processor_group_size ] ;
 112
 113     _send_requests.resize( _processor_group_size ) ;
 114     _recv_requests.resize( _processor_group_size ) ;
 115
 116     _send_MPI_tag = new int[ _processor_group_size ] ;
 117     _recv_MPI_Tag = new int[ _processor_group_size ] ;
 118     int i ;
 119     for (i = 0 ; i < _processor_group_size ; i++ )
 120       {
 121         _send_request[ i ] = _max_request ;
 122         _recv_request[ i ] = _max_request ;
 123         _send_requests[ i ].resize(0) ;
 124         _recv_requests[ i ].resize(0) ;
 125         _send_MPI_tag[ i ] = _max_MPI_tag ;
 126         _recv_MPI_Tag[ i ] = _max_MPI_tag ;
 127       }
 128     MPI_Datatype array_of_types[3] ;
 129     array_of_types[0] = MPI_DOUBLE ;
 130     array_of_types[1] = MPI_DOUBLE ;
 131     array_of_types[2] = MPI_INT ;
 132     int array_of_blocklengths[3] ;
 133     array_of_blocklengths[0] = 1 ;
 134     array_of_blocklengths[1] = 1 ;
 135     array_of_blocklengths[2] = 1 ;
 136     MPI_Aint array_of_displacements[3] ;
 137     array_of_displacements[0] = 0 ;
 138     array_of_displacements[1] = sizeof(double) ;
 139     array_of_displacements[2] = 2*sizeof(double) ;
 140     MPI_Type_create_struct(3, array_of_blocklengths, array_of_displacements,
 141                     array_of_types, &_MPI_TIME) ;
 142     MPI_Type_commit(&_MPI_TIME) ;
 143   }
 144
 145   MPIAccess::~MPIAccess()
 146   {
 147     delete [] _send_request ;
 148     delete [] _recv_request ;
 149     delete [] _send_MPI_tag ;
 150     delete [] _recv_MPI_Tag ;
 151     MPI_Type_free(&_MPI_TIME) ;
 152   }
 153
 154   /*
 155     MPIAccess and "RequestIds" :
 156     ============================
 157
 158     . WARNING : In the specification document, the distinction
 159     between "MPITags" and "RequestIds" is not clear. "MPITags"
 160     are arguments of calls to MPI. "RequestIds" does not concern
 161     calls to MPI. "RequestIds" are named "tag"as arguments in/out
 162     in the MPIAccess API in the specification documentation.
 163     But in the implementation we have the right name RequestId (or
 164     RecvRequestId/SendRequestId).
 165
 166     . When we have a MPI write/read request via MPIAccess, we get
 167     an identifier "RequestId".
 168     That identifier matches a  structure RequestStruct of
 169     MPIAccess. The access to that structure is done with the map
 170     "_MapOfRequestStruct".
 171     That structure RequestStruct give the possibility to manage
 172     the structures MPI_Request and MPI_Status * of MPI. It give
 173     also the possibility to get information about that request :
 174     target, send/recv, tag, [a]synchronous, type, outcount.
 175
 176     . That identifier is used to control an asynchronous request
 177     via MPIAccess : Wait, Test, Probe, etc...
 178
 179     . In practise "RequestId" is simply an integer fo the interval
 180     [0 , 2**32-1]. There is only one such a cyclic for
 181     [I]Sends and [I]Recvs.
 182
 183     . That "RequestIds" and their associated structures give an easy
 184     way to manage asynchronous communications.
 185     For example we have mpi_access->Wait( int RequestId ) instead of
 186     MPI_Wait(MPI_Request *request, MPI_Status *status).
 187
 188     . The API of MPIAccess may give the "SendRequestIds" of a "target",
 189     the "RecvRequestIds" from a "source" or the "SendRequestIds" of
 190     all "targets" or the "RecvRequestIds" of all "sources".
 191     That avoid to manage them in Presentation-ParaMEDMEM.
 192   */
 193
 194   int MPIAccess::newRequest( MPI_Datatype datatype, int tag , int destsourcerank ,
 195                              bool fromsourcerank , bool asynchronous )
 196   {
 197     RequestStruct *mpiaccessstruct = new RequestStruct;
 198     mpiaccessstruct->MPITag = tag ;
 199     mpiaccessstruct->MPIDatatype = datatype ;
 200     mpiaccessstruct->MPITarget = destsourcerank ;
 201     mpiaccessstruct->MPIIsRecv = fromsourcerank ;
 202     MPI_Status *aStatus = new MPI_Status ;
 203     mpiaccessstruct->MPIStatus = aStatus ;
 204     mpiaccessstruct->MPIAsynchronous = asynchronous ;
 205     mpiaccessstruct->MPICompleted = !asynchronous ;
 206     mpiaccessstruct->MPIOutCount = -1 ;
 207     if ( !asynchronous )
 208       {
 209         mpiaccessstruct->MPIRequest = MPI_REQUEST_NULL ;
 210         mpiaccessstruct->MPIStatus->MPI_SOURCE = destsourcerank ;
 211         mpiaccessstruct->MPIStatus->MPI_TAG = tag ;
 212         mpiaccessstruct->MPIStatus->MPI_ERROR = MPI_SUCCESS ;
 213       }
 214     if ( _request == _max_request )
 215       _request = _base_request ;
 216     _request += 1 ;
 217     _map_of_request_struct[_request] = mpiaccessstruct ;
 218     if ( fromsourcerank )
 219       _recv_request[ destsourcerank ] = _request;
 220     else
 221       _send_request[ destsourcerank ] = _request;
 222     if ( _trace )
 223       cout << "NewRequest" << _my_rank << "( " << _request << " ) "
 224            << mpiaccessstruct << endl ;
 225     return _request ;
 226   }
 227
 228   /*
 229     MPIAccess and "tags" (or "MPITags") :
 230     =====================================
 231
 232     . The constructor give the possibility to choose an interval of
 233     tags to use : [BaseTag , MaxTag].
 234     The default is [ 0 , MPI_TAG_UB], MPI_TAG_UB being the maximum
 235     value in an implementation of MPI (minimum 32767 = 2**15-1).
 236     On awa with the implementation lam MPI_TAG_UB value is
 237     7353944. The norm MPI specify that value is the same in all
 238     processes started by mpirun.
 239     In the case of the use of the same IntraCommunicator in a process
 240     for several distinct data flows (or for several IntraCommunicators
 241     with common processes), that permits to avoid ambiguity
 242     and may help debug.
 243
 244     . In MPIAccess the tags have two parts (#define MODULO_TAG 10) :
 245     + The last decimal digit decimal correspond to MPI_DataType ( 1 for
 246     TimeMessages, 2 for MPI_INT and 3 for MPI_DOUBLE)
 247     + The value of other digits correspond to a circular number for each
 248     message.
 249     + A TimeMessage and the associated DataMessage have the same number
 250     (but the types are different and the tags also).
 251
 252     . For a Send of a message from a process "source" to a process
 253     "target", we have _send_MPI_tag[target] in the process
 254     source (it contains the last "tag" used for the Send of a
 255     message to the process target).
 256     And in the process "target" which receive that message, we have
 257     _recv_MPI_Tag[source] (it contains the last "tag" used for the Recv
 258     of messages from the process source).
 259     Naturally in the MPI norm the values of that tags must be the same.
 260   */
 261   int MPIAccess::newSendTag( MPI_Datatype datatype, int destrank , int method ,
 262                              bool asynchronous, int &RequestId )
 263   {
 264     int tag ;
 265     tag = incrTag( _send_MPI_tag[destrank] ) ;
 266     tag = valTag( tag, method ) ;
 267     _send_MPI_tag[ destrank ] = tag ;
 268     RequestId = newRequest( datatype, tag, destrank , false , asynchronous ) ;
 269     _send_request[ destrank ] = RequestId ;
 270     _send_requests[ destrank ].push_back( RequestId ) ;
 271     return tag ;
 272   }
 273
 274   int MPIAccess::newRecvTag( MPI_Datatype datatype, int sourcerank , int method ,
 275                              bool asynchronous, int &RequestId )
 276   {
 277     int tag ;
 278     tag = incrTag( _recv_MPI_Tag[sourcerank] ) ;
 279     tag = valTag( tag, method ) ;
 280     _recv_MPI_Tag[ sourcerank ] = tag ;
 281     RequestId = newRequest( datatype, tag , sourcerank , true , asynchronous ) ;
 282     _recv_request[ sourcerank ] = RequestId ;
 283     _recv_requests[ sourcerank ].push_back( RequestId ) ;
 284     return tag ;
 285   }
 286
 287   // Returns the number of all SendRequestIds that may be used to allocate
 288   // ArrayOfSendRequests for the call to SendRequestIds
 289   int MPIAccess::sendRequestIdsSize()
 290   {
 291     int size = 0;
 292     for (int i = 0 ; i < _processor_group_size ; i++ )
 293       size += _send_requests[ i ].size() ;
 294     return size ;
 295   }
 296
 297   // Returns in ArrayOfSendRequests with the dimension "size" all the
 298   // SendRequestIds
 299   int MPIAccess::sendRequestIds(int size, int *ArrayOfSendRequests)
 300   {
 301     int destrank ;
 302     int i = 0 ;
 303     for ( destrank = 0 ; destrank < _processor_group_size ; destrank++ )
 304       {
 305         list< int >::const_iterator iter ;
 306         for (iter = _send_requests[ destrank ].begin() ; iter != _send_requests[destrank].end() ; iter++ )
 307           ArrayOfSendRequests[i++] = *iter ;
 308       }
 309     return i ;
 310   }
 311
 312   // Returns the number of all RecvRequestIds that may be used to allocate
 313   // ArrayOfRecvRequests for the call to RecvRequestIds
 314   int MPIAccess::recvRequestIdsSize()
 315   {
 316     int size = 0 ;
 317     for (int i = 0 ; i < _processor_group_size ; i++ )
 318       size += _recv_requests[ i ].size() ;
 319     return size ;
 320   }
 321
 322   // Returns in ArrayOfRecvRequests with the dimension "size" all the
 323   // RecvRequestIds
 324   int MPIAccess::recvRequestIds(int size, int *ArrayOfRecvRequests)
 325   {
 326     int sourcerank ;
 327     int i = 0 ;
 328     for ( sourcerank = 0 ; sourcerank < _processor_group_size ; sourcerank++ )
 329       {
 330         list< int >::const_iterator iter ;
 331         for (iter = _recv_requests[ sourcerank ].begin() ; iter != _recv_requests[sourcerank].end() ; iter++ )
 332           ArrayOfRecvRequests[i++] = *iter ;
 333       }
 334     return i ;
 335   }
 336
 337   // Returns in ArrayOfSendRequests with the dimension "size" all the
 338   // SendRequestIds to a destination rank
 339   int MPIAccess::sendRequestIds(int destrank, int size, int *ArrayOfSendRequests)
 340   {
 341     if (size < (int)_send_requests[destrank].size() )
 342       throw INTERP_KERNEL::Exception("wrong call to MPIAccess::SendRequestIds");
 343     int i = 0 ;
 344     list< int >::const_iterator iter ;
 345     for (iter = _send_requests[ destrank ].begin() ; iter != _send_requests[destrank].end() ; iter++ )
 346       ArrayOfSendRequests[i++] = *iter ;
 347     return _send_requests[destrank].size() ;
 348   }
 349
 350   // Returns in ArrayOfRecvRequests with the dimension "size" all the
 351   // RecvRequestIds from a sourcerank
 352   int MPIAccess::recvRequestIds(int sourcerank, int size, int *ArrayOfRecvRequests)
 353   {
 354     if (size < (int)_recv_requests[sourcerank].size() )
 355       throw INTERP_KERNEL::Exception("wrong call to MPIAccess::RecvRequestIds");
 356     int i = 0 ;
 357     list< int >::const_iterator iter ;
 358     _recv_requests[ sourcerank ] ;
 359     for (iter = _recv_requests[ sourcerank ].begin() ; iter != _recv_requests[sourcerank].end() ; iter++ )
 360       ArrayOfRecvRequests[i++] = *iter ;
 361     return _recv_requests[sourcerank].size() ;
 362   }
 363
 364   // Send in synchronous mode count values of type datatype from buffer to target
 365   // (returns RequestId identifier even if the corresponding structure is deleted :
 366   // it is only in order to have the same signature as the asynchronous mode)
 367   int MPIAccess::send(void* buffer, int count, MPI_Datatype datatype, int target, int &RequestId)
 368   {
 369     int sts = MPI_SUCCESS ;
 370     RequestId = -1 ;
 371     if ( count )
 372       {
 373         _MessageIdent aMethodIdent = methodId( datatype ) ;
 374         int MPItag = newSendTag( datatype, target , aMethodIdent , false , RequestId ) ;
 375         if ( aMethodIdent == _message_time )
 376           {
 377             TimeMessage *aTimeMsg = (TimeMessage *) buffer ;
 378             aTimeMsg->tag = MPItag ;
 379           }
 380         deleteRequest( RequestId ) ;
 381         sts = _comm_interface.send(buffer, count, datatype, target, MPItag,
 382                                   *_intra_communicator ) ;
 383         if ( _trace )
 384           cout << "MPIAccess::Send" << _my_rank << " SendRequestId "
 385                << RequestId << " count " << count << " target " << target
 386                << " MPItag " << MPItag << endl ;
 387       }
 388     return sts ;
 389   }
 390
 391   // Receive (read) in synchronous mode count values of type datatype in buffer from source
 392   // (returns RequestId identifier even if the corresponding structure is deleted :
 393   // it is only in order to have the same signature as the asynchronous mode)
 394   // The output argument OutCount is optional : *OutCount <= count
 395   int MPIAccess::recv(void* buffer, int count, MPI_Datatype datatype, int source, int &RequestId, int *OutCount)
 396   {
 397     int sts = MPI_SUCCESS ;
 398     RequestId = -1 ;
 399     if ( OutCount != NULL )
 400       *OutCount = -1 ;
 401     if ( count )
 402       {
 403         _MessageIdent aMethodIdent = methodId( datatype ) ;
 404         int MPItag = newRecvTag( datatype, source , aMethodIdent , false , RequestId ) ;
 405         sts =  _comm_interface.recv(buffer, count, datatype, source, MPItag,
 406                                    *_intra_communicator , MPIStatus( RequestId ) ) ;
 407         int outcount = 0 ;
 408         if ( sts == MPI_SUCCESS )
 409           {
 410             MPI_Datatype datatype = MPIDatatype( RequestId ) ;
 411             _comm_interface.getCount(MPIStatus( RequestId ), datatype, &outcount ) ;
 412             setMPIOutCount( RequestId , outcount ) ;
 413             setMPICompleted( RequestId , true ) ;
 414             deleteStatus( RequestId ) ;
 415           }
 416         if ( OutCount != NULL )
 417           *OutCount = outcount ;
 418         if ( _trace )
 419           cout << "MPIAccess::Recv" << _my_rank << " RecvRequestId "
 420                << RequestId << " count " << count << " source " << source
 421                << " MPItag " << MPItag << endl ;
 422         deleteRequest( RequestId ) ;
 423       }
 424     return sts ;
 425   }
 426
 427   // Send in asynchronous mode count values of type datatype from buffer to target
 428   // Returns RequestId identifier.
 429   int MPIAccess::ISend(void* buffer, int count, MPI_Datatype datatype, int target, int &RequestId)
 430   {
 431     int sts = MPI_SUCCESS ;
 432     RequestId = -1 ;
 433     if ( count )
 434       {
 435         _MessageIdent aMethodIdent = methodId( datatype ) ;
 436         int MPItag = newSendTag( datatype, target , aMethodIdent , true , RequestId ) ;
 437         if ( aMethodIdent == _message_time )
 438           {
 439             TimeMessage *aTimeMsg = (TimeMessage *) buffer ;
 440             aTimeMsg->tag = MPItag ;
 441           }
 442         MPI_Request *aSendRequest = MPIRequest( RequestId ) ;
 443         if ( _trace )
 444           {
 445             cout << "MPIAccess::ISend" << _my_rank << " ISendRequestId "
 446                  << RequestId << " count " << count << " target " << target
 447                  << " MPItag " << MPItag << endl ;
 448             if ( MPItag == 1 )
 449               cout << "MPIAccess::ISend" << _my_rank << " time "
 450                    << ((TimeMessage *)buffer)->time << " " << ((TimeMessage *)buffer)->deltatime
 451                    << endl ;
 452           }
 453         sts = _comm_interface.Isend(buffer, count, datatype, target, MPItag,
 454                                    *_intra_communicator , aSendRequest) ;
 455       }
 456     return sts ;
 457   }
 458
 459   // Receive (read) in asynchronous mode count values of type datatype in buffer from source
 460   // returns RequestId identifier.
 461   int MPIAccess::IRecv(void* buffer, int count, MPI_Datatype datatype, int source, int &RequestId)
 462   {
 463     int sts = MPI_SUCCESS ;
 464     RequestId = -1 ;
 465     if ( count )
 466       {
 467         _MessageIdent aMethodIdent = methodId( datatype ) ;
 468         int MPItag = newRecvTag( datatype, source , aMethodIdent , true , RequestId ) ;
 469         MPI_Request *aRecvRequest = MPIRequest( RequestId ) ;
 470         if ( _trace )
 471           {
 472             cout << "MPIAccess::IRecv" << _my_rank << " IRecvRequestId "
 473                  << RequestId << " count " << count << " source " << source
 474                  << " MPItag " << MPItag << endl ;
 475             if ( MPItag == 1 )
 476               cout << "MPIAccess::ISend" << _my_rank << " time "
 477                    << ((TimeMessage *)buffer)->time << " " << ((TimeMessage *)buffer)->deltatime
 478                    << endl ;
 479           }
 480         sts = _comm_interface.Irecv(buffer, count, datatype, source, MPItag,
 481                                    *_intra_communicator , aRecvRequest) ;
 482       }
 483     return sts ;
 484   }
 485
 486   // Perform a Send and a Recv in synchronous mode
 487   int MPIAccess::sendRecv(void* sendbuf, int sendcount, MPI_Datatype sendtype,
 488                           int dest, int &SendRequestId,
 489                           void* recvbuf, int recvcount, MPI_Datatype recvtype,
 490                           int source, int &RecvRequestId, int *OutCount)
 491   {
 492     int sts = MPI_SUCCESS ;
 493     SendRequestId = -1 ;
 494     RecvRequestId = -1 ;
 495     if ( recvcount )
 496       sts = IRecv(recvbuf, recvcount, recvtype, source, RecvRequestId) ;
 497     int outcount = -1 ;
 498     if ( _trace )
 499       cout << "MPIAccess::SendRecv" << _my_rank << " IRecv RecvRequestId "
 500            << RecvRequestId << endl ;
 501     if ( sts == MPI_SUCCESS )
 502       {
 503         if ( sendcount )
 504           sts = send(sendbuf, sendcount, sendtype, dest, SendRequestId) ;
 505         if ( _trace )
 506           cout << "MPIAccess::SendRecv" << _my_rank << " Send SendRequestId "
 507                << SendRequestId << endl ;
 508         if ( sts == MPI_SUCCESS && recvcount )
 509           {
 510             sts = wait( RecvRequestId ) ;
 511             outcount = MPIOutCount( RecvRequestId ) ;
 512             if ( _trace )
 513               cout << "MPIAccess::SendRecv" << _my_rank << " IRecv RecvRequestId "
 514                    << RecvRequestId << " outcount " << outcount << endl ;
 515           }
 516       }
 517     if ( OutCount != NULL )
 518       {
 519         *OutCount = outcount ;
 520         if ( _trace )
 521           cout << "MPIAccess::SendRecv" << _my_rank << " *OutCount = " << *OutCount
 522                << endl ;
 523       }
 524     deleteRequest( RecvRequestId ) ;
 525     return sts ;
 526   }
 527
 528   // Perform a Send and a Recv in asynchronous mode
 529   int MPIAccess::ISendRecv(void* sendbuf, int sendcount, MPI_Datatype sendtype,
 530                            int dest, int &SendRequestId,
 531                            void* recvbuf, int recvcount, MPI_Datatype recvtype,
 532                            int source, int &RecvRequestId)
 533   {
 534     int sts = MPI_SUCCESS ;
 535     SendRequestId = -1 ;
 536     RecvRequestId = -1 ;
 537     if ( recvcount )
 538       sts = IRecv(recvbuf, recvcount, recvtype, source, RecvRequestId) ;
 539     if ( sts == MPI_SUCCESS )
 540       if ( sendcount )
 541         sts = ISend(sendbuf, sendcount, sendtype, dest, SendRequestId) ;
 542     return sts ;
 543   }
 544
 545   // Perform a wait of a Send or Recv asynchronous Request
 546   // Do nothing for a synchronous Request
 547   // Manage MPI_Request * and MPI_Status * structure
 548   int MPIAccess::wait( int RequestId )
 549   {
 550     int status = MPI_SUCCESS ;
 551     if ( !MPICompleted( RequestId ) )
 552       {
 553         if ( *MPIRequest( RequestId ) != MPI_REQUEST_NULL )
 554           {
 555             if ( _trace )
 556               cout << "MPIAccess::Wait" << _my_rank << " -> wait( " << RequestId
 557                    << " ) MPIRequest " << MPIRequest( RequestId ) << " MPIStatus "
 558                    << MPIStatus( RequestId ) << " MPITag " << MPITag( RequestId )
 559                    << " MPIIsRecv " << MPIIsRecv( RequestId ) << endl ;
 560             status = _comm_interface.wait(MPIRequest( RequestId ), MPIStatus( RequestId )) ;
 561           }
 562         else
 563           {
 564             if ( _trace )
 565               cout << "MPIAccess::Wait" << _my_rank << " MPIRequest == MPI_REQUEST_NULL"
 566                    << endl ;
 567           }
 568         setMPICompleted( RequestId , true ) ;
 569         if ( MPIIsRecv( RequestId ) && MPIStatus( RequestId ) )
 570           {
 571             MPI_Datatype datatype = MPIDatatype( RequestId ) ;
 572             int outcount ;
 573             status = _comm_interface.getCount(MPIStatus( RequestId ), datatype,
 574                                              &outcount ) ;
 575             if ( status == MPI_SUCCESS )
 576               {
 577                 setMPIOutCount( RequestId , outcount ) ;
 578                 deleteStatus( RequestId ) ;
 579                 if ( _trace )
 580                   cout << "MPIAccess::Wait" << _my_rank << " RequestId " << RequestId
 581                        << "MPIIsRecv " << MPIIsRecv( RequestId ) << " outcount " << outcount
 582                        << endl ;
 583               }
 584             else
 585               {
 586                 if ( _trace )
 587                   cout << "MPIAccess::Wait" << _my_rank << " MPIIsRecv "
 588                        << MPIIsRecv( RequestId ) << " outcount " << outcount << endl ;
 589               }
 590           }
 591         else
 592           {
 593             if ( _trace )
 594               cout << "MPIAccess::Wait" << _my_rank << " MPIIsRecv " << MPIIsRecv( RequestId )
 595                    << " MPIOutCount " << MPIOutCount( RequestId ) << endl ;
 596           }
 597       }
 598     if ( _trace )
 599       cout << "MPIAccess::Wait" << _my_rank << " RequestId " << RequestId
 600            << " Request " << MPIRequest( RequestId )
 601            << " Status " << MPIStatus( RequestId ) << " MPICompleted "
 602            << MPICompleted( RequestId ) << " MPIOutCount " << MPIOutCount( RequestId )
 603            << endl ;
 604     return status ;
 605   }
 606
 607   // Perform a "test" of a Send or Recv asynchronous Request
 608   // If the request is done, returns true in the flag argument
 609   // If the request is not finished, returns false in the flag argument
 610   // Do nothing for a synchronous Request
 611   // Manage MPI_request * and MPI_status * structure
 612   int MPIAccess::test(int RequestId, int &flag)
 613   {
 614     int status = MPI_SUCCESS ;
 615     flag = MPICompleted( RequestId ) ;
 616     if ( _trace )
 617       cout << "MPIAccess::Test" << _my_rank << " flag " << flag ;
 618     if ( MPIIsRecv( RequestId ) )
 619       {
 620         if ( _trace )
 621           cout << " Recv" ;
 622       }
 623     else
 624       {
 625         if ( _trace )
 626           cout << " Send" ;
 627       }
 628     if( _trace )
 629       cout << "Request" << RequestId << " " << MPIRequest( RequestId )
 630            << " Status " << MPIStatus( RequestId ) << endl ;
 631     if ( !flag )
 632       {
 633         if ( *MPIRequest( RequestId ) != MPI_REQUEST_NULL )
 634           {
 635             if ( _trace )
 636               cout << "MPIAccess::Test" << _my_rank << " -> test( " << RequestId
 637                    << " ) MPIRequest " << MPIRequest( RequestId )
 638                    << " MPIStatus " << MPIStatus( RequestId )
 639                    << " MPITag " << MPITag( RequestId )
 640                    << " MPIIsRecv " << MPIIsRecv( RequestId ) << endl ;
 641             status = _comm_interface.test(MPIRequest( RequestId ), &flag,
 642                                          MPIStatus( RequestId )) ;
 643           }
 644         else
 645           {
 646             if ( _trace )
 647               cout << "MPIAccess::Test" << _my_rank << " MPIRequest == MPI_REQUEST_NULL"
 648                    << endl ;
 649           }
 650         if ( flag )
 651           {
 652             setMPICompleted( RequestId , true ) ;
 653             if ( MPIIsRecv( RequestId ) && MPIStatus( RequestId ) )
 654               {
 655                 int outcount ;
 656                 MPI_Datatype datatype = MPIDatatype( RequestId ) ;
 657                 status = _comm_interface.getCount( MPIStatus( RequestId ), datatype,
 658                                                   &outcount ) ;
 659                 if ( status == MPI_SUCCESS )
 660                   {
 661                     setMPIOutCount( RequestId , outcount ) ;
 662                     deleteStatus( RequestId ) ;
 663                     if ( _trace )
 664                       cout << "MPIAccess::Test" << _my_rank << " MPIIsRecv "
 665                            << MPIIsRecv( RequestId ) << " outcount " << outcount << endl ;
 666                   }
 667                 else
 668                   {
 669                     if ( _trace )
 670                       cout << "MPIAccess::Test" << _my_rank << " MPIIsRecv "
 671                            << MPIIsRecv( RequestId ) << " outcount " << outcount << endl ;
 672                   }
 673               }
 674             else
 675               {
 676                 if ( _trace )
 677                   cout << "MPIAccess::Test" << _my_rank << " MPIIsRecv "
 678                        << MPIIsRecv( RequestId ) << " MPIOutCount "
 679                        << MPIOutCount( RequestId ) << endl ;
 680               }
 681           }
 682       }
 683     if ( _trace )
 684       cout << "MPIAccess::Test" << _my_rank << " RequestId " << RequestId
 685            << " flag " << flag << " MPICompleted " << MPICompleted( RequestId )
 686            << " MPIOutCount " << MPIOutCount( RequestId ) << endl ;
 687     return status ;
 688   }
 689
 690   int MPIAccess::waitAny(int count, int *array_of_RequestIds, int &RequestId)
 691   {
 692     int status = MPI_ERR_OTHER ;
 693     RequestId = -1 ;
 694     cout << "MPIAccess::WaitAny not yet implemented" << endl ;
 695     return status ;
 696   }
 697
 698   int MPIAccess::testAny(int count, int *array_of_RequestIds, int &RequestId, int &flag)
 699   {
 700     int status = MPI_ERR_OTHER ;
 701     RequestId = -1 ;
 702     flag = 0 ;
 703     cout << "MPIAccess::TestAny not yet implemented" << endl ;
 704     return status ;
 705   }
 706
 707   // Perform a wait of each Send or Recv asynchronous Request of the array
 708   // array_of_RequestIds of size "count".
 709   // That array may be filled with a call to SendRequestIdsSize or RecvRequestIdsSize
 710   // Do nothing for a synchronous Request
 711   // Manage MPI_Request * and MPI_Status * structure
 712   int MPIAccess::waitAll(int count, int *array_of_RequestIds)
 713   {
 714     if ( _trace )
 715       cout << "WaitAll" << _my_rank << " : count " << count << endl ;
 716     int status ;
 717     int retstatus = MPI_SUCCESS ;
 718     int i ;
 719     for ( i = 0 ; i < count ; i++ )
 720       {
 721         if ( _trace )
 722           cout << "WaitAll" << _my_rank << " " << i << " -> Wait( "
 723                << array_of_RequestIds[i] << " )" << endl ;
 724         status = wait( array_of_RequestIds[i] ) ;
 725         if ( status != MPI_SUCCESS )
 726           retstatus = status ;
 727       }
 728     if ( _trace )
 729       cout << "EndWaitAll" << _my_rank << endl ;
 730     return retstatus ;
 731   }
 732
 733   // Perform a "test" of each Send or Recv asynchronous Request of the array
 734   // array_of_RequestIds of size "count".
 735   // That array may be filled with a call to SendRequestIdsSize or RecvRequestIdsSize
 736   // If all requests are done, returns true in the flag argument
 737   // If all requests are not finished, returns false in the flag argument
 738   // Do nothing for a synchronous Request
 739   // Manage MPI_Request * and MPI_Status * structure
 740   int MPIAccess::testAll(int count, int *array_of_RequestIds, int &flag)
 741   {
 742     if ( _trace )
 743       cout << "TestAll" << _my_rank << " : count " << count << endl ;
 744     int status ;
 745     int retstatus = MPI_SUCCESS ;
 746     bool retflag = true ;
 747     int i ;
 748     for ( i = 0 ; i < count ; i++ )
 749       {
 750         status = test( array_of_RequestIds[i] , flag ) ;
 751         retflag = retflag && (flag != 0) ;
 752         if ( status != MPI_SUCCESS )
 753           retstatus = status ;
 754       }
 755     flag = retflag ;
 756     if ( _trace )
 757       cout << "EndTestAll" << _my_rank << endl ;
 758     return retstatus ;
 759   }
 760
 761   int MPIAccess::waitSome(int count, int *array_of_RequestIds, int outcount,
 762                           int *outarray_of_RequestIds)
 763   {
 764     int status = MPI_ERR_OTHER ;
 765     cout << "MPIAccess::WaitSome not yet implemented" << endl ;
 766     return status ;
 767   }
 768
 769   int MPIAccess::testSome(int count, int *array_of_RequestIds, int outcounts,
 770                           int *outarray_of_RequestIds)
 771   {
 772     int status = MPI_ERR_OTHER ;
 773     cout << "MPIAccess::TestSome not yet implemented" << endl ;
 774     return status ;
 775   }
 776
 777   // Probe checks if a message is available for read from FromSource rank.
 778   // Returns the corresponding source, MPITag, datatype and outcount
 779   // Probe is a blocking call which wait until a message is available
 780   int MPIAccess::probe(int FromSource, int &source, int &MPITag,
 781                        MPI_Datatype &myDatatype, int &outcount)
 782   {
 783     MPI_Status aMPIStatus ;
 784     int sts =  _comm_interface.probe( FromSource, MPI_ANY_TAG,
 785                                      *_intra_communicator , &aMPIStatus ) ;
 786     if ( sts == MPI_SUCCESS )
 787       {
 788         source = aMPIStatus.MPI_SOURCE ;
 789         MPITag = aMPIStatus.MPI_TAG ;
 790         int MethodId = (MPITag % MODULO_TAG) ;
 791         myDatatype = datatype( (MEDCoupling::_MessageIdent) MethodId ) ;
 792         _comm_interface.getCount(&aMPIStatus, myDatatype, &outcount ) ;
 793         if ( _trace )
 794           cout << "MPIAccess::Probe" << _my_rank << " FromSource " << FromSource
 795                << " source " << source << " MPITag " << MPITag << " MethodId "
 796                << MethodId << " datatype " << myDatatype << " outcount " << outcount
 797                << endl ;
 798       }
 799     else
 800       {
 801         source = -1 ;
 802         MPITag = -1 ;
 803         myDatatype = 0 ;
 804         outcount = -1 ;
 805       }
 806     return sts ;
 807   }
 808
 809   // IProbe checks if a message is available for read from FromSource rank.
 810   // If there is a message available, returns the corresponding source,
 811   // MPITag, datatype and outcount with flag = true
 812   // If not, returns flag = false
 813   int MPIAccess::IProbe(int FromSource, int &source, int &MPITag,
 814                         MPI_Datatype &myDataType, int &outcount, int &flag)
 815   {
 816     MPI_Status aMPIStatus ;
 817     int sts =  _comm_interface.Iprobe( FromSource, MPI_ANY_TAG,
 818                                       *_intra_communicator , &flag,
 819                                       &aMPIStatus ) ;
 820     if ( sts == MPI_SUCCESS && flag )
 821       {
 822         source = aMPIStatus.MPI_SOURCE ;
 823         MPITag = aMPIStatus.MPI_TAG ;
 824         int MethodId = (MPITag % MODULO_TAG) ;
 825         myDataType = datatype( (MEDCoupling::_MessageIdent) MethodId ) ;
 826         _comm_interface.getCount(&aMPIStatus, myDataType, &outcount ) ;
 827         if ( _trace )
 828           cout << "MPIAccess::IProbe" << _my_rank << " FromSource " << FromSource
 829                << " source " << source << " MPITag " << MPITag << " MethodId "
 830                << MethodId << " datatype " << myDataType << " outcount " << outcount
 831                << " flag " << flag << endl ;
 832       }
 833     else
 834       {
 835         source = -1 ;
 836         MPITag = -1 ;
 837         myDataType = 0 ;
 838         outcount = -1 ;
 839       }
 840     return sts ;
 841   }
 842
 843   // Cancel concerns a "posted" asynchronous IRecv
 844   // Returns flag = true if the receiving request was successfully canceled
 845   // Returns flag = false if the receiving request was finished but not canceled
 846   // Use cancel, wait and test_cancelled of the MPI API
 847   int MPIAccess::cancel( int RecvRequestId, int &flag )
 848   {
 849     flag = 0 ;
 850     int sts = _comm_interface.cancel( MPIRequest( RecvRequestId ) ) ;
 851     if ( sts == MPI_SUCCESS )
 852       {
 853         sts = _comm_interface.wait( MPIRequest( RecvRequestId ) ,
 854                                    MPIStatus( RecvRequestId ) ) ;
 855         if ( sts == MPI_SUCCESS )
 856           sts = _comm_interface.testCancelled( MPIStatus( RecvRequestId ) , &flag ) ;
 857       }
 858     return sts ;
 859   }
 860
 861   // Cancel concerns a "pending" receiving message (without IRecv "posted")
 862   // Returns flag = true if the message was successfully canceled
 863   // Returns flag = false if the receiving request was finished but not canceled
 864   // Use Irecv, cancel, wait and test_cancelled of the MPI API
 865   int MPIAccess::cancel( int source, int theMPITag, MPI_Datatype datatype, int outcount, int &flag )
 866   {
 867     int sts ;
 868     MPI_Aint extent, lbound ;
 869     flag = 0 ;
 870     sts =  MPI_Type_get_extent( datatype , &lbound, &extent ) ;
 871     if ( sts == MPI_SUCCESS )
 872       {
 873         void * recvbuf = malloc( extent*outcount ) ;
 874         MPI_Request aRecvRequest ;
 875         if ( _trace )
 876           cout << "MPIAccess::Cancel" << _my_rank << " Irecv extent " << extent
 877                << " datatype " << datatype << " source " << source << " theMPITag "
 878                << theMPITag << endl ;
 879         sts = _comm_interface.Irecv( recvbuf, outcount, datatype, source, theMPITag,
 880                                     *_intra_communicator , &aRecvRequest ) ;
 881         if ( sts == MPI_SUCCESS )
 882           {
 883             sts = _comm_interface.cancel( &aRecvRequest ) ;
 884             if ( _trace )
 885               cout << "MPIAccess::Cancel" << _my_rank << " theMPITag " << theMPITag
 886                    << " cancel done" << endl ;
 887             if ( sts == MPI_SUCCESS )
 888               {
 889                 MPI_Status aStatus ;
 890                 if ( _trace )
 891                   cout << "MPIAccess::Cancel" << _my_rank << " wait" << endl ;
 892                 sts = _comm_interface.wait( &aRecvRequest , &aStatus ) ;
 893                 if ( sts == MPI_SUCCESS )
 894                   {
 895                     if ( _trace )
 896                       cout << "MPIAccess::Cancel" << _my_rank << " test_cancelled" << endl ;
 897                     sts = _comm_interface.testCancelled( &aStatus , &flag ) ;
 898                   }
 899               }
 900           }
 901         if ( _trace && datatype == timeType() )
 902           cout << "MPIAccess::Cancel" << _my_rank << " time "
 903                << ((TimeMessage *) recvbuf)->time << " "
 904                << ((TimeMessage *) recvbuf)->deltatime << endl ;
 905         free( recvbuf ) ;
 906       }
 907     if ( _trace )
 908       cout << "MPIAccess::Cancel" << _my_rank << " flag " << flag << endl ;
 909     return sts ;
 910   }
 911
 912
 913   // CancelAll concerns all "pending" receiving message (without IRecv "posted")
 914   // CancelAll use IProbe and Cancel (see obove)
 915   int MPIAccess::cancelAll()
 916   {
 917     int sts = MPI_SUCCESS ;
 918     int target ;
 919     int source ;
 920     int MPITag ;
 921     MPI_Datatype datatype ;
 922     int outcount ;
 923     int flag ;
 924     for ( target = 0 ; target < _processor_group_size ; target++ )
 925       {
 926         sts = IProbe(target, source, MPITag, datatype, outcount, flag) ;
 927         if ( sts == MPI_SUCCESS && flag )
 928           {
 929             sts = cancel(source, MPITag, datatype, outcount, flag) ;
 930             if ( _trace )
 931               cout << "MPIAccess::CancelAll" << _my_rank << " source " << source
 932                    << " MPITag " << MPITag << " datatype " << datatype
 933                    << " outcount " << outcount << " Cancel flag " << flag << endl ;
 934             if ( sts != MPI_SUCCESS )
 935               break ;
 936           }
 937         else if ( sts != MPI_SUCCESS )
 938           break ;
 939       }
 940     return sts ;
 941   }
 942
 943   // Same as barrier of MPI API
 944   int MPIAccess::barrier()
 945   {
 946     int status = _comm_interface.barrier( *_intra_communicator ) ;
 947     return status ;
 948   }
 949
 950   // Same as Error_string of MPI API
 951   int MPIAccess::errorString(int errorcode, char *string, int *resultlen) const
 952   {
 953     return _comm_interface.errorString( errorcode, string, resultlen) ;
 954   }
 955
 956   // Returns source, tag, error and outcount corresponding to receiving RequestId
 957   // By default the corresponding structure of RequestId is deleted
 958   int MPIAccess::status(int RequestId, int &source, int &tag, int &error,
 959                         int &outcount, bool keepRequestStruct)
 960   {
 961     MPI_Status *myStatus = MPIStatus( RequestId ) ;
 962     if ( _trace )
 963       cout << "MPIAccess::status" << _my_rank << " RequestId " << RequestId
 964            << " status " << myStatus << endl ;
 965     if ( myStatus != NULL && MPIAsynchronous( RequestId ) &&
 966          MPICompleted( RequestId ) )
 967       {
 968         if ( MPIIsRecv( RequestId ) )
 969           {
 970             source = myStatus->MPI_SOURCE ;
 971             tag = myStatus->MPI_TAG ;
 972             error = myStatus->MPI_ERROR ;
 973             MPI_Datatype datatype = MPIDatatype( RequestId ) ;
 974             _comm_interface.getCount(myStatus, datatype, &outcount ) ;
 975             if ( _trace )
 976               cout << "MPIAccess::status" << _my_rank << " RequestId " << RequestId
 977                    << " status " << myStatus << " outcount " << outcount << endl ;
 978             setMPIOutCount( RequestId , outcount ) ;
 979           }
 980         else
 981           {
 982             source = MPITarget( RequestId ) ;
 983             tag = MPITag( RequestId ) ;
 984             error = 0 ;
 985             outcount = MPIOutCount( RequestId ) ;
 986           }
 987         if ( !keepRequestStruct )
 988           deleteRequest( RequestId ) ;
 989         return MPI_SUCCESS ;
 990       }
 991     else
 992       {
 993         source = MPITarget( RequestId ) ;
 994         tag = MPITag( RequestId ) ;
 995         error = 0 ;
 996         outcount = MPIOutCount( RequestId ) ;
 997       }
 998     return MPI_SUCCESS ;
 999   }
1000
1001   int MPIAccess::requestFree( MPI_Request *request )
1002   {
1003     return _comm_interface.requestFree( request ) ;
1004   }
1005
1006   // Print all information of all known requests for debugging purpose
1007   void MPIAccess::check() const
1008   {
1009     int i = 0 ;
1010     map< int , RequestStruct * >::const_iterator MapOfRequestStructiterator ;
1011     cout << "MPIAccess::Check" << _my_rank << "_map_of_request_struct_size "
1012          << _map_of_request_struct.size() << endl ;
1013     for ( MapOfRequestStructiterator = _map_of_request_struct.begin() ;
1014           MapOfRequestStructiterator != _map_of_request_struct.end() ;
1015           MapOfRequestStructiterator++ )
1016       {
1017         if ( MapOfRequestStructiterator->second != NULL )
1018           {
1019             cout << "    Check" << _my_rank << " " << i << ". Request"
1020                  << MapOfRequestStructiterator->first << "-->" ;
1021             if ( (MapOfRequestStructiterator->second)->MPIAsynchronous )
1022               cout << "I" ;
1023             if ( (MapOfRequestStructiterator->second)->MPIIsRecv )
1024               cout << "Recv from " ;
1025             else
1026               cout << "Send to " ;
1027             cout << (MapOfRequestStructiterator->second)->MPITarget
1028                  << " MPITag " << (MapOfRequestStructiterator->second)->MPITag
1029                  << " DataType " << (MapOfRequestStructiterator->second)->MPIDatatype
1030                  << " Request " << (MapOfRequestStructiterator->second)->MPIRequest
1031                  << " Status " << (MapOfRequestStructiterator->second)->MPIStatus
1032                  << " Completed " << (MapOfRequestStructiterator->second)->MPICompleted
1033                  << endl ;
1034           }
1035         i++ ;
1036       }
1037   }
1038
1039   // Returns the MPI size of a TimeMessage
1040   MPI_Aint MPIAccess::timeExtent() const
1041   {
1042     MPI_Aint aextent, lbound ;
1043     MPI_Type_get_extent( _MPI_TIME , &lbound, &aextent ) ;
1044     return aextent ;
1045   }
1046
1047   // Returns the MPI size of a MPI_INT
1048   MPI_Aint MPIAccess::intExtent() const
1049   {
1050     MPI_Aint aextent, lbound ;
1051     MPI_Type_get_extent( MPI_INT , &lbound, &aextent ) ;
1052     return aextent ;
1053   }
1054
1055   // Returns the MPI size of a MPI_DOUBLE
1056   MPI_Aint MPIAccess::doubleExtent() const
1057   {
1058     MPI_Aint aextent, lbound ;
1059     MPI_Type_get_extent( MPI_DOUBLE , &lbound, &aextent ) ;
1060     return aextent ;
1061   }
1062
1063   // Outputs fields of a TimeMessage structure
1064   ostream & operator<< (ostream & f ,const TimeMessage & aTimeMsg )
1065   {
1066     f << " time " << aTimeMsg.time << " deltatime " << aTimeMsg.deltatime
1067       << " tag " << aTimeMsg.tag ;
1068     return f;
1069   }
1070
1071   // Outputs the DataType coded in a Tag
1072   ostream & operator<< (ostream & f ,const _MessageIdent & methodtype )
1073   {
1074     switch (methodtype)
1075       {
1076       case _message_time :
1077         f << " MethodTime ";
1078         break;
1079       case _message_int :
1080         f << " MPI_INT ";
1081         break;
1082       case _message_double :
1083         f << " MPI_DOUBLE ";
1084         break;
1085       default :
1086         f << " UnknownMethodType ";
1087         break;
1088       }
1089     return f;
1090   }
1091 }