The functionality of looking up and performing a Net payload action is now provided by static functions through the NetInterface, simplifying net implementations

Author: giltirn

include/chimbuko/net.hpp

@@ -131,6 +131,26 @@ namespace chimbuko {
      */
     void list_payloads(std::ostream &os) const;
 
+    typedef std::unordered_map<MessageKind, std::unordered_map<MessageType,  std::unique_ptr<NetPayloadBase> > > PayloadMapType; /**< Map of message kind/type to payloads */
+    typedef std::unordered_map<int, PayloadMapType> WorkerPayloadMapType; /**< Map of worker index and message type to payloads */
+
+    /**
+     * @brief Find the action associated with the given worker_id and message type and perform the action
+     * @param worker_id The worker index
+     * @param msg_reply The reply message
+     * @param msg The input message
+     * @param payloads The map of worker/message type to payload
+     */
+    static void find_and_perform_action(int worker_id, Message &msg_reply, const Message &msg, const WorkerPayloadMapType &payloads);
+
+    /**
+     * @brief Find the action associated with the given message type and perform the action
+     * @param msg_reply The reply message
+     * @param msg The input message
+     * @param payloads The map of worker/message type to payload
+     */
+    static void find_and_perform_action(Message &msg_reply, const Message &msg, const PayloadMapType &payloads);
+
   protected:
     /**
      * @brief initialize thread pool 
@@ -139,14 +159,9 @@ namespace chimbuko {
      */
     virtual void init_thread_pool(int nt) = 0;
 
-  protected:
     int              m_nt;    /**< The number of threads in the pool */
-    
-    std::unordered_map<int,
-		       std::unordered_map<MessageKind,
-					  std::unordered_map<MessageType,  std::unique_ptr<NetPayloadBase> >
-					  >
-		       > m_payloads; /**< Map of worker index (implementation defined), message kind and message type to a payload*/
+
+    WorkerPayloadMapType m_payloads; /**< Map of worker index (implementation defined), message kind and message type to a payload*/
   };
 
   namespace DefaultNetInterface

src/net.cpp

@@ -1,6 +1,7 @@
 #include "chimbuko/net.hpp"
 #include "chimbuko/net/mpi_net.hpp"
 #include "chimbuko/net/zmq_net.hpp"
+#include "chimbuko/util/error.hpp"
 
 #include <iostream>
 #include <chrono>
@@ -43,3 +44,21 @@ void NetInterface::list_payloads(std::ostream &os) const{
     }
   }
 }
+
+void NetInterface::find_and_perform_action(int worker_id, Message &msg_reply, const Message &msg, const NetInterface::WorkerPayloadMapType &payloads){
+  auto iit = payloads.find(worker_id);
+  if(iit == payloads.end()){ fatal_error(std::string("Cannot find payloads with worker index ") + std::to_string(worker_id) ); }
+  find_and_perform_action(msg_reply, msg, iit->second);
+}
+
+void NetInterface::find_and_perform_action(Message &msg_reply, const Message &msg, const NetInterface::PayloadMapType &payloads){  
+  auto kit = payloads.find((MessageKind)msg.kind());
+  if(kit == payloads.end()) fatal_error("No payload associated with the message kind provided. Message: " + msg.buf() + " (did you add the payload to the server?)");
+  auto pit = kit->second.find((MessageType)msg.type());
+  if(pit == kit->second.end()) fatal_error("No payload associated with the message type provided. Mesage: " + msg.buf() + " (did you add the payload to the server?)");
+
+  msg_reply = msg.createReply();
+  pit->second->action(msg_reply, msg);
+}
+
+

src/net/zmq_net.cpp

@@ -139,37 +139,28 @@ void doWork(void* context,
     }
     perf.add(perf_prefix + "poll_time_ms", timer.elapsed_ms());
 
-    //Receive the message into a json-formatted string
-    timer.start();
     std::string strmsg;
+
+    //Receive the message
+    timer.start();
     ZMQNet::recv(socket, strmsg);
     perf.add(perf_prefix + "receive_from_front_ms", timer.elapsed_ms());
-
     verboseStream << "ZMQNet worker thread " << thr_idx << " received message of size " << strmsg.size() << std::endl;
 
     //Parse the message and instantiate a reply message with appropriate sender
     timer.start();
     Message msg, msg_reply;
     msg.set_msg(strmsg, true);
-
-    msg_reply = msg.createReply();
-    perf.add(perf_prefix + "message_parse_reply_create_ms", timer.elapsed_ms());
+    perf.add(perf_prefix + "message_parse_ms", timer.elapsed_ms());
 
     timer.start();
-    auto kit = payloads.find((MessageKind)msg.kind());
-    if(kit == payloads.end()) fatal_error("ZMQNet::doWork : No payload associated with the message kind provided. Message: " + strmsg + " (did you add the payload to the server?)");
-    auto pit = kit->second.find((MessageType)msg.type());
-    if(pit == kit->second.end()) fatal_error("ZMQNet::doWork : No payload associated with the message type provided. Mesage: " + strmsg + " (did you add the payload to the server?)");
-    perf.add(perf_prefix + "message_kind_type_lookup_ms", timer.elapsed_ms());
+    NetInterface::find_and_perform_action(msg_reply, msg, payloads);
+    perf.add(perf_prefix + "find_and_perform_action_ms", timer.elapsed_ms());
 
-    //Apply the payload
-    timer.start();
-    pit->second->action(msg_reply, msg);
+    //Send the reply
     strmsg = msg_reply.data();
     verboseStream << "ZMQNet worker thread " << thr_idx << " sending response of size " << strmsg.size() << std::endl;
-    perf.add(perf_prefix + "perform_action_ms", timer.elapsed_ms());
 
-    //Send the reply
     timer.start();
     ZMQNet::send(socket, strmsg);
     perf.add(perf_prefix + "send_to_front_ms", timer.elapsed_ms());

src/net/zmqme_net.cpp

@@ -134,24 +134,16 @@ void doWork(std::unordered_map<int,
       Message msg, msg_reply;
       msg.set_msg(strmsg, true);
 
-      msg_reply = msg.createReply();
-      
-      timer.start();
-      auto kit = payloads_t.find((MessageKind)msg.kind());
-      if(kit == payloads_t.end()) throw std::runtime_error("ZMQMENet::doWork : No payload associated with the message kind provided (did you add the payload to the server?)");
-      auto pit = kit->second.find((MessageType)msg.type());
-      if(pit == kit->second.end()) throw std::runtime_error("ZMQMENet::doWork : No payload associated with the message type provided (did you add the payload to the server?)");
-      perf.add(perf_prefix + "message_kind_type_lookup_ms", timer.elapsed_ms());
-
-      //Apply the payload
       timer.start();
-      pit->second->action(msg_reply, msg);
-      verboseStream << "ZMQMENet Worker thread " << thr_idx << " sending response of size " << msg_reply.data().size() << std::endl;
-      perf.add(perf_prefix + "perform_action_ms", timer.elapsed_ms());
+      NetInterface::find_and_perform_action(msg_reply, msg, payloads_t);
+      perf.add(perf_prefix + "find_and_perform_action_ms", timer.elapsed_ms());
+      
+      strmsg = msg_reply.data();
+      verboseStream << "ZMQMENet Worker thread " << thr_idx << " sending response of size " << strmsg.size() << std::endl;
 
       //Send the reply
       timer.start();
-      ZMQNet::send(socket, msg_reply.data());
+      ZMQNet::send(socket, strmsg);
       perf.add(perf_prefix + "send_to_front_ms", timer.elapsed_ms());
     } //while(<work in queue>)