merge sm_release

Author: sandeepmittal

Makefile.am

@@ -1,3 +1,3 @@
-SUBDIRS = include src app test sphinx 3rdparty scripts
+SUBDIRS = include src app test sphinx 3rdparty scripts sim
 nobase_include_HEADERS = chimbuko_config.h
 

app/Makefile.am

@@ -21,6 +21,9 @@ hpserver_LDADD = $(LDADD)
 pshutdown_SOURCES = pshutdown.cpp
 pshutdown_LDADD = $(LDADD)
 
+provdb_dump_state_SOURCES = provdb_dump_state.cpp
+provdb_dump_state_LDADD = $(LDADD)
+
 sstSinker_SOURCES = sstSinker.cpp
 sstSinker_LDADD = $(LDADD)
 
@@ -32,7 +35,7 @@ bpfile_replay_LDADD = $(LDADD)
 
 
 if ENABLE_PROVDB
-bin_PROGRAMS += provdb_admin provdb_query provdb_shutdown provdb_rebuild_ascii
+bin_PROGRAMS += provdb_admin provdb_query provdb_shutdown provdb_rebuild_ascii provdb_dump_state provdb_commit
 
 provdb_admin_SOURCES = provdb_admin.cpp
 provdb_admin_LDADD = $(LDADD)
@@ -46,6 +49,10 @@ provdb_shutdown_LDADD = $(LDADD)
 provdb_rebuild_ascii_SOURCES = provdb_rebuild_ascii.cpp
 provdb_rebuild_ascii_LDADD = $(LDADD)
 
+provdb_commit_SOURCES = provdb_commit.cpp
+provdb_commit_LDADD = $(LDADD)
+
+
 else
 	echo "Provenance DB not being built"
 endif

app/provdb_admin.cpp

@@ -6,6 +6,7 @@
 #endif
 #include "chimbuko/util/commandLineParser.hpp"
 #include "chimbuko/util/string.hpp"
+#include "chimbuko/util/time.hpp"
 #include <iostream>
 #include <sonata/Admin.hpp>
 #include <sonata/Provider.hpp>
@@ -16,10 +17,11 @@
 #include <iostream>
 #include <csignal>
 #include <cassert>
+#include <nlohmann/json.hpp>
 
 #include <spdlog/spdlog.h>
 #include "chimbuko/verbose.hpp"
-
+#include <thallium/serialization/stl/string.hpp>
 
 namespace tl = thallium;
 
@@ -38,6 +40,9 @@ bool a_client_has_connected = false; //has an AD rank previously connected?
 bool pserver_connected = false; //is the pserver connected
 bool pserver_has_connected = false; //did the pserver ever connect
 
