Merge branch 'sm_release' into ckelly_develop

Author: giltirn

app/Makefile.am

@@ -1,5 +1,5 @@
 AM_CPPFLAGS = -I$(top_srcdir)/include -I$(top_srcdir)/3rdparty @PS_FLAGS@
-LDADD = ../src/libchimbuko.la -lstdc++fs 
+LDADD = ../src/libchimbuko.la -lstdc++fs
 
 bin_PROGRAMS = driver pclient pclient_stats hpserver pserver pshutdown sstSinker sst_view bpfile_replay
 

app/driver.cpp

@@ -71,6 +71,7 @@ optionalArgsParser & getOptionalArgsParser(){
     addOptionalCommandLineArg(p, hbos_use_global_threshold, "Set true to use a global threshold in HBOS algorithm. Dafault is true.");
     addOptionalCommandLineArg(p, program_idx, "Set the index associated with the instrumented program. Use to label components of a workflow. (default 0)");
     addOptionalCommandLineArg(p, outlier_sigma, "Set the number of standard deviations that defines an anomalous event (default 6)");
+    addOptionalCommandLineArg(p, net_recv_timeout, "Timeout (in ms) for blocking receives on client from parameter server");
     addOptionalCommandLineArg(p, pserver_addr, "Set the address of the parameter server. If empty (default) the pserver will not be used.");
     addOptionalCommandLineArg(p, hpserver_nthr, "Set the number of threads used by the hierarchical PS. This parameter is used to compute a port offset for the particular endpoint that this AD rank connects to (default 1)");
     addOptionalCommandLineArg(p, interval_msec, "Force the AD to pause for this number of ms at the end of each IO step (default 0)");
@@ -129,6 +130,7 @@ ChimbukoParams getParamsFromCommandLine(int argc, char** argv, const int mpi_wor
   params.ad_algorithm = "hbos";
   params.hbos_threshold = 0.99;
   params.hbos_use_global_threshold = true;
+  params.net_recv_timeout = 30000;
   params.pserver_addr = "";  //don't use pserver by default
   params.hpserver_nthr = 1;
   params.outlier_sigma = 6.0;     // anomaly detection algorithm parameter

app/provdb_admin.cpp

@@ -147,7 +147,7 @@ int main(int argc, char** argv) {
       progressStream << "ProvDB Admin: Enabling verbose debug output" << std::endl;
       enableVerboseLogging() = true;
       spdlog::set_level(spdlog::level::trace); //enable logging of Sonata
-    }       
+    }
 
     //argv[1] should specify the ip address and port (the only way to fix the port that I'm aware of)
     //Should be of form <ip address>:<port>   eg. 127.0.0.1:1234
@@ -239,7 +239,7 @@ int main(int argc, char** argv) {
 
       //Initialize provider
       sonata::Provider provider(engine, 0);
-    
+
       progressStream << "ProvDB Admin: Provider is running on " << addr << std::endl;
 
       { //Scope in which admin object is active
@@ -269,7 +269,7 @@ int main(int argc, char** argv) {
 	//Create the collections
 	{ //scope in which client is active
 	  sonata::Client client(engine);
-	  
+
 	  //Initialize the provdb shards
 	  std::vector<sonata::Database> db(args.nshards);
 	  for(int s=0;s<args.nshards;s++){
@@ -297,7 +297,7 @@ int main(int argc, char** argv) {
 	  Clock::time_point commit_timer_start = Clock::now();
 
 	  //Spin quietly until SIGTERM sent
-	  signal(SIGTERM, termSignalHandler);  
+	  signal(SIGTERM, termSignalHandler);
 	  progressStream << "ProvDB Admin: main thread waiting for completion" << std::endl;
 	  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
@@ -313,13 +313,13 @@ int main(int argc, char** argv) {
 	      glob_db.commit();
 	      commit_timer_start = Clock::now();
 	    }
-      
+
 	    //If at least one client has previously connected but none are now connected, shutdown the server
 	    //If all clients disconnected we must also wait for the pserver to disconnect (if it is connected)
 
 	    //If args.autoshutdown is disabled we can force shutdown via a "stop_server" RPC
 	    if(
-	       (args.autoshutdown || cmd_shutdown)  && 
+	       (args.autoshutdown || cmd_shutdown)  &&
 	       ( a_client_has_connected && connected.size() == 0 ) &&
 	       ( !pserver_has_connected || (pserver_has_connected && !pserver_connected) ) &&
 	       ( !committer_has_connected || (committer_has_connected && !committer_connected) )
@@ -336,7 +336,7 @@ int main(int argc, char** argv) {
 	  progressStream << "ProvDB Admin: destroying pserver database as it didn't connect (connection is optional)" << std::endl;
 	  admin.destroyDatabase(addr, 0, glob_db_name);
 	}
-	
+
 	progressStream << "ProvDB Admin: ending admin scope" << std::endl;
 	margo_dump("margo_dump_end_admin_scope");
       }//admin scope
@@ -346,10 +346,10 @@ int main(int argc, char** argv) {
     }//provider scope
 
     progressStream << "ProvDB Admin: shutting down server engine" << std::endl;
-    delete mtx; //delete mutex prior to engine finalize    
+    delete mtx; //delete mutex prior to engine finalize
     engine.finalize();
-    progressStream << "ProvDB Admin: finished, exiting engine scope" << std::endl;  
+    progressStream << "ProvDB Admin: finished, exiting engine scope" << std::endl;
   }
-  progressStream << "ProvDB Admin: finished, exiting main scope" << std::endl;    
+  progressStream << "ProvDB Admin: finished, exiting main scope" << std::endl;
   return 0;
 }

app/sstSinker.cpp

@@ -20,10 +20,10 @@ int main(int argc, char ** argv){
 
   int world_rank, world_size;
   MPI_Comm_rank(MPI_COMM_WORLD, &world_rank);
-  MPI_Comm_size(MPI_COMM_WORLD, &world_size);    
+  MPI_Comm_size(MPI_COMM_WORLD, &world_size);
 
   bool error = false;
-  try 
+  try
     {
       commandLineParser<SinkerArgs> cmdline;
       addOptionalCommandLineArg(cmdline, timeout, "Specify the SST connect timeout in seconds (Default 60s)");
@@ -35,19 +35,19 @@ int main(int argc, char ** argv){
 		  << "Options:" << std::endl;
 	cmdline.help(std::cout);
 	return 0;
-      }      
+      }
 
       std::string engineType = argv[1]; // BPFile or SST
       std::string data_dir = argv[2]; // *.bp location
       std::string prefix = argv[3]; // "tau-metrics-nwchem"
       std::string inputFile = prefix + "-" + std::to_string(world_rank) + ".bp";
       int fetch_data = atoi(argv[4]);
-      
+
       SinkerArgs args;
       cmdline.parse(args, argc-5, (const char**)(argv+5) );
 
       if (world_rank == 0) {
-	std::cout << "\n" 
+	std::cout << "\n"
 		  << "rank       : " << world_rank << "\n"
 		  << "Engine     : " << engineType << "\n"
 		  << "BP in dir  : " << data_dir << "\n"
@@ -62,7 +62,7 @@ int main(int argc, char ** argv){
       // -----------------------------------------------------------------------
       ADParser * parser;
 
-      // int step = 0; 
+      // int step = 0;
 
       // -----------------------------------------------------------------------
       // Measurement variables
@@ -92,7 +92,7 @@ int main(int argc, char ** argv){
 	  if (!parser->getStatus())
 	    {
 	      // No more steps available.
-	      break;                
+	      break;
 	    }
 
 	  // step = parser->getCurrentStep();
@@ -128,7 +128,7 @@ int main(int argc, char ** argv){
 	total_processing_time = global_measures[0];
 	total_frames = global_measures[1];
       }
-        
+
       if (false && world_rank == 0) {
 	std::cout << "\n"
 		  << "Avg. num. frames     : " << (double)total_frames/(double)world_size << "\n"

docker/ubuntu18.04/openmpi4.0.4/Dockerfile.mochi

@@ -22,4 +22,3 @@ WORKDIR /
 
 ENTRYPOINT [ "/bin/bash" ]
 #To setup spack you need to run "source /spack/spack/share/spack/setup-env.sh" inside your image
-

include/chimbuko/ad/ADLocalCounterStatistics.hpp

@@ -83,7 +83,7 @@ namespace chimbuko{
      *
      * The message string is the output of net_serialize()
      */
-    std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client) const;
+    std::pair<size_t, size_t> updateGlobalStatistics(ADThreadNetClient &net_client) const;
 
     /**
      * @brief Attach a PerfStats object into which performance metrics are accumulated
@@ -149,7 +149,7 @@ namespace chimbuko{
      * @param step step (or frame) number
      * @return std::pair<size_t, size_t> [sent, recv] message size
      */
-    static std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client, const std::string &l_stats, int step);
+    static std::pair<size_t, size_t> updateGlobalStatistics(ADThreadNetClient &net_client, const std::string &l_stats, int step);
 
     unsigned long m_program_idx; /**< Program idx*/
     int m_step; /**< io step */

include/chimbuko/ad/ADLocalFuncStatistics.hpp

@@ -134,7 +134,7 @@ namespace chimbuko{
      *
      * The message communicated is the string dump of the output of get_json_state()
      */
-    std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client) const;
+    std::pair<size_t, size_t> updateGlobalStatistics(ADThreadNetClient &net_client) const;
 
 
     /**
@@ -206,7 +206,7 @@ namespace chimbuko{
      * @param step step (or frame) number
      * @return std::pair<size_t, size_t> [sent, recv] message size
      */
-    static std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client, const std::string &l_stats, int step);
+    static std::pair<size_t, size_t> updateGlobalStatistics(ADThreadNetClient &net_client, const std::string &l_stats, int step);
 
     AnomalyData m_anom_data; /**< AnomalyData instance holding information about the anomalies */
     std::unordered_map<unsigned long, FuncStats> m_funcstats; /**< map of function index to function profile statistics */