temp debug changes: needs fix

Author: sandeepmittal

app/sstSinker.cpp

@@ -4,6 +4,10 @@
 #include "chimbuko/util/commandLineParser.hpp"
 #include "chimbuko/util/string.hpp"
 
+#ifndef ENABLE_PROVDB
+#error "Provdb flag not set"
+#endif
+
 using namespace chimbuko;
 using namespace std::chrono;
 

include/chimbuko/ad/ADLocalCounterStatistics.hpp

@@ -5,6 +5,7 @@
 #include <chimbuko/ad/ADEvent.hpp>
 #include <chimbuko/ad/ADCounter.hpp>
 #include "chimbuko/util/PerfStats.hpp"
+//#include "chimbuko/chimbuko.hpp"
 
 namespace chimbuko{
 
@@ -76,7 +77,7 @@ namespace chimbuko{
      *
      * The message string is the output of get_json_state() in string format
      */
-    std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client) const;
+    std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client, int rank, std::string pserver_addr) const;
 
     /**
      * @brief Attach a PerfStats object into which performance metrics are accumulated
@@ -116,7 +117,7 @@ namespace chimbuko{
      * @param step step (or frame) number
      * @return std::pair<size_t, size_t> [sent, recv] message size
      */
-    static std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client, const std::string &l_stats, int step);
+    static std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client, const std::string &l_stats, int step, int rank, std::string pserver_addr);
 
     unsigned long m_program_idx; /**< Program idx*/
     int m_step; /**< io step */

include/chimbuko/ad/ADLocalFuncStatistics.hpp

@@ -5,6 +5,7 @@
 #include <chimbuko/ad/AnomalyData.hpp>
 #include <chimbuko/util/Anomalies.hpp>
 #include "chimbuko/util/PerfStats.hpp"
+#include "chimbuko/chimbuko.hpp"
 
 namespace chimbuko{
 
@@ -75,7 +76,7 @@ namespace chimbuko{
      *
      * The message communicated is the string dump of the output of get_json_state()
      */
-    std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client) const;
+    std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client, int rank, std::string pserver_addr) const;
 
 
     /**
@@ -107,7 +108,7 @@ namespace chimbuko{
      * @param step step (or frame) number
      * @return std::pair<size_t, size_t> [sent, recv] message size
      */
-    static std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client, const std::string &l_stats, int step);
+    static std::pair<size_t, size_t> updateGlobalStatistics(ADNetClient &net_client, const std::string &l_stats, int step, int rank, std::string pserver_addr);
 
     int m_step; /**< io step */
     unsigned long m_rank; /**< Rank*/

include/chimbuko/ad/ADNetClient.hpp

@@ -7,7 +7,10 @@
 #endif
 #include "chimbuko/message.hpp"
 #include "chimbuko/util/PerfStats.hpp"
-
+#include "chimbuko/util/string.hpp"
+#include "chimbuko/util/error.hpp"
+#include "chimbuko/util/time.hpp"
+ 
 namespace chimbuko{
 
 
@@ -113,6 +116,163 @@ namespace chimbuko{
 
 
 
+  //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::cout << "Worker is waiting for "<< wait_ms << "ms" << std::endl;
+      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){
+      std::cout << "Performing blocking send and receive" << std::endl;
+      client.send_and_receive(*recv, *send);
+      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){
+      std::cout << "Performing non-blocking send and receive" << std::endl;
+      Message recv;
+      client.send_and_receive(recv, send);
+      std::cout << "Non-blocking send returned " << recv.buf() << std::endl;
+    }
+    bool do_delete() const{ return true; }
+  };
+
+  //ADNetClient inside a worker thread with blocking send/receive and non-blocking send
+  class ADThreadNetClient{
+    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;
+    }
+
+    void run(){
+      std::cout << "Starting worker thread" << std::endl;
+      worker = std::thread([&](){
+          ADNetClient client;
+          bool shutdown = false;
+
+          while(!shutdown){
+            size_t nwork = getNwork();
+            while(nwork > 0){
+              ClientAction* work_item = getWorkItem();
+              work_item->perform(client);
+              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!");
+              std::cout << "Worker received shutdown request" << std::endl;
+            }else{
+              std::this_thread::sleep_for(std::chrono::milliseconds(10));
+            }  
+          }
+        });
+    }
+
+  public:
+    ADThreadNetClient(){
+      run();
+    }
+    
+    //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"){
+      enqueue_action(new ClientActionConnect(rank, srank,sname));
+    }
+    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(){
+      std::cout << "Joining worker thread" << std::endl;
+      worker.join();
+    }
+
+  };
 