+bool committer_connected = false; //is the committer connected?
+bool committer_has_connected = false; //did the committer ever connect?
+
 //Allows a client to register with the provider
 void client_hello(const tl::request& req, const int rank) {
   std::lock_guard<tl::mutex> lock(*mtx);
@@ -66,6 +71,33 @@ void pserver_goodbye(const tl::request& req) {
   progressStream << "ProvDB Admin: Pserver has said goodbye" << std::endl;
 }
 
+//Allows the committer to register with the provider
+void committer_hello(const tl::request& req) {
+  std::lock_guard<tl::mutex> lock(*mtx);
+  committer_connected = true;
+  committer_has_connected = true;
+  progressStream << "ProvDB Admin: Committer has said hello" << std::endl;
+}
+//Allows the committer to deregister from the provider
+void committer_goodbye(const tl::request& req) {
+  std::lock_guard<tl::mutex> lock(*mtx);
+  committer_connected = false;
+  progressStream << "ProvDB Admin: Committer has said goodbye" << std::endl;
+}
+
+//Get the connection status as a string of format 
+//"<current clients> <a client has connected (0/1)> <pserver connected (0/1)> <pserver has connected (0/1)> <committer connected (0/1)> <commiter has connected (0/1)>"
+void connection_status(const tl::request& req) {
+  std::lock_guard<tl::mutex> lock(*mtx);
+  std::stringstream ss;
+  ss << connected.size() << " " << int(a_client_has_connected) << " " 
+     << int(pserver_connected) << " " << int(pserver_has_connected) << " "
+     << int(committer_connected) << " " << int(committer_has_connected);
+  req.respond(ss.str());
+}
+
+
+
 bool cmd_shutdown = false; //true if a client has requested that the server shut down
 
 void client_stop_rpc(const tl::request& req) {
@@ -74,6 +106,18 @@ void client_stop_rpc(const tl::request& req) {
   progressStream << "ProvDB Admin: Received shutdown request from client" << std::endl;
 }
 
+margo_instance_id margo_id;
+
+void margo_dump(const std::string &stub){
+  std::string fn = stub + "." + getDateTimeFileExt();
+  progressStream << "ProvDB Admin: margo dump to " << fn << std::endl;
+  margo_state_dump(margo_id, fn.c_str(), 0, nullptr);
+}
+
+void margo_dump_rpc(const tl::request& req) {
+  const static std::string stub("margo_dump");
+  margo_dump(stub);
+}
 
 
 
@@ -87,8 +131,11 @@ struct ProvdbArgs{
   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), db_write_dir("."){}
+  bool db_in_mem; //database is in-memory not written to disk, for testing
+  std::string db_base_config;
+  std::string db_margo_config;
+  
+  ProvdbArgs(): engine("ofi+tcp"), autoshutdown(true), nshards(1), db_type("unqlite"), nthreads(1), db_commit_freq(10000), db_write_dir("."), db_in_mem(false), db_base_config(""), db_margo_config(""){}
 };
 
 
@@ -112,8 +159,11 @@ int main(int argc, char** argv) {
     addOptionalCommandLineArg(parser, nshards, "Specify the number of database shards (default 1)");
     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_commit_freq, "Specify the frequency at which the database flushes to disk in ms (default 10000). 0 disables the flush until the end.");
     addOptionalCommandLineArg(parser, db_write_dir, "Specify the directory in which the database shards will be written (default \".\")");
+    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)");
 
     if(argc-1 < parser.nMandatoryArgs() || (argc == 2 && std::string(argv[1]) == "-help")){
       parser.help(std::cout);
@@ -124,16 +174,38 @@ int main(int argc, char** argv) {
     parser.parseCmdLineArgs(args, argc, argv);
 
     if(args.nshards < 1) throw std::runtime_error("Must have at least 1 database shard");
+    if(args.db_in_mem && args.db_type != "unqlite") throw std::runtime_error("-db_in_mem option not valid for backends other than unqlite");
+
+    if(args.db_in_mem){ progressStream << "Using in-memory database" << std::endl; }
 
     std::string eng_opt = args.engine;
     if(args.ip.size() > 0){
       eng_opt += std::string("://") + args.ip;
     }
 
+    //Get Sonata config
+    nlohmann::json base_config;
+    if(args.db_base_config.size() > 0){
+      std::ifstream in(args.db_base_config);
+      if(in.fail()){ throw std::runtime_error("Failed to read db config file " + args.db_base_config); }
+      base_config = nlohmann::json::parse(in);
+    }
+
     progressStream << "ProvDB Admin: initializing thallium with address: " << eng_opt << std::endl;
 
+    //Get Margo config
+    std::string margo_config = stringize("{ \"rpc_thread_count\" : %d, \"use_progress_thread\" : true }", args.nthreads);
+    if(args.db_margo_config.size() > 0){
+      std::ifstream in(args.db_margo_config);
+      if(in.fail()){ throw std::runtime_error("Failed to read Margo config file " + args.db_margo_config); }
+      auto cfg = nlohmann::json::parse(in);
+      margo_config = cfg.dump();
+    }
+    progressStream << "ProvDB Admin: Margo configuration " << margo_config << std::endl;
+
     //Initialize provider engine
-    tl::engine engine(eng_opt, THALLIUM_SERVER_MODE, true, args.nthreads);
+    tl::engine engine(eng_opt, THALLIUM_SERVER_MODE, margo_config); 
+    margo_id = engine.get_margo_instance();
 
 #ifdef _PERF_METRIC
     //Get Margo to output profiling information
@@ -154,7 +226,12 @@ 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("committer_hello",committer_hello).disable_response();
+    engine.define("committer_goodbye",committer_goodbye).disable_response();
     engine.define("stop_server",client_stop_rpc).disable_response();
+    engine.define("margo_dump", margo_dump_rpc).disable_response();
+    engine.define("connection_status", connection_status);
+
 
     std::string addr = (std::string)engine.self();  //ip and port of admin
 
@@ -167,19 +244,25 @@ int main(int argc, char** argv) {
 
       { //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/%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);
 
+	nlohmann::json glob_db_config = base_config;
+	glob_db_config["path"] = args.db_in_mem ? ":mem:" : stringize("%s/%s.unqlite", args.db_write_dir.c_str(), glob_db_name.c_str());
+
+	progressStream << "ProvDB Admin: creating global data database: " << glob_db_name << " " << glob_db_config.dump() << " " << args.db_type << std::endl;
+	admin.createDatabase(addr, 0, glob_db_name, args.db_type, glob_db_config.dump());
+	
 	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/%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);
+
+	  nlohmann::json config = base_config;
+	  config["path"] = args.db_in_mem ? ":mem:" : stringize("%s/%s.unqlite", args.db_write_dir.c_str(), db_name.c_str());
+
+	  progressStream << "ProvDB Admin: Shard " << s << ": " << db_name << " " << config.dump() << " " << args.db_type << std::endl;
+	  admin.createDatabase(addr, 0, db_name, args.db_type, config.dump());
 	  db_shard_names[s] = db_name;
 	}
 
@@ -213,18 +296,20 @@ int main(int argc, char** argv) {
 	  typedef std::chrono::high_resolution_clock Clock;
 	  Clock::time_point commit_timer_start = Clock::now();
 
-
 	  //Spin quietly until SIGTERM sent
 	  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
 
 	    unsigned long commit_timer_ms = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now() - commit_timer_start).count();
-	    if(commit_timer_ms >= args.db_commit_freq){
+	    if(args.db_commit_freq > 0 && commit_timer_ms >= args.db_commit_freq){
 	      verboseStream << "ProvDB Admin: committing database to disk" << std::endl;
-	      for(int s=0;s<args.nshards;s++)
+	      for(int s=0;s<args.nshards;s++){		
+		progressStream << "ProvDB Admin: committing shard " << s << std::endl;
 		db[s].commit();
+	      }
+	      progressStream << "ProvDB Admin: committing global db" << std::endl;
 	      glob_db.commit();
 	      commit_timer_start = Clock::now();
 	    }
@@ -236,9 +321,11 @@ int main(int argc, char** argv) {
 	    if(
 	       (args.autoshutdown || cmd_shutdown)  &&
 	       ( a_client_has_connected && connected.size() == 0 ) &&
-	       ( !pserver_has_connected || (pserver_has_connected && !pserver_connected) )
+	       ( !pserver_has_connected || (pserver_has_connected && !pserver_connected) ) &&
+	       ( !committer_has_connected || (committer_has_connected && !committer_connected) )
 	       ){
 	      progressStream << "ProvDB Admin: detected all clients disconnected, shutting down" << std::endl;
+	      margo_dump("margo_dump_all_client_disconnected");
 	      break;
 	    }
 	  }
@@ -251,9 +338,11 @@ int main(int argc, char** argv) {
 	}
 
 	progressStream << "ProvDB Admin: ending admin scope" << std::endl;
+	margo_dump("margo_dump_end_admin_scope");
       }//admin scope
 
       progressStream << "ProvDB Admin: ending provider scope" << std::endl;
+      margo_dump("margo_dump_end_provide_scope");
     }//provider scope
 
     progressStream << "ProvDB Admin: shutting down server engine" << std::endl;