Improved handling of events for which no global model information yet exists. The model can be empty if the current step is the first step on which a function has been seen, and/or if function data has not had time to pass through the parameter server. Previously such events were just labeled as normal, on which is reasonable in most cases. However this approach causes problems for GPU events, particularly for short-running programs, as the GPU event traces often appear as a big dump on a single step long after the function was executed, resulting in GPU events not being properly labeled. To combat this, events for which no model data is available will be left unlabeled (this is no longer cause for an error) and held until a future step where a model becomes available. Associated call-stack events are also maintained so as to ensure complete provenance output.

Author: giltirn

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

@@ -248,7 +248,7 @@ namespace chimbuko {
 	 * throws a runtime error if the call is not present in the call-list
 	 */
 	CallListIterator_t getCallData(const eventID &event_id) const override;
-
+	
 	/**
 	 * @brief Get a pair of iterators marking the start and one-past-the-end of a window of size (up to) win_size events
 	 *        on either size around the given event occurring on the same thread
@@ -329,6 +329,7 @@ namespace chimbuko {
 	  size_t n_kept_protected; /**< Number of calls maintained because they have been protected*/
 	  size_t n_kept_incomplete; /**< Number of calls maintained because they have not yet completed*/
 	  size_t n_kept_window; /**< Number of calls maintained because they may be needed for provenance window capture on next io step*/
+	  size_t n_kept_unlabeled; /**< Number of calls maintained because they have not yet been labeled*/
 	};
 
 
