src/ParaMEDMEMTest/test_MPI_Access_Time_0.cxx

   1 // Copyright (C) 2007-2015  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 <string>
  21 #include <vector>
  22 #include <map>
  23 #include <iostream>
  24 #include <mpi.h>
  25
  26 #include "MPIAccessTest.hxx"
  27 #include <cppunit/TestAssert.h>
  28
  29 //#include "CommInterface.hxx"
  30 //#include "ProcessorGroup.hxx"
  31 //#include "MPIProcessorGroup.hxx"
  32 #include "MPIAccess.hxx"
  33
  34 // use this define to enable lines, execution of which leads to Segmentation Fault
  35 #define ENABLE_FAULTS
  36
  37 // use this define to enable CPPUNIT asserts and fails, showing bugs
  38 #define ENABLE_FORCED_FAILURES
  39
  40 using namespace std;
  41 using namespace MEDCoupling;
  42
  43 void chksts( int sts , int myrank , MEDCoupling::MPIAccess * mpi_access ) {
  44   char msgerr[MPI_MAX_ERROR_STRING] ;
  45   int lenerr ;
  46   if ( sts != MPI_SUCCESS ) {
  47     mpi_access->errorString(sts, msgerr, &lenerr) ;
  48     debugStream << "test" << myrank << " lenerr " << lenerr << " "
  49          << msgerr << endl ;
  50     ostringstream strstream ;
  51     strstream << "==========================================================="
  52               << "test" << myrank << " KO"
  53               << "==========================================================="
  54               << endl ;
  55     debugStream << strstream.str() << endl ;
  56     CPPUNIT_FAIL( strstream.str() ) ;
  57   }
  58 return ;
  59 }
  60
  61 void MPIAccessTest::test_MPI_Access_Time_0() {
  62
  63   debugStream << "test_MPI_Access_Time_0" << endl ;
  64
  65 //  MPI_Init(&argc, &argv) ;
  66
  67   int size ;
  68   int myrank ;
  69   MPI_Comm_size(MPI_COMM_WORLD,&size) ;
  70   MPI_Comm_rank(MPI_COMM_WORLD,&myrank) ;
  71
  72   if ( size < 2 ) {
  73     ostringstream strstream ;
  74     strstream << "usage :" << endl
  75               << "mpirun -np <nbprocs> test_MPI_Access_Time_0" <<endl
  76               << " nbprocs =2" << endl
  77               << "test must be runned with 2 procs" << endl ;
  78     cerr << strstream.str() << endl ;
  79     //CPPUNIT_FAIL( strstream.str() ) ;
  80     return;
  81   }
  82
  83 #define maxreq 100
  84
  85   double t ;
  86   double dt[2] = {2., 1.} ;
  87   double maxt = maxreq/dt[myrank] ;
  88
  89   debugStream << "test_MPI_Access_Time_0 rank" << myrank << endl ;
  90
  91   MEDCoupling::CommInterface interface ;
  92
  93   MEDCoupling::MPIProcessorGroup* group = new MEDCoupling::MPIProcessorGroup(interface) ;
  94
  95   MEDCoupling::MPIAccess * mpi_access = new MEDCoupling::MPIAccess( group ) ;
  96
  97   if ( myrank >= 2 ) {
  98     debugStream << "test_MPI_Access_Time_0 rank" << myrank << " --> mpi_access->barrier" << endl ;
  99     mpi_access->barrier() ;
 100     debugStream << "test_MPI_Access_Time_0 rank" << myrank << " <-- mpi_access->barrier" << endl ;
 101     debugStream << "test_MPI_Access_Time_0 rank" << myrank << " --> mpi_access->barrier" << endl ;
 102     mpi_access->barrier() ;
 103     debugStream << "test_MPI_Access_Time_0 rank" << myrank << " <-- mpi_access->barrier" << endl ;
 104     delete group ;
 105     delete mpi_access ;
 106     debugStream << "test_MPI_Access_Time" << myrank << " OK" << endl ;
 107     return ;
 108   }
 109
 110   int target = 1 - myrank ;
 111   int SendTimeRequestId[maxreq] ;
 112   int RecvTimeRequestId[maxreq] ;
 113   int SendRequestId[maxreq] ;
 114   int RecvRequestId[maxreq] ;
 115   int sts ;
 116   int sendbuf[maxreq] ;
 117   int recvbuf[maxreq] ;
 118   MEDCoupling::TimeMessage aSendTimeMsg[maxreq] ;
 119   int lasttime = -1 ;
 120   MEDCoupling::TimeMessage RecvTimeMessages[maxreq+1] ;
 121   MEDCoupling::TimeMessage *aRecvTimeMsg = &RecvTimeMessages[1] ;
 122 //  mpi_access->Trace() ;
 123   int istep = 0 ;
 124   for ( t = 0 ; t < maxt ; t = t+dt[myrank] ) {
 125      debugStream << "test" << myrank << " ==========================TIME " << t
 126           << " ==========================" << endl ;
 127      if ( myrank == 0 ) {
 128        aSendTimeMsg[istep].time = t ;
 129        aSendTimeMsg[istep].deltatime = dt[myrank] ;
 130        //aSendTimeMsg[istep].maxtime = maxt ;
 131        if ( t+dt[myrank] >= maxt ) {
 132          aSendTimeMsg[istep].deltatime = 0 ;
 133        }
 134        sts = mpi_access->ISend( &aSendTimeMsg[istep] , 1 ,
 135                                mpi_access->timeType() , target ,
 136                                SendTimeRequestId[istep]) ;
 137        debugStream << "test" << myrank << " ISend TimeRequestId " << SendTimeRequestId[istep]
 138             << " tag " << mpi_access->MPITag(SendTimeRequestId[istep]) << endl ;
 139        chksts( sts , myrank , mpi_access ) ;
 140        sendbuf[istep] = istep ;
 141        sts = mpi_access->ISend(&sendbuf[istep],1,MPI_INT,target, SendRequestId[istep]) ;
 142        debugStream << "test" << myrank << " ISend Data RequestId " << SendRequestId[istep]
 143             << " tag " << mpi_access->MPITag(SendRequestId[istep]) << endl ;
 144        chksts( sts , myrank , mpi_access ) ;
 145 //CheckSent
 146 //=========
 147        int sendrequests[2*maxreq] ;
 148        int sendreqsize = mpi_access->sendRequestIds( target , 2*maxreq ,
 149                                                     sendrequests ) ;
 150        int j , flag ;
 151        for ( j = 0 ; j < sendreqsize ; j++ ) {
 152           sts = mpi_access->test( sendrequests[j] , flag ) ;
 153           chksts( sts , myrank , mpi_access ) ;
 154           if ( flag ) {
 155             mpi_access->deleteRequest( sendrequests[j] ) ;
 156             debugStream << "test" << myrank << " " << j << ". " << sendrequests[j]
 157                  << " sendrequest deleted" << endl ;
 158           }
 159        }
 160      }
 161      else {
 162 //InitRecv
 163 //========
 164        if ( t == 0 ) {
 165          aRecvTimeMsg[lasttime].time = 0 ;
 166          sts = mpi_access->IRecv( &aRecvTimeMsg[lasttime+1] , 1 ,
 167                                  mpi_access->timeType() ,
 168                                  target , RecvTimeRequestId[lasttime+1]) ;
 169          debugStream << "test" << myrank << " t == 0 IRecv TimeRequestId "
 170               << RecvTimeRequestId[lasttime+1]
 171               << " MPITag " << mpi_access->MPITag( RecvTimeRequestId[lasttime+1] )
 172               << " MPICompleted "
 173               << mpi_access->MPICompleted( RecvTimeRequestId[lasttime+1] ) << endl ;
 174          chksts( sts , myrank , mpi_access ) ;
 175        }
 176        else {
 177          debugStream << "test" << myrank << " t # 0 lasttime " << lasttime << endl ;
 178 //InitialOutTime
 179 //==============
 180          bool outtime = false ;
 181          if ( lasttime != -1 ) {
 182            if ( t <= aRecvTimeMsg[lasttime-1].time ) {
 183              ostringstream strstream ;
 184              strstream << "==========================================================="
 185                        << endl << "test" << myrank << " t " << t << " <= "
 186                        << "aRecvTimeMsg[ " << lasttime << "-1 ].time "
 187                        << aRecvTimeMsg[lasttime-1].time << " KO" << endl
 188                        << "==========================================================="
 189                        << endl ;
 190              debugStream << strstream.str() << endl ;
 191              CPPUNIT_FAIL( strstream.str() ) ;
 192            }
 193            else {
 194              debugStream << "==========================================================="
 195                   << endl << "test" << myrank << " t " << t << " > "
 196                   << "aRecvTimeMsg[ " << lasttime << "-1 ].time "
 197                   << aRecvTimeMsg[lasttime-1].time << " OK" << endl
 198                   << "==========================================================="
 199                   << endl ;
 200            }
 201            //outtime = ((aRecvTimeMsg[lasttime].time +
 202            //            aRecvTimeMsg[lasttime].deltatime) >=
 203            //           aRecvTimeMsg[lasttime].maxtime) ;
 204            outtime = aRecvTimeMsg[lasttime].deltatime == 0 ;
 205          }
 206 // CheckRecv - CheckTime
 207 // On a lasttime tel que :
 208 // aRecvTimeMsg[ lasttime-1 ].time < T(i-1) <= aRecvTimeMsg[ lasttime ].time
 209 // On cherche lasttime tel que :
 210 // aRecvTimeMsg[ lasttime-1 ].time < T(i) <= aRecvTimeMsg[ lasttime ].time
 211          if ( t <= aRecvTimeMsg[lasttime].time ) {
 212            outtime = false ;
 213          }
 214          debugStream << "test" << myrank << " while outtime( " << outtime << " && t " << t
 215               << " > aRecvTimeMsg[ " << lasttime << " ] "
 216               << aRecvTimeMsg[lasttime].time << " )" << endl ;
 217          while ( !outtime && (t > aRecvTimeMsg[lasttime].time) ) {
 218               lasttime += 1 ;
 219 //TimeMessage
 220 //===========
 221               sts = mpi_access->wait( RecvTimeRequestId[lasttime] ) ;
 222               chksts( sts , myrank , mpi_access ) ;
 223               debugStream << "test" << myrank << " Wait done RecvTimeRequestId "
 224                    << RecvTimeRequestId[lasttime] << " lasttime " << lasttime
 225                    << " tag " << mpi_access->MPITag(RecvTimeRequestId[lasttime])
 226                    << aRecvTimeMsg[lasttime] << endl ;
 227               if ( lasttime == 0 ) {
 228                 aRecvTimeMsg[lasttime-1] = aRecvTimeMsg[lasttime] ;
 229               }
 230               mpi_access->deleteRequest( RecvTimeRequestId[lasttime] ) ;
 231
 232               double deltatime = aRecvTimeMsg[lasttime].deltatime ;
 233               //double maxtime = aRecvTimeMsg[lasttime].maxtime ;
 234               double nexttime = aRecvTimeMsg[lasttime].time + deltatime ;
 235               debugStream << "test" << myrank << " t " << t << " lasttime " << lasttime
 236                    << " deltatime " << deltatime
 237                    << " nexttime " << nexttime << endl ;
 238               //if ( nexttime < maxtime && t > nexttime ) {
 239               if ( deltatime != 0 && t > nexttime ) {
 240 //CheckRecv :
 241 //=========
 242                 //while ( nexttime < maxtime && t > nexttime ) {
 243                 while ( deltatime != 0 && t > nexttime ) {
 244                      int source, MPITag, outcount ;
 245                      MPI_Datatype datatype ;
 246                      sts = mpi_access->probe( target , source, MPITag, datatype,
 247                                              outcount ) ;
 248                      chksts( sts , myrank , mpi_access ) ;
 249 // Cancel DataMessages jusqu'a un TimeMessage
 250                      int cancelflag ;
 251                      while ( !mpi_access->isTimeMessage( MPITag ) ) {
 252                           sts = mpi_access->cancel( source, MPITag, datatype, outcount ,
 253                           //sts = mpi_access->cancel( source, datatype, outcount ,
 254                                                    //RecvRequestId[lasttime] ,
 255                                                    cancelflag ) ;
 256                           debugStream << "test" << myrank << " Recv TO CANCEL RequestId "
 257                                << RecvRequestId[lasttime]
 258                                << " tag " << mpi_access->recvMPITag( target )
 259                                << " cancelflag " << cancelflag << endl ;
 260                           chksts( sts , myrank , mpi_access ) ;
 261                           sts = mpi_access->probe( target , source, MPITag, datatype,
 262                                                   outcount ) ;
 263                           chksts( sts , myrank , mpi_access ) ;
 264                      }
 265 //On peut avancer en temps
 266                      nexttime += deltatime ;
 267                      //if ( nexttime < maxtime && t > nexttime ) {
 268                      if ( deltatime != 0 && t > nexttime ) {
 269 // Cancel du TimeMessage
 270                        sts = mpi_access->cancel( source, MPITag, datatype, outcount ,
 271                        //sts = mpi_access->cancel( source, datatype, outcount ,
 272                                                 //RecvRequestId[lasttime] ,
 273                                                 cancelflag ) ;
 274                        debugStream << "test" << myrank << " Time TO CANCEL RequestId "
 275                             << RecvRequestId[lasttime]
 276                             << " tag " << mpi_access->recvMPITag( target )
 277                             << " cancelflag " << cancelflag << endl ;
 278                        chksts( sts , myrank , mpi_access ) ;
 279                      }
 280                 }
 281               }
 282               else {
 283 //DoRecv
 284 //======
 285                 debugStream << "test" << myrank << " Recv target " << target
 286                      << " lasttime " << lasttime
 287                      << " lasttime-1 " << aRecvTimeMsg[lasttime-1]
 288                      << " lasttime " << aRecvTimeMsg[lasttime]
 289                      << endl ;
 290                 sts = mpi_access->recv(&recvbuf[lasttime],1,MPI_INT,target,
 291                                        RecvRequestId[lasttime]) ;
 292                 debugStream << "test" << myrank << " Recv RequestId "
 293                      << RecvRequestId[lasttime]
 294                      << " tag " << mpi_access->recvMPITag( target )
 295                      << endl ;
 296                 chksts( sts , myrank , mpi_access ) ;
 297               }
 298               //outtime = ((aRecvTimeMsg[lasttime].time +
 299               //            aRecvTimeMsg[lasttime].deltatime) >=
 300               //           aRecvTimeMsg[lasttime].maxtime) ;
 301               outtime = aRecvTimeMsg[lasttime].deltatime == 0 ;
 302               if ( !outtime ) {
 303 // Une lecture asynchrone d'un message temps a l'avance
 304                 sts = mpi_access->IRecv( &aRecvTimeMsg[lasttime+1] , 1 ,
 305                                         mpi_access->timeType() , target ,
 306                                         RecvTimeRequestId[lasttime+1]) ;
 307                 debugStream << "test" << myrank << " IRecv TimeRequestId "
 308                      << RecvTimeRequestId[lasttime+1] << " MPITag "
 309                      << mpi_access->MPITag( RecvTimeRequestId[lasttime+1] )
 310                      << " MPICompleted "
 311                      << mpi_access->MPICompleted( RecvTimeRequestId[lasttime+1] )
 312                      << endl ;
 313                 chksts( sts , myrank , mpi_access ) ;
 314               }
 315               else if ( t <= aRecvTimeMsg[lasttime].time ) {
 316                 outtime = false ;
 317               }
 318          }
 319
 320          //printf("DEBUG t %.15f Msg[lasttime-1] %.15f Msg[lasttime] %.15f \n",t,
 321          //       aRecvTimeMsg[lasttime-1].time,aRecvTimeMsg[lasttime].time) ;
 322          if ( ((t <= aRecvTimeMsg[lasttime-1].time) ||
 323                (t > aRecvTimeMsg[lasttime].time)) && !outtime ) {
 324            ostringstream strstream ;
 325            strstream << "==========================================================="
 326                      << endl << "test" << myrank << " t " << t << " <= "
 327                      << "aRecvTimeMsg[ " << lasttime << "-1 ].time "
 328                      << aRecvTimeMsg[lasttime-1].time << " ou t " << t << " > "
 329                      << "aRecvTimeMsg[ " << lasttime << " ].time "
 330                      << aRecvTimeMsg[lasttime].time << endl
 331                      << " ou bien outtime " << outtime << " KO RequestTimeIds "
 332                      << RecvTimeRequestId[lasttime-1] << " " << RecvTimeRequestId[lasttime]
 333                      << " RequestIds "
 334                      << RecvRequestId[lasttime-1] << " " << RecvRequestId[lasttime] << endl
 335                      << "==========================================================="
 336                      << endl ;
 337            debugStream << strstream.str() << endl ;
 338            CPPUNIT_FAIL( strstream.str() ) ;
 339          }
 340          else {
 341            debugStream << "==========================================================="
 342                 << endl << "test" << myrank
 343                 << " aRecvTimeMsg[ " << lasttime << "-1 ].time "
 344                 << aRecvTimeMsg[lasttime-1].time << " < t " << t << " <= "
 345                 << "aRecvTimeMsg[ " << lasttime << " ].time "
 346                 << aRecvTimeMsg[lasttime].time << endl
 347                 << " ou bien outtime " << outtime << " OK RequestTimeIds "
 348                 << RecvTimeRequestId[lasttime-1] << " " << RecvTimeRequestId[lasttime]
 349                 << " RequestIds "
 350                 << RecvRequestId[lasttime-1] << " " << RecvRequestId[lasttime] << endl
 351                 << "==========================================================="
 352                 << endl ;
 353          }
 354        }
 355      }
 356      chksts( sts , myrank , mpi_access ) ;
 357      istep = istep + 1 ;
 358   }
 359
 360   debugStream << "test" << myrank << " Barrier :" << endl ;
 361   mpi_access->barrier() ;
 362
 363   if (MPI_ACCESS_VERBOSE) mpi_access->check() ;
 364
 365   if ( myrank == 0 ) {
 366 //CheckFinalSent
 367 //==============
 368     debugStream << "test" << myrank << " CheckFinalSent :" << endl ;
 369     int sendrequests[2*maxreq] ;
 370     int sendreqsize = mpi_access->sendRequestIds( target , 2*maxreq , sendrequests ) ;
 371     int j ;
 372     for ( j = 0 ; j < sendreqsize ; j++ ) {
 373        sts = mpi_access->wait( sendrequests[j] ) ;
 374        chksts( sts , myrank , mpi_access ) ;
 375        mpi_access->deleteRequest( sendrequests[j] ) ;
 376        debugStream << "test" << myrank << " " << j << ". " << sendrequests[j] << " deleted"
 377             << endl ;
 378     }
 379   }
 380   else {
 381     debugStream << "test" << myrank << " CheckFinalRecv :" << endl ;
 382     int recvrequests[2*maxreq] ;
 383     int recvreqsize = mpi_access->recvRequestIds( target , 2*maxreq , recvrequests ) ;
 384     int cancelflag ;
 385     int j ;
 386     for ( j = 0 ; j < recvreqsize ; j++ ) {
 387        sts = mpi_access->cancel( recvrequests[j] , cancelflag ) ;
 388        chksts( sts , myrank , mpi_access ) ;
 389        mpi_access->deleteRequest( recvrequests[j] ) ;
 390        debugStream << "test" << myrank << " " << j << ". " << recvrequests[j] << " deleted"
 391             << " cancelflag " << cancelflag << endl ;
 392     }
 393     int source, MPITag, outcount , flag ;
 394     MPI_Datatype datatype ;
 395     sts = mpi_access->IProbe( target , source, MPITag, datatype,
 396                              outcount , flag ) ;
 397     chksts( sts , myrank , mpi_access ) ;
 398     while ( flag ) {
 399          sts = mpi_access->cancel( source, MPITag, datatype, outcount ,
 400          //sts = mpi_access->cancel( source, datatype, outcount ,
 401                                   //RecvRequestId[lasttime] ,
 402                                   cancelflag ) ;
 403          debugStream << "test" << myrank << " TO CANCEL RequestId "
 404               << RecvRequestId[lasttime]
 405               << " tag " << mpi_access->recvMPITag( target )
 406               << " cancelflag " << cancelflag << endl ;
 407          chksts( sts , myrank , mpi_access ) ;
 408          sts = mpi_access->IProbe( target , source, MPITag, datatype,
 409                                   outcount , flag ) ;
 410          chksts( sts , myrank , mpi_access ) ;
 411     }
 412   }
 413   if(MPI_ACCESS_VERBOSE) mpi_access->check() ;
 414
 415   if ( myrank == 0 ) {
 416     int sendrequests[2*maxreq] ;
 417     int sendreqsize = mpi_access->sendRequestIds( target , 2*maxreq , sendrequests ) ;
 418     if ( sendreqsize != 0 ) {
 419       ostringstream strstream ;
 420       strstream << "=========================================================" << endl
 421                 << "test" << myrank << " sendreqsize " << sendreqsize << " KO" << endl
 422                 << "=========================================================" << endl ;
 423       debugStream << strstream.str() << endl ;
 424       CPPUNIT_FAIL( strstream.str() ) ;
 425     }
 426     else {
 427       debugStream << "=========================================================" << endl
 428            << "test" << myrank << " sendreqsize " << sendreqsize << " OK" << endl
 429            << "=========================================================" << endl ;
 430     }
 431   }
 432   else {
 433     int recvrequests[2*maxreq] ;
 434     int recvreqsize = mpi_access->recvRequestIds( target , 2*maxreq , recvrequests ) ;
 435     if ( recvreqsize != 0 ) {
 436       ostringstream strstream ;
 437       strstream << "=========================================================" << endl
 438                 << "test" << myrank << " recvreqsize " << recvreqsize << " KO" << endl
 439                 << "=========================================================" << endl ;
 440       debugStream << strstream.str() << endl ;
 441       CPPUNIT_FAIL( strstream.str() ) ;
 442     }
 443     else {
 444       debugStream << "=========================================================" << endl
 445            << "test" << myrank << " recvreqsize " << recvreqsize << " OK" << endl
 446            << "=========================================================" << endl ;
 447     }
 448   }
 449
 450   int i ;
 451   for ( i = 0 ; i <= lasttime ; i++ ) {
 452      debugStream << "test" << myrank << " " << i << ". RecvTimeMsg "
 453           << aRecvTimeMsg[i].time << " recvbuf " << recvbuf[i] << endl ;
 454   }
 455
 456   debugStream << "test_MPI_Access_Time_0 rank" << myrank << " --> mpi_access->barrier" << endl ;
 457   mpi_access->barrier() ;
 458   debugStream << "test_MPI_Access_Time_0 rank" << myrank << " <-- mpi_access->barrier" << endl ;
 459
 460   delete group ;
 461   delete mpi_access ;
 462
 463 //  MPI_Finalize();
 464
 465   debugStream << "test" << myrank << " OK" << endl ;
 466
 467   return ;
 468 }
 469
 470
 471
 472