ADIOS2 reader no longer uses MPI (it is not needed)

Modified code such that Chimbuko can now optionally be compiled entirely without MPI. User will have to feed the AD instance the appropriate rank idx (eg using environment variables).
	Most integration tests are broken with this option. Unit tests are fine.
Added an example to benchmark_client/func_multimodal where the AD is used in non-MPI mode inside a wrapper script
Improved ADThreadNetClient:
	Command to disconnect and command to terminate worker thread are now separate
	Connection and disconnection are now blocking operations such that we don't have situations where the main thread thinks the connection is ready when it is not
	Queueing actions now checks the worker thread is active

Author: giltirn

app/bpfile_replay.cpp

@@ -1,4 +1,3 @@
-#include <mpi.h>
 #include <adios2.h>
 #include <string>
 #include <cassert>
@@ -22,7 +21,6 @@ int main(int argc, char** argv){
     enableVerboseLogging() = true;
   }       
 
-  MPI_Init(&argc, &argv);
   if(argc < 3){
     std::cout << "Usage: bpfile_replay <input BPfile filename> <output step freq (ms)> <options>\n"
 	      << "Output will be on SST under the same filename ${input BPfile filename}. The temporary .sst file created will thus be ${input BPfile filename}.sst\n"
@@ -77,7 +75,7 @@ int main(int argc, char** argv){
   }
 
 
-  adios2::ADIOS ad = adios2::ADIOS(MPI_COMM_SELF, adios2::DebugON);
+  adios2::ADIOS ad = adios2::ADIOS(adios2::DebugON);
 
   adios2::IO io_out = ad.DeclareIO("writer");
 
@@ -217,6 +215,5 @@ int main(int argc, char** argv){
   std::cout << "Shutting down" << std::endl;
   wr.Close();
 
-  MPI_Finalize();
   return 0;
 };

app/driver.cpp

@@ -1,3 +1,7 @@
+#include<chimbuko_config.h>
+#ifdef USE_MPI
+#include<mpi.h>
+#endif
 #include "chimbuko/chimbuko.hpp"
 #include "chimbuko/verbose.hpp"
 #include "chimbuko/util/string.hpp"
@@ -89,7 +93,15 @@ optionalArgsParser & getOptionalArgsParser(){
     addOptionalCommandLineArg(p, trace_connect_timeout, "(For SST mode) Set the timeout in seconds on the connection to the TAU-instrumented binary (default 60s)");
     addOptionalCommandLineArg(p, parser_beginstep_timeout, "Set the timeout in seconds on waiting for the next ADIOS2 timestep (default 30s)");
 
-    addOptionalCommandLineArg(p, rank, "Set the rank index of the trace data. Used for verification unless override_rank is set. A value < 0 signals the value to be equal to the MPI rank of Chimbuko driver (default)");
+    addOptionalCommandLineArg(p, rank, 
+#ifdef USE_MPI
+			      "Set the rank index of the trace data. Used for verification unless override_rank is set. A value < 0 signals the value to be equal to the MPI rank of Chimbuko driver (default)"
+#else
+			      "Set the rank index of the trace data (default 0)"
+#endif
+			      );
+
+
     p.addOptionalArg(new overrideRankArg); //-override_rank <idx>
     p.addOptionalArg(new setLoggingHeadRankArg); //-logging_head_rank <rank>
 
@@ -110,7 +122,11 @@ void printHelp(){
   getOptionalArgsParser().help(std::cout);
 }
 
-ChimbukoParams getParamsFromCommandLine(int argc, char** argv, const int mpi_world_rank){
+ChimbukoParams getParamsFromCommandLine(int argc, char** argv
+#ifdef USE_MPI
+, const int mpi_world_rank
+#endif
+){
   if(argc < 4){
     std::cerr << "Expected at least 4 arguments: <exe> <BPFile/SST> <.bp location> <bp file prefix>" << std::endl;
     exit(-1);
@@ -153,8 +169,13 @@ ChimbukoParams getParamsFromCommandLine(int argc, char** argv, const int mpi_wor
 
   //By default assign the rank index of the trace data as the MPI rank of the AD process
   //Allow override by user
-  if(params.rank < 0)
+  if(params.rank < 0){
+#ifdef USE_MPI
     params.rank = mpi_world_rank;
+#else
+    params.rank = 0; //default to 0 for non-MPI applications
+#endif
+  }
 
   params.verbose = params.rank == 0; //head node produces verbose output
 
@@ -187,26 +208,35 @@ int main(int argc, char ** argv){
     return 0;
   }
 
+#ifdef USE_MPI
   assert( MPI_Init(&argc, &argv) == MPI_SUCCESS );
 
   int mpi_world_rank, mpi_world_size;
   MPI_Comm_rank(MPI_COMM_WORLD, &mpi_world_rank);
   MPI_Comm_size(MPI_COMM_WORLD, &mpi_world_size);
+#endif
 
   //Parse environment variables
   if(const char* env_p = std::getenv("CHIMBUKO_VERBOSE"))
     enableVerboseLogging() = true;
 
   //Parse Chimbuko parameters
-  ChimbukoParams params = getParamsFromCommandLine(argc, argv, mpi_world_rank);
+  ChimbukoParams params = getParamsFromCommandLine(argc, argv
+#ifdef USE_MPI
+    , mpi_world_rank
+#endif
+    );
+
   if(params.rank == progressHeadRank()) params.print();
 
   if(enableVerboseLogging()){
     headProgressStream(params.rank) << "Driver rank " << params.rank << ": Enabling verbose debug output" << std::endl;
   }
 
   verboseStream << "Driver rank " << params.rank << ": waiting at pre-run barrier" << std::endl;
+#ifdef USE_MPI
   MPI_Barrier(MPI_COMM_WORLD);
+#endif
 
   bool error = false;
 
@@ -248,6 +278,7 @@ int main(int argc, char ** argv){
       // -----------------------------------------------------------------------
       // Average analysis time and total number of outliers
       // -----------------------------------------------------------------------
+#ifdef USE_MPI
       verboseStream << "Driver rank " << params.rank << ": waiting at post-run barrier" << std::endl;
       MPI_Barrier(MPI_COMM_WORLD);
       processing_time = duration_cast<milliseconds>(t2 - t1).count();
@@ -274,6 +305,19 @@ int main(int argc, char ** argv){
 	total_n_comm_events = global_measures[1];
 	total_n_counter_events = global_measures[2];
       }
+#else
+      //Without MPI only report local parameters. In principle we could aggregate using the Pserver if we really want the global information
+      total_processing_time = processing_time;
+      total_n_outliers = n_outliers;
+      total_frames = frames;
+
+      total_n_func_events = n_func_events;
+      total_n_comm_events = n_comm_events;
+      total_n_counter_events = n_counter_events;
+      
+      int mpi_world_size = 1;
+#endif
+
 
       headProgressStream(params.rank) << "Driver rank " << params.rank << ": Final report\n"
 				  << "Avg. num. frames over MPI ranks : " << (double)total_frames/(double)mpi_world_size << "\n"
@@ -301,7 +345,9 @@ int main(int argc, char ** argv){
     error = true;
   }
 
+#ifdef USE_MPI
   MPI_Finalize();
+#endif
   headProgressStream(params.rank) << "Driver is exiting" << std::endl;
   return error ? 1 : 0;
 }

app/hpserver.cpp

@@ -13,7 +13,9 @@ int main(void){
 
 #include <chimbuko/pserver.hpp>
 #include <chimbuko/net/zmqme_net.hpp>
+#ifdef USE_MPI
 #include <mpi.h>
+#endif
 #include <chimbuko/param/sstd_param.hpp>
 #include <chimbuko/util/commandLineParser.hpp>
 #include <chimbuko/verbose.hpp>
@@ -104,7 +106,10 @@ int main (int argc, char ** argv){
   }
 
   ZMQMENet net;
+#ifdef USE_MPI
   MPI_Init(&argc, &argv);
+#endif
+
   PSstatSender stat_sender(args.stat_send_freq);
 
   try {
@@ -170,7 +175,7 @@ int main (int argc, char ** argv){
 
   std::cout << "HPserver finalizing the network" << std::endl;
   net.finalize();
-#ifdef _USE_ZMQNET
+#if defined(_USE_ZMQNET) && defined(USE_MPI)
   MPI_Finalize();
 #endif
 

app/pclient.cpp

@@ -1,4 +1,5 @@
 //A test application that mocks part of the anomaly detection modules, acting as a client for the parameter server and sending it anomaly information
+#include <chimbuko_config.h>
 
 #include "chimbuko/param/sstd_param.hpp"
 #include "chimbuko/message.hpp"
@@ -7,6 +8,9 @@
 #include "chimbuko/net/mpi_net.hpp"
 #else
 #include "chimbuko/net/zmq_net.hpp"
+#endif
+
+#ifdef USE_MPI
 #include <mpi.h>
 #endif
 
@@ -18,11 +22,13 @@ using namespace chimbuko;
 
 int main (int argc, char** argv)
 {
-    int size, rank;
+    int size=1, rank=0;
 
+#ifdef USE_MPI
     MPI_Init(&argc, &argv);
     MPI_Comm_size(MPI_COMM_WORLD, &size);
     MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+#endif
 
 #ifdef _USE_MPINET
     int count;
@@ -114,7 +120,9 @@ s     */
         //g_param.assign(msg.data_buffer());
     }
 
+#ifdef USE_MPI
     MPI_Barrier(MPI_COMM_WORLD);
+#endif
 
 #ifdef _USE_MPINET
     if (rank == 0) {
@@ -159,6 +167,8 @@ s     */
     zmq_ctx_term(context);
 #endif
 
+#ifdef USE_MPI
     MPI_Finalize();
+#endif
     return EXIT_SUCCESS;
 }

app/pclient_stats.cpp

@@ -1,4 +1,5 @@
 //A test application that mocks part of the anomaly detection modules, acting as a client for the parameter server and sending it function statistics information
+#include <chimbuko_config.h>
 
 #include "chimbuko/param/sstd_param.hpp"
 #include "chimbuko/message.hpp"
@@ -9,6 +10,9 @@
 #include "chimbuko/net/mpi_net.hpp"
 #else
 #include "chimbuko/net/zmq_net.hpp"
+#endif
+
+#ifdef USE_MPI
 #include <mpi.h>
 #endif
 
@@ -21,8 +25,9 @@ using namespace chimbuko;
 int main (int argc, char** argv)
 {
     const int N_MPI_PROCESSORS = 10;
-    int size, rank;
+    int size=1, rank=0;
 
+#ifdef USE_MPI
     MPI_Init(&argc, &argv);
     MPI_Comm_size(MPI_COMM_WORLD, &size);
     MPI_Comm_rank(MPI_COMM_WORLD, &rank);
@@ -33,6 +38,7 @@ int main (int argc, char** argv)
         MPI_Finalize();
         return EXIT_SUCCESS;
     }
+#endif
 
 #ifdef _USE_MPINET
     throw std::runtime_error("Not implemented yet.");
@@ -103,8 +109,10 @@ int main (int argc, char** argv)
         std::this_thread::sleep_for(std::chrono::milliseconds(10));
     }
 
+#ifdef USE_MPI
     MPI_Barrier(MPI_COMM_WORLD);
-    
+#endif    
+
     // terminate parameter server
 #ifdef _USE_MPINET
     throw std::runtime_error("Not implemented yet.");
@@ -127,6 +135,8 @@ int main (int argc, char** argv)
     zmq_ctx_term(context);
 #endif
 
+#ifdef USE_MPI
     MPI_Finalize();
+#endif
     return EXIT_SUCCESS;
 }

app/pserver.cpp

@@ -4,7 +4,6 @@
 #include <chimbuko/pserver.hpp>
 
 #ifdef _USE_MPINET
-#include <mpi.h>
 #include <chimbuko/net/mpi_net.hpp>
 #else
 #include <chimbuko/net/zmq_net.hpp>