Added setRecvTimeout method to ADNetClient interface to fix compile bug

ADThreadNetClient now has an option to use a local network client, used in the sim
Fixed ADThreadNetClient not calling disconnect_ps in destructor, preventing thread join unless this function is called manually

Author: giltirn

include/chimbuko/ad/ADNetClient.hpp

@@ -78,6 +78,11 @@ namespace chimbuko{
      */
     void linkPerf(PerfStats* perf){ m_perf = perf; }
 
+    /**
+     * @brief Set the timeout for receiving messages (implementation dependent)
+     */
+    virtual void setRecvTimeout(const int timeout_ms){ }
+
   protected:
     bool m_use_ps;                           /**< true if the parameter server is in use */
     int m_rank;                              /**< MPI rank of current process */
@@ -125,7 +130,7 @@ namespace chimbuko{
     /**
      * @brief Set the timeout on blocking receives. Must be called prior to connecting
      */
-    void setRecvTimeout(const int timeout_ms){ m_recv_timeout_ms = timeout_ms; }
+    void setRecvTimeout(const int timeout_ms) override{ m_recv_timeout_ms = timeout_ms; }
 
     /**
      * @brief Get the zeroMQ socket
@@ -328,20 +333,29 @@ namespace chimbuko{
       return work_item;
     }
 
-    void run(){
+    /**
+     * @brief Create the worker thread
+     * @param local Use a local (in process) communicator if true, otherwise use the default network communicator
+     */
+    void run(bool local = false){
       worker = std::thread([&](){
+	  ADNetClient *client = nullptr;
+	  if(local){
+	    client = new ADLocalNetClient;
+	  }else{
 #ifdef _USE_MPINET
-    	  ADMPINetClient client;
+	    client = new ADMPINetClient;
 #else
-	  ADZMQNetClient client;
+	    client = new ADZMQNetClient;
 #endif
+	  }
           bool shutdown = false;
 
           while(!shutdown){
             size_t nwork = getNwork();
             while(nwork > 0){
               ClientAction* work_item = getWorkItem();
-              work_item->perform(client);
+              work_item->perform(*client);
               shutdown = shutdown || work_item->shutdown_worker();
 
               if(work_item->do_delete()) delete work_item;
@@ -353,12 +367,17 @@ namespace chimbuko{
               std::this_thread::sleep_for(std::chrono::milliseconds(80));
             }  
           }
+	  delete client;
         });
     }
 
   public:
-    ADThreadNetClient(){
-      run();
+    /**
+     * @brief Constructor
+     * @param local Use a local (in process) communicator if true, otherwise use the default network communicator
+     */
+    ADThreadNetClient(bool local = false){
+      run(local);
     }
 
     //Use only if you know what you are doing!
@@ -398,6 +417,7 @@ namespace chimbuko{
     }    
 
     ~ADThreadNetClient(){
+      disconnect_ps();
       worker.join();
     }
 

sim/include/sim/ad.hpp

@@ -30,7 +30,7 @@ namespace chimbuko_sim{
     ADNormalEventProvenance m_normal_events;
     ADMetadataParser m_metadata;
     std::unique_ptr<ADProvenanceDBclient> m_pdb_client;
-    ADLocalNetClient m_net_client; /**< The local net client. Only activated if an AD algorithm is used */
+    ADThreadNetClient *m_net_client; /**< The local net client. Only activated if an AD algorithm is used */
     ADOutlier* m_outlier; /**< The local outlier algorithm instance, if used*/
   public:
     /** 
@@ -54,11 +54,14 @@ namespace chimbuko_sim{
     ADsim(int window_size, int pid, int rid, unsigned long program_start, unsigned long step_freq): ADsim(){
       init(window_size, pid, rid, program_start, step_freq);
     }
-    ADsim(): m_outlier(nullptr){}
+    ADsim(): m_outlier(nullptr), m_net_client(nullptr){}
 
     ADsim(ADsim &&r);
 
-    ~ADsim(){ if(m_outlier) delete m_outlier; }
+    ~ADsim(){ 
+      if(m_outlier) delete m_outlier; 
+      if(m_net_client) delete m_net_client;
+    }
 
     ADProvenanceDBclient &getProvDBclient(){ return *m_pdb_client; }
 

sim/main/example2.cpp

@@ -73,5 +73,6 @@ int main(int argc, char **argv){
   }
 
   MPI_Finalize();
+  std::cout << "Done" << std::endl;
   return 0;
 }

sim/src/ad.cpp

@@ -29,9 +29,11 @@ ADsim::ADsim(ADsim &&r):
   m_normal_events(std::move(r.m_normal_events)),
   m_metadata(std::move(r.m_metadata)),
   m_pdb_client(std::move(r.m_pdb_client)),
-  m_outlier(r.m_outlier)
+  m_outlier(r.m_outlier),
+  m_net_client(r.m_net_client)
 {
   r.m_outlier = nullptr;
+  r.m_net_client = nullptr;
 }
 
 
@@ -50,8 +52,9 @@ void ADsim::init(int window_size, int pid, int rid, unsigned long program_start,
   if(p.algorithm != "none"){    
     m_outlier = ADOutlier::set_algorithm(p.stat, p.algorithm, p.hbos_thres, p.glob_thres, p.sstd_sigma);
     getPserver(); //force construction of pserver
-    m_net_client.connect_ps(m_rid);
-    m_outlier->linkNetworkClient(&m_net_client);
+    m_net_client = new ADThreadNetClient(true); //use local comms
+    m_net_client->connect_ps(m_rid);
+    m_outlier->linkNetworkClient(m_net_client);
   }
 }