Restored a number of things that got lost in the recent merge

Author: giltirn

app/Makefile.am

@@ -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;
@@ -143,6 +154,7 @@ 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
 
@@ -219,8 +231,10 @@ int main(int argc, char** argv) {
 
 	    //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) )
 	       ){

app/provdb_shutdown.cpp

@@ -0,0 +1,22 @@
+#include "chimbuko/ad/ADProvenanceDBclient.hpp"
+
+using namespace chimbuko;
+
+int main(int argc, char** argv){
+#ifdef ENABLE_PROVDB
+  if(argc != 2){
+    std::cout << "Usage: provdb_shutdown <server address e.g. ofi+tcp;ofi_rxm://172.17.0.4:5000>" << std::endl;
+    return 1;
+  }
+  ADProvenanceDBclient ad(0);
+  ad.connect(argv[1], 1);
+  ad.stopServer();
+  ad.disconnect();
+
+  return 0;
+#else
+  std::cout << "ProvDB not in use" << std::endl;
+  return 1;
+#endif
+}
+

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,7 +256,7 @@ 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;
   }
 

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
+}

include/chimbuko/ad/ADEvent.hpp

@@ -1,6 +1,7 @@
 #pragma once
 #include "chimbuko/ad/ADDefine.hpp"
 #include "chimbuko/ad/ExecData.hpp"
+#include "chimbuko/ad/ADMetadataParser.hpp"
 #include "chimbuko/util/map.hpp"
 #include <string>
 #include <vector>
@@ -164,6 +165,13 @@ namespace chimbuko {
      */
     void linkCounterMap(const std::unordered_map<int, std::string>* m) { m_counterMap = m; }
 
+
+    /**
+     * @brief Optional: give the event manager knowledge of which threads are GPU threads, improves error checking
+     */
+    void linkGPUthreadMap(const std::unordered_map<unsigned long, GPUvirtualThreadInfo> *m){ m_gpu_thread_Map = m; }
+
+
     /**
      * @brief Get the Func Map object
      *
@@ -273,6 +281,10 @@ namespace chimbuko {
      */
     CallListMap_p_t* trimCallList(int n_keep_thread = 0);
 
+    /**
+     * @brief Get the total number of function events in the call list over all pid/rid/tid
+     */
+    size_t getCallListSize() const;
 
     /**
      * @brief purge all function calls that are completed (i.e. a pair of ENTRY and EXIT events are observed)
@@ -318,6 +330,11 @@ namespace chimbuko {
     int m_eidx_comm_recv; /**< If previously seen, the eid corresponding to the comm recv event (-1 otherwise)*/
 
 
+    /**
+     * @brief Optional: give the event manager knowledge of which threads are GPU threads, improves error checking
+     */
+    const std::unordered_map<unsigned long, GPUvirtualThreadInfo> *m_gpu_thread_Map;
+
 
     /**
      * @brief communication event stack. Once a function call has exited, all comms events are associated with that call and the stack is cleared

include/chimbuko/ad/ADNetClient.hpp

@@ -58,7 +58,7 @@ namespace chimbuko{
      * @param msg The message
      * @return The response message in serialized format. Use Message::set_msg( <serialized_msg>,  true )  to unpack
      */       
-    std::string send_and_receive(const Message &msg);
+    std::string send_and_receive(const Message &msg) const;
 
     /**
      * @brief Send a message to the parameter server and receive the response both as Message objects
@@ -90,6 +90,10 @@ namespace chimbuko{
      */
     void * getZMQcontext(){ return m_context; }
 
+    /**
+     * @brief Issue a stop command to the server. The server will then stop once all clients have disconnected and all messages processed
+     */
+    void stopServer() const;
 #endif
 
 

include/chimbuko/ad/ADOutlier.hpp

@@ -9,7 +9,6 @@
 #include "chimbuko/util/PerfStats.hpp"
 #include "chimbuko/util/Anomalies.hpp"
 
-
 namespace chimbuko {
   /**
    * @brief abstract class for anomaly detection algorithms

include/chimbuko/ad/ADProvenanceDBclient.hpp

@@ -90,6 +90,7 @@ namespace chimbuko{
     thallium::endpoint m_server; /**< Endpoint for provDB comms*/
     thallium::remote_procedure *m_client_hello; /**< RPC to register client with provDB */
     thallium::remote_procedure *m_client_goodbye; /**< RPC to deregister client with provDB */
+    thallium::remote_procedure *m_stop_server; /** RPC to shutdown server*/
     bool m_perform_handshake; /**< Optionally disable the client->server registration */
 
     PerfStats *m_stats; /**< Performance data gathering*/
@@ -233,6 +234,12 @@ namespace chimbuko{
      */
     size_t getNoutstandingAsyncReqs(){ return anom_send_man.getNoutstanding(); }
 
+
+    /**
+     * @brief Issue a stop request to the server
+     */
+    void stopServer() const;
+
   };
 
 };