To ADNetClient interface added function prototype for async_send, defaults to blocking send

ADThreadNetClient is now a proper derived class of the ADNetClient interface

Author: giltirn

include/chimbuko/ad/ADNetClient.hpp

@@ -62,7 +62,7 @@ namespace chimbuko{
      * @param msg The message
      * @return The response message in serialized format. Use Message::set_msg( <serialized_msg>,  true )  to unpack
      */       
-    virtual std::string send_and_receive(const Message &msg) const = 0;
+    virtual std::string send_and_receive(const Message &msg) = 0;
 
     /**
      * @brief Send a message to the parameter server and receive the response both as Message objects
@@ -73,6 +73,18 @@ namespace chimbuko{
      */       
     void send_and_receive(Message &recv, const Message &send);
 
+    /**
+     * @brief Perform a non-blocking send operation
+     *
+     * Not all net clients support non-blocking sends, in which case it will default to a blocking send
+     */
+    virtual void async_send(const Message &send);
+
+    /**
+     * @brief Check if a net client supports non-blocking sends
+     */
+    virtual bool async_send_supported() const{ return false; }
+
     /**
      * @brief If linked timing and packet size information will be gathered
      */
@@ -124,7 +136,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) const override;
+    std::string send_and_receive(const Message &msg) override;
 
 
     /**
@@ -145,7 +157,7 @@ namespace chimbuko{
     /**
      * @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;
+    void stopServer();
 
   private:
     void* m_context;                         /**< ZeroMQ context */
@@ -184,7 +196,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) const override;
+    std::string send_and_receive(const Message &msg) override;
 
 
   private:
@@ -222,14 +234,14 @@ 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) const override;
+    std::string send_and_receive(const Message &msg) override;
   };
 
 
   /**
    * @brief ADNetClient inside a worker thread with blocking send/receive and non-blocking send
    */
-  class ADThreadNetClient{
+  class ADThreadNetClient: public ADNetClient{
   public:
     /**
      * @brief Virtual class representing actions performed by the worker thread
@@ -257,12 +269,7 @@ namespace chimbuko{
     std::thread worker;
     mutable std::mutex m;
     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
      */
@@ -302,35 +309,24 @@ namespace chimbuko{
      */
     void disconnect_ps();
 
+    using ADNetClient::send_and_receive;
+
     /**
      * @brief Perform a blocking send and receive operation
+     * @param msg The message
+     * @return The response message in serialized format. Use Message::set_msg( <serialized_msg>,  true )  to unpack
      */
-    void send_and_receive(Message &recv, const Message &send);
+    std::string send_and_receive(const Message &send) override;
 
     /**
      * @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; }
+    void async_send(const Message &send) override;
 
     /**
-     * @brief Get the MPI rank of the server (MPINet)
+     * @brief Check if a net client supports non-blocking sends
      */
-    int get_server_rank() const{ return m_srank; }
-    
-    /**
-     * @brief Get the MPI rank of the client
-     */
-    int get_client_rank() const{ return m_rank; }
+    bool async_send_supported() const override{ return true; }
 
     /**
      * @brief Set a timeout (in ms) on receiving a response message

src/ad/ADNetClient.cpp

@@ -13,6 +13,10 @@ void ADNetClient::send_and_receive(Message &recv,  const Message &send){
   recv.set_msg( send_and_receive(send), true );
 }
 
+void ADNetClient::async_send(const Message &send){
+  send_and_receive(send);
+}
+
 
 #ifdef _USE_ZMQNET
 
@@ -89,7 +93,7 @@ void ADZMQNetClient::disconnect_ps() {
 }
 
 
-std::string ADZMQNetClient::send_and_receive(const Message &msg) const{  
+std::string ADZMQNetClient::send_and_receive(const Message &msg){
   PerfTimer timer;
   std::string send_msg = msg.data(), recv_msg;
   //Send local parameters to PS
@@ -109,7 +113,7 @@ std::string ADZMQNetClient::send_and_receive(const Message &msg) const{
   return recv_msg;
 }
 
-void ADZMQNetClient::stopServer() const{
+void ADZMQNetClient::stopServer(){
   verboseStream << "Client is sending stop request to server" << std::endl;
   Message msg;
   msg.set_info(0, 0, MessageType::REQ_QUIT, MessageKind::CMD);
@@ -190,7 +194,7 @@ void ADMPINetClient::disconnect_ps() {
   m_use_ps = false;
 }
 
-std::string ADMPINetClient::send_and_receive(const Message &msg) const{  
+std::string ADMPINetClient::send_and_receive(const Message &msg){  
   PerfTimer timer;
   std::string send_msg = msg.data(), recv_msg;
   MessageType req_type = MessageType(msg.type());
@@ -245,7 +249,7 @@ void ADLocalNetClient::disconnect_ps(){
 }
 
 
-std::string ADLocalNetClient::send_and_receive(const Message &msg) const{  
+std::string ADLocalNetClient::send_and_receive(const Message &msg){
   if(!m_use_ps) fatal_error("User should call connect_ps prior to sending/receiving messages");
 
   PerfTimer timer;    
@@ -302,14 +306,14 @@ struct ClientActionWait: public ADThreadNetClient::ClientAction{
 struct ClientActionBlockingSendReceive: public ADThreadNetClient::ClientAction{
   std::mutex m;
   std::condition_variable cv;
-  Message *recv;
+  std::string *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){}
+  ClientActionBlockingSendReceive(std::string *recv, Message const *send): send(send), recv(recv), complete(false){}
 
   void perform(ADNetClient &client){
-    client.send_and_receive(*recv, *send);
+    *recv = client.send_and_receive(*send);
 
     {
       std::unique_lock<std::mutex> lk(m);
@@ -418,10 +422,12 @@ void ADThreadNetClient::connect_ps(int rank, int srank, std::string sname){
 void ADThreadNetClient::disconnect_ps(){
   enqueue_action(new ClientActionDisconnect());
 }
-void ADThreadNetClient::send_and_receive(Message &recv, const Message &send){
+std::string ADThreadNetClient::send_and_receive(const Message &send){
+  std::string recv;
   ClientActionBlockingSendReceive action(&recv, &send);
   enqueue_action(&action);
   action.wait_for();
+  return recv;
 }
 void ADThreadNetClient::async_send(const Message &send){
   enqueue_action(new ClientActionAsyncSend(send));