Fixed a bug where empty data sets in ExecData_map would lead to ADLocalFuncStatistics sending uninitialized function stats data objects to the pserver and hence to the global database

Author: giltirn

include/chimbuko/modules/performance_analysis/ad/ADExecDataInterface.hpp

@@ -56,7 +56,7 @@ namespace chimbuko {
 	/**
 	 * @brief Return the function index associated with a given data set
 	 */
-	size_t getDataSetModelIndex(size_t dset_index) const override{ return m_dset_fid_map[dset_index]; }
+	size_t getDataSetModelIndex(size_t dset_index) const override;
 
 	/**
 	 * @brief Return the data set index associated with a given function index

src/core/chimbuko.cpp

@@ -335,7 +335,7 @@ bool ChimbukoBase::runFrame(){
 #endif
 
     int nout = iface->nEventsRecorded(ADDataInterface::EventType::Outlier);
-    int nnormal = iface->nEvents() - nout; //this is the total number of normal events, not just of those that were recorded
+    int nnormal = iface->nEvents() - nout; //this is the total number of normal events (also unlabeled events), not just of those that were recorded
     m_run_stats.n_outliers += nout;
     m_accum_prd.n_outliers += nout;
 

src/modules/performance_analysis/ad/ADAnomalyProvenance.cpp

@@ -236,6 +236,8 @@ void ADAnomalyProvenance::getProvenanceEntries(std::vector<nlohmann::json> &anom
     for(auto const &e : iface.getResults(dset_idx).getEventsRecorded(ADDataInterface::EventType::Outlier)){
       timer2.start();
       auto anom_it = iface.getExecDataEntry(dset_idx, e.index);
+
+      //verboseStream << "Collecting provenance for anomaly on dset " << dset_idx << " index " << e.index << " with content:\n" << anom_it->get_json().dump(2) << std::endl;
 
       if(anom_it->get_exclusive() < m_min_anom_time) continue; //skip executions with too short runtimes to avoid filling the database with irrelevant anomalies
 

src/modules/performance_analysis/ad/ADExecDataInterface.cpp

@@ -9,6 +9,12 @@ ADExecDataInterface::ADExecDataInterface(ExecDataMap_t const* execDataMap, Outli
   size_t dset_idx = 0;
   for(auto it = execDataMap->begin(); it != execDataMap->end(); ++it)
     m_dset_fid_map[dset_idx++] = it->first;
+
+  if(enableVerboseLogging()){
+    std::cout << "ADExecDataInterface created with #datasets=" << this->nDataSets() << " (exec-data map size " << execDataMap->size() << ")" << " with dset_idx:fid mapping ";
+    for(int d=0;d<this->nDataSets();d++) std::cout << d << ":" << m_dset_fid_map[d] << " ";
+    std::cout << std::endl;
+  }
 }
 
 double ADExecDataInterface::getStatisticValue(const ExecData_t &e) const{
@@ -30,6 +36,14 @@ void ADExecDataInterface::setIgnoreFunction(const std::string &func){
   m_func_ignore.insert(func);
 }
 
+size_t ADExecDataInterface::getDataSetModelIndex(size_t dset_index) const{
+  if(dset_index >= m_dset_fid_map.size()){
+    std::string err = "Invalid dset_index " +std::to_string(dset_index);
+    fatal_error(err);
+  }
+  return m_dset_fid_map[dset_index];
+}
+
 size_t ADExecDataInterface::getDataSetIndexOfFunction(size_t fid) const{
   for(size_t dset_idx = 0; dset_idx < m_dset_fid_map.size(); dset_idx++)
     if(m_dset_fid_map[dset_idx] == fid) return dset_idx;
@@ -39,6 +53,7 @@ size_t ADExecDataInterface::getDataSetIndexOfFunction(size_t fid) const{
 CallListIterator_t ADExecDataInterface::getExecDataEntry(size_t dset_index, size_t elem_index) const{
   auto it = m_execDataMap->find(m_dset_fid_map[dset_index]);
   if(it == m_execDataMap->end()){ fatal_error("Invalid dset_idx"); }
+  if(elem_index >= it->second.size()){ fatal_error("Invalid elem_index"); }
   return it->second[elem_index];
 }
 
@@ -78,6 +93,7 @@ void ADExecDataInterface::recordDataSetLabelsInternal(const std::vector<Elem> &d
   auto it = m_execDataMap->find(m_dset_fid_map[dset_index]);
   if(it == m_execDataMap->end()){ fatal_error("Invalid dset_idx"); }
   for(auto const &e: data){
+    if(e.index >= it->second.size()){ fatal_error("Invalid element index"); }
     CallListIterator_t eint = it->second[e.index];
     eint->set_outlier_score(e.score);
     if(e.label != EventType::Unassigned){

src/modules/performance_analysis/ad/ADLocalFuncStatistics.cpp

@@ -2,6 +2,7 @@
 #include <chimbuko/modules/performance_analysis/ad/AnomalyData.hpp>
 #include <chimbuko/modules/performance_analysis/pserver/PScommon.hpp>
 #include <chimbuko/core/util/serialize.hpp>
+#include <chimbuko/core/verbose.hpp>
 #include <climits>
 #include <algorithm>
 
@@ -23,7 +24,7 @@ void ADLocalFuncStatistics::gatherStatistics(const ExecDataMap_t* exec_data){
     //Create new entry if it doesn't exist
     if(fstats_it == m_funcstats.end()){
       const std::string &name = it.second.front()->get_funcname(); //it.second has already been checked to have size >= 1
-      fstats_it = m_funcstats.insert( std::unordered_map<unsigned long, FuncStats>::value_type(func_id, FuncStats(m_anom_data.get_app(), func_id, name)) ).first;
+      fstats_it = m_funcstats.insert( std::unordered_map<unsigned long, FuncStats>::value_type(func_id, FuncStats(m_anom_data.get_app(), func_id, name)) ).first;     
     }
 
     for (auto itt : it.second) { //loop over events for that function
@@ -33,6 +34,8 @@ void ADLocalFuncStatistics::gatherStatistics(const ExecDataMap_t* exec_data){
       min_ts = std::min(min_ts, static_cast<unsigned long>(itt->get_entry()) );
       max_ts = std::max(max_ts, static_cast<unsigned long>(itt->get_exit()));
     }
+
+    verboseStream << "ADLocalFuncStatistics::gatherStatistics generated stats for func_id " << func_id << ":\n" << fstats_it->second.get_json().dump(2) << std::endl;
   }
 
   m_anom_data.set_min_ts(min_ts);
@@ -42,13 +45,29 @@ void ADLocalFuncStatistics::gatherStatistics(const ExecDataMap_t* exec_data){
 void ADLocalFuncStatistics::gatherAnomalies(const ADExecDataInterface &iface){
   //Gather information on the number of anomalies and stats on their scores
   size_t nanom_tot = 0;
+  verboseStream << "ADLocalFuncStatistics::gatherAnomalies processing anomalies for " << iface.nDataSets() << std::endl;
+  
   for(size_t dset_idx =0 ; dset_idx < iface.nDataSets(); dset_idx++){
     size_t fid = iface.getDataSetModelIndex(dset_idx);
     auto const & r = iface.getResults(dset_idx);
+
+    //Because some data sets may be empty or contain events that don't get labeled this step we should skip those data sets
+    //This is particularly important for empty data sets as these do not have entries in m_funcstats, hence trying to push anomaly counts (even 0) will lead to uninitialized entries
+    if(r.nEventsRecorded(ADDataInterface::EventType::Outlier) == 0 && r.nEventsRecorded(ADDataInterface::EventType::Normal) == 0)
+      continue;    
+    
     auto const &anom = r.getEventsRecorded(ADDataInterface::EventType::Outlier);
     size_t nanom = anom.size();
     nanom_tot += nanom;
-    m_funcstats[fid].n_anomaly += nanom; //increment func anomalies count                                                                                                                                                             
+
+    verboseStream << "ADLocalFuncStatistics::gatherAnomalies for dset_idx=" << dset_idx << " and fid=" << fid << " with " << nanom << " anomalies" << std::endl;
+
+    auto fit = m_funcstats.find(fid);
+    if(fit == m_funcstats.end()){
+      std::stringstream ss; ss << "ADLocalFuncStatistics::gatherAnomalies cannot find funcstats entry for function index " << fid << " with " << nanom << " anomalies";
+      fatal_error(ss.str());
+    }    
+    fit->second.n_anomaly += nanom; //increment func anomalies count                                                                                                                                                             
     for(auto const &e : anom) m_anom_data.add_outlier_score(e.score);
   }
   m_anom_data.incr_n_anomalies(nanom_tot);
@@ -64,6 +83,7 @@ nlohmann::json ADLocalFuncStatistics::get_json() const{
     g_info["func"].push_back(e.second.get_json());
   }
   g_info["anomaly"] = m_anom_data.get_json();
+
   return g_info;
 }
 

src/modules/performance_analysis/ad/ADcombinedPSdata.cpp

@@ -1,6 +1,7 @@
 #include<chimbuko/modules/performance_analysis/ad/ADcombinedPSdata.hpp>
 #include <chimbuko/modules/performance_analysis/pserver/PScommon.hpp>
 #include<chimbuko/core/util/serialize.hpp>
+#include<chimbuko/core/verbose.hpp>
 
 using namespace chimbuko;
 using namespace chimbuko::modules::performance_analysis;
@@ -87,6 +88,13 @@ std::pair<size_t, size_t> ADcombinedPSdataArray::send(ADNetClient &net_client) c
     latest_step = std::max(step, latest_step);
   }
 
+  if(enableVerboseLogging()){
+    for(ADLocalFuncStatistics const &f : m_func_stats){
+      std::cout << "ADcombinedPSdataArray::send sending func stats\n" << f.get_json().dump(2) << std::endl;
+    }
+
+  }
+  
   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.setContent(net_serialize());

src/modules/performance_analysis/chimbuko.cpp

@@ -358,6 +358,10 @@ void Chimbuko::bufferStorePSdata(const ADDataInterface &anomalies,  const int st
     ADLocalFuncStatistics func_stats(m_program_idx, rank, step, &perf);
     func_stats.gatherStatistics(m_event->getExecDataMap());
     func_stats.gatherAnomalies(ei);
+
+    verboseStream << "Chimbuko::bufferStorePSdata func_stats has " << func_stats.getFuncStats().size() << " entries" << std::endl;
+    verboseStream << "Chimbuko::bufferStorePSdata generated func_stats " << func_stats.get_json().dump(2) << std::endl;
+    
     m_funcstats_buf.emplace_back(std::move(func_stats));
     perf.add("ad_gather_ps_data_gather_profile_stats_time_ms", timer.elapsed_ms());
 
@@ -374,6 +378,8 @@ void Chimbuko::bufferStorePSdata(const ADDataInterface &anomalies,  const int st
     m_anom_metrics_buf.emplace_back(std::move(metrics));
     perf.add("ad_gather_ps_data_gather_metrics_time_ms", timer.elapsed_ms());
     perf.add("ad_gather_ps_data_total_time_ms", timer.elapsed_ms());
+
+    verboseStream << "Chimbuko::bufferStorePSdata buffered func_stats " << m_funcstats_buf.back().get_json().dump(2) << std::endl;
   }
 }
 

src/modules/performance_analysis/pserver/GlobalAnomalyStats.cpp

@@ -72,6 +72,9 @@ void GlobalAnomalyStats::add_anomaly_data(const ADLocalFuncStatistics& data){
 
   for (auto const &fp: data.getFuncStats()) {
     const FuncStats &f = fp.second;
+
+    verboseStream << "GlobalAnomalyStats adding FuncStats " << f.get_json().dump(2) << std::endl;
+    
     update_func_stat(f.pid, f.id, f.name, f.n_anomaly, f.inclusive, f.exclusive);
   }
 }