Merge branch 'ckelly_develop' into sm_release

Author: sandeepmittal

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
 
@@ -32,14 +32,17 @@ bpfile_replay_LDADD = $(LDADD)
 
 
 if ENABLE_PROVDB
-bin_PROGRAMS += provdb_admin provdb_query
+bin_PROGRAMS += provdb_admin provdb_query provdb_shutdown
 
 provdb_admin_SOURCES = provdb_admin.cpp
 provdb_admin_LDADD = $(LDADD)
 
 provdb_query_SOURCES = provdb_query.cpp
 provdb_query_LDADD = $(LDADD)
 
+provdb_shutdown_SOURCES = provdb_shutdown.cpp
+provdb_shutdown_LDADD = $(LDADD)
+
 else
 	echo "Provenance DB not being built"
 endif

app/driver.cpp

@@ -1,8 +1,8 @@
-// #include "chimbuko/AD.hpp"
 #include "chimbuko/chimbuko.hpp"
 #include "chimbuko/verbose.hpp"
 #include "chimbuko/util/string.hpp"
 #include "chimbuko/util/commandLineParser.hpp"
+#include "chimbuko/util/error.hpp"
 #include <chrono>
 #include <cstdlib>
 
@@ -285,14 +285,17 @@ int main(int argc, char ** argv){
     }
   catch (const std::invalid_argument &e){
     std::cout << '[' << getDateTime() << ", rank " << params.rank << "] Driver : caught invalid argument: " << e.what() << std::endl;
+    if(params.err_outputpath.size()) recoverable_error(std::string("Driver : caught invalid argument: ") + e.what()); //ensure errors also written to error logs
     error = true;
   }
   catch (const std::ios_base::failure &e){
     std::cout << '[' << getDateTime() << ", rank " << params.rank << "] Driver : I/O base exception caught: " << e.what() << std::endl;
+    if(params.err_outputpath.size()) recoverable_error(std::string("Driver : I/O base exception caught: ") + e.what());
     error = true;
   }
   catch (const std::exception &e){
     std::cout << '[' << getDateTime() << ", rank " << params.rank << "] Driver : Exception caught: " << e.what() << std::endl;
+    if(params.err_outputpath.size()) recoverable_error(std::string("Driver : Exception caught: ") + e.what());
     error = true;
   }
 

app/provdb_admin.cpp

@@ -66,6 +66,17 @@ void pserver_goodbye(const tl::request& req) {
   progressStream << "ProvDB Admin: Pserver has said goodbye" << std::endl;
 }
 
