Fixed bug caused by the bool 'local' parameter of ADThreadNetClient::run not being captured by copy leaving a dangling reference

ADThreadNetClient cleanup:
	Moved function implementations to cpp file
	ClientAction is now a member class of ADThreadNetClient
	Added documentation for many functions

Author: giltirn

include/chimbuko/ad/ADNetClient.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <chimbuko_config.h>
 #ifdef _USE_MPINET
 #include "chimbuko/net/mpi_net.hpp"
 #else
@@ -193,7 +194,8 @@ namespace chimbuko{
 
 #endif
 
-  
+
+ 
   /**
    * @brief Implementation of ADNetClient for intraprocess communications
    */
@@ -224,208 +226,119 @@ namespace chimbuko{
     std::string send_and_receive(const Message &msg) const override;
   };
 
-  //Actions performed by the worker thread
-  struct ClientAction{
-    virtual void perform(ADNetClient &client) = 0;
-    virtual bool do_delete() const = 0; //whether to delete the work object after completion
-    virtual bool shutdown_worker() const{ return false; } //whether to shutdown the worker after completing the action
-    virtual ~ClientAction(){}
-  };
-  
-  struct ClientActionConnect: public ClientAction{
-    int rank;
-    int srank;
-    std::string sname;
-
-    ClientActionConnect(int rank, int srank, const std::string &sname): rank(rank), srank(srank), sname(sname){}
-
-    void perform(ADNetClient &client){
-      std::cout << "Connecting to client" << std::endl;
-      client.connect_ps(rank, srank, sname);
-    }
-    bool do_delete() const{ return true; }
-  };
-
-  struct ClientActionDisconnect: public ClientAction{
-    void perform(ADNetClient &client){
-      std::cout << "Disconnecting from client" << std::endl;
-      client.disconnect_ps();
-    }
-    bool do_delete() const{ return true; }
-    bool shutdown_worker() const{ return true; }
-  };
-
-  //Make the worker wait for some time, for testing
-  struct ClientActionWait: public ClientAction{
-    size_t wait_ms;
-    ClientActionWait(size_t wait_ms): wait_ms(wait_ms){}
-
-    void perform(ADNetClient &client){
-      std::this_thread::sleep_for(std::chrono::milliseconds(wait_ms));
-    }
-    bool do_delete() const{ return true; }
-  };
-
-  struct ClientActionBlockingSendReceive: public ClientAction{
-    std::mutex m;
-    std::condition_variable cv;
-    Message *recv;
-    Message const *send;  //it's blocking so we know that the object will live long enough
-    bool complete;
-
-    ClientActionBlockingSendReceive(Message *recv, Message const *send): send(send), recv(recv), complete(false){}
-
-    void perform(ADNetClient &client){
-      client.send_and_receive(*recv, *send);
-
-      {
-        std::unique_lock<std::mutex> lk(m);
-        complete = true;
-        cv.notify_one();
-      }
-    }
-    bool do_delete() const{ return false; }
-
-    void wait_for(){
-      std::unique_lock<std::mutex> l(m);
-      cv.wait(l, [&]{ return complete; });
-    }
-  };
-
-  //Return message is just dumped
-  struct ClientActionAsyncSend: public ClientAction{
-    Message send;  //copy of send message because we don't know how long it will be before it sends
-
-    ClientActionAsyncSend(const Message &send): send(send){}
-
-    void perform(ADNetClient &client){
-      Message recv;
-      client.send_and_receive(recv, send);
-    }
-    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
+  /**
+   * @brief ADNetClient inside a worker thread with blocking send/receive and non-blocking send
+   */
   class ADThreadNetClient{
+  public:
+    /**
+     * @brief Virtual class representing actions performed by the worker thread
+     */
+    struct ClientAction{
+      /** 
+       * @brief Perform the action utilizing the underlying net implementation
+       */
+      virtual void perform(ADNetClient &client) = 0;
+      
+      /**
+       * @brief Whether to delete the work object (instance of ClientAction) after completion
+       */
+      virtual bool do_delete() const = 0;
+      
+      /**
+       * @brief Whether to shutdown the worker thread after completing the action
+       */
+      virtual bool shutdown_worker() const{ return false; }
+
+      virtual ~ClientAction(){}
+    };
+
+  private:
     std::thread worker;
     mutable std::mutex m;
-    std::queue<ClientAction*> queue;
-
-    size_t getNwork() const{
-      std::lock_guard<std::mutex> l(m);
-      return queue.size();
-    }
-    ClientAction* getWorkItem(){
-      std::lock_guard<std::mutex> l(m);
-      ClientAction *work_item = queue.front();
-      queue.pop();
-      return work_item;
-    }
+    std::queue<ClientAction*> queue; /**< The queue of net operations*/
+
+    int m_rank;
+    int m_srank;
+    bool m_use_ps;
+    PerfStats * m_perf;
+    
+    /**
+     * @brief Get the number of outstanding net operations
+     */
+    size_t getNwork() const;
+    /**
+     * @brief Get the next net operation
+     */
+    ClientAction* getWorkItem();
 
     /**
      * @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
-	    client = new ADMPINetClient;
-#else
-	    client = new ADZMQNetClient;
-#endif
-	  }
-          bool shutdown = false;
-
-          while(!shutdown){
-            size_t nwork = getNwork();
-            while(nwork > 0){
-              ClientAction* work_item = getWorkItem();
-              work_item->perform(*client);
-              shutdown = shutdown || work_item->shutdown_worker();
-
-              if(work_item->do_delete()) delete work_item;
-              nwork = getNwork();
-            }
-            if(shutdown){
-              if(nwork > 0) fatal_error("Worker was shut down before emptying its queue!");
-            }else{
-              std::this_thread::sleep_for(std::chrono::milliseconds(80));
-            }  
-          }
-	  delete client;
-        });
-    }
+    void run(bool local = false);
 
   public:
     /**
      * @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!
-    void enqueue_action(ClientAction *action){
-      std::lock_guard<std::mutex> l(m);
-      queue.push(action);
-    }
-
-    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());
-    }
-    void send_and_receive(Message &recv, const Message &send){
-      ClientActionBlockingSendReceive action(&recv, &send);
-      enqueue_action(&action);
-      action.wait_for();
-    }
-    void async_send(const Message &send){
-      enqueue_action(new ClientActionAsyncSend(send));
-    }
+    ADThreadNetClient(bool local = false);
+
+    /**
+     * @brief Add an action to the queue
+     *
+     *Use only if you know what you are doing!
+     */
+    void enqueue_action(ClientAction *action);
+
+    /**
+     * @brief Connect to the parameter server
+     */
+    void connect_ps(int rank, int srank = 0, std::string sname="MPINET");
+
+    /**
+     * @brief Disconnect from the parameter server
+     */
+    void disconnect_ps();
+
+    /**
+     * @brief Perform a blocking send and receive operation
+     */
+    void send_and_receive(Message &recv, const Message &send);
 
+    /**
+     * @brief Perform a non-blocking send operation
+     */
+    void async_send(const Message &send);
+
+    /**
+     * @brief Is the parameter server in use / connected?
+     */
     bool use_ps() const { return m_use_ps; }
 
+    /**
+     * @brief Link a performance monitoring instance
+     */
     void linkPerf(PerfStats* perf){ m_perf = perf; }
 
+    /**
+     * @brief Get the MPI rank of the server (MPINet)
+     */
     int get_server_rank() const{ return m_srank; }
 
+    /**
+     * @brief Get the MPI rank of the client
+     */
     int get_client_rank() const{ return m_rank; }
 
-    void setRecvTimeout(const int timeout_ms) {
-      enqueue_action(new ClientActionSetRecvTimeout(timeout_ms));
-    }    
+    /**
+     * @brief Set a timeout (in ms) on receiving a response message
+     */
+    void setRecvTimeout(const int timeout_ms);
 
-    ~ADThreadNetClient(){
-      disconnect_ps();
-      worker.join();
-    }
-   
-   private:
-    int m_rank;
-    int m_srank;
-    bool m_use_ps;
-    PerfStats * m_perf;
+    ~ADThreadNetClient();
   };