For the new non-unit frequency options for PS stats send, AD model update and provDB data send, the sending is now staggered by the rank index to make the network node less "spiky"

Author: giltirn

include/chimbuko/ad/ADOutlier.hpp

@@ -51,17 +51,17 @@ namespace chimbuko {
      * @brief Construct a new ADOutlier object
      *
      */
-    ADOutlier(OutlierStatistic stat = ExclusiveRuntime);
+    ADOutlier(int rank, OutlierStatistic stat = ExclusiveRuntime);
     /**
      * @brief Destroy the ADOutlier object
      *
      */
     virtual ~ADOutlier();
 
     /**
-     * @brief Fatory method to select AD algorithm at runtime
+     * @brief Factory method to select AD algorithm at runtime
      */
-    static ADOutlier *set_algorithm(const std::string & algorithm, const AlgoParams &params);
+    static ADOutlier *set_algorithm(int rank, const std::string & algorithm, const AlgoParams &params);
 
     /**
      * @brief check if the parameter server is in use
@@ -170,6 +170,7 @@ namespace chimbuko {
     int m_sync_call_count;                   /**< count of calls to sync_param */
     int m_global_model_sync_freq;            /**< how often the local model is pushed and synchronized with the globel model (default 1)*/
 
+    int m_rank;                              /**< rank index*/
     PerfStats *m_perf;
   private:
     OutlierStatistic m_statistic; /** Which statistic to use for outlier detection */
@@ -187,7 +188,7 @@ namespace chimbuko {
      * @brief Construct a new ADOutlierSSTD object
      *
      */
-    ADOutlierSSTD(OutlierStatistic stat = ExclusiveRuntime, double sigma = 6.0);
+    ADOutlierSSTD(int rank, OutlierStatistic stat = ExclusiveRuntime, double sigma = 6.0);
     /**
      * @brief Destroy the ADOutlierSSTD object
      *
@@ -239,13 +240,14 @@ namespace chimbuko {
 
     /**
      * @brief Construct a new ADOutlierHBOS object
+     * @param rank Rank of AD instance
      * @param stat Which statistic to use
      * @param threshold The threshold defining an outlier
      * @param use_global_threshold The threshold is maintained as part of the global model
      * @param maxbins The maximum number of bins in the histograms
      *
      */
-    ADOutlierHBOS(OutlierStatistic stat = ExclusiveRuntime, double threshold = 0.99, bool use_global_threshold = true, int maxbins = 200);
+    ADOutlierHBOS(int rank, OutlierStatistic stat = ExclusiveRuntime, double threshold = 0.99, bool use_global_threshold = true, int maxbins = 200);
 
     /**
      * @brief Destroy the ADOutlierHBOS object
@@ -337,7 +339,7 @@ namespace chimbuko {
      * @brief Construct a new ADOutlierCOPOD object
      *
      */
-    ADOutlierCOPOD(OutlierStatistic stat = ExclusiveRuntime, double threshold = 0.99, bool use_global_threshold = true);
+    ADOutlierCOPOD(int rank, OutlierStatistic stat = ExclusiveRuntime, double threshold = 0.99, bool use_global_threshold = true);
 
     /**
      * @brief Destroy the ADOutlierCOPOD object

sim/src/ad.cpp

@@ -75,7 +75,7 @@ void ADsim::init(int window_size, int pid, int rid, unsigned long program_start,
     params.glob_thres = p.glob_thres;
     params.sstd_sigma = p.sstd_sigma;
 
-    m_outlier = ADOutlier::set_algorithm(p.algorithm, params);
+    m_outlier = ADOutlier::set_algorithm(rid, p.algorithm, params);
     getPserver(); //force construction of pserver
     m_net_client = new ADThreadNetClient(true); //use local comms
     m_net_client->connect_ps(m_rid);

src/ad/ADOutlier.cpp

@@ -19,8 +19,8 @@ ADOutlier::AlgoParams::AlgoParams(): stat(ADOutlier::ExclusiveRuntime), sstd_sig
 /* ---------------------------------------------------------------------------
  * Implementation of ADOutlier class
  * --------------------------------------------------------------------------- */
-ADOutlier::ADOutlier(OutlierStatistic stat)
-  : m_execDataMap(nullptr), m_param(nullptr), m_local_param(nullptr), m_use_ps(false), m_perf(nullptr), m_statistic(stat), m_sync_call_count(0), m_global_model_sync_freq(1)
+ADOutlier::ADOutlier(int rank, OutlierStatistic stat)
+  : m_execDataMap(nullptr), m_param(nullptr), m_local_param(nullptr), m_use_ps(false), m_perf(nullptr), m_statistic(stat), m_sync_call_count(0), m_global_model_sync_freq(1), m_rank(rank)
 {
 }
 
@@ -49,17 +49,17 @@ void loadPerFunctionThresholds(ADOutlierType* into, const std::string &filename)
 }
 
 
-ADOutlier *ADOutlier::set_algorithm(const std::string & algorithm, const AlgoParams &params) {
+ADOutlier *ADOutlier::set_algorithm(int rank, const std::string & algorithm, const AlgoParams &params) {
   if (algorithm == "sstd" || algorithm == "SSTD") {
-    return new ADOutlierSSTD(params.stat, params.sstd_sigma);
+    return new ADOutlierSSTD(rank, params.stat, params.sstd_sigma);
   }
   else if (algorithm == "hbos" || algorithm == "HBOS") {
-    ADOutlierHBOS* alg = new ADOutlierHBOS(params.stat, params.hbos_thres, params.glob_thres, params.hbos_max_bins);
+    ADOutlierHBOS* alg = new ADOutlierHBOS(rank, params.stat, params.hbos_thres, params.glob_thres, params.hbos_max_bins);
     loadPerFunctionThresholds(alg,params.func_threshold_file);
     return alg;
   }
   else if (algorithm == "copod" || algorithm == "COPOD") {
-    ADOutlierCOPOD* alg = new ADOutlierCOPOD(params.stat, params.hbos_thres);
+    ADOutlierCOPOD* alg = new ADOutlierCOPOD(rank, params.stat, params.hbos_thres);
     loadPerFunctionThresholds(alg,params.func_threshold_file);
     return alg;   
   }
@@ -111,8 +111,8 @@ void ADOutlier::updateGlobalModel()
   PerfTimer timer;
   timer.start();
 
-  if ( m_sync_call_count %  m_global_model_sync_freq == 0 ){    
-    verboseStream << "ADOutlier performing synchronization of local and global model" << std::endl;
+  if ( m_sync_call_count == 0 || (m_sync_call_count + m_rank) %  m_global_model_sync_freq == 0 ){ //apart from on first step, stagger updates over ranks by rank index
+    verboseStream << "ADOutlier rank " << m_rank << " performing synchronization of local and global model on call count " << m_sync_call_count << std::endl;
     PerfTimer utimer;
     utimer.start();
 
@@ -144,7 +144,7 @@ void ADOutlier::setIgnoreFunction(const std::string &func){
 /* ---------------------------------------------------------------------------
  * Implementation of ADOutlierSSTD class
  * --------------------------------------------------------------------------- */
-ADOutlierSSTD::ADOutlierSSTD(OutlierStatistic stat, double sigma) : ADOutlier(stat), m_sigma(sigma) {
+ADOutlierSSTD::ADOutlierSSTD(int rank, OutlierStatistic stat, double sigma) : ADOutlier(rank, stat), m_sigma(sigma) {
     m_param = new SstdParam();
     m_local_param = new SstdParam();
 }
@@ -281,7 +281,7 @@ unsigned long ADOutlierSSTD::compute_outliers(Anomalies &outliers,
 /* ---------------------------------------------------------------------------
  * Implementation of ADOutlierHBOS class
  * --------------------------------------------------------------------------- */
-ADOutlierHBOS::ADOutlierHBOS(OutlierStatistic stat, double threshold, bool use_global_threshold, int maxbins) : ADOutlier(stat), m_alpha(78.88e-32), m_threshold(threshold), m_use_global_threshold(use_global_threshold), m_maxbins(maxbins) {
+ADOutlierHBOS::ADOutlierHBOS(int rank, OutlierStatistic stat, double threshold, bool use_global_threshold, int maxbins) : ADOutlier(rank, stat), m_alpha(78.88e-32), m_threshold(threshold), m_use_global_threshold(use_global_threshold), m_maxbins(maxbins) {
     m_param = new HbosParam();
     m_local_param = new HbosParam();
 }
@@ -531,7 +531,7 @@ unsigned long ADOutlierHBOS::compute_outliers(Anomalies &outliers,
 /* ---------------------------------------------------------------------------
  * Implementation of ADOutlierCOPOD class
  * --------------------------------------------------------------------------- */
-ADOutlierCOPOD::ADOutlierCOPOD(OutlierStatistic stat, double threshold, bool use_global_threshold) : ADOutlier(stat), m_alpha(78.88e-32), m_threshold(threshold), m_use_global_threshold(use_global_threshold) {
+ADOutlierCOPOD::ADOutlierCOPOD(int rank, OutlierStatistic stat, double threshold, bool use_global_threshold) : ADOutlier(rank, stat), m_alpha(78.88e-32), m_threshold(threshold), m_use_global_threshold(use_global_threshold) {
     m_param = new CopodParam();
     m_local_param = new CopodParam();
 }

src/chimbuko.cpp

@@ -248,7 +248,7 @@ void Chimbuko::init_outlier(){
   else if(m_params.outlier_statistic == "inclusive_runtime") params.stat = ADOutlier::InclusiveRuntime;
   else{ fatal_error("Invalid statistic"); }
 
-  m_outlier = ADOutlier::set_algorithm(m_params.ad_algorithm, params);
+  m_outlier = ADOutlier::set_algorithm(m_params.rank, m_params.ad_algorithm, params);
   m_outlier->linkExecDataMap(m_event->getExecDataMap()); //link the map of function index to completed calls such that they can be tagged as outliers if appropriate
   if(m_net_client) m_outlier->linkNetworkClient(m_net_client);
   m_outlier->linkPerf(&m_perf);
@@ -534,9 +534,10 @@ void Chimbuko::sendProvenance(const int step, bool force){
 #endif
       )
      && 
-     ( step % m_params.prov_io_freq == 0 || force )
+     ( (step + m_params.rank) % m_params.prov_io_freq == 0 || force ) //stagger sends over ranks by offsetting by rank index
      ){
     //Get the provenance data
+    verboseStream << "Chimbuko rank " << m_params.rank << " performing send of provenance data on step " << step << std::endl;
 
     PerfTimer timer;
     //Write and send provenance data
@@ -633,8 +634,9 @@ void Chimbuko::gatherPSdata(const Anomalies &anomalies,
 }
 
 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) ){
+  if(m_net_client && m_net_client->use_ps() && m_funcstats_buf.size() && ( (step + m_params.rank) % m_params.ps_send_stats_freq == 0 || force) ){ //stagger sends by offsetting by rank
     //Send the data in a single communication
+    verboseStream << "Chimbuko rank " << m_params.rank << " performing send of PS stats data on step " << step << std::endl;
     PerfTimer timer;
     timer.start();
     ADcombinedPSdataArray comb_stats(m_funcstats_buf, m_countstats_buf, m_anom_metrics_buf, &m_perf); 

test/unit_tests/ad/COPODOutlier.cpp

@@ -264,7 +264,7 @@ TEST(CopodADOutlierTest, TestFunctionThresholdOverride){
   int func_id2 = 202;
 
   double default_threshold = 0.99;
-  ADOutlierCOPOD ad(ADOutlier::ExclusiveRuntime, default_threshold);
+  ADOutlierCOPOD ad(0, ADOutlier::ExclusiveRuntime, default_threshold);
   ad.overrideFuncThreshold("my_func",0.77);
 
   EXPECT_EQ(ad.getFunctionThreshold("my_func"), 0.77);

test/unit_tests/ad/HBOSOutlier.cpp

@@ -287,7 +287,7 @@ TEST(HBOSADOutlierTest, TestFunctionThresholdOverride){
   int func_id2 = 202;
 
   double default_threshold = 0.99;
-  ADOutlierHBOS ad(ADOutlier::ExclusiveRuntime, default_threshold);
+  ADOutlierHBOS ad(0, ADOutlier::ExclusiveRuntime, default_threshold);
   ad.overrideFuncThreshold("my_func",0.77);
 
   EXPECT_EQ(ad.getFunctionThreshold("my_func"), 0.77);