Added "provdb_commit" main program that remotely requests the provDB to commit its database to disk at some regular frequency

provdb changes:
	Added RPCs to register connection/disconnection of commit client
	Added RPC that returns the status of connected clients
	Setting db_commmit_freq cmdline option to 0 now disables the periodic commit called within provdb_admin
Modified launch script to disable the periodic commit inside provdb_admin and use provdb_commit instead. This appears to workaround the provdb hang

Author: giltirn

app/Makefile.am

@@ -35,7 +35,7 @@ bpfile_replay_LDADD = $(LDADD)
 
 
 if ENABLE_PROVDB
-bin_PROGRAMS += provdb_admin provdb_query provdb_shutdown provdb_rebuild_ascii provdb_dump_state
+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)
@@ -49,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

@@ -20,7 +20,7 @@
 
 #include <spdlog/spdlog.h>
 #include "chimbuko/verbose.hpp"
-
+#include <thallium/serialization/stl/string.hpp>
 
 namespace tl = thallium;
 
@@ -39,6 +39,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);
@@ -67,6 +70,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) {
@@ -120,7 +150,7 @@ 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 \".\")");
 
     if(argc-1 < parser.nMandatoryArgs() || (argc == 2 && std::string(argv[1]) == "-help")){
@@ -140,8 +170,13 @@ int main(int argc, char** argv) {
 
     progressStream << "ProvDB Admin: initializing thallium with address: " << eng_opt << std::endl;
 
+    //Disable loopback so that RPC calls from this process are forced to go through the net interface
+    hg_init_info info;
+    memset(&info, 0, sizeof(info));
+    info.no_loopback = HG_TRUE;
+
     //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, &info); 
     margo_id = engine.get_margo_instance();
 
 #ifdef _PERF_METRIC
@@ -163,8 +198,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
 
@@ -230,7 +269,7 @@ int main(int argc, char** argv) {
 	    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++){		
 		progressStream << "ProvDB Admin: committing shard " << s << std::endl;
@@ -248,7 +287,8 @@ 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;
 	      break;

app/provdb_commit.cpp

@@ -0,0 +1,117 @@
+#include<atomic>
+#include<chrono>
+#include<thread>
+#include<cassert>
+#include<chimbuko/ad/ADProvenanceDBengine.hpp>
+#include<chimbuko/verbose.hpp>
+#include<chimbuko/util/string.hpp>
+#include <thallium/serialization/stl/string.hpp>
+
+using namespace chimbuko;
+
+bool stop_wait_loop = false; //trigger breakout of main thread spin loop
+
+void termSignalHandler( int signum ) {
+  stop_wait_loop = true;
+  progressStream << "ProvDB Commit caught SIGTERM, exiting " << std::endl;
+}
+
+bool exit_check(thallium::endpoint &server, 
+		thallium::remote_procedure &connection_status){
+  std::string conn = connection_status.on(server)();
+  int connected =-1;
+  int a_client_has_connected=-1;
+  int pserver_connected=-1;
+  int pserver_has_connected=-1;
+  std::stringstream of(conn);
+  of >> connected >> a_client_has_connected >> pserver_connected >> pserver_has_connected;
+  assert(!of.fail());
+
+  progressStream << "ProvDB Commit determined status: clients connected=" << connected << " (a client has connected ? " << a_client_has_connected << ") "
+		 << "pserver connected=" << pserver_connected << " (pserver has connected ? " << pserver_has_connected << ")" << std::endl;
+  return 
+    ( a_client_has_connected && connected == 0 ) &&
+    ( !pserver_has_connected || (pserver_has_connected && !pserver_connected) );
+}
+
+
+int main(int argc, char** argv){  
+#ifdef ENABLE_PROVDB
+  if(argc != 4){
+    std::cout << "Usage: provdb_commit <server address e.g. ofi+tcp;ofi_rxm://172.17.0.4:5000> <nshards> <freq (ms)>" << std::endl;
+    return 1;
+  }
+  std::string addr = argv[1];
+  int nshards = std::stoi(argv[2]);
+  int freq_ms = std::stoi(argv[3]);
+
+  std::string protocol = ADProvenanceDBengine::getProtocolFromAddress(addr);
+  if(ADProvenanceDBengine::getProtocol().first != protocol){
+    int mode = ADProvenanceDBengine::getProtocol().second;
+    verboseStream << "DB client reinitializing engine with protocol \"" << protocol << "\"" << std::endl;
+    ADProvenanceDBengine::finalize();
+    ADProvenanceDBengine::setProtocol(protocol,mode);      
+  }      
+
+  thallium::engine &eng = ADProvenanceDBengine::getEngine();
+
+  thallium::endpoint server = eng.lookup(addr);
+
+  thallium::remote_procedure client_hello(eng.define("committer_hello").disable_response());
+  thallium::remote_procedure client_goodbye(eng.define("committer_goodbye").disable_response());
+  thallium::remote_procedure connection_status(eng.define("connection_status"));
+  client_hello.on(server)();
+
+  {
+    sonata::Client client = sonata::Client(eng);
+    {    
+      std::vector<sonata::Database> databases(nshards);
+      for(int i=0;i<nshards;i++){
+	std::string db_name = stringize("provdb.%d", i);
+	databases[i] = client.open(addr, 0, db_name);
+      }
+      sonata::Database glob_db  = client.open(addr, 0, "provdb.global");
+    
+      typedef std::chrono::high_resolution_clock Clock;
+      Clock::time_point commit_timer_start = Clock::now();
+
+      signal(SIGTERM, termSignalHandler);  
+      progressStream << "ProvDB Commit:  starting commit loop" << std::endl;    
+
+      int iter = 0;
+      while(!stop_wait_loop) { //stop wait loop will be set by SIGTERM handler
+	thallium::thread::sleep(eng, 1000); //Thallium engine sleeps but listens for rpc requests
+
+	//Check every 10 seconds for connection status
+	if(iter % 10 == 0 &&
+	   exit_check(server, connection_status ) ){
+	  progressStream << "ProvDB Commit: no clients connected, exiting" << std::endl;
+	  break;
+	}
+
+	unsigned long commit_timer_ms = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now() - commit_timer_start).count();
+	if(commit_timer_ms >= freq_ms){
+	  verboseStream << "ProvDB Commit: committing database to disk" << std::endl;
+	  for(int s=0;s<nshards;s++){		
+	    progressStream << "ProvDB Commit: committing shard " << s << std::endl;
+	    databases[s].commit();
+	  }
+	  progressStream << "ProvDB Commit: committing global db" << std::endl;
+	  glob_db.commit();
+	  commit_timer_start = Clock::now();
+	}
+	++iter;
+      }//while(!stop_wait_loop)
+    } //db scope
+  }//client scope
+
+  progressStream << "ProvDB Commit sending goodbye handshake" << std::endl;
+  client_goodbye.on(server)();
+  client_hello.deregister();
+  client_goodbye.deregister();
+  connection_status.deregister();
+#endif
+
+  progressStream << "ProvDB Commit done" << std::endl;
+  return 0;
+}

scripts/launch/run_services.sh

@@ -110,7 +110,7 @@ if (( ${use_provdb} == 1 )); then
     #Enable better error reporting from Mercury
     export HG_LOG_SUBSYS=hg export HG_LOG_LEVEL=error
 
-    provdb_admin "${provdb_addr}" ${provdb_extra_args} -engine ${provdb_engine} -nshards ${provdb_nshards} -nthreads ${provdb_nthreads} -db_write_dir ${provdb_writedir} 2>&1 | tee ${log_dir}/provdb.log &
+    provdb_admin "${provdb_addr}" ${provdb_extra_args} -engine ${provdb_engine} -nshards ${provdb_nshards} -nthreads ${provdb_nthreads} -db_write_dir ${provdb_writedir} -db_commit_freq 0 2>&1 | tee ${log_dir}/provdb.log &
     provdb_pid=$!
 
     start_time=$SECONDS
@@ -135,6 +135,16 @@ else
     ps_extra_args+=" -prov_outputpath ${provdb_writedir}"
 fi
 
+#Committer
+if (( ${use_provdb} == 1 )); then
+    echo "==========================================="
+    echo "Instantiating committer"
+    echo "==========================================="
+    prov_add=$(cat ${provdb_dir}/provider.address)
+    provdb_commit "${prov_add}" ${provdb_nshards} 10000 2>&1 | tee ${log_dir}/committer.log &
+    sleep 3
+fi
+
 #Visualization
 if (( ${use_viz} == 1 )); then
     if (( ${use_pserver} != 1 )); then

src/ad/ADParser.cpp

@@ -83,7 +83,7 @@ int ADParser::beginStep(bool verbose) {
       }else if(status == adios2::StepStatus::NotReady){ 
 	double elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now() - start).count();
 	if(elapsed > m_beginstep_timeout * 1000){
-	  recoverable_error("ADParser::beginStep : ADIOS2::BeginStep timed out waiting for next step to be ready\n"); 
+	  recoverable_error("ADParser::beginStep : ADIOS2::BeginStep timed out after " + std::to_string(elapsed) + " ms waiting for next step to be ready\n"); 
 	  m_status = false;
 	  m_current_step = -1;
 	  break;