Added a new LocalNet net implementation for intraprocess communication + unit test

Added an ADLocalNetClient implementation associated with the above + unit test

Author: giltirn

include/chimbuko/ad/ADNetClient.hpp

@@ -5,6 +5,7 @@
 #else
 #include "chimbuko/net/zmq_net.hpp"
 #endif
+#include "chimbuko/net/local_net.hpp"
 #include "chimbuko/message.hpp"
 #include "chimbuko/util/PerfStats.hpp"
 
@@ -168,6 +169,8 @@ namespace chimbuko{
      */
     void disconnect_ps() override;
 
+    using ADNetClient::send_and_receive;
+
     /**
      * @brief Send a message to the parameter server and receive the response in a serialized format
      * @param msg The message
@@ -183,6 +186,35 @@ namespace chimbuko{
 #endif
 
 
+  /**
+   * @brief Implementation of ADNetClient for intraprocess communications
+   */
+  class ADLocalNetClient: public ADNetClient{
+  public:
+    /**
+     * @brief connect to the parameter server
+     * 
+     * @param rank Ignored
+     * @param srank Ignored
+     * @param sname Ignored
+     */
+    void connect_ps(int rank, int srank = 0, std::string sname="MPINET") override;
+    /**
+     * @brief disconnect from the connected parameter server
+     * 
+     * Called automatically by destructor if not previously called
+     */
+    void disconnect_ps() override;
+
+    using ADNetClient::send_and_receive;
+    
+    /**
+     * @brief Send a message to the parameter server and receive the response in a serialized format
+     * @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;
+  };
 
 
 

include/chimbuko/net/local_net.hpp

@@ -0,0 +1,71 @@
+#pragma once
+
+#include <chimbuko/net.hpp>
+#include <chimbuko/util/PerfStats.hpp>
+
+namespace chimbuko {
+  
+  /**
+     @brief A simple interface for a network in which the sender and receiver are on the same process
+  */
+  class LocalNet : public NetInterface {
+  public:
+
+    //Pointer the the current instance of LocalNet. Only one instance can exist at any given time
+    static LocalNet * & globalInstance();
+
+    LocalNet();
+    ~LocalNet();    
+
+    /**
+     * @brief (virtual) initialize network interface
+     * 
+     * @param argc command line argc 
+     * @param argv command line argv
+     * @param nt the number of threads for a thread pool
+     */
+    void init(int* argc = nullptr, char*** argv = nullptr, int nt=1) override{}
+
+    /**
+     * @brief (virtual) finalize network
+     * 
+     */
+    void finalize() override{}
+
+#ifdef _PERF_METRIC
+    /**
+     * @brief (virtual) Run network server with performance logging to provided directory
+     */
+    void run(std::string logdir="./") override{}
+#else
+    /**
+     * @brief (virtual) Run network server
+     */    
+    void run() override{}
+#endif
+    /**
+     * @brief (virtual) stop network server
+     * 
+     */
+    void stop() override{}
+
+    /**
+     * @brief (virtual) name of network server
+     * 
+     * @return std::string name of network server
+     */
+    std::string name() const override{ return "LocalNet"; }
+
+
+    /**
+     * @brief Send a message and receive the response
+     */
+    static std::string send_and_receive(const std::string &send_str);
+    
+  protected:
+    void init_thread_pool(int nt){}
+
+  };
+
+
+}

src/Makefile.am

@@ -1,5 +1,5 @@
 AM_CPPFLAGS = -I$(top_srcdir)/include -I$(top_srcdir)/3rdparty @PS_FLAGS@
 
 lib_LTLIBRARIES = libchimbuko.la
-libchimbuko_la_SOURCES = chimbuko.cpp param.cpp pserver/global_anomaly_stats.cpp pserver/global_counter_stats.cpp util/ADIOS2parseUtils.cpp util/RunStats.cpp util/DispatchQueue.cpp util/Anomalies.cpp util/map.cpp util/string.cpp util/error.cpp util/memutils.cpp util/time.cpp util/PerfStats.cpp ad/ADOutlier.cpp ad/ADParser.cpp ad/ADio.cpp ad/utils.cpp ad/ADEvent.cpp ad/AnomalyData.cpp ad/ExecData.cpp ad/ADCounter.cpp ad/ADAnomalyProvenance.cpp ad/ADNetClient.cpp ad/ADLocalFuncStatistics.cpp ad/ADLocalCounterStatistics.cpp ad/ADProvenanceDBengine.cpp ad/ADProvenanceDBclient.cpp ad/ADMetadataParser.cpp ad/ADglobalFunctionIndexMap.cpp ad/ADNormalEventProvenance.cpp pserver/PSstatSender.cpp pserver/PSglobalFunctionIndexMap.cpp pserver/PSProvenanceDBclient.cpp pserver/AnomalyStat.cpp message.cpp net/zmq_net.cpp net/mpi_net.cpp net/zmqme_net.cpp net.cpp param/hbos_param.cpp param/sstd_param.cpp
+libchimbuko_la_SOURCES = chimbuko.cpp param.cpp pserver/global_anomaly_stats.cpp pserver/global_counter_stats.cpp util/ADIOS2parseUtils.cpp util/RunStats.cpp util/DispatchQueue.cpp util/Anomalies.cpp util/map.cpp util/string.cpp util/error.cpp util/memutils.cpp util/time.cpp util/PerfStats.cpp ad/ADOutlier.cpp ad/ADParser.cpp ad/ADio.cpp ad/utils.cpp ad/ADEvent.cpp ad/AnomalyData.cpp ad/ExecData.cpp ad/ADCounter.cpp ad/ADAnomalyProvenance.cpp ad/ADNetClient.cpp ad/ADLocalFuncStatistics.cpp ad/ADLocalCounterStatistics.cpp ad/ADProvenanceDBengine.cpp ad/ADProvenanceDBclient.cpp ad/ADMetadataParser.cpp ad/ADglobalFunctionIndexMap.cpp ad/ADNormalEventProvenance.cpp pserver/PSstatSender.cpp pserver/PSglobalFunctionIndexMap.cpp pserver/PSProvenanceDBclient.cpp pserver/AnomalyStat.cpp message.cpp net/zmq_net.cpp net/mpi_net.cpp net/zmqme_net.cpp net/local_net.cpp net.cpp param/hbos_param.cpp param/sstd_param.cpp
 libchimbuko_la_LDFLAGS = -version-info 3:0:0

src/ad/ADNetClient.cpp

@@ -231,3 +231,34 @@ std::string ADMPINetClient::send_and_receive(const Message &msg) const{
 }
 
 #endif
+
+
+
+void ADLocalNetClient::connect_ps(int rank, int srank, std::string sname){
+  if(LocalNet::globalInstance() == nullptr) fatal_error("No LocalNet instance exists");
+  m_use_ps = true;
+}
+
+
+void ADLocalNetClient::disconnect_ps(){
+  m_use_ps = false;
+}
+
+
+std::string ADLocalNetClient::send_and_receive(const Message &msg) const{  
+  if(!m_use_ps) fatal_error("User should call connect_ps prior to sending/receiving messages");
+
+  PerfTimer timer;    
+  std::string send_msg = msg.data();
+  std::string recv_msg = LocalNet::send_and_receive(send_msg);
+
+#ifdef _PERF_METRIC
+  if(m_perf){
+    m_perf->add("net_client_send_recv_time_ms", timer.elapsed_ms());
+    m_perf->add("net_client_send_bytes", send_msg.size()); //1 char = 1 byte
+    m_perf->add("net_client_recv_bytes", recv_msg.size());
+  }
+#endif
+
+  return recv_msg;
+}

src/net/local_net.cpp

@@ -0,0 +1,30 @@
+#include "chimbuko/message.hpp"
+#include "chimbuko/verbose.hpp"
+#include "chimbuko/util/string.hpp"
+#include "chimbuko/util/error.hpp"
+#include "chimbuko/net/local_net.hpp"
+#include <iostream>
+#include <string.h>
+
+using namespace chimbuko;
+
+LocalNet * & LocalNet::globalInstance(){ static LocalNet* net = nullptr; return net; }
+
+LocalNet::LocalNet(){
+  if(globalInstance() != nullptr) fatal_error("Cannot have 2 instances of LocalNet active");
+  globalInstance() = this;
+}
+
+LocalNet::~LocalNet(){
+  globalInstance() = nullptr;
+}
+
+std::string LocalNet::send_and_receive(const std::string &send_str){
+  if(globalInstance() == nullptr){ fatal_error("An instance of LocalNet must exist"); }
+
+  Message msg, msg_reply;
+  msg.set_msg(send_str, true);
+  
+  NetInterface::find_and_perform_action(0, msg_reply, msg, globalInstance()->m_payloads);
+  return msg_reply.data();
+}

test/unit_tests/ad/ADNetClient.cpp

@@ -359,3 +359,40 @@ TEST(ADNetClient, TestRemoteStop){
 
 
 
+
+TEST(ADLocalNetClientTest, SendRecv){  
+  ADLocalNetClient client;
+  bool err = false;
+  try{
+    client.connect_ps(1,2,"arg");
+  }catch(const std::exception &e){
+    err = true;
+  }
+  EXPECT_EQ(err,true);
+  EXPECT_EQ(client.use_ps(),false);
+
+  err = false;
+  LocalNet net;
+  net.add_payload(new NetPayloadBounceback);
+  
+  try{
+    client.connect_ps(1,2,"arg");
+  }catch(const std::exception &e){
+    err = true;
+  }
+  
+  EXPECT_EQ(err, false);
+  EXPECT_EQ(client.use_ps(), true);
+  
+  Message msg, msg_reply;
+  msg.set_info(0,0,REQ_ECHO,CMD);
+  msg.set_msg("Hello!!");
+
+  client.send_and_receive(msg_reply, msg);
+
+  EXPECT_EQ(msg_reply.buf(), "Hello!!");
+
+  client.disconnect_ps();
+  EXPECT_EQ(client.use_ps(), false);
+}
+

test/unit_tests/net/LocalNet.cpp

@@ -0,0 +1,72 @@
+#include "gtest/gtest.h"
+#include "../unit_test_common.hpp"
+#include <chimbuko/net/local_net.hpp>
+
+using namespace chimbuko;
+
+TEST(TestLocalNet, TestSingleton){
+  LocalNet *net1 = new LocalNet;
+  bool err = false;
+
+  try{
+    LocalNet net2;
+  }catch(const std::exception &e){
+    err = true;
+  }
+  
+  EXPECT_EQ(err, true);
+
+  delete net1;
+
+  err = false;
+  try{
+    LocalNet net2;
+  }catch(const std::exception &e){
+    err = true;
+  }
+
+  EXPECT_EQ(err, false);
+}
+
+class NetPayloadBounceback: public NetPayloadBase{
+public:
+  MessageKind kind() const{ return MessageKind::CMD; }
+  MessageType type() const{ return MessageType::REQ_ECHO; }
+  void action(Message &response, const Message &message) override{
+    check(message);
+    std::cout << "Bounce received: '" << message.buf() << "'" << std::endl;
+    response.set_msg(message.buf(), false);
+  };
+};
+
+
+TEST(TestLocalNet, SendAndReceive){
+  //Test it throws if no active instance
+  bool err = false;
+  try{
+    LocalNet::send_and_receive("hello");
+  }catch(const std::exception &e){
+    err = true;
+  }
+  EXPECT_EQ(err, true);
+
+  err = false;
+
+  LocalNet net;
+  net.add_payload(new NetPayloadBounceback);
+
+  std::string the_message = "Hello!";
+  Message msg;
+  msg.set_info(0,0, MessageType::REQ_ECHO, MessageKind::CMD);
+  msg.set_msg(the_message);
+
+  std::string str_reply = LocalNet::send_and_receive(msg.data());
+  Message reply;
+  reply.set_msg(str_reply, true);
+
+  EXPECT_EQ(reply.buf(), the_message);
+}
+
+
+
+

test/unit_tests/net/Makefile.am

@@ -2,10 +2,14 @@ AM_CPPFLAGS = -I$(top_srcdir)/include -I$(top_srcdir)/3rdparty @PS_FLAGS@
 LDADD = $(top_builddir)/src/libchimbuko.la -lgtest -lstdc++fs 
 
 testdir = $(prefix)/test/unit_tests/net
-test_PROGRAMS = ZMQMENet ZMQNet
+test_PROGRAMS = ZMQMENet ZMQNet LocalNet
 
 ZMQMENet_SOURCES = ZMQMENet.cpp ../unit_test_main_mpi.cpp
 ZMQMENet_LDADD = $(LDADD)
 
 ZMQNet_SOURCES = ZMQNet.cpp ../unit_test_main_mpi.cpp
 ZMQNet_LDADD = $(LDADD)
+
+LocalNet_SOURCES = LocalNet.cpp ../unit_test_main_mpi.cpp
+LocalNet_LDADD = $(LDADD)
+

test/unit_tests/run_all.sh

@@ -33,3 +33,4 @@ set -o pipefail
 ./pserver/global_anomaly_stats
 ./net/ZMQMENet
 ./net/ZMQNet
+./net/LocalNet