Merge pull request #25 from CODARcode/sm_release

Copod algorithm

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())
+  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

@@ -1,4 +1,5 @@
 //The hierarchical parameter server main program. This program collects statistics from the node-instances of the anomaly detector
+#include <chimbuko_config.h>
 #ifndef _USE_ZMQNET
 #include<iostream>
 #warning "Hierarchical parameter server requires ZMQNet"
@@ -12,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>
@@ -103,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 {
@@ -169,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/provdb_admin.cpp

@@ -300,9 +300,12 @@ int main(int argc, char** argv) {
 	  //Spin quietly until SIGTERM sent
 	  signal(SIGTERM, termSignalHandler);
 	  progressStream << "ProvDB Admin: main thread waiting for completion" << std::endl;
+
+	  size_t iter = 0;
 	  while(!stop_wait_loop) { //stop wait loop will be set by SIGTERM handler
 	    tl::thread::sleep(engine, 1000); //Thallium engine sleeps but listens for rpc requests
-
+	    if(iter % 20 == 0){ verboseStream << "ProvDB Admin heartbeat" << std::endl; }
+	    
 	    unsigned long commit_timer_ms = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now() - commit_timer_start).count();
 	    if(args.db_commit_freq > 0 && commit_timer_ms >= args.db_commit_freq){
 	      verboseStream << "ProvDB Admin: committing database to disk" << std::endl;
@@ -331,7 +334,9 @@ int main(int argc, char** argv) {
 #endif
 	      break;
 	    }
-	  }
+
+	    ++iter;
+	  }//wait loop
 	}//client scope
 
 	//If the pserver didn't connect (it is optional), delete the empty database

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>
@@ -125,6 +124,7 @@ int main (int argc, char ** argv){
   }
   GlobalAnomalyStats global_func_stats; //global anomaly statistics
   GlobalCounterStats global_counter_stats; //global counter statistics
+  GlobalAnomalyMetrics global_anom_metrics; //global anomaly metrics
   PSglobalFunctionIndexMap global_func_index_map; //mapping of function name to global index
 
   //Optionally load previously-computed AD algorithm statistics
@@ -187,14 +187,14 @@ int main (int argc, char ** argv){
 
     net.add_payload(new NetPayloadUpdateParams(param, args.freeze_params));
     net.add_payload(new NetPayloadGetParams(param));
-    net.add_payload(new NetPayloadUpdateAnomalyStats(&global_func_stats));
-    net.add_payload(new NetPayloadUpdateCounterStats(&global_counter_stats));
+    net.add_payload(new NetPayloadRecvCombinedADdata(&global_func_stats, &global_counter_stats, &global_anom_metrics));
     net.add_payload(new NetPayloadGlobalFunctionIndexMapBatched(&global_func_index_map));
     net.init(nullptr, nullptr, args.nt);
 
     //Start sending anomaly statistics to viz
     stat_sender.add_payload(new PSstatSenderGlobalAnomalyStatsPayload(&global_func_stats));
     stat_sender.add_payload(new PSstatSenderGlobalCounterStatsPayload(&global_counter_stats));
+    stat_sender.add_payload(new PSstatSenderGlobalAnomalyMetricsPayload(&global_anom_metrics));
     stat_sender.run_stat_sender(args.ws_addr, args.stat_outputdir);
 
     //Register a signal handler that prevents the application from exiting on SIGTERM; instead this signal will be handled by ZeroMQ and will cause the pserver to shutdown gracefully
@@ -277,5 +277,8 @@ int main (int argc, char ** argv){
   }
 
   progressStream << "Pserver: finished" << std::endl;
+
+  delete param;
+
   return 0;
 }