+bool cmd_shutdown = false; //true if a client has requested that the server shut down
+
+void client_stop_rpc(const tl::request& req) {
+  std::lock_guard<tl::mutex> lock(*mtx);
+  cmd_shutdown = true;
+  progressStream << "ProvDB Admin: Received shutdown request from client" << std::endl;
+}
+
+
+
+
 
 struct ProvdbArgs{
   std::string ip;
@@ -75,8 +86,9 @@ struct ProvdbArgs{
   int nthreads;
   std::string db_type;
   unsigned long db_commit_freq;
+  std::string db_write_dir;
 
-  ProvdbArgs(): engine("ofi+tcp"), autoshutdown(true), nshards(1), db_type("unqlite"), nthreads(1), db_commit_freq(10000){}
+  ProvdbArgs(): engine("ofi+tcp"), autoshutdown(true), nshards(1), db_type("unqlite"), nthreads(1), db_commit_freq(10000), db_write_dir("."){}
 };
 
 
@@ -88,7 +100,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
@@ -101,6 +113,7 @@ int main(int argc, char** argv) {
     addOptionalCommandLineArg(parser, nthreads, "Specify the number of RPC handler threads (default 1)");
     addOptionalCommandLineArg(parser, db_type, "Specify the Sonata database type (default \"unqlite\")");
     addOptionalCommandLineArg(parser, db_commit_freq, "Specify the frequency at which the database flushes to disk in ms (default 10000)");
+    addOptionalCommandLineArg(parser, db_write_dir, "Specify the directory in which the database shards will be written (default \".\")");
 
     if(argc-1 < parser.nMandatoryArgs() || (argc == 2 && std::string(argv[1]) == "-help")){
       parser.help(std::cout);
@@ -120,7 +133,7 @@ int main(int argc, char** argv) {
     progressStream << "ProvDB Admin: initializing thallium with address: " << eng_opt << std::endl;
 
     //Initialize provider engine
-    tl::engine engine(eng_opt, THALLIUM_SERVER_MODE, true, args.nthreads); 
+    tl::engine engine(eng_opt, THALLIUM_SERVER_MODE, true, args.nthreads);
 
 #ifdef _PERF_METRIC
     //Get Margo to output profiling information
@@ -141,29 +154,30 @@ int main(int argc, char** argv) {
     engine.define("client_goodbye",client_goodbye).disable_response();
     engine.define("pserver_hello",pserver_hello).disable_response();
     engine.define("pserver_goodbye",pserver_goodbye).disable_response();
+    engine.define("stop_server",client_stop_rpc).disable_response();
 
     std::string addr = (std::string)engine.self();  //ip and port of admin
 
     { //Scope in which provider is active
 
       //Initialize provider
       sonata::Provider provider(engine, 0);
-    
+
       progressStream << "ProvDB Admin: Provider is running on " << addr << std::endl;
 
       { //Scope in which admin object is active
 	sonata::Admin admin(engine);
 	progressStream << "ProvDB Admin: creating global data database" << std::endl;
 	std::string glob_db_name = "provdb.global";
-	std::string glob_db_config = stringize("{ \"path\" : \"./%s.unqlite\" }", glob_db_name.c_str());
+	std::string glob_db_config = stringize("{ \"path\" : \"%s/%s.unqlite\" }", args.db_write_dir.c_str(), glob_db_name.c_str());
 	admin.createDatabase(addr, 0, glob_db_name, args.db_type, glob_db_config);
-	
+
 	progressStream << "ProvDB Admin: creating " << args.nshards << " database shards" << std::endl;
 
 	std::vector<std::string> db_shard_names(args.nshards);
 	for(int s=0;s<args.nshards;s++){
 	  std::string db_name = stringize("provdb.%d",s);
-	  std::string config = stringize("{ \"path\" : \"./%s.unqlite\" }", db_name.c_str());
+	  std::string config = stringize("{ \"path\" : \"%s/%s.unqlite\" }", args.db_write_dir.c_str(), db_name.c_str());
 	  progressStream << "ProvDB Admin: Shard " << s << ": " << db_name << " " << config << " " << args.db_type << std::endl;
 	  admin.createDatabase(addr, 0, db_name, args.db_type, config);
 	  db_shard_names[s] = db_name;
@@ -172,7 +186,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++){
@@ -201,7 +215,7 @@ int main(int argc, char** argv) {
 
 
 	  //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
@@ -214,11 +228,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 && 
+	       (args.autoshutdown || cmd_shutdown)  &&
 	       ( a_client_has_connected && connected.size() == 0 ) &&
 	       ( !pserver_has_connected || (pserver_has_connected && !pserver_connected) )
 	       ){
@@ -233,18 +249,18 @@ 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;
       }//admin scope
 
       progressStream << "ProvDB Admin: ending provider scope" << std::endl;
     }//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/pserver.cpp

@@ -131,7 +131,7 @@ int main (int argc, char ** argv){
     nlohmann::json in_p;
     in >> in_p;
     global_func_index_map.deserialize(in_p["func_index_map"]);
-    param->assign(in_p["alg_params"].dump()); //param.assign(in_p["alg_params"].dump());
+    param->assign(in_p["alg_params"].dump());
   }
 
 #ifdef _USE_MPINET
@@ -181,8 +181,8 @@ int main (int argc, char ** argv){
       if(args.stat_outputdir.size()) std::cout << "(dir @ " << args.stat_outputdir << ")";
     }
 
-    net.add_payload(new NetPayloadUpdateParams(param, args.freeze_params)); //new NetPayloadUpdateParams(&param, args.freeze_params));
-    net.add_payload(new NetPayloadGetParams(param)); //new NetPayloadGetParams(&param));
+    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 NetPayloadGlobalFunctionIndexMapBatched(&global_func_index_map));
@@ -227,7 +227,7 @@ int main (int argc, char ** argv){
     o.open(args.logdir + "/parameters.txt");
     if (o.is_open())
       {
-	param->show(o); //param.show(o);
+	param->show(o);
 	o.close();
       }
   }
@@ -256,13 +256,10 @@ int main (int argc, char ** argv){
     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->serialize()); //param.serialize());
+    out_p["alg_params"] = nlohmann::json::parse(param->serialize());
     out << out_p;
   }
 
   progressStream << "Pserver: finished" << std::endl;
-
-  delete param;
-
   return 0;
 }

app/pshutdown.cpp

@@ -1,46 +1,23 @@
-#include "chimbuko/net/zmq_net.hpp"
-#include "chimbuko/message.hpp"
+#include "chimbuko/verbose.hpp"
+#include "chimbuko/util/error.hpp"
+#include "chimbuko/ad/ADNetClient.hpp"
 
 using namespace chimbuko;
 
-int main (int argc, char ** argv)
-{
-    void * context;
-    void * socket;
-    std::string addr;
-    Message msg;
-    std::string strmsg;
-
-    if (argc > 1)
-        addr = argv[1];
-
-    context = zmq_ctx_new();
-    socket = zmq_socket(context, ZMQ_REQ);
-    zmq_connect(socket, addr.c_str());
-
-    // test connection
-    msg.clear();
-    msg.set_info(-1, -1, MessageType::REQ_ECHO, MessageKind::DEFAULT);
-    msg.set_msg("Hello!");
-
-    ZMQNet::send(socket, msg.data());
-
-    msg.clear();
-    ZMQNet::recv(socket, strmsg);
-    msg.set_msg(strmsg, true);
-
-    // if (msg.data_buffer().compare("Hello!>I am ZMQNET!") != 0)
-    // {
-    //     std::cerr << "Connect error to parameter server (ZMQNET)!\n";
-    //     exit(1);
-    // }    
-
-    // shutdown
-    zmq_send(socket, nullptr, 0, 0);
-
-    // finalize
-    zmq_close(socket);
-    zmq_ctx_term(context);
-
-    return EXIT_SUCCESS; 
-}
+int main(int argc, char** argv){
+#ifdef _USE_ZMQNET
+  if(argc != 2){
+    std::cout << "Usage: pshutdown <server address e.g. tcp://localhost:5559>" << std::endl;
+    return 1;
+  }
+  ADNetClient ad;
+  ad.connect_ps(0,0,argv[1]);
+  ad.stopServer();
+  ad.disconnect_ps();
+
+  return 0;
+#else
+  std::cout << "Not implemented for MPINET" << std::endl;
+  return 1;
+#endif
+}