+ 
 
 };

include/chimbuko/ad/old_ADNetClient.hpp

@@ -0,0 +1,118 @@
+#pragma once
+
+#ifdef _USE_MPINET
+#include "chimbuko/net/mpi_net.hpp"
+#else
+#include "chimbuko/net/zmq_net.hpp"
+#endif
+#include "chimbuko/message.hpp"
+#include "chimbuko/util/PerfStats.hpp"
+
+namespace chimbuko{
+
+
+  /**
+   * @brief A wrapper class to facilitate communications between the AD and the parameter server
+   */
+  class ADNetClient{
+  public:
+    ADNetClient();
+
+    ~ADNetClient();
+    
+    /**
+     * @brief check if the parameter server is in use
+     * 
+     * @return true if the parameter server is in use
+     * @return false if the parameter server is not in use
+     */
+    bool use_ps() const { return m_use_ps; }
+
+    /**
+     * @brief connect to the parameter server
+     * 
+     * @param rank this process rank
+     * @param srank server process rank. If using ZMQnet this is not applicable
+     * @param sname server name. If using ZMQNet this is the server ip address, for MPINet it is not applicable
+     */
+    void connect_ps(int rank, int srank = 0, std::string sname="MPINET");
+    /**
+     * @brief disconnect from the connected parameter server
+     * 
+     * Called automatically by destructor if not previously called
+     */
+    void disconnect_ps();
+
+    /**
+     * @brief Return the MPI rank of the parameter server
+     */
+    int get_server_rank() const{ return m_srank; }
+
+    /**
+     * @brief Return the MPI rank of this client
+     */
+    int get_client_rank() const{ return m_rank; }
+
+    /**
+     * @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;
+
+    /**
+     * @brief Send a message to the parameter server and receive the response both as Message objects
+     * @param send The sent message
+     * @param recv The received message
+     *
+     * Note recv and send can be the same object
+     */       
+    void send_and_receive(Message &recv, const Message &send);
+
+    /**
+     * @brief If linked timing and packet size information will be gathered
+     */
+    void linkPerf(PerfStats* perf){ m_perf = perf; }
+
+#ifdef _USE_ZMQNET
+    /**
+     * @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; }
+
+    /**
+     * @brief Get the zeroMQ socket
+     */
+    void * getZMQsocket(){ return m_socket; }
+
+    /**
+     * @brief Get the zeroMQ context
+     */
+    void * getZMQcontext(){ return m_context; }
+    
+    /**
+     * @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;
+#endif
+
+
+  private:
+    bool m_use_ps;                           /**< true if the parameter server is in use */
+    int m_rank;                              /**< MPI rank of current process */
+    int m_srank;                             /**< server process rank                    */
+#ifdef _USE_MPINET
+    MPI_Comm m_comm;                         /**< Instance of the MPI communicator */
+#else
+    void* m_context;                         /**< ZeroMQ context */
+    void* m_socket;                          /**< ZeroMQ socket */
+    int m_recv_timeout_ms;                   /**< Timeout (in ms) on blocking receives (default 30s)*/
+#endif
+    PerfStats * m_perf;                      /**< Performance monitoring */
+  };
+
+  
+
+
+
+};

include/chimbuko/chimbuko.hpp

@@ -94,6 +94,10 @@ namespace chimbuko {
      */
     bool use_ps() const { return m_outlier->use_ps(); }
 
+#ifndef ENABLE_PROVDB
+#error "Provdb flag not set"
+#endif
+
 #ifdef ENABLE_PROVDB
     /**
      * @brief Whether the provenance database is in use