Moved generation of JSON from AggregatedAnomalyData from GlobalAnomalyStats to a method of AggregatedAnomalyData

Improved thread safety of GlobalAnomalyStats and marked functions in header as thread safe or not thread safe
Made outlier_severity a method of ExecData

Author: giltirn

include/chimbuko/ad/ExecData.hpp

@@ -482,10 +482,15 @@ class ExecData_t {
     unsigned long get_n_counter() const { return m_counters.size(); }
 
     /**
-     * @brief Return the outlier score assigned to the data point
+     * @brief Return the outlier score assigned to the data point, representing how unlikely an event is
      */
     double get_outlier_score() const{ return m_score; }
 
+    /**
+     * @brief Return the outlier severity, representing how important the outlier is
+     */  
+    double get_outlier_severity() const{ return (double)get_runtime(); }   
+
     /**
      * @brief Set the label
      *

include/chimbuko/pserver/AggregateAnomalyData.hpp

@@ -64,6 +64,13 @@ namespace chimbuko {
      */
     std::list<AnomalyData>* get_data() const;
 
+    /**
+     * @brief Get the current statistics in JSON format and flush the list of AnomalyData collected since previous call to get()
+     * @param pid The program index
+     * @param rid The rank index
+     */
+    nlohmann::json get_json_and_flush(int pid, int rid);
+
     /**
      * @brief Return the number of JSON-formatted strings of serialized incoming AnomalyData since the last flush
      */

include/chimbuko/pserver/GlobalAnomalyStats.hpp

@@ -23,6 +23,8 @@ namespace chimbuko{
 
     /**
      * @brief Merge internal statistics with those contained within the input ADLocalFuncStatistics object
+     *
+     * Thread safe
      */
     void add_anomaly_data(const ADLocalFuncStatistics& data);
 
@@ -31,13 +33,17 @@ namespace chimbuko{
      * @brief Get the JSON-formatted string corresponding to the anomaly statistics (RunStats instance) for a given program/rank
      * @param pid program index
      * @param rid rank
+     *
+     * Thread safe
      */
     std::string get_anomaly_stat(const int pid, const unsigned long rid) const;
 
     /**
      * @brief Get the RunStats object corresponding to the anomaly statistics for a given program/rank (throw error if not present)
      * @param pid program index
      * @param rid rank
+     * 
+     * Thread safe
      */
     RunStats get_anomaly_stat_obj(const int pid, const unsigned long rid) const;
 
@@ -46,19 +52,25 @@ namespace chimbuko{
      * @brief Const accessor to the AggregateAnomalyData instance corresponding to a particular stat_id (throw error if not present)
      * @param pid program index
      * @param rid rank
+     *
+     * NOT thread safe
      */    
     const AggregateAnomalyData & get_anomaly_stat_container(const int pid, const unsigned long rid) const;
 
     /**
      * @brief Get the number of anomaly data objects collected since the last flush for a given program/rank
      * @param pid program index
      * @param rid rank
+     *
+     * Thread safe
      */
     size_t get_n_anomaly_data(const int pid, const unsigned long rid) const;
 
     /**
      * @brief Update internal data to include additional information
-     * @parag anom The anomaly data
+     * @param anom The anomaly data
+     *
+     * Thread safe
      */
     void update_anomaly_stat(const AnomalyData &anom);
 
@@ -70,6 +82,8 @@ namespace chimbuko{
      * @param n_anomaly The number of anomalies detected
      * @param inclusive Statistics on inclusive timings
      * @param exclusive Statistics on exclusive timings
+     *
+     * Thread safe
      */
     void update_func_stat(int pid,
 			  unsigned long fid, 
@@ -82,32 +96,44 @@ namespace chimbuko{
      * @brief Get the FuncStats object containing the profile information for the specified function
      * @param pid Program index
      * @param fid Function index
+     *
+     * NOT thread safe
      */
     const AggregateFuncStats & get_func_stats(int pid, unsigned long fid) const;
 
     /**
      * @brief Collect anomaly statistics into JSON object and flush the m_anomaly_stats statistics
+     *
+     * Thread safe
      */
     nlohmann::json collect_stat_data();
 
     /**
      * @brief Collect aggregated function profile statistics into JSON object
+     *
+     * Thread safe
      */
     nlohmann::json collect_func_data() const;
 
     /**
      * @brief Collect anomaly statistics and function statistics. Flushes the m_anomaly_stats statistics
      * @return JSON object containing anomaly and function data
+     *
+     * Thread safe
      */
     nlohmann::json collect();
 
     /**
      * @brief Comparison operator
+     *
+     * NOT thread safe
      */
     bool operator==(const GlobalAnomalyStats &r) const{ return m_anomaly_stats == r.m_anomaly_stats && m_funcstats == r.m_funcstats; }
 
     /**
      * @brief Inequality operator
+     *
+     * NOT thread safe
      */
     bool operator!=(const GlobalAnomalyStats &r) const{ return !( *this == r ); }
 

src/ad/ADAnomalyProvenance.cpp

@@ -179,7 +179,7 @@ nlohmann::json ADAnomalyProvenance::get_json() const{
 					  {"io_step_tstart", m_io_step_tstart},
 					    {"io_step_tend", m_io_step_tend},
 					      {"outlier_score", m_call.get_outlier_score() },
-						{"outlier_severity", m_call.get_runtime() },
+						{"outlier_severity", m_call.get_outlier_severity() },
 						  {"hostname", getHostname() }
   };
 }

src/pserver/AggregateAnomalyData.cpp

@@ -1,4 +1,5 @@
 #include "chimbuko/pserver/AggregateAnomalyData.hpp"
+#include "chimbuko/util/string.hpp"
 #include <sstream>
 
 using namespace chimbuko;
@@ -89,3 +90,36 @@ bool AggregateAnomalyData::operator==(const AggregateAnomalyData &r) const{
   if(*m_data != *r.m_data) return false;
   return true;
 }
+
+nlohmann::json AggregateAnomalyData::get_json_and_flush(int pid, int rid){
+  auto stats = this->get(); //returns a std::pair<RunStats, std::list<AnomalyData>*>,  and flushes the state of pair.second. 
+  //We now own the std::list<AnomalyData>* pointer and have to delete it
+      
+  nlohmann::json object;
+
+  if(stats.second){
+    //Decide whether to include the data for this pid/rid
+    //Do this only if any anomalies were seen since the last call
+    bool include = false;
+    for(const AnomalyData &adata: *stats.second){
+      if(adata.get_n_anomalies() > 0){
+	include = true;
+	break;
+      }
+    }
+
+    if(include){
+      object["key"] = stringize("%d:%d", pid,rid);
+      object["stats"] = stats.first.get_json(); //statistics on anomalies to date for this pid/rid
+      
+      object["data"] = nlohmann::json::array();
+      for (const AnomalyData &adata: *stats.second){
+	//Don't include data for which there are no anomalies
+	if(adata.get_n_anomalies()>0)
+	  object["data"].push_back(adata.get_json());
+      }
+    }
+    delete stats.second;
+  }
+  return object;
+}

src/pserver/GlobalAnomalyStats.cpp

@@ -42,68 +42,44 @@ const AggregateAnomalyData & GlobalAnomalyStats::get_anomaly_stat_container(cons
 }
 
 RunStats GlobalAnomalyStats::get_anomaly_stat_obj(const int pid, const unsigned long rid) const{
+  std::lock_guard<std::mutex> _(m_mutex_anom);
   return get_anomaly_stat_container(pid, rid).get_stats();  
 }
 
 std::string GlobalAnomalyStats::get_anomaly_stat(const int pid, const unsigned long rid) const{
-  RunStats stat;
-  try{
-    stat = get_anomaly_stat_obj(pid,rid);
-  }catch(const std::exception &e){
-    return "";
-  }
-  return stat.get_json().dump();
+  std::lock_guard<std::mutex> _(m_mutex_anom);
+  auto pit = m_anomaly_stats.find(pid);
+  if(pit == m_anomaly_stats.end()) return "";
+  auto rit = pit->second.find(rid);
+  if(rit == pit->second.end()) return "";
+  return rit->second.get_stats().get_json().dump();
 }
 
 size_t GlobalAnomalyStats::get_n_anomaly_data(const int pid, const unsigned long rid) const{
-  AggregateAnomalyData const* s;
-  try{
-    s = &get_anomaly_stat_container(pid,rid);
-  }catch(const std::exception &e){
-    return 0;
-  }  
-  return s->get_n_data();
+  std::lock_guard<std::mutex> _(m_mutex_anom);
+  auto pit = m_anomaly_stats.find(pid);
+  if(pit == m_anomaly_stats.end()) return 0;
+  auto rit = pit->second.find(rid);
+  if(rit == pit->second.end()) return 0;
+  return rit->second.get_n_data();
 }
 
 nlohmann::json GlobalAnomalyStats::collect_stat_data(){
   nlohmann::json jsonObjects = nlohmann::json::array();
-
-  //m_anomaly_stats is a map of app_idx/rank to AggregateAnomalyData instances
-  //AggregateAnomalyData contains statistics on the number of anomalies found per io step and also a set of AnomalyData objects
-  //that have been collected from that rank since the last flush
-  for(auto & pp : m_anomaly_stats){
-    int pid = pp.first; //pid
-    for(auto & rp: pp.second){
-      unsigned long rid = rp.first; //rank
-      
-      auto stats = rp.second.get(); //returns a std::pair<RunStats, std::list<AnomalyData>*>,  and flushes the state of pair.second. 
-      //We now own the std::list<AnomalyData>* pointer and have to delete it
-      
-      if(stats.second){
-	//Decide whether to include the data for this pid/rid
-	//Do this only if any anomalies were seen since the last call
-	bool include = false;
-	for(const AnomalyData &adata: *stats.second){
-	  if(adata.get_n_anomalies() > 0){
-	    include = true;
-	    break;
-	  }
-	}
-
-	if(include){
-	  nlohmann::json object;
-	  object["key"] = stringize("%d:%d", pid,rid);
-	  object["stats"] = stats.first.get_json(); //statistics on anomalies to date for this pid/rid
-	  
-	  object["data"] = nlohmann::json::array();
-	  for (const AnomalyData &adata: *stats.second){
-	    //Don't include data for which there are no anomalies
-	    if(adata.get_n_anomalies()>0)
-	      object["data"].push_back(adata.get_json());
-	  }
-	  jsonObjects.push_back(object);
-	}
-	delete stats.second;
+  {
+    std::lock_guard<std::mutex> _(m_mutex_anom);
+
+    //m_anomaly_stats is a map of app_idx/rank to AggregateAnomalyData instances
+    //AggregateAnomalyData contains statistics on the number of anomalies found per io step and also a set of AnomalyData objects
+    //that have been collected from that rank since the last flush
+    for(auto & pp : m_anomaly_stats){
+      int pid = pp.first; //pid
+      for(auto & rp: pp.second){
+	unsigned long rid = rp.first; //rank
+	
+	nlohmann::json object = rp.second.get_json_and_flush(pid,rid);
+	if(!object.empty())
+	  jsonObjects.push_back(std::move(object));
       }
     }
   }
@@ -153,7 +129,6 @@ void GlobalAnomalyStats::update_func_stat(int pid, unsigned long fid, const std:
 }
 
 const AggregateFuncStats & GlobalAnomalyStats::get_func_stats(int pid, unsigned long fid) const{
-  std::lock_guard<std::mutex> _(m_mutex_func);
   auto pit = m_funcstats.find(pid);
   if(pit == m_funcstats.end()) fatal_error("Could not find program index");
   auto fit = pit->second.find(fid);

test/unit_tests/unit_test_common.hpp

@@ -10,6 +10,7 @@
 #include <chimbuko/message.hpp>
 #include <zmq.h>
 #include <chimbuko/util/barrier.hpp>
+#include <chimbuko/util/RunStats.hpp>
 
 namespace chimbuko{
   /**
@@ -404,4 +405,21 @@ namespace chimbuko{
   };
 
 
+  bool compare(const RunStats &a, const RunStats &b, const double tol = 1e-12){
+    RunStats::RunStatsValues sa = a.get_stat_values(), sb = b.get_stat_values();
+    bool ret = true;
+#define COM(A) if(fabs( sa. A - sb. A ) > tol){ std::cout << #A << " " << sa. A << " " << sb. A << std::endl; ret = false; }
+    COM(count);
+    COM(minimum);
+    COM(maximum);
+    COM(accumulate);
+    COM(mean);
+    COM(stddev);
+    COM(skewness);
+    COM(kurtosis);
+    return ret;
+#undef COM
+  }
+
+
 }