setRecvTimeout can be set from config file

Author: sandeepmittal

app/driver.cpp

@@ -71,6 +71,7 @@ optionalArgsParser & getOptionalArgsParser(){
     addOptionalCommandLineArg(p, hbos_use_global_threshold, "Set true to use a global threshold in HBOS algorithm. Dafault is true.");
     addOptionalCommandLineArg(p, program_idx, "Set the index associated with the instrumented program. Use to label components of a workflow. (default 0)");
     addOptionalCommandLineArg(p, outlier_sigma, "Set the number of standard deviations that defines an anomalous event (default 6)");
+    addOptionalCommandLineArg(p, net_recv_timeout, "Timeout (in ms) for blocking receives on client from parameter server");
     addOptionalCommandLineArg(p, pserver_addr, "Set the address of the parameter server. If empty (default) the pserver will not be used.");
     addOptionalCommandLineArg(p, hpserver_nthr, "Set the number of threads used by the hierarchical PS. This parameter is used to compute a port offset for the particular endpoint that this AD rank connects to (default 1)");
     addOptionalCommandLineArg(p, interval_msec, "Force the AD to pause for this number of ms at the end of each IO step (default 0)");
@@ -129,6 +130,7 @@ ChimbukoParams getParamsFromCommandLine(int argc, char** argv, const int mpi_wor
   params.ad_algorithm = "hbos";
   params.hbos_threshold = 0.99;
   params.hbos_use_global_threshold = true;
+  params.net_recv_timeout = 30000;
   params.pserver_addr = "";  //don't use pserver by default
   params.hpserver_nthr = 1;
   params.outlier_sigma = 6.0;     // anomaly detection algorithm parameter

include/chimbuko/ad/ADLocalCounterStatistics.hpp

@@ -77,7 +77,8 @@ namespace chimbuko{
      *
      * The message string is the output of get_json_state() in string format
      */
-    std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client, int rank, std::string pserver_addr) const;
+    //std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client, int rank, std::string pserver_addr) const;
+    std::pair<size_t, size_t> updateGlobalStatistics(ADThreadNetClient &net_client, int rank, std::string pserver_addr) const;
 
     /**
      * @brief Attach a PerfStats object into which performance metrics are accumulated
@@ -117,7 +118,8 @@ namespace chimbuko{
      * @param step step (or frame) number
      * @return std::pair<size_t, size_t> [sent, recv] message size
      */
-    static std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client, const std::string &l_stats, int step, int rank, std::string pserver_addr);
+    //static std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client, const std::string &l_stats, int step, int rank, std::string pserver_addr);
+    static std::pair<size_t, size_t> updateGlobalStatistics(ADThreadNetClient &net_client, const std::string &l_stats, int step, int rank, std::string pserver_addr);
 
     unsigned long m_program_idx; /**< Program idx*/
     int m_step; /**< io step */

include/chimbuko/ad/ADLocalFuncStatistics.hpp

@@ -76,8 +76,8 @@ namespace chimbuko{
      *
      * The message communicated is the string dump of the output of get_json_state()
      */
-    std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client, int rank, std::string pserver_addr) const;
-
+    //std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client, int rank, std::string pserver_addr) const;
+    std::pair<size_t, size_t> updateGlobalStatistics(ADThreadNetClient &net_client, int rank, std::string pserver_addr) const;
 
     /**
      * @brief Get the current state as a JSON object
@@ -108,7 +108,8 @@ namespace chimbuko{
      * @param step step (or frame) number
      * @return std::pair<size_t, size_t> [sent, recv] message size
      */
-    static std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client, const std::string &l_stats, int step, int rank, std::string pserver_addr);
+    //static std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client, const std::string &l_stats, int step, int rank, std::string pserver_addr);
+    static std::pair<size_t, size_t> updateGlobalStatistics(ADThreadNetClient &net_client, const std::string &l_stats, int step, int rank, std::string pserver_addr);
 
     int m_step; /**< io step */
     unsigned long m_rank; /**< Rank*/

include/chimbuko/ad/ADNetClient.hpp

