mpi_progress.cpp

#include <iostream>
#include <mpi.h>
#include <vector>
#include <progresscpp/ProgressBar.hpp>

int main( int argc, char *argv[] )
{
   int size, rank;
   MPI_Init( &argc, &argv );
   MPI_Comm_size( MPI_COMM_WORLD, &size );
   MPI_Comm_rank( MPI_COMM_WORLD, &rank );

   int n = 1000;
   int Tag = 0;
   int EndTag = 1;
   MPI_Request request;

   // initialize the bar
   progresscpp::ProgressBar progressBar(n, 70);


   ///
   /// iprobe + receive
   ///
   auto iprobeRecv = [&rank, &Tag, &EndTag, &progressBar](int source, int tag)
   {
      MPI_Status status;
      int flag;
      // rank=sourceからtagメッセージが来ていないかを確認
      MPI_Iprobe(source, tag, MPI_COMM_WORLD, &flag, &status);
      if ( flag  == 1 )
      {
         if ( tag == Tag )
         {
            // メッセージが来ていたら受信
            int progressCount = 0;
            MPI_Recv(&progressCount, 1, MPI_INT, source, tag, MPI_COMM_WORLD, &status);
            ++progressBar;
         }
         else if ( tag == EndTag )
         {
            MPI_Recv(NULL, 0, MPI_BYTE, source, tag, MPI_COMM_WORLD, &status);
         }
      }
      return flag;
   };


   ///
   /// main
   ///
   for ( int i = rank; i < n; i += size )
   {
      // 何かしらのjobをした
      ++progressBar;

      if( rank == 0 )
      {
         // 受信
         for ( int source = 1; source < size; source++ )
            iprobeRecv(source, Tag);
         progressBar.display();
      }
      else
      {
         // 行ったjob-Idiを送信
         int p = 1;
         MPI_Send(&p, 1, MPI_INT, 0, Tag, MPI_COMM_WORLD);
      }
   }


   ///
   /// Post Process
   ///
   if ( rank > 0 )
   {
      // jobを終えたworkerはend_tagを送信
      MPI_Send(NULL, 0, MPI_BYTE, 0, EndTag, MPI_COMM_WORLD);
   }
   else
   {
      for ( int source = 1; source < size; source++ )
      {
         // end_tagを受け取るまでiprob
         while ( !iprobeRecv(source, EndTag) )
            iprobeRecv(source, Tag);
         progressBar.display();
      }
      progressBar.display();
   }


   MPI_Finalize();
   return 0;
}