Merge branch 'ckelly_develop' into provdb_multiprovider

Author: giltirn

include/chimbuko/ad/ADOutlier.hpp

@@ -238,7 +238,7 @@ namespace chimbuko {
 
   private:
     double m_alpha; /**< Used to prevent log2 overflow */
-    double m_threshold; /**< Threshold used to filter anomalies in HBOS*/
+    double m_threshold; /**< Threshold used to filter anomalies in COPOD*/
     bool m_use_global_threshold; /**< Flag to use global threshold*/
     //double m_threshold; /** sync with global threshold */
     OutlierStatistic m_statistic; /** Which statistic to use for outlier detection */
@@ -253,13 +253,13 @@ namespace chimbuko {
   public:
 
     /**
-     * @brief Construct a new ADOutlierHBOS object
+     * @brief Construct a new ADOutlierCOPOD object
      *
      */
     ADOutlierCOPOD(OutlierStatistic stat = ExclusiveRuntime, double threshold = 0.99, bool use_global_threshold = true);
 
     /**
-     * @brief Destroy the ADOutlierHBOS object
+     * @brief Destroy the ADOutlierCOPOD object
      *
      */
     ~ADOutlierCOPOD();
@@ -308,11 +308,12 @@ namespace chimbuko {
 
   private:
     double m_alpha; /**< Used to prevent log2 overflow */
-    double m_threshold; /**< Threshold used to filter anomalies in HBOS*/
+    double m_threshold; /**< Threshold used to filter anomalies in COPOD*/
     bool m_use_global_threshold; /**< Flag to use global threshold*/
-    //double m_threshold; /** sync with global threshold */
-    OutlierStatistic m_statistic; /** Which statistic to use for outlier detection */
+    //double m_threshold; /**< sync with global threshold */
+    OutlierStatistic m_statistic; /**< Which statistic to use for outlier detection */
 
+    std::unordered_map<unsigned long, int> m_skewness; /**< skewness for each function*/
   };
 
 

include/chimbuko/ad/ExecData.hpp

@@ -489,7 +489,7 @@ class ExecData_t {
     /**
      * @brief Return the outlier severity, representing how important the outlier is
      */  
-    double get_outlier_severity() const{ return (double)get_runtime(); }   
+    double get_outlier_severity() const{ return (double)get_exclusive(); }
 
     /**
      * @brief Set the label

src/ad/ADOutlier.cpp

@@ -266,16 +266,26 @@ Anomalies ADOutlierHBOS::run(int step) {
     }
     if (runtimes.size() > 0) {
       if (!g.find(func_id)) { // If func_id does not exist
-        const int r = param[func_id].create_histogram(runtimes);
-        if (r < 0) {return outliers;}
+        
+	const int r = param[func_id].create_histogram(runtimes);
+        if (r < 0) {
+		recoverable_error(std::string("AD: Func_ID does not exist"));
+		continue;
+	}
       }
       else { //merge with exisiting func_id, not overwrite
 
         const int r = param[func_id].merge_histograms(g[func_id], runtimes);
-	if (r < 0) {return outliers;}
+	if (r < 0) {
+		recoverable_error(std::string("AD: Merging error received "));
+		continue;
+	}
       }
     }
-    else { return outliers;}
+    else { 
+      //recoverable_error(std::string("AD: Zero function runtimes "));
+	    continue;
+    }
   }
 
   //Update temp runstats to include information collected previously (synchronizes with the parameter server if connected)
@@ -543,15 +553,42 @@ Anomalies ADOutlierCOPOD::run(int step) {
     if (runtimes.size() > 0) {
       if (!g.find(func_id)) { // If func_id does not exist
         const int r = param[func_id].create_histogram(runtimes);
-        if (r < 0) {return outliers;}
+        if (r < 0) {
+		recoverable_error(std::string("AD: Func_ID does not exist "));
+		continue;
+	}
       }
       else { //merge with exisiting func_id, not overwrite
 
         const int r = param[func_id].merge_histograms(g[func_id], runtimes);
-	if (r < 0) {return outliers;}
+	if (r < 0) {
+		recoverable_error(std::string("AD: Merging error received "));
+		continue;
+	}
       }
     }
-    else { return outliers;}
+    else { 
+      ///recoverable_error(std::string("AD: Zero function runtimes "));
+	    continue;
+    }
+    verboseStream << "Size of runtimes: " << runtimes.size() << ", func_id: " << func_id << std::endl;
+    
+    //calculate skewness of runtimes for func_id
+    const double mu = std::accumulate(runtimes.begin(), runtimes.end(), 0.0) / runtimes.size();
+
+    std::vector<double> diff = std::vector<double>(runtimes.size());
+    std::transform(runtimes.begin(), runtimes.end(), diff.begin(), [mu](double x) {return x - mu;});
+    const double sq_sum = std::inner_product(diff.begin(), diff.end(), diff.begin(), 0.0);
+    const double stdev = std::sqrt(sq_sum / runtimes.size());
+
+    double summ = 0;
+    for (int i=0; i<runtimes.size(); i++) {
+	summ += std::pow((runtimes.at(i) - mu), 3);
+    }
+
+    const double abs_skewness = summ / ((runtimes.size() - 1) * std::pow(stdev,3));
+    m_skewness[func_id] = (abs_skewness < 0) ? -1 : (abs_skewness > 0) ? 1 : 0;
+
   }
 
   //Update temp runstats to include information collected previously (synchronizes with the parameter server if connected)
@@ -579,6 +616,7 @@ unsigned long ADOutlierCOPOD::compute_outliers(Anomalies &outliers,
 					      std::vector<CallListIterator_t>& data){
 
   verboseStream << "Finding outliers in events for func " << func_id << std::endl;
+  verboseStream << "data Size: " << data.size() << std::endl;
 
   CopodParam& param = *(CopodParam*)m_param;
 
@@ -589,47 +627,72 @@ unsigned long ADOutlierCOPOD::compute_outliers(Anomalies &outliers,
   //std::vector<double> prob_counts = std::vector<double>(param[func_id].counts().size(), 0.0);
   double tot_runtimes = std::accumulate(param[func_id].counts().begin(), param[func_id].counts().end(), 0.0);
 
+  if (tot_runtimes <= 0 ) {
+	  return n_outliers;
+  }
   std::vector<double> recon_p_runtimes = std::vector<double>(tot_runtimes, 0.0);
   std::vector<double> recon_n_runtimes = std::vector<double>(tot_runtimes, 0.0);
   int recon_idx = 0;
-  //verboseStream << "Unwrapping Merged Histogram. Size: " << param[func_id].counts().size() << std::endl;
+  verboseStream << "Unwrapping Merged Histogram. Size: " << param[func_id].counts().size() << std::endl;
   for(int i=0; i < param[func_id].counts().size(); i++){
     int count = param[func_id].counts().at(i);
-    //verboseStream << "Count: " << count << ", Value: " << param[func_id].bin_edges().at(i) << std::endl;
+    verboseStream << "Count: " << count << ", Value: " << param[func_id].bin_edges().at(i) << std::endl;
     for(int j=0; j<count; j++){
 
       recon_p_runtimes.at(recon_idx) = param[func_id].bin_edges().at(i);
       recon_n_runtimes.at(recon_idx) = -1 * param[func_id].bin_edges().at(i);
-      //verboseStream << "recon_p_runtimes.at(recon_idx): " << recon_p_runtimes.at(recon_idx) << ", recon_n_runtimes.at(recon_idx): " << recon_n_runtimes.at(recon_idx) << std::endl;
-      //verboseStream << "recon_idx: " << recon_idx << std::endl;
+      verboseStream << "recon_idx: " << recon_idx << std::endl;
+      verboseStream << "recon_p_runtimes.at(recon_idx): " << recon_p_runtimes.at(recon_idx) << ", recon_n_runtimes.at(recon_idx): " << recon_n_runtimes.at(recon_idx) << std::endl;
       recon_idx++;
     }
   }
 
+
   std::vector<double> func_p_ecdf = empiricalCDF(recon_p_runtimes, true);
   std::vector<double> func_n_ecdf = empiricalCDF(recon_n_runtimes, true);
+  
+  verboseStream << "Size of empiricalCDF(recon_p_runtimes): " << func_p_ecdf.size() << std::endl;
+  verboseStream << "Size of empiricalCDF(recon_n_runtimes): " << func_n_ecdf.size() << std::endl;
 
   std::vector<double> mean_pn_ecdf = std::vector<double>(func_p_ecdf.size(), 0.0);
+  verboseStream << "Size of mean_pn_ecdf: " << mean_pn_ecdf.size() << ", func_id: " << func_id << std::endl;
+
   for(int i=0; i < mean_pn_ecdf.size(); i++){
-    mean_pn_ecdf.at(i) = (func_p_ecdf.at(i) + func_n_ecdf.at(i)) / 2;
+    mean_pn_ecdf.at(i) = (func_p_ecdf.at(i) + func_n_ecdf.at(i)) / 2.0;
+    verboseStream << "mean_pn_ecdf.at(i): " << mean_pn_ecdf.at(i) << ", func_p_ecdf.at(i): " << func_p_ecdf.at(i) << ", func_n_ecdf.at(i): " << func_n_ecdf.at(i) << std::endl;
   }
 
+  //use skewness
+  std::vector<double> skewness_arr = std::vector<double>(func_p_ecdf.size(), 0.0);
+  const int p_sign = (m_skewness[func_id] - 1) < 0 ? -1 : (m_skewness[func_id] - 1) > 0 ? 1 : 0;
+  const int n_sign = (m_skewness[func_id] + 1) < 0 ? -1 : (m_skewness[func_id] + 1) > 0 ? 1 : 0;
+  
+  for (int i = 0; i< func_p_ecdf.size(); i++) {
+	skewness_arr.at(i) = (func_p_ecdf.at(i) * -1 * p_sign) + (func_n_ecdf.at(i) * n_sign);
+  	verboseStream << "skewness_arr.at(" << i << "): " << skewness_arr.at(i) << std::endl;
+  }
 
-  //for(int i=0; i < param[func_id].counts().size(); i++){
-  //  int count = param[func_id].counts().at(i);
-  //  double p = count / tot_runtimes;
-  //  prob_counts.at(i) += p;
-  //}
+  std::vector<double> final_comp = std::vector<double>(skewness_arr.size(), 0.0);
+  for (int i = 0; i < skewness_arr.size(); i++) {
+	final_comp.at(i) = std::max(skewness_arr.at(i), mean_pn_ecdf.at(i));
+  }
+
+
+  //for(int i=0; i<mean_pn_ecdf.size(); i++)
+	//  verboseStream << "mean_pn_ecdf at " << i << ": " << mean_pn_ecdf.at(i) << std::endl;
 
   //Create COPOD score vector
-  std::vector<double> out_scores_i;
+  std::vector<double> out_scores_i = std::vector<double>(final_comp.size(), 0.0);
+  verboseStream << "m_alpha: " << m_alpha << std::endl;
+
   double min_score = -1 * log2(0.0 + m_alpha);
-  double max_score = -1 * log2(1.0 + m_alpha);
+  double max_score = log2(1.0 + m_alpha) - min_score;
+  verboseStream << "Initializaing min_score: " << min_score << ", max_score: " << max_score <<std::endl;
   verboseStream << "out_scores_i: " << std::endl;
-  for(int i=0; i < mean_pn_ecdf.size(); i++){
-    double l = -1 * log2(mean_pn_ecdf.at(i) + m_alpha);
-    out_scores_i.push_back(l);
-    //verboseStream << "Count: " << param[func_id].counts().at(i) << ", Probability: " << prob_counts.at(i) << ", score: "<< l << std::endl;
+  for(int i=0; i < final_comp.size(); i++){
+    double l = -1 * log2(final_comp.at(i) + m_alpha);
+    out_scores_i.at(i) = l;
+    verboseStream << "Final_comp at " << i << ": " << final_comp.at(i) << ", score: "<< l << std::endl;
     //if(prob_counts.at(i) > 0) {
       if(l < min_score){
         min_score = l;
@@ -648,9 +711,10 @@ unsigned long ADOutlierCOPOD::compute_outliers(Anomalies &outliers,
 
   //compute threshold
   verboseStream << "Global threshold before comparison with local threshold =  " << param[func_id].get_threshold() << std::endl;
-  double l_threshold = min_score + (m_threshold * (max_score - min_score));
+  double l_threshold = (max_score < 0) ? (-1 * m_threshold * (max_score - min_score)) : min_score + (m_threshold * (max_score - min_score));
+  verboseStream << "l_threshold computed: " << l_threshold << std::endl;
   if(m_use_global_threshold) {
-    if(l_threshold < param[func_id].get_threshold()) {
+    if(l_threshold < param[func_id].get_threshold() && param[func_id].get_threshold() > (-1 * log2(1.00001))) {
       l_threshold = param[func_id].get_threshold();
     } else {
       param[func_id].set_glob_threshold(l_threshold); //.get_histogram().glob_threshold = l_threshold;
@@ -659,9 +723,6 @@ unsigned long ADOutlierCOPOD::compute_outliers(Anomalies &outliers,
   }
 
   //Compute COPOD based score for each datapoint
-  //const double bin_width = param[func_id].bin_edges().at(1) - param[func_id].bin_edges().at(0);
-  //const int num_bins = param[func_id].counts().size();
-  //verboseStream << "Bin width: " << bin_width << std::endl;
 
   int top_out = 0;
   int running_idx = 0;
@@ -670,12 +731,21 @@ unsigned long ADOutlierCOPOD::compute_outliers(Anomalies &outliers,
 
       const double runtime_i = this->getStatisticValue(*itt); //runtimes.push_back(this->getStatisticValue(*itt));
       double ad_score;
-
-      //verboseStream << "mean_pn_ecdf.at(running_idx++): " << mean_pn_ecdf.at(running_idx) << std::endl;
-      if (mean_pn_ecdf.at(running_idx++) < 0.99)
-	      ad_score = l_threshold + 1;
-      else
-	      ad_score = l_threshold - 1;
+      
+      if (running_idx < final_comp.size()) {
+      	verboseStream << "final_comp.at(" << running_idx << "): " << final_comp.at(running_idx) << ", func_id: " << func_id << ", runtime: " << runtime_i << std::endl;
+     
+        if (out_scores_i.at(running_idx) > l_threshold) //(final_comp.at(running_idx) > 0) // < 0.9)
+	        ad_score = l_threshold + 1;
+        else
+	        ad_score = l_threshold - 1;
+        
+	running_idx++;
+      }
+      else {
+	recoverable_error("AD: COPOD: runtime Index");
+        continue;
+      }
 
       itt->set_outlier_score(ad_score);
       verboseStream << "ad_score: " << ad_score << ", l_threshold: " << l_threshold << std::endl;

src/pserver/AggregateFuncStats.cpp

@@ -4,6 +4,8 @@ using namespace chimbuko;
 
 AggregateFuncStats::AggregateFuncStats(int pid, int fid, const std::string &func): m_pid(pid), m_fid(fid), m_func(func){
   m_func_anomaly.set_do_accumulate(true); //accumulate number of anomalies as 'accumulate' field
+  m_inclusive.set_do_accumulate(true); //also accumulate total runtimes over all rank, thread
+  m_exclusive.set_do_accumulate(true);
 }
 
 void AggregateFuncStats::add(unsigned long n_anomaly, const RunStats& inclusive, const RunStats& exclusive){

src/util/RunStats.cpp

@@ -199,8 +199,11 @@ RunStats chimbuko::operator+(const RunStats &a, const RunStats &b)
     bool do_accumulate = false;
     if (a.m_do_accumulate || b.m_do_accumulate)
     {
-        sum_acc = a.m_state.acc + b.m_state.acc;
-        do_accumulate = true;
+      double a_acc = a.m_do_accumulate ? a.m_state.acc : a.m_state.eta * a.m_state.count; //reconstruct sum
+      double b_acc = b.m_do_accumulate ? b.m_state.acc : b.m_state.eta * b.m_state.count;
+      sum_acc = a_acc + b_acc;
+      //sum_acc = a.m_state.acc + b.m_state.acc;
+      do_accumulate = true;
     }
 
     RunStats combined(do_accumulate);

test/test_param.cpp

@@ -289,7 +289,7 @@ TEST(GlobalAnomalyStatsTest, AnomalyStatTest2)
     EXPECT_EQ(0, param.collect_func_data().size());
 
     std::unordered_map<unsigned long, RunStats> g_inclusive, g_exclusive;
-    RunStats l_inclusive, l_exclusive;
+    RunStats l_inclusive(true), l_exclusive(true); //accumulate total runtimes
 
     l_inclusive.clear();
     l_inclusive.push(10); l_inclusive.push(10.5); l_inclusive.push(9.5);

test/unit_tests/ad/COPODOutlierADs.cpp

@@ -28,7 +28,7 @@ class ADOutlierCOPODTest: public ADOutlierCOPOD{
 };
 
 
-TEST(HBOSADOutlierTestSyncParamWithPSComputeOutliers, Works){
+TEST(COPODADOutlierTestSyncParamWithPSComputeOutliers, Works){
   CopodParam global_params_ps; //parameters held in the parameter server
   CopodParam local_params_ad, local_params_ad2; //parameters collected by AD
 
@@ -49,7 +49,7 @@ TEST(HBOSADOutlierTestSyncParamWithPSComputeOutliers, Works){
     std::vector<double> runtimes;
     Histogram &r = local_params_ad[0];
     for(int i=0;i<N;i++) {
-      double val = i==N-1 ? 1000 : double(dist(gen));
+      double val = i==N-1 ? 10000 : double(dist(gen));
       call_list.push_back( createFuncExecData_t(0,0,0,  0, "my_func", 1000*(i+1), val) );
       runtimes.push_back(val);
       std::cout << "vals in localhist 1: " << val << std::endl;
@@ -65,7 +65,7 @@ TEST(HBOSADOutlierTestSyncParamWithPSComputeOutliers, Works){
     std::vector<double> runtimes;
     Histogram &r = local_params_ad2[0];
     for(int i=0;i<N;i++) {
-      double val = i==N-1 ? 1000 : double(dist(gen));
+      double val = i==N-1 ? 20000 : double(dist(gen));
       call_list2.push_back( createFuncExecData_t(0,0,0,  0, "my_func", 1000*(i+1), val) );
       runtimes.push_back(val);
       std::cout << "vals in localhist 2: " << val << std::endl;

test/unit_tests/ad/HBOSOutlierTestBPFile.cpp

@@ -115,7 +115,7 @@ TEST(HBOSADOutlierBPFileWithoutPServer, Works) {
     //Parameters for the connection to the instrumented binary trace output
     params.trace_engineType = "BPFile";  // BPFile or SST
     params.trace_data_dir = trace_data_dir; // *.bp location
-    std::string bp_prefix = "tau-metrics";  // bp file prefix (e.g. tau-metrics-[nwchem])
+    std::string bp_prefix = "tau-metrics"; //"tau-metrics-xgc-es-cpp-gpu"; // bp file prefix (e.g. tau-metrics-[nwchem])
 
     //The remainder are optional arguments. Enable using the appropriate command line switch
     params.program_idx = 0;

test/unit_tests/ad/SSTDOutlierTestBPFile.cpp

@@ -97,7 +97,7 @@ TEST(SSTDADOutlierBPFileWithoutPServer, Works) {
   if(_argc < 2){
     throw std::runtime_error("Path to trace data directory must be provided as an argument!");
   }
-  std::string trace_data_dir = _argv[1];
+  std::string trace_data_dir = _argv[1]; //"/home/src/chimbuko_ad/96rank_98step_traces_11_8_21/"; //_argv[1];
 
   //int file_suffix = 1;
   int ranks = 4;
@@ -116,7 +116,7 @@ TEST(SSTDADOutlierBPFileWithoutPServer, Works) {
     //Parameters for the connection to the instrumented binary trace output
     params.trace_engineType = "BPFile"; // argv[1]; // BPFile or SST
     params.trace_data_dir = trace_data_dir; // *.bp location
-    std::string bp_prefix = "tau-metrics"; //argv[3]; // bp file prefix (e.g. tau-metrics-[nwchem])
+    std::string bp_prefix = "tau-metrics"; //"tau-metrics-xgc-es-cpp-gpu"; //argv[3]; // bp file prefix (e.g. tau-metrics-[nwchem])
 
     //The remainder are optional arguments. Enable using the appropriate command line switch
     params.program_idx = 0;