@@ -153,7 +153,7 @@ namespace chimbuko{
     ClientActionWait(size_t wait_ms): wait_ms(wait_ms){}
 
     void perform(ADNetClient &client){
-      std::cout << "Worker is waiting for "<< wait_ms << "ms" << std::endl;
+      //std::cout << "Worker is waiting for "<< wait_ms << "ms" << std::endl;
       std::this_thread::sleep_for(std::chrono::milliseconds(wait_ms));
     }
     bool do_delete() const{ return true; }
@@ -169,7 +169,7 @@ namespace chimbuko{
     ClientActionBlockingSendReceive(Message *recv, Message const *send): send(send), recv(recv), complete(false){}
 
     void perform(ADNetClient &client){
-      std::cout << "Performing blocking send and receive" << std::endl;
+      //std::cout << "Performing blocking send and receive" << std::endl;
       client.send_and_receive(*recv, *send);
       complete = true;
       cv.notify_one();
@@ -189,14 +189,25 @@ namespace chimbuko{
     ClientActionAsyncSend(const Message &send): send(send){}
 
     void perform(ADNetClient &client){
-      std::cout << "Performing non-blocking send and receive" << std::endl;
+      //std::cout << "Performing non-blocking send and receive" << std::endl;
       Message recv;
       client.send_and_receive(recv, send);
-      std::cout << "Non-blocking send returned " << recv.buf() << std::endl;
+      //std::cout << "Non-blocking send returned " << recv.buf() << std::endl;
     }
     bool do_delete() const{ return true; }
   };
 
+  struct ClientActionSetRecvTimeout: public ClientAction{
+    int timeout;
+    ClientActionSetRecvTimeout(const int timeout): timeout(timeout){}
+
+    void perform(ADNetClient &client){
+      client.setRecvTimeout(timeout);
+    }
+
+    bool do_delete() const{return true;}
+  };
+
   //ADNetClient inside a worker thread with blocking send/receive and non-blocking send
   class ADThreadNetClient{
     std::thread worker;
@@ -213,9 +224,9 @@ namespace chimbuko{
       queue.pop();
       return work_item;
     }
-
+    
     void run(){
-      std::cout << "Starting worker thread" << std::endl;
+      //std::cout << "Starting worker thread" << std::endl;
       worker = std::thread([&](){
           ADNetClient client;
           bool shutdown = false;
@@ -232,9 +243,9 @@ namespace chimbuko{
             }
             if(shutdown){
               if(nwork > 0) fatal_error("Worker was shut down before emptying its queue!");
-              std::cout << "Worker received shutdown request" << std::endl;
+              //std::cout << "Worker received shutdown request" << std::endl;
             }else{
-              std::this_thread::sleep_for(std::chrono::milliseconds(20));
+              std::this_thread::sleep_for(std::chrono::milliseconds(80));
             }  
           }
         });
@@ -252,7 +263,10 @@ namespace chimbuko{
     }
 
     void connect_ps(int rank, int srank = 0, std::string sname="MPINET"){
+      m_rank = rank;
+      m_srank = srank;
       enqueue_action(new ClientActionConnect(rank, srank,sname));
+      m_use_ps = true;
     }
     void disconnect_ps(){
         enqueue_action(new ClientActionDisconnect());
@@ -266,11 +280,28 @@ namespace chimbuko{
       enqueue_action(new ClientActionAsyncSend(send));
     }
 
+    bool use_ps() const { return m_use_ps; }
+
+    void linkPerf(PerfStats* perf){ m_perf = perf; }
+
+    int get_server_rank() const{ return m_srank; }
+    
+    int get_client_rank() const{ return m_rank; }
+
+    void setRecvTimeout(const int timeout_ms) {
+      enqueue_action(new ClientActionSetRecvTimeout(timeout_ms));
+    }    
+
     ~ADThreadNetClient(){
-      std::cout << "Joining worker thread" << std::endl;
+      //std::cout << "Joining worker thread" << std::endl;
       worker.join();
     }
-
+   
+   private:
+    int m_rank;
+    int m_srank;
+    bool m_use_ps;
+    PerfStats * m_perf;
   };
 
 

include/chimbuko/ad/ADOutlier.hpp

@@ -57,7 +57,8 @@ namespace chimbuko {
     /**
      * @brief Link the interface for communicating with the parameter server
      */
-    void linkNetworkClient(ADNetClient *client);
+    //void linkNetworkClient(ADNetClient *client);
+    void linkNetworkClient(ADThreadNetClient *client);
 
     /**
      * @brief abstract method to run the implemented anomaly detection algorithm
@@ -112,8 +113,9 @@ namespace chimbuko {
   protected:
     int m_rank;                              /**< this process rank                      */
     bool m_use_ps;                           /**< true if the parameter server is in use */
-    ADNetClient* m_net_client;                 /**< interface for communicating to parameter server */
-
+    //ADNetClient* m_net_client;                 /**< interface for communicating to parameter server */
+    ADThreadNetClient* m_net_client;                 /**< interface for communicating to parameter server */
+    
     std::unordered_map< std::array<unsigned long, 4>, size_t, ArrayHasher<unsigned long,4> > m_local_func_exec_count; /**< Map(program id, rank id, thread id, func id) -> number of times encountered on this node*/
 
     const ExecDataMap_t * m_execDataMap;     /**< execution data map */

include/chimbuko/ad/ADParser.hpp

@@ -43,8 +43,8 @@ namespace chimbuko {
      *
      * If this is performed, the parser will replace the local with global index in the incoming data stream
      */
-    void linkNetClient(ADNetClient *net_client){ m_global_func_idx_map.linkNetClient(net_client); }
-
+    //void linkNetClient(ADNetClient *net_client){ m_global_func_idx_map.linkNetClient(net_client); }
+    void linkNetClient(ADThreadNetClient *net_client){ m_global_func_idx_map.linkNetClient(net_client); }
     /**
      * @brief If linked, performance information will be gathered
      */

include/chimbuko/ad/ADglobalFunctionIndexMap.hpp

@@ -14,7 +14,8 @@ namespace chimbuko{
    * If the parameter server is not connected it will simply return the local index
    */
   class ADglobalFunctionIndexMap{
-    ADNetClient *m_net_client;
+    //ADNetClient *m_net_client;
+    ADThreadNetClient *m_net_client;
     std::unordered_map<unsigned long, unsigned long> m_idxmap; /**< Map of local function index to global function index*/
     unsigned long m_pid; /**< Program index*/
   public:
@@ -26,7 +27,8 @@ namespace chimbuko{
      *
      *If a pointer to the net client is not provided the local index will not be synchronized betwee nodes
      */
-    ADglobalFunctionIndexMap(unsigned long pid, ADNetClient *net_client = nullptr): m_net_client(net_client), m_pid(pid){}
+    //ADglobalFunctionIndexMap(unsigned long pid, ADNetClient *net_client = nullptr): m_net_client(net_client), m_pid(pid){}
+    ADglobalFunctionIndexMap(unsigned long pid, ADThreadNetClient *net_client = nullptr): m_net_client(net_client), m_pid(pid){}
 
     /**
      * @brief Check if the pserver is connected
@@ -36,7 +38,8 @@ namespace chimbuko{
     /**
      * @brief Link the net client
      */
-    void linkNetClient(ADNetClient *net_client){ m_net_client = net_client; }
+    //void linkNetClient(ADNetClient *net_client){ m_net_client = net_client; }
+    void linkNetClient(ADThreadNetClient *net_client){ m_net_client = net_client; }
 
     /**
      * @brief Lookup the global index corresponding to the input local index
@@ -63,7 +66,8 @@ namespace chimbuko{
     /**
      * @brief Return a pointer to the net client
      */
-    ADNetClient* getNetClient(){ return m_net_client; }
+    //ADNetClient* getNetClient(){ return m_net_client; }
+    ADThreadNetClient* getNetClient(){ return m_net_client; }
   };
 
 }

include/chimbuko/chimbuko.hpp

@@ -57,6 +57,8 @@ namespace chimbuko {
     std::string outlier_statistic; /**< Set the statistic used in outlier detection*/
 
     int step_report_freq; /**<Steps between Chimbuko reporting IO step progress. Use 0 to deactivate this logging entirely (default 1)*/
+    
+    int net_recv_timeout; /**< Timeout (in ms) used for blocking receives functionality on client (driver) of parameter server */
 
     ChimbukoParams();
 
@@ -207,7 +209,8 @@ namespace chimbuko {
     ADCounter * m_counter;     /**< counter event manager */
     ADOutlier * m_outlier;     /**< outlier detection algorithm */
     ADio * m_io;               /**< output writer */
-    ADNetClient * m_net_client; /**< client for comms with parameter server */
+    //ADNetClient * m_net_client; /**< client for comms with parameter server */
+    ADThreadNetClient * m_net_client; /**< client for comms with parameter server */
     ADMetadataParser *m_metadata_parser; /**< parser for metadata */
 #ifdef ENABLE_PROVDB
     ADProvenanceDBclient *m_provdb_client; /**< provenance DB client*/

src/ad/ADLocalCounterStatistics.cpp

@@ -70,7 +70,8 @@ ADLocalCounterStatistics::State ADLocalCounterStatistics::get_state() const{
 }
 
 
-std::pair<size_t, size_t> ADLocalCounterStatistics::updateGlobalStatistics(ADNetClient &net_client, int rank, std::string pserver_addr) const{
+//std::pair<size_t, size_t> ADLocalCounterStatistics::updateGlobalStatistics(ADNetClient &net_client, int rank, std::string pserver_addr) const{
+std::pair<size_t, size_t> ADLocalCounterStatistics::updateGlobalStatistics(ADThreadNetClient &net_client, int rank, std::string pserver_addr) const{
   //nlohmann::json state = get_json_state();
   State state = get_state();
   PerfTimer timer;
@@ -86,23 +87,25 @@ std::pair<size_t, size_t> ADLocalCounterStatistics::updateGlobalStatistics(ADNet
   return msgsz;
 }
 
-std::pair<size_t, size_t> ADLocalCounterStatistics::updateGlobalStatistics(ADNetClient &net_client, const std::string &l_stats, int step, int rank, std::string pserver_addr){
+//std::pair<size_t, size_t> ADLocalCounterStatistics::updateGlobalStatistics(ADNetClient &net_client, const std::string &l_stats, int step, int rank, std::string pserver_addr){
+std::pair<size_t, size_t> ADLocalCounterStatistics::updateGlobalStatistics(ADThreadNetClient &net_client, const std::string &l_stats, int step, int rank, std::string pserver_addr){
   if (!net_client.use_ps())
     return std::make_pair(0, 0);
 
-  ADThreadNetClient thrnet;
-  thrnet.connect_ps(rank, 0, pserver_addr);
+  //ADThreadNetClient thrnet;
+  //thrnet.connect_ps(rank, 0, pserver_addr);
 
   Message msg;
   msg.set_info(net_client.get_client_rank(), net_client.get_server_rank(), MessageType::REQ_ADD, MessageKind::COUNTER_STATS, step);
   msg.set_msg(l_stats);
 
   size_t sent_sz = msg.size();
   //std::string strmsg = net_client.send_and_receive(msg);
-  thrnet.async_send(msg);
-  size_t recv_sz = msg.size(); //strmsg.size();
+  net_client.async_send(msg);
+  //thrnet.async_send(msg);
+  size_t recv_sz = 0; //msg.size(); //strmsg.size();
 
-  thrnet.disconnect_ps();  
+  //thrnet.disconnect_ps();  
   return std::make_pair(sent_sz, recv_sz);
 }
 

src/ad/ADLocalFuncStatistics.cpp

@@ -95,7 +95,8 @@ ADLocalFuncStatistics::State ADLocalFuncStatistics::get_state() const{
 }
 
 
-std::pair<size_t, size_t> ADLocalFuncStatistics::updateGlobalStatistics(ADNetClient &net_client, int rank, std::string pserver_addr) const{
+//std::pair<size_t, size_t> ADLocalFuncStatistics::updateGlobalStatistics(ADNetClient &net_client, int rank, std::string pserver_addr) const{
+std::pair<size_t, size_t> ADLocalFuncStatistics::updateGlobalStatistics(ADThreadNetClient &net_client, int rank, std::string pserver_addr) const{
   // func id --> (name, # anomaly, inclusive run stats, exclusive run stats)
   //nlohmann::json g_info = get_json_state();
   State g_info = get_state();
@@ -113,22 +114,24 @@ std::pair<size_t, size_t> ADLocalFuncStatistics::updateGlobalStatistics(ADNetCli
   return msgsz;
 }
 
-std::pair<size_t, size_t> ADLocalFuncStatistics::updateGlobalStatistics(ADNetClient &net_client, const std::string &l_stats, int step, int rank, std::string pserver_addr){
+//std::pair<size_t, size_t> ADLocalFuncStatistics::updateGlobalStatistics(ADNetClient &net_client, const std::string &l_stats, int step, int rank, std::string pserver_addr){
+std::pair<size_t, size_t> ADLocalFuncStatistics::updateGlobalStatistics(ADThreadNetClient &net_client, const std::string &l_stats, int step, int rank, std::string pserver_addr){
   if (!net_client.use_ps())
     return std::make_pair(0, 0);
 
-  ADThreadNetClient thrnet;
-  thrnet.connect_ps(rank, 0, pserver_addr);
+  //ADThreadNetClient thrnet;
+  //thrnet.connect_ps(rank, 0, pserver_addr);
 
   Message msg;
   msg.set_info(net_client.get_client_rank(), net_client.get_server_rank(), MessageType::REQ_ADD, MessageKind::ANOMALY_STATS, step);
   msg.set_msg(l_stats);
 
   size_t sent_sz = msg.size();
   //std::string strmsg = net_client.send_and_receive(msg);
-  thrnet.async_send(msg);
-  size_t recv_sz = msg.size(); //strmsg.size();
+  net_client.async_send(msg);
+  //thrnet.async_send(msg);
+  size_t recv_sz =0;// msg.size(); //strmsg.size();
 
-  thrnet.disconnect_ps();
+  //thrnet.disconnect_ps();
   return std::make_pair(sent_sz, recv_sz);
 }