Merge branch 'ckelly_develop'

Author: giltirn

Makefile.am

@@ -1,3 +1,12 @@
 SUBDIRS = include src app test sphinx 3rdparty scripts sim
 nobase_include_HEADERS = chimbuko_config.h
 
+install-provdb-python:
+	python3 -m pip install --user $(top_srcdir)/scripts/provdb_python/
+#python3 -m pip install --prefix $(prefix) $(top_srcdir)/scripts/provdb_python/
+
+uninstall-provdb-python:
+	python3 -m pip uninstall -y provdb-python
+#Below are needed if installing into prefixc becase pip doesn't allow uninstall of packages installed with --prefix?????!?!?	 (I HATE PYTHON)
+#	rm -rf $(prefix)/lib/python3.6
+#	rm $(prefix)/bin/provdb-python

app/bpfile_replay.cpp

@@ -75,7 +75,7 @@ int main(int argc, char** argv){
   }
 
 
-  adios2::ADIOS ad = adios2::ADIOS(adios2::DebugON);
+  adios2::ADIOS ad = adios2::ADIOS();
 
   adios2::IO io_out = ad.DeclareIO("writer");
 

app/driver.cpp

@@ -88,6 +88,7 @@ optionalArgsParser & getOptionalArgsParser(){
     addOptionalCommandLineArg(p, provdb_addr_dir, "Directory in which the provenance database outputs its address files. If empty (default) the provenance DB will not be used.");
     addOptionalCommandLineArg(p, nprovdb_shards, "Number of provenance database shards. Clients connect to shards round-robin by rank (default 1)");
     addOptionalCommandLineArg(p, nprovdb_instances, "Number of provenance database instances. Shards are divided uniformly over instances. (default 1)");
+    addOptionalCommandLineArg(p, provdb_mercury_auth_key, "Set the Mercury authorization key for connection to the provDB (default \"\")");
 #endif
 #ifdef _PERF_METRIC
     addOptionalCommandLineArg(p, perf_outputpath, "Output path for AD performance monitoring data. If an empty string (default) no output is written.");

app/pclient_stats.cpp

@@ -80,15 +80,14 @@ int main (int argc, char** argv)
         int n_anomalies = std::max(0, (int)dist(generator));
         AnomalyData d(0, rank, step, 0, 0, n_anomalies);
 
-	ADLocalFuncStatistics::State state;
-	state.anomaly = d;
+	ADLocalFuncStatistics fstat;
+	fstat.setAnomalyData(d);
 
         // create message
         msg.clear();
         msg.set_info(rank, 0, MessageType::REQ_ADD, MessageKind::ANOMALY_STATS, step);
         msg.set_msg(
-		    state.serialize_cerealpb(), false
-		    //nlohmann::json::object({{"anomaly", d.get_json()}}).dump(), false
+		    fstat.net_serialize(), false
         );
 
 #ifdef _USE_MPINET

app/provdb_admin.cpp

@@ -141,14 +141,15 @@ struct ProvdbArgs{
   bool db_in_mem; //database is in-memory not written to disk, for testing
   std::string db_base_config;
   std::string db_margo_config;
-
+  std::string db_mercury_auth_key;
+  
   bool db_use_aggregator; /**< Use the aggregator backend*/
   int db_batch_size; /**< Batch size for "aggregator" backend*/
 
   bool db_bypass_unqlite;
 
   ProvdbArgs(): engine("ofi+tcp"), autoshutdown(true), server_instance(0), ninstances(1), nshards(1), db_type("unqlite"), db_commit_freq(10000), db_write_dir("."), db_in_mem(false), db_base_config(""), db_margo_config(""),
-		db_use_aggregator(false), db_batch_size(64), db_bypass_unqlite(true){}
+		db_use_aggregator(false), db_batch_size(64), db_bypass_unqlite(true), db_mercury_auth_key(""){}
 };
 
 
@@ -176,6 +177,7 @@ int main(int argc, char** argv) {
     addOptionalCommandLineArg(parser, db_in_mem, "Use an in-memory database rather than writing to disk (*unqlite backend only*) (default false)");
     addOptionalCommandLineArg(parser, db_base_config, "Provide the *absolute path* to a JSON file to use as the base configuration of the Sonata databases. The database path will be appended automatically (default \"\" - not used)");
     addOptionalCommandLineArg(parser, db_margo_config, "Provide the *absolute path* to a JSON file containing the Margo configuration (default \"\" - not used)");
+    addOptionalCommandLineArg(parser, db_mercury_auth_key, "Provide an authentication key for the Mercury configuration (default \"\" - not used)");
     addOptionalCommandLineArg(parser, db_use_aggregator, "Use the \"aggregator\" backend layer (default false)");
     addOptionalCommandLineArg(parser, db_batch_size, "Provide the batch size for the \"aggregator\" backend if in use (default 64)");
     addOptionalCommandLineArg(parser, db_bypass_unqlite, "Use Sonata's bypass method for faster unqlite stores (default true)");    
@@ -244,7 +246,12 @@ int main(int argc, char** argv) {
 
     free(config); //yuck c-strings
     margo_finalize(margo_id);
- 
+
+    //Apply auth_key if provided
+    if(args.db_mercury_auth_key.size()){
+      config_j["mercury"]["auth_key"] = args.db_mercury_auth_key;
+    }
+        
     //Initial number of pools should be 1: the primary pool
     assert(config_j["argobots"]["pools"].size() == 1);
 
@@ -439,16 +446,19 @@ int main(int argc, char** argv) {
 	      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)
+	    bool do_break = false;
+
+	    //Shut down if told to, as long as no clients are still connected
+	    //Force shutdown is called via a "stop_server" RPC
+	    if(cmd_shutdown && connected.size() == 0 && !pserver_connected && !committer_connected){
+	      do_break = true;
+	    }
+	    //If using auto-shutdown, we wait until a) at least one client has previously connected b) no clients are currently connected  c) The pserver and committer are not connected (their connection is optional)
+	    if(args.autoshutdown && a_client_has_connected && connected.size() == 0 && !pserver_connected && !committer_connected){
+	      do_break = true;
+	    }
 
-	    //If args.autoshutdown is disabled we can force shutdown via a "stop_server" RPC
-	    if(
-	       (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) )
-	       ){
+	    if(do_break){
 	      PSprogressStream << "detected all clients disconnected, shutting down" << std::endl;
 #ifdef ENABLE_MARGO_STATE_DUMP
 	      margo_dump("margo_dump_all_client_disconnected." + std::to_string(instance));

app/provdb_commit.cpp

@@ -2,6 +2,7 @@
 #include<chrono>
 #include<thread>
 #include<cassert>
+#include<csignal>
 #include<chimbuko/ad/ADProvenanceDBengine.hpp>
 #include<chimbuko/verbose.hpp>
 #include<chimbuko/util/string.hpp>
@@ -48,8 +49,9 @@ struct Args{
   int ninstances;
   int nshards;
   int freq_ms;
-
-  Args(): addr_file_dir("."), instance(0), ninstances(1), nshards(1), freq_ms(30000){}
+  std::string provdb_mercury_auth_key; //An authorization key for initializing Mercury (optional, default "")
+  
+  Args(): addr_file_dir("."), instance(0), ninstances(1), nshards(1), freq_ms(30000), provdb_mercury_auth_key(""){}
 };
 
 
@@ -62,7 +64,8 @@ int main(int argc, char** argv){
   addOptionalCommandLineArg(parser, ninstances, "Specify the number of server instances (default 1)");
   addOptionalCommandLineArg(parser, nshards, "Specify the total number of database shards (default 1)");
   addOptionalCommandLineArg(parser, freq_ms, "Specify the frequency in ms at which commit is called (default 30000)");
-
+  addOptionalCommandLineArg(parser, provdb_mercury_auth_key, "Set the Mercury authorization key for connection to the provDB (default \"\")");
+  
   if(argc-1 < parser.nMandatoryArgs() || (argc == 2 && std::string(argv[1]) == "-help")){
     parser.help(std::cout);
     return 0;    
@@ -77,6 +80,11 @@ int main(int argc, char** argv){
     return 0;
   } 
 
+  if(args.provdb_mercury_auth_key != ""){
+    CprogressStream << "setting Mercury authorization key to \"" << args.provdb_mercury_auth_key << "\"" << std::endl;
+    ADProvenanceDBengine::setMercuryAuthorizationKey(args.provdb_mercury_auth_key);
+  }
+  
   ProvDBsetup setup(args.nshards, args.ninstances);
   std::string addr = setup.getInstanceAddress(args.instance, args.addr_file_dir);
   int instance_shard_offset = setup.getShardOffsetInstance(args.instance);

app/pserver.cpp

@@ -54,6 +54,7 @@ struct pserverArgs{
 
 #ifdef ENABLE_PROVDB
   std::string provdb_addr_dir;
+  std::string provdb_mercury_auth_key; //An authorization key for initializing Mercury (optional, default "")
 #endif
 
   std::string prov_outputpath;
@@ -63,7 +64,7 @@ struct pserverArgs{
 	       , max_pollcyc_msg(10), zmq_io_thr(1), autoshutdown(true)
 #endif
 #ifdef ENABLE_PROVDB
-	       , provdb_addr_dir("")
+	       , provdb_addr_dir(""), provdb_mercury_auth_key("")
 #endif
   {}
 
@@ -88,6 +89,7 @@ struct pserverArgs{
 #endif
 #ifdef ENABLE_PROVDB
       addOptionalCommandLineArg(p, provdb_addr_dir, "The directory containing the address file written out by the provDB server. An empty string will disable the connection to the global DB.  (default empty, disabled)");
+      addOptionalCommandLineArg(p, provdb_mercury_auth_key, "Set the Mercury authorization key for connection to the provDB (default \"\")");
 #endif
       addOptionalCommandLineArg(p, prov_outputpath, "Output global provenance data to this directory. Can be used in place of or in conjunction with the provenance database. An empty string \"\" (default) disables this output");
       addOptionalCommandLineArg(p, model_update_freq, "The frequency in ms at which the global AD model is updated (default 1000ms)");
@@ -136,14 +138,7 @@ int main (int argc, char ** argv){
   //Optionally load previously-computed AD algorithm statistics
   if(args.load_params_set){
     progressStream << "Pserver: Loading parameters from input file " << args.load_params << std::endl;
-    std::ifstream in(args.load_params);
-    if(!in.good()) throw std::runtime_error("Could not load anomaly algorithm parameters from the file provided");
-    nlohmann::json in_p;
-    in >> in_p;
-    global_func_index_map.deserialize(in_p["func_index_map"]);
-
-    //When the global model is updated the previous model is discarded. Thus if we want to bring in params from a previous run we need to set the state of *one* of the worker threads (which are never flushed)
-    param.updateWorkerModel(in_p["alg_params"].dump(),0);
+    restoreModel(global_func_index_map,  param, args.load_params);
   }
 
   //Start the aggregator thread for the global model
@@ -165,6 +160,11 @@ int main (int argc, char ** argv){
   PSstatSender stat_sender(args.stat_send_freq);
 
 #ifdef ENABLE_PROVDB
+  if(args.provdb_mercury_auth_key != ""){
+    progressStream << "Pserver: setting Mercury authorization key to \"" << args.provdb_mercury_auth_key << "\"" << std::endl;
+    ADProvenanceDBengine::setMercuryAuthorizationKey(args.provdb_mercury_auth_key);
+  }
+  
   PSProvenanceDBclient provdb_client;
 #endif
 
@@ -299,12 +299,7 @@ int main (int argc, char ** argv){
   //Optionally save the final AD algorithm parameters
   if(args.save_params_set){
     progressStream << "PServer: Saving parameters to output file " << args.save_params << std::endl;
-    std::ofstream out(args.save_params);
-    if(!out.good()) throw std::runtime_error("Could not write anomaly algorithm parameters to the file provided");
-    nlohmann::json out_p;
-    out_p["func_index_map"] = global_func_index_map.serialize();
-    out_p["alg_params"] = nlohmann::json::parse(param.getSerializedGlobalModel()); //todo  - store in human readable format rather than net-serialize format
-    out << out_p;
+    writeModel(args.save_params, global_func_index_map, param);
   }
 
   progressStream << "Pserver: finished" << std::endl;

app/sst_view.cpp

@@ -48,7 +48,7 @@ int main(int argc, char** argv){
   adios2::IO io;
   adios2::Engine eng;
 
-  ad = adios2::ADIOS(adios2::DebugON);
+  ad = adios2::ADIOS();
   io = ad.DeclareIO("tau-metrics");
   if(!offline) io.SetEngine("SST");
   io.SetParameters({

benchmark_suite/benchmark_hist_merge/Makefile

@@ -0,0 +1,5 @@
+benchmark: benchmark.cpp
+	mpic++ -std=c++17 -O3 -g -I/src/develop2/PerformanceAnalysis/test/unit_tests -I/install/AD/develop2/include -I/install/AD/develop2/include/chimbuko/3rdparty -L/install/AD/develop2/lib benchmark.cpp -o benchmark -lchimbuko -lstdc++fs
+benchmark_tau: benchmark.cpp
+	tau_cxx.sh -std=c++17 -O3 -g -I/src/develop2/PerformanceAnalysis/test/unit_tests -I/install/AD/develop2/include -I/install/AD/develop2/include/chimbuko/3rdparty -L/install/AD/develop2/lib benchmark.cpp -o benchmark -lchimbuko -lstdc++fs
+

benchmark_suite/benchmark_hist_merge/benchmark.cpp

@@ -0,0 +1,109 @@
+//A fake AD that sends data to the pserver at a regular cadence
+#include<mpi.h>
+#include<chimbuko/ad/ADNetClient.hpp>
+#include<chimbuko/ad/utils.hpp>
+#include<chimbuko/util/commandLineParser.hpp>
+#include<chimbuko/param/sstd_param.hpp>
+#include<chimbuko/param/hbos_param.hpp>
+#include<chimbuko/param/copod_param.hpp>
+#include<chimbuko/ad/ADEvent.hpp>
+#include<chimbuko/util/Anomalies.hpp>
+#include<chimbuko/ad/ADLocalFuncStatistics.hpp>
+#include<chimbuko/ad/ADLocalCounterStatistics.hpp>
+#include<chimbuko/ad/ADLocalAnomalyMetrics.hpp>
+#include<chimbuko/ad/ADcombinedPSdata.hpp>
+#include<chimbuko/ad/ADglobalFunctionIndexMap.hpp>
+#include<chimbuko/verbose.hpp>
+#include "gtest/gtest.h"
+#include<unit_test_common.hpp>
+
+using namespace chimbuko;
+
+struct Args{
+  int nfuncs;
+  std::string algorithm;
+  int hbos_bins;
+  int cycles;
+  
+  Args(){
+    nfuncs = 100;
+    algorithm = "hbos";
+    hbos_bins = 20;
+    cycles = 10000;
+  }
+};
+
+int main(int argc, char **argv){
+  MPI_Init(&argc, &argv);
+  
+  if(const char* env_p = std::getenv("CHIMBUKO_VERBOSE")){
+    std::cout << "Enabling verbose debug output" << std::endl;
+    enableVerboseLogging() = true;
+  }       
+
+  commandLineParser<Args> cmdline;
+  addOptionalCommandLineArgDefaultHelpString(cmdline, cycles);
+  addOptionalCommandLineArgDefaultHelpString(cmdline, nfuncs);
+  addOptionalCommandLineArgDefaultHelpString(cmdline, algorithm); //algorithm, default "sstd"
+  addOptionalCommandLineArgDefaultHelpString(cmdline, hbos_bins); //default 20
+
+  if(argc == 1 || (argc == 2 && std::string(argv[1]) == "-help")){
+    cmdline.help();
+    MPI_Finalize();
+    return 0;
+  }
+
+  Args args;
+  cmdline.parse(args, argc-1, (const char**)(argv+1));
+
+  //Set up a params object with the required number of params
+  ParamInterface *params;
+  if(args.algorithm == "sstd"){
+    SstdParam *p = new SstdParam;
+    for(int i=0;i<args.nfuncs;i++){
+      RunStats &r = (*p)[i];
+      for(int j=0;j<100;j++)
+	r.push(double(j));
+    }
+    params = p;
+  }else if(args.algorithm == "hbos" || args.algorithm == "copod"){
+    std::vector<unsigned int> counts(args.hbos_bins);
+    for(int i=0;i<args.hbos_bins;i++) counts[i] = i;
+    Histogram hd;
+    hd.set_histogram(counts, 0.0001, args.hbos_bins, 0, 1);
+
+    if(args.algorithm == "hbos"){
+      HbosParam *p = new HbosParam;
+      for(int i=0;i<args.nfuncs;i++) (*p)[i].getHistogram() = hd;
+      params = p;
+    }else{
+      CopodParam *p = new CopodParam;
+      for(int i=0;i<args.nfuncs;i++) (*p)[i].getHistogram() = hd;
+      params = p;
+    }
+  }else{
+    fatal_error("Unknown AD algorithm");
+  }
+
+  PerfTimer cyc_timer;
+  PerfTimer timer;
+
+  //To make the benchmark as lightweight as possible, precompute the messages and send the same each cycle
+  std::string params_msg = params->serialize(); 
+  RunStats r;  
+
+  for(int c=0;c<args.cycles;c++){
+    cyc_timer.start();
+    std::cout << c << std::endl;
+      
+    timer.start();
+    params->update(params_msg, false);
+    r.push(timer.elapsed_ms());    
+  }//cycle loop
+  std::cout << r.mean() << " " << r.stddev() << std::endl;
+
+  delete params;
+  MPI_Finalize();
+  
+  return 0;
+}