Stats sent from AD to pserver can now be aggregated over multiple steps before being sent + unit test

Added driver option -ps_send_stats_freq to control the above frequency

Author: giltirn

app/driver.cpp

@@ -112,6 +112,7 @@ optionalArgsParser & getOptionalArgsParser(){
 
     addOptionalCommandLineArg(p, outlier_statistic, "Set the statistic used for outlier detection. Options: exclusive_runtime (default), inclusive_runtime");
     addOptionalCommandLineArg(p, global_model_sync_freq, "Set the frequency in steps between updates of the global model (default 1)");
+    addOptionalCommandLineArg(p, ps_send_stats_freq, "Set how often in steps the statistics data is uploaded to the pserver (default 1)");
     addOptionalCommandLineArg(p, step_report_freq, "Set the steps between Chimbuko reporting IO step progress. Use 0 to deactivate this logging entirely (default 1)");
     addOptionalCommandLineArg(p, prov_record_startstep, "If != -1, the IO step on which to start recording provenance information for anomalies (for testing, default -1)");
     addOptionalCommandLineArg(p, prov_record_stopstep, "If != -1, the IO step on which to stop recording provenance information for anomalies (for testing, default -1)");

app/pserver.cpp

@@ -194,7 +194,7 @@ int main (int argc, char ** argv){
     for(int i=0;i<args.nt;i++){
       net.add_payload(new NetPayloadUpdateParamManager(&param, i, args.freeze_params), i);
       net.add_payload(new NetPayloadGetParamsFromManager(&param),i);
-      net.add_payload(new NetPayloadRecvCombinedADdata(&global_func_stats[i], &global_counter_stats[i], &global_anom_metrics[i]),i); //each worker thread writes to a separate stats object which are aggregated only at viz send time
+      net.add_payload(new NetPayloadRecvCombinedADdataArray(&global_func_stats[i], &global_counter_stats[i], &global_anom_metrics[i]),i); //each worker thread writes to a separate stats object which are aggregated only at viz send time
       net.add_payload(new NetPayloadGlobalFunctionIndexMapBatched(&global_func_index_map),i);
       net.add_payload(new NetPayloadPing,i);
     }

include/chimbuko/ad/ADLocalCounterStatistics.hpp

@@ -24,7 +24,9 @@ namespace chimbuko{
 			     const std::unordered_set<std::string> *which_counters, PerfStats *perf = nullptr):
       m_program_idx(program_idx), m_step(step), m_which_counter(which_counters), m_perf(perf)
     {}
-				
+
+    ADLocalCounterStatistics(): m_program_idx(0), m_step(-1), m_which_counter(nullptr), m_perf(nullptr){}
+	
     /**
      * @brief Add counters to internal statistics
      */

include/chimbuko/ad/ADcombinedPSdata.hpp

@@ -70,4 +70,70 @@ namespace chimbuko{
   };
 
 
+
+  /** 
+   * @brief A class that combines all data that is sent to the PS collected over multiple steps into a single send transmission
+   */
+  class ADcombinedPSdataArray{
+  public:
+    ADcombinedPSdataArray(std::vector<ADLocalFuncStatistics> &func_stats, 
+			  std::vector<ADLocalCounterStatistics> &counter_stats, 
+			  std::vector<ADLocalAnomalyMetrics> &anom_metrics,
+		     PerfStats* perf=nullptr): 
+      m_func_stats(func_stats), m_counter_stats(counter_stats), m_anom_metrics(anom_metrics), m_perf(perf){}
+    
+    std::vector<ADLocalCounterStatistics> &get_counter_stats(){ return m_counter_stats; }
+    const std::vector<ADLocalCounterStatistics> &get_counter_stats() const{ return m_counter_stats; }
+    
+    std::vector<ADLocalFuncStatistics> &get_func_stats(){ return m_func_stats; }
+    const std::vector<ADLocalFuncStatistics> &get_func_stats() const{ return m_func_stats; }
+
+    std::vector<ADLocalAnomalyMetrics> &get_anom_metrics(){ return m_anom_metrics; }
+    const std::vector<ADLocalAnomalyMetrics> &get_anom_metrics() const{ return m_anom_metrics; }
+
+    /**
+     * @brief Serialize using cereal
+     */
+    template<class Archive>
+    void serialize(Archive & archive){
+      archive(m_func_stats , m_counter_stats, m_anom_metrics);
+    }
+
+    /**
+     * Serialize into Cereal portable binary format
+     */
+    std::string serialize_cerealpb() const;
+      
+    /**
+     * Serialize from Cereal portable binary format
+     */     
+    void deserialize_cerealpb(const std::string &strstate);
+
+    /**
+     * @brief Serialize this class for communication over the network
+     */
+    std::string net_serialize() const;
+
+    /**
+     * @brief Unserialize this class after communication over the network
+     */
+    void net_deserialize(const std::string &s);
+      
+      
+    /**
+     * @brief Send the data to the pserver
+     * @param net_client The network client object
+     * @return std::pair<size_t, size_t> [sent, recv] message size
+     */
+    std::pair<size_t, size_t> send(ADNetClient &net_client) const;
+  
+
+  private:
+    std::vector<ADLocalFuncStatistics> &m_func_stats;
+    std::vector<ADLocalCounterStatistics> &m_counter_stats;
+    std::vector<ADLocalAnomalyMetrics> &m_anom_metrics;
+    PerfStats* m_perf;
+  };
+
+
 }

include/chimbuko/chimbuko.hpp

@@ -39,6 +39,8 @@ namespace chimbuko {
 				 < If no parameter server is in use, this string should be empty (length zero)
 				 < If using ZmqNet (default) this is a tcp address of the form "tcp://${ADDRESS}:${PORT}"
 			      */
+    int ps_send_stats_freq; /**< How often in steps the statistics data is uploaded to the pserver (default 1) */
+
     int hpserver_nthr;        /**< If using the hierarchical pserver, this parameter is used to compute a port offset for the particular endpoint that this AD rank connects to */
 
     std::string prov_outputpath; /**< Directory where provenance data is written (in conjunction with provDB if active). Blank string indicates no output*/
@@ -260,14 +262,18 @@ namespace chimbuko {
      */
     void sendProvenance(const int step, bool force = false);
 
+    /**
+     * @brief Gather and buffer the required statistics data for the pserver
+     */
+    void gatherPSdata(const Anomalies &anomalies,
+		      const int step,
+		      const unsigned long first_event_ts,
+		      const unsigned long last_event_ts);
 
     /**
-     * @brief Gather and send the required data to the pserver
+     * @brief Send buffered pserver statistics data when step matches frequency or force==true
      */
-    void gatherAndSendPSdata(const Anomalies &anomalies,
-			     const int step,
-			     const unsigned long first_event_ts,
-			     const unsigned long last_event_ts) const;
+    void sendPSdata(const int step, bool force = false);
 
     /**
      * @brief Send new metadata entries collected during current fram to provenance DB
@@ -309,6 +315,10 @@ namespace chimbuko {
     std::vector<nlohmann::json> m_normalevent_prov_buf; /**<Buffered normal event provenance data waiting to be sent*/
 
     std::set<unsigned long> m_exec_ignore_counters; /**< Counter indices in this list are ignored by the event manager (but will still be picked up by other components)*/
+
+    std::vector<ADLocalFuncStatistics> m_funcstats_buf; /**< Buffered function statistics data waiting to be sent*/
+    std::vector<ADLocalCounterStatistics> m_countstats_buf;  /**< Buffered counter statistics data waiting to be sent*/
+    std::vector<ADLocalAnomalyMetrics> m_anom_metrics_buf;  /**< Buffered anomaly metrics data waiting to be sent*/    
   };
 
 }

include/chimbuko/pserver/NetPayloadRecvCombinedADdata.hpp

@@ -6,7 +6,7 @@
 
 namespace chimbuko{
 
-    /**
+  /**
    * @brief Net payload for receiving the combined data payload from the AD
    */
   class NetPayloadRecvCombinedADdata: public NetPayloadBase{
@@ -23,6 +23,23 @@ namespace chimbuko{
     void action(Message &response, const Message &message) override;
   };
 
+  /**
+   * @brief Net payload for receiving the combined data payload over multiple steps from the AD
+   */
+  class NetPayloadRecvCombinedADdataArray: public NetPayloadBase{
+    GlobalAnomalyStats * m_global_anom_stats;
+    GlobalCounterStats * m_global_counter_stats;
+    GlobalAnomalyMetrics * m_global_anom_metrics;
+  public:
+    NetPayloadRecvCombinedADdataArray(GlobalAnomalyStats * global_anom_stats, GlobalCounterStats * global_counter_stats,
+				      GlobalAnomalyMetrics * global_anom_metrics): m_global_anom_stats(global_anom_stats),
+										   m_global_counter_stats(global_counter_stats),
+										   m_global_anom_metrics(global_anom_metrics){}
+    MessageKind kind() const override{ return MessageKind::AD_PS_COMBINED_STATS; }
+    MessageType type() const override{ return MessageType::REQ_ADD; }
+    void action(Message &response, const Message &message) override;
+  };
+
 
 
 

src/ad/ADcombinedPSdata.cpp

@@ -40,8 +40,63 @@ std::pair<size_t, size_t> ADcombinedPSdata::send(ADNetClient &net_client) const{
 
   if(m_perf != nullptr){
     m_perf->add("ps_combine_send_update_ms", timer.elapsed_ms());
-    m_perf->add("ps_combine_send_sent_MB", (double)sent_sz / 1000000.0); // MB
-    m_perf->add("ps_combine_send_recv_MB", (double)recv_sz / 1000000.0); // MB
+    m_perf->add("ps_combine_send_sent_MB", (double)sent_sz / 1024./1024.); // MB
+    m_perf->add("ps_combine_send_recv_MB", (double)recv_sz / 1024./1024.); // MB
+  }
+  
+  return std::make_pair(sent_sz, recv_sz);
+}
+
+
+
+
+
+std::string ADcombinedPSdataArray::serialize_cerealpb() const{
+  return cereal_serialize(*this);
+}
+
+void ADcombinedPSdataArray::deserialize_cerealpb(const std::string &strstate){
+  cereal_deserialize(*this, strstate);
+}
+
+std::string ADcombinedPSdataArray::net_serialize() const{
+  return serialize_cerealpb();
+}
+
+void ADcombinedPSdataArray::net_deserialize(const std::string &s){
+  deserialize_cerealpb(s);
+}
+
+std::pair<size_t, size_t> ADcombinedPSdataArray::send(ADNetClient &net_client) const{
+  PerfTimer timer;
+  timer.start();
+
+  if (!net_client.use_ps())
+    return std::make_pair(0, 0);
+
+  size_t N = m_func_stats.size();
+  if(m_counter_stats.size() != N || m_anom_metrics.size() != N) fatal_error("Expect arrays to have the same size");
+
+  int latest_step = -1;
+  for(size_t i=0;i<N;i++){
+    int step = m_func_stats[i].getAnomalyData().get_step();
+    if(m_counter_stats[i].getIOstep() != step) fatal_error("Step value mismatch between counter stats and func anomaly stats");
+    if(m_anom_metrics[i].get_step() != step) fatal_error("Step value mismatch between anomaly metrics and func anomaly stats");
+    latest_step = std::max(step, latest_step);
+  }
+
+  Message msg;
+  msg.set_info(net_client.get_client_rank(), net_client.get_server_rank(), MessageType::REQ_ADD, MessageKind::AD_PS_COMBINED_STATS, latest_step);
+  msg.set_msg(net_serialize());
+  
+  size_t sent_sz = msg.size();
+  net_client.async_send(msg);
+  size_t recv_sz =0;
+
+  if(m_perf != nullptr){
+    m_perf->add("ps_combine_send_update_ms", timer.elapsed_ms());
+    m_perf->add("ps_combine_send_sent_MB", (double)sent_sz / 1024./1024.); // MB
+    m_perf->add("ps_combine_send_recv_MB", (double)recv_sz / 1024./1024.); // MB
   }
 
   return std::make_pair(sent_sz, recv_sz);

src/chimbuko.cpp

@@ -40,7 +40,8 @@ ChimbukoParams::ChimbukoParams(): rank(-1234),  //not set!
                                   monitoring_counter_prefix(""),
                                   prov_min_anom_time(0),
                                   prov_io_freq(1),
-                                  global_model_sync_freq(1)
+                                  global_model_sync_freq(1),
+                                  ps_send_stats_freq(1)
 {}
 
 void ChimbukoParams::print() const{
@@ -599,39 +600,52 @@ void Chimbuko::sendNewMetadataToProvDB(int step) const{
   }
 }
 
-void Chimbuko::gatherAndSendPSdata(const Anomalies &anomalies,
-				   const int step,
-				   const unsigned long first_event_ts,
-				   const unsigned long last_event_ts) const{
+void Chimbuko::gatherPSdata(const Anomalies &anomalies,
+			    const int step,
+			    const unsigned long first_event_ts,
+			    const unsigned long last_event_ts){
   if(m_net_client && m_net_client->use_ps()){
     PerfTimer timer;
 
     //Gather function profile and anomaly statistics
     timer.start();
-    ADLocalFuncStatistics prof_stats(m_params.program_idx, m_params.rank, step, &m_perf);
-    prof_stats.gatherStatistics(m_event->getExecDataMap());
-    prof_stats.gatherAnomalies(anomalies);
-    m_perf.add("ad_gather_send_ps_data_gather_profile_stats_time_ms", timer.elapsed_ms());
+    ADLocalFuncStatistics func_stats(m_params.program_idx, m_params.rank, step, &m_perf);
+    func_stats.gatherStatistics(m_event->getExecDataMap());
+    func_stats.gatherAnomalies(anomalies);
+    m_funcstats_buf.emplace_back(std::move(func_stats));
+    m_perf.add("ad_gather_ps_data_gather_profile_stats_time_ms", timer.elapsed_ms());
 
     //Gather counter statistics
     timer.start();
     ADLocalCounterStatistics count_stats(m_params.program_idx, step, nullptr, &m_perf); //currently collect all counters
     count_stats.gatherStatistics(m_counter->getCountersByIndex());
-    m_perf.add("ad_gather_send_ps_data_gather_counter_stats_time_ms", timer.elapsed_ms());
+    m_countstats_buf.emplace_back(std::move(count_stats));
+    m_perf.add("ad_gather_data_gather_counter_stats_time_ms", timer.elapsed_ms());
 
     //Gather anomaly metrics
     timer.start();
     ADLocalAnomalyMetrics metrics(m_params.program_idx, m_params.rank, step, first_event_ts, last_event_ts, anomalies);
-    m_perf.add("ad_gather_send_ps_data_gather_metrics_time_ms", timer.elapsed_ms());
+    m_anom_metrics_buf.emplace_back(std::move(metrics));
+    m_perf.add("ad_gather_ps_data_gather_metrics_time_ms", timer.elapsed_ms());
 
+    m_perf.add("ad_gather_ps_data_total_time_ms", timer.elapsed_ms());
+  }
+}
+
+void Chimbuko::sendPSdata(const int step, bool force){
+  if(m_net_client && m_net_client->use_ps() && m_funcstats_buf.size() && (step % m_params.ps_send_stats_freq == 0 || force) ){
     //Send the data in a single communication
+    PerfTimer timer;
     timer.start();
-    ADcombinedPSdata comb_stats(prof_stats, count_stats, metrics, &m_perf);
-    comb_stats.send(*m_net_client);
-    m_perf.add("ad_gather_send_ps_data_gather_send_all_stats_to_ps_time_ms", timer.elapsed_ms());
+    ADcombinedPSdataArray comb_stats(m_funcstats_buf, m_countstats_buf, m_anom_metrics_buf, &m_perf); 
+    comb_stats.send(*m_net_client); //serializes and puts result in buffer awaiting send so we are free to flush the buffers
+    m_funcstats_buf.clear(); m_countstats_buf.clear(); m_anom_metrics_buf.clear();
+    m_perf.add("ad_send_ps_data_total_time_ms", timer.elapsed_ms());
   }
 }
 
+
+
 bool Chimbuko::runFrame(unsigned long long& n_func_events,
 			unsigned long long& n_comm_events,
 			unsigned long long& n_counter_events,
@@ -720,7 +734,7 @@ bool Chimbuko::runFrame(unsigned long long& n_func_events,
 
     //Gather and send statistics and data to the pserver
     timer.start();
-    gatherAndSendPSdata(anomalies, step, m_execdata_first_event_ts, m_execdata_last_event_ts);
+    gatherPSdata(anomalies, step, m_execdata_first_event_ts, m_execdata_last_event_ts);
     m_perf.add("ad_run_gather_send_ps_data_time_ms", timer.elapsed_ms());
 
     //Trim the call list
@@ -739,7 +753,9 @@ bool Chimbuko::runFrame(unsigned long long& n_func_events,
     if(do_step_report){ headProgressStream(m_params.rank) << "driver rank " << m_params.rank << " function execution analysis complete: total=" << nout + nnormal << " normal=" << nnormal << " anomalous=" << nout << std::endl; }
   }//if(do_run_analysis)
 
-  sendProvenance(step); //only actually sends if the step aligns with the send frequency. Want to do this even if not analyzing this step
+  //Send provenance and PS data if the step aligns with the send frequency. Want to do this even if not analyzing this step
+  sendProvenance(step); 
+  sendPSdata(step);
 
   m_perf.add("ad_run_total_step_time_excl_parse_ms", step_timer.elapsed_ms());
 
@@ -805,5 +821,7 @@ void Chimbuko::run(unsigned long long& n_func_events,
     if (m_params.interval_msec)
       std::this_thread::sleep_for(std::chrono::milliseconds(m_params.interval_msec));
   }
-  sendProvenance(m_parser->getCurrentStep(), true); //send any outstanding buffered provenance data (second arg forces send)
+  //send any outstanding buffered provenance and PS data (second arg forces send)
+  sendProvenance(m_parser->getCurrentStep(), true);
+  sendPSdata(m_parser->getCurrentStep(), true); 
 }

src/pserver/NetPayloadRecvCombinedADdata.cpp

@@ -22,4 +22,29 @@ void NetPayloadRecvCombinedADdata::action(Message &response, const Message &mess
 
   response.set_msg("", false);
 }
+
+void NetPayloadRecvCombinedADdataArray::action(Message &response, const Message &message){
+  check(message);
+  if(m_global_anom_stats == nullptr) throw std::runtime_error("Cannot update global anomaly statistics as stats object has not been linked");
+  if(m_global_counter_stats == nullptr) throw std::runtime_error("Cannot update global counter statistics as stats object has not been linked");
+  if(m_global_anom_metrics == nullptr) throw std::runtime_error("Cannot update global anomaly metrics as stats object has not been linked");
+  
+  std::vector<ADLocalFuncStatistics> locf;
+  std::vector<ADLocalCounterStatistics> locc;
+  std::vector<ADLocalAnomalyMetrics> locm;
+
+  ADcombinedPSdataArray comb(locf,locc,locm);
+  comb.net_deserialize(message.buf());
+
+  int N = locf.size();
+  if(locc.size() != N || locm.size() != N) fatal_error("Expect consistent size for arrays");
+
+  for(int i=0;i<N;i++){
+    m_global_anom_stats->add_anomaly_data(locf[i]);
+    m_global_counter_stats->add_counter_data(locc[i]);
+    m_global_anom_metrics->add(locm[i]);
+  }
+
+  response.set_msg("", false);
+}