@@ -414,7 +415,7 @@ namespace chimbuko {
 	/**
 	 * @brief map of function index to an array of complete calls to this function during this IO step
 	 *
-	 * In practise this map is cleared every IO step by calls to trimCallList
+	 * In practise, labeled events are cleared every IO step by calls to purgeCallList
 	 */
 	ExecDataMap_t     m_execDataMap;
 
@@ -440,6 +441,12 @@ namespace chimbuko {
 	 */
 	std::unordered_set<std::string> m_ignoreCorrelationID;
 
+
+	/**
+	 * @brief Set of events that have been stack locked because they weren't able to be labeled
+	 */
+	std::unordered_set<eventID> m_stackLockedUnlabeled;
+	
 	/**
 	 * @brief verbose
 	 *

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

@@ -602,7 +602,7 @@ namespace chimbuko {
 	/**
 	 * @brief Determine whether the event can be deleted by the garbage collection at the end of the io step
 	 */
-	inline bool can_delete() const{ return m_references == 0;}
+	inline bool can_delete() const{ return m_references == 0 && m_label != 0;}
 
 	/**
 	 * @brief Increment the external reference counter, preventing object deletion

src/core/ad/ADDataInterface.cpp

@@ -8,10 +8,11 @@ void ADDataInterface::DataSetAnomalies::recordEvent(const Elem &event){
     //Store a max of one normal event. Select that with the lower score (more normal)
     if(m_normal_events.size() == 0) m_normal_events.push_back(event);    
     else if(event.score < m_normal_events[0].score) m_normal_events[0] = event;
-  }else{ //Outlier
+    ++m_labeled_events; //increment even if not recorded
+  }else if(event.label == EventType::Outlier){ //Outlier
     m_anomalies.push_back(event);
+    ++m_labeled_events; //increment even if not recorded
   }
-  ++m_labeled_events; //increment even if not recorded
 }
 
 size_t ADDataInterface::nEventsRecorded(EventType type) const{
@@ -29,9 +30,6 @@ size_t ADDataInterface::nEvents() const{
 }
 
 void ADDataInterface::recordDataSetLabels(const std::vector<Elem> &data, size_t dset_index){
-  //Check first for unassigned labels before we modify anything
-  for(auto const &e : data) if(e.label == EventType::Unassigned){ fatal_error("Encountered unassigned label when recording; the AD algorithm should label all data points assigned!"); }
-
   auto &danom = m_dset_anom[dset_index];
   for(auto const &e : data) danom.recordEvent(e);
   this->recordDataSetLabelsInternal(data,dset_index);

src/core/ad/ADOutlier.cpp

@@ -269,32 +269,22 @@ double ADOutlierSSTD::computeScore(double value, size_t model_idx, const SstdPar
   double std_dev = it->second.stddev();
   if(std_dev == 0.) std_dev = 1e-10; //distribution throws an error if std.dev = 0
 
-  //boost::math::normal_distribution<double> dist(mean, std_dev);
-  //double cdf_val = boost::math::cdf(dist, runtime); // P( X <= x ) for random variable X
-  //double score = std::min(cdf_val, 1-cdf_val); //two-tailed
-
-  //Using the CDF gives scores ~0 for basically any global outlier
-  //Instead we will use the difference in runtime compared to the avg in units of the standard deviation
+   //We use the difference in runtime compared to the avg in units of the standard deviation
   double score = fabs( value - mean ) / std_dev;
   return score;
 }
 
 
 void ADOutlierSSTD::labelData(std::vector<ADDataInterface::Elem> &data_vals, size_t dset_idx, size_t model_idx){
-
-  verboseStream << "Finding outliers in events for data set " << dset_idx << " of size " << data_vals.size() << std::endl;
-
+  verboseStream << "Finding outliers in events for data set " << dset_idx << " of size " << data_vals.size() << " and model idx " << model_idx << std::endl;
+  
   if(data_vals.size() == 0) return;
 
   SstdParam& param = *(SstdParam*)m_param;
   auto & fparam = param[model_idx];
 
-  if (fparam.count() < 2){
-    verboseStream << "Less than 2 events in stats associated with data set, stats not complete" << std::endl;
-    for(auto &e : data_vals){ //still need to label all events
-      e.label = ADDataInterface::EventType::Normal;
-      e.score = 0;
-    }
+  if (fparam.count() < 2){ //TODO: This will probably cause problems if we only ever see <2 calls to the function!
+    verboseStream << "Less than 2 events in stats associated with data set, stats not complete, delaying evaluation" << std::endl;
     return;
   }
 
@@ -387,9 +377,8 @@ void ADOutlierHBOS::labelData(std::vector<ADDataInterface::Elem> &data_vals, siz
   //Check that the histogram contains bins
   if(nbin == 0){
     //As the pserver global model update is delayed, initially the clients may receive an empty model from the pserver for this data set
-    //Given that the model at this stage is unreliable anyway, we simply skip the set and label the data as normal
+    //To combat this, we hold off on labeling the data until we get a global model back
     verboseStream << "Global model is empty, skipping outlier evaluation for data set " << dset_idx << std::endl;
-    for(auto &e : data_vals) e.label = ADDataInterface::EventType::Normal;
     return;
   }
 
@@ -628,8 +617,7 @@ inline double copod_score(const double value_i, const Histogram &hist, const His
 
 
 void ADOutlierCOPOD::labelData(std::vector<ADDataInterface::Elem> &data_vals, size_t dset_idx, size_t model_idx){
-  verboseStream << "Finding outliers in events for data set " << dset_idx << std::endl;
-  verboseStream << "data Size: " << data_vals.size() << std::endl;
+  verboseStream << "Finding outliers in events for data set " << dset_idx << " of size " << data_vals.size() << " and model idx " << model_idx << std::endl;
 
   if(data_vals.size() == 0) return;
 
@@ -648,9 +636,8 @@ void ADOutlierCOPOD::labelData(std::vector<ADDataInterface::Elem> &data_vals, si
   //Check that the histogram contains bins
   if(nbin == 0){
     //As the pserver global model update is delayed, initially the clients may receive an empty model from the pserver for this function
-    //Given that the model at this stage is unreliable anyway, we simply skip the function and label the data as normal
+    //To combat this we delay evaluation of the score until we get some data
     verboseStream << "Global model is empty, skipping outlier evaluation for data set " << dset_idx << std::endl;
-    for(auto &e : data_vals) e.label = ADDataInterface::EventType::Normal;
     return;
   }
 

src/modules/performance_analysis/ad/ADEvent.cpp

@@ -46,37 +46,51 @@ void ADEvent::stackProtectGC(CallListIterator_t it){
   verboseStream << "ADEvent::stackProtectGC incrementing register count of " << it->get_id().toString() << " to " << it->reference_count()+1 << std::endl;
   it->register_reference();
 
-  eventID parent = it->get_parent();
-  while(parent != eventID::root()){
-    CallListIterator_t pit;
-    try{
-      pit = getCallData(parent);
-    }catch(const std::exception &e){
-      recoverable_error("Could not find parent " + parent.toString() + " in call list due to : " + e.what());
-      break;
+  std::vector<eventID> parents(1, it->get_parent());
+  //Also lock GPU event parents
+  if(m_gpu_thread_Map != nullptr && m_gpu_thread_Map->count(it->get_tid()) && it->n_GPU_correlationID_partner() == 1){
+    parents.push_back(it->get_GPU_correlationID_partner(0));
+  }
+  
+  for(auto parent : parents){
+    while(parent != eventID::root()){
+      CallListIterator_t pit;
+      try{
+	pit = getCallData(parent);
+      }catch(const std::exception &e){
+	recoverable_error("Could not find parent " + parent.toString() + " in call list due to : " + e.what());
+	break;
+      }
+      verboseStream << "ADEvent::stackProtectGC incrementing register count of " << pit->get_id().toString() << " to " << pit->reference_count()+1 << std::endl;
+      pit->register_reference();
+      parent = pit->get_parent();
     }
-    verboseStream << "ADEvent::stackProtectGC incrementing register count of " << pit->get_id().toString() << " to " << pit->reference_count()+1 << std::endl;
-    pit->register_reference();
-    parent = pit->get_parent();
   }
 }
 
 void ADEvent::stackUnProtectGC(CallListIterator_t it){
   verboseStream << "ADEvent::stackUnProtectGC decrementing register count of " << it->get_id().toString() << " to " << it->reference_count()-1 << std::endl;
   it->deregister_reference();
 
-  eventID parent = it->get_parent();
-  while(parent != eventID::root()){
-    CallListIterator_t pit;
-    try{
-      pit = getCallData(parent);
-    }catch(const std::exception &e){
-      recoverable_error("Could not find parent " + parent.toString() + " in call list due to : " + e.what());
-      break;
+  std::vector<eventID> parents(1, it->get_parent());
+  //Also unlock GPU event parents
+  if(m_gpu_thread_Map != nullptr && m_gpu_thread_Map->count(it->get_tid()) && it->n_GPU_correlationID_partner() == 1){
+    parents.push_back(it->get_GPU_correlationID_partner(0));
+  }
+  
+  for(auto parent : parents){
+    while(parent != eventID::root()){
+      CallListIterator_t pit;
+      try{
+	pit = getCallData(parent);
+      }catch(const std::exception &e){
+	recoverable_error("Could not find parent " + parent.toString() + " in call list due to : " + e.what());
+	break;
+      }
+      verboseStream << "ADEvent::stackUnProtectGC decrementing register count of " << pit->get_id().toString() << " to " << pit->reference_count()-1 << std::endl;
+      pit->deregister_reference();
+      parent = pit->get_parent();
     }
-    verboseStream << "ADEvent::stackUnProtectGC decrementing register count of " << pit->get_id().toString() << " to " << pit->reference_count()-1 << std::endl;
-    pit->deregister_reference();
-    parent = pit->get_parent();
   }
 }
 
@@ -366,6 +380,17 @@ static unsigned long nested_map_size(const T& m) {
   return n_elements;
 }
 
+static void clearUnlabeled(ExecDataMap_t &execDataMap){
+  for(auto &fid_p : execDataMap){
+    std::vector<CallListIterator_t> keep;
+    for(const CallListIterator_t &cit : fid_p.second)
+      if(cit->get_label() == 0) keep.push_back(cit);
+    fid_p.second = std::move(keep);
+    verboseStream << "clearUnlabeled kept " << fid_p.second.size() << " unlabeled points for fid " << fid_p.first << std::endl;
+  }
+}
+
+
 CallListMap_p_t* ADEvent::trimCallList(int n_keep_thread) {
   //Remove completed entries from the call list
   CallListMap_p_t* cpListMap = new CallListMap_p_t;
@@ -400,13 +425,13 @@ CallListMap_p_t* ADEvent::trimCallList(int n_keep_thread) {
       }
     }
   }
-  m_execDataMap.clear();
+  clearUnlabeled(m_execDataMap);
   return cpListMap;
 }
 
 
 void ADEvent::purgeCallList(int n_keep_thread, purgeReport* report) {
-  size_t n_purged=0, n_kept_protected=0, n_kept_incomplete=0, n_kept_window=0;
+  size_t n_purged=0, n_kept_protected=0, n_kept_incomplete=0, n_kept_window=0, n_kept_unlabeled=0;
 
   //Remove completed entries from the call list
   for (auto& it_p : m_callList) {
@@ -425,6 +450,14 @@ void ADEvent::purgeCallList(int n_keep_thread, purgeReport* report) {
 	auto one_past_last = std::prev(cl.end(),n_keep_thread);
 
 	while (it != one_past_last) {
+	  //Check if entry and stack was locked due to being unlabeled, and if now labeled, unlock
+	  std::unordered_set<eventID>::iterator sl_it;
+	  if(it->get_label() != 0 && (sl_it = m_stackLockedUnlabeled.find(it->get_id()) ) != m_stackLockedUnlabeled.end()){
+	    verboseStream << "Previously unlabeled event " << it->get_id().toString() << " has now been labeled, unlocking stack" << std::endl;
+	    stackUnProtectGC(it);
+	    m_stackLockedUnlabeled.erase(sl_it);
+	  }
+	  
 	  if (it->can_delete() && it->get_exit() != 0) {
 	    //Remove completed event from map of event index string to call list
 	    m_callIDMap.erase(it->get_id());
@@ -434,7 +467,16 @@ void ADEvent::purgeCallList(int n_keep_thread, purgeReport* report) {
 	  }
 	  else {
 	    if(it->get_exit() == 0) ++n_kept_incomplete;
-	    else if(!it->can_delete()) ++n_kept_protected;	    
+	    else if(!it->can_delete()){
+	      if(it->get_label() == 0 && m_stackLockedUnlabeled.count(it->get_id()) == 0){
+		//Lock the stack to ensure that events referred to don't get erased
+		verboseStream << "Event " << it->get_id().toString() << " is not labeled, locking stack" << std::endl;
+		stackProtectGC(it);
+		m_stackLockedUnlabeled.insert(it->get_id());		
+		++n_kept_unlabeled;
+	      }
+	      else ++n_kept_protected;
+	    }
 	    it++;
 	  }
 	}
@@ -446,9 +488,9 @@ void ADEvent::purgeCallList(int n_keep_thread, purgeReport* report) {
     report->n_kept_protected = n_kept_protected;
     report->n_kept_incomplete = n_kept_incomplete;
     report->n_kept_window = n_kept_window;
+    report->n_kept_unlabeled = n_kept_unlabeled;
   }
-
-  m_execDataMap.clear();
+  clearUnlabeled(m_execDataMap);
 }
 
 size_t ADEvent::getCallListSize() const{

src/modules/performance_analysis/ad/ADExecDataInterface.cpp

@@ -80,7 +80,8 @@ void ADExecDataInterface::recordDataSetLabelsInternal(const std::vector<Elem> &d
   for(auto const &e: data){
     CallListIterator_t eint = it->second[e.index];
     eint->set_outlier_score(e.score);
-    if(e.label == EventType::Unassigned){ fatal_error("Encountered an unassigned label when recording"); }
-    eint->set_label( e.label == EventType::Outlier ? -1 : 1 );
+    if(e.label != EventType::Unassigned){
+      eint->set_label( e.label == EventType::Outlier ? -1 : 1 );
+    }
   }
 }