Reworked how io steps are treated in the AD sim to resolve logic errors dealing with function executions that span io steps.

Now the user generates the full program trace and iterates through the IO steps afterwards.

Author: giltirn

sim/include/sim/ad.hpp

@@ -5,6 +5,7 @@
 #include<chimbuko/ad/ADNormalEventProvenance.hpp>
 #include<chimbuko/ad/ADEvent.hpp>
 #include<chimbuko/ad/ADCounter.hpp>
+#include<map>
 
 namespace chimbuko_sim{
   using namespace chimbuko;
@@ -13,32 +14,41 @@ namespace chimbuko_sim{
 
   //An object that represents a rank of the AD
   class ADsim{
-    std::unordered_map<unsigned long, std::list<ExecData_t> > m_all_execs; //map of thread to execs, never flushed
-    std::unordered_map<eventID, CallListIterator_t> m_eventid_map; //map from event index to iterator ; never flushed
-    std::list<CallListIterator_t> m_step_exec_its; //iterators to events on this io step; flushed at end of step
-    std::unordered_map<unsigned long, std::vector<CallListIterator_t>> m_step_func_exec_its; //map of function id to iterators on this io step; flushed at end of step
+    std::unordered_map<unsigned long, std::list<ExecData_t> > m_all_execs; /**< Map of thread to execs */
+    std::unordered_map<eventID, CallListIterator_t> m_eventid_map; /**< Map from event index to iterator */
+    std::map<unsigned long, std::list<CallListIterator_t> > m_step_execs; /**< Events in any given step */
+
+    unsigned long m_largest_step; /**< Largest step index of an inserted function execution thus far encountered */
+    unsigned long m_program_start;
+    unsigned long m_step_freq; /**< Frequency of IO steps */
 
-    bool m_step_is_open; /**< Is an io step currently open? */
-    int m_step;
     int m_window_size;
     int m_pid;
     int m_rid;
-    unsigned long m_step_start_time;
     ADNormalEventProvenance m_normal_events;
-    ADCounter m_counters;
     ADMetadataParser m_metadata;
     std::unique_ptr<ADProvenanceDBclient> m_pdb_client;
   public:
-    void init(int window_size, int pid, int rid);
+    /** 
+     * @brief Initialize the AD simulator
+     * @param window_size The number of events around an anomaly that are recorded in the provDB
+     * @param pid The program index
+     * @param rid The rank index
+     * @param program_start Timestamp of program start
+     * @param step_freq Frequency at which IO steps are to occur
+     */
+    void init(int window_size, int pid, int rid, unsigned long program_start, unsigned long step_freq);
 
     /** 
      * @brief Instantiate the AD simulator
      * @param window_size The number of events around an anomaly that are recorded in the provDB
      * @param pid The program index
      * @param rid The rank index
+     * @param program_start Timestamp of program start
+     * @param step_freq Frequency at which IO steps are to occur
      */
-    ADsim(int window_size, int pid, int rid): ADsim(){
-      init(window_size, pid, rid);
+    ADsim(int window_size, int pid, int rid, unsigned long program_start, unsigned long step_freq): ADsim(){
+      init(window_size, pid, rid, program_start, step_freq);
     }
     ADsim(){}
 
@@ -101,16 +111,26 @@ namespace chimbuko_sim{
     void bindCPUparentGPUkernel(CallListIterator_t cpu_parent, CallListIterator_t gpu_kern);
 
     /**
-     * @brief Begin an IO step at the provided time
+     * @brief Bind a parent and child execution
      */
-    void beginStep(const unsigned long step_start_time);
+    static void bindParentChild(CallListIterator_t parent, CallListIterator_t child);
 
     /**
-     * @brief End the IO step at the provided time
+     * @brief Analyze the given IO step, sending the data to the pserver/provdb
      * 
      * When this is called the executions added since the start of the step will be gathered and analyzed, and the data send to the pserver/provdb
      */
-    void endStep(const unsigned long step_end_time);
+    void step(const unsigned long step);
+
+    /**
+     * @brief Get the largest step index in the trace
+     */
+    unsigned long largest_step() const{ return m_largest_step; }
+
+    /**
+     * @brief Get the number of executions in a given step
+     */
+    size_t nStepExecs(unsigned long step) const;
   };
 
   /**

sim/main/example1.cpp

@@ -9,12 +9,14 @@ int main(int argc, char **argv){
 
   int window_size = 5; //number of events to record around an anomaly in the provenance data
   int pid = 0; //program index
-  
+  unsigned long program_start = 100;
+  unsigned long step_freq = 1000;
+
   //Setup the "AD" instances
   int n_ranks = 1;
   std::vector<ADsim> ad;
   for(int r=0;r<n_ranks;r++)
-    ad.push_back(ADsim(window_size, pid, r));
+    ad.push_back(ADsim(window_size, pid, r, program_start, step_freq));
 
   //Setup some functions, providing some fake statistics to attach to their provenance data
   registerFunc("main", 1000, 200, 20);
@@ -31,22 +33,29 @@ int main(int argc, char **argv){
     ad[0].registerGPUthread(gpu_threads[g]);
   }
 
+  //Generate the program trace
   threadExecution thr0(0, ad[0]); //thread 0 (a cpu thread) on first ad
 
-  ad[0].beginStep(50);
-
   thr0.enterNormal("main", 100, "tag_main"); //events can be tagged and retrieved once exit has been called, allowing for external manipulation
+
   thr0.addCounter("main_counter", 1234, 120);
   thr0.addComm(CommType::Send, 1, 1024, 130);
   thr0.enterAnomaly("child1", 140, 1.99);
   thr0.exit(149);
-  thr0.exit(150);
+  
+  thr0.exit(1500);
 
   auto it = thr0.getTagged("tag_main");
   std::cout << it << std::endl;
 
-  ad[0].endStep(200);
+  //Run the simulation
+  int nstep = ad[0].largest_step() + 1;
+  std::cout << "Got data for " << nstep << " steps" << std::endl;
 
+  for(int i=0;i<nstep;i++){
+    std::cout << "Step " << i << " contains " << ad[0].nStepExecs(i) << " execs" << std::endl;
+    ad[0].step(i);
+  }
 
   MPI_Finalize();
   return 0;

sim/src/ad.cpp

@@ -15,16 +15,16 @@
 using namespace chimbuko;
 using namespace chimbuko_sim;
 
-void ADsim::init(int window_size, int pid, int rid){
-  m_step = 0;
+void ADsim::init(int window_size, int pid, int rid, unsigned long program_start, unsigned long step_freq){
   m_window_size = window_size;
   m_pid = pid;
   m_rid = rid;
   m_pdb_client.reset(new ADProvenanceDBclient(rid));
   m_pdb_client->setEnableHandshake(false);
   m_pdb_client->connect(getProvDB().getAddr(), getProvDB().getNshards());
-  m_counters.linkCounterMap(getCidxManager().getCounterMap());
-  m_step_is_open = false;
+  m_program_start = program_start;
+  m_step_freq = step_freq;
+  m_largest_step = 0;
 }
 
 CallListIterator_t ADsim::addExec(const int thread,
@@ -33,17 +33,20 @@ CallListIterator_t ADsim::addExec(const int thread,
 				  unsigned long runtime,
 				  bool is_anomaly,
 				  double outlier_score){
-  if(!m_step_is_open) fatal_error("No step is currently open");
-  if(start + runtime < m_step_start_time) fatal_error("Function end time is not within the current step");
+  if(start < m_program_start) fatal_error("Exec start is before program start");
 
   auto it = funcIdxMap().find(func_name);
   if(it == funcIdxMap().end()) assert(0);
   unsigned long func_id = it->second;
 
-  static long event_idx_s = 0;
-  long event_idx = event_idx_s++;
+  unsigned long exit = start + runtime;
+  unsigned long step = ( exit - m_program_start ) / m_step_freq;
+
+  long event_idx = m_step_execs[step].size();
 
-  eventID id(m_rid, m_step, event_idx);
+  m_largest_step = std::max(m_largest_step, step);
+
+  eventID id(m_rid, step, event_idx);
 
   ExecData_t exec(id, m_pid, m_rid, thread, func_id, func_name, start, start+runtime);
   exec.set_label( is_anomaly ? -1 : 1 );
@@ -53,9 +56,7 @@ CallListIterator_t ADsim::addExec(const int thread,
   CallListIterator_t exec_it = m_all_execs[thread].insert(m_all_execs[thread].end(), exec);
   m_eventid_map[id] = exec_it;
 
-  m_step_exec_its.push_back(exec_it);
-  m_step_func_exec_its[func_id].push_back(exec_it);
-
+  m_step_execs[step].push_back(exec_it);
   return exec_it;
 }
 
@@ -69,7 +70,6 @@ void ADsim::attachCounter(const std::string &counter_name,
   assert(ts <= to->get_exit());
 
   CounterData_t cdata(m_pid, m_rid, to->get_tid(), cid, counter_name, value, ts);
-  m_counters.addCounter(cdata);
   to->add_counter(cdata);
 }
 
@@ -110,25 +110,48 @@ void ADsim::bindCPUparentGPUkernel(CallListIterator_t cpu_parent, CallListIterat
   cpu_parent->set_GPU_correlationID_partner(gpu_kern->get_id());
 }
 
-void ADsim::beginStep(const unsigned long step_start_time){
-  m_step_start_time = step_start_time;
-  m_step_is_open = true;
+void ADsim::bindParentChild(CallListIterator_t parent, CallListIterator_t child){
+  parent->inc_n_children();
+  parent->update_exclusive(child->get_runtime());
+  //Tell the child who it's parent is
+  child->set_parent(parent->get_id());
 }
 
-void ADsim::endStep(const unsigned long step_end_time){
+
+void ADsim::step(const unsigned long step){
+  unsigned long step_start_time = m_program_start + step * m_step_freq;
+  unsigned long step_end_time = step_start_time + m_step_freq;
+
+  //Collect events in the timestep
+  auto mse = m_step_execs.find(step);
+  if(mse == m_step_execs.end()){
+    std::cout << "Step " << step << " contains no executions" << std::endl;
+    return;
+  }
+  std::list<CallListIterator_t> const& this_step_execs = mse->second;
+
   //Collect outliers and 1 normal event/func into Anomalies object
+  //and collect the map of function index to execs
   Anomalies anom;
+  std::unordered_map<unsigned long, std::vector<CallListIterator_t> > this_step_func_execs;
+  ADCounter counters;
+  counters.linkCounterMap(getCidxManager().getCounterMap());
 
   int nanom=0, nnorm=0;
-  for(const auto &exec_it : m_step_exec_its){
-    if(exec_it->get_exit() > step_end_time) fatal_error("Event " + exec_it->get_json(true, true).dump(4) + " has an exit time later than the end of the current step, " + std::to_string(step_end_time) );
-
+  for(const auto &exec_it : this_step_execs){
     if(exec_it->get_label() == -1){ anom.insert(exec_it, Anomalies::EventType::Outlier); nanom++; }
     else if(anom.nFuncEvents(exec_it->get_fid(), Anomalies::EventType::Normal) == 0){ anom.insert(exec_it, Anomalies::EventType::Normal); nnorm++; }
+
+    this_step_func_execs[exec_it->get_fid()].push_back(exec_it);
+    
+    const std::deque<CounterData_t>& ect = exec_it->get_counters();
+    for(const CounterData_t &c : ect)
+      counters.addCounter(c);
   }
 
-  std::cout << "Rank " << m_rid << " " << nanom << " anomalies and " << nnorm << " normal events" << std::endl;
-  std::cout << "TEST " << anom.nEvents(Anomalies::EventType::Outlier) << " " << anom.nEvents(Anomalies::EventType::Normal) << std::endl;
+  //TODO: Normal events need to be processed correctly
+
+  std::cout << "Step " << step << " rank " << m_rid << " " << nanom << " anomalies and " << nnorm << " normal events" << std::endl;
 
   //Extract provenance data and send to the provDB
   if(nanom > 0){
@@ -138,8 +161,8 @@ void ADsim::endStep(const unsigned long step_end_time){
     eventIDmap evmap(m_all_execs, m_eventid_map);
 
     ADAnomalyProvenance::getProvenanceEntries(anomalous_events, normal_events, m_normal_events,
-					      anom, m_step, m_step_start_time, step_end_time, m_window_size,
-					      globalParams(), evmap, m_counters, m_metadata);
+					      anom, step, step_start_time, step_end_time, m_window_size,
+					      globalParams(), evmap, counters, m_metadata);
     //Send to provdb
     if(anomalous_events.size() > 0)
       m_pdb_client->sendMultipleData(anomalous_events, ProvenanceDataType::AnomalyData); 
@@ -150,20 +173,22 @@ void ADsim::endStep(const unsigned long step_end_time){
 
   //Update the pserver
   {
-    ADLocalCounterStatistics count_stats(m_pid, m_step, nullptr);
-    count_stats.gatherStatistics(m_counters.getCountersByIndex());
+    ADLocalCounterStatistics count_stats(m_pid, step, nullptr);
+    count_stats.gatherStatistics(counters.getCountersByIndex());
     getPserver().addCounterData(count_stats);
 
-    ADLocalFuncStatistics prof_stats(m_pid, m_rid, m_step);
-    prof_stats.gatherStatistics(&m_step_func_exec_its); //takes map of fid -> vector of ExecData iterators
+    ADLocalFuncStatistics prof_stats(m_pid, m_rid, step);
+    prof_stats.gatherStatistics(&this_step_func_execs); //takes map of fid -> vector of ExecData iterators
     prof_stats.gatherAnomalies(anom);
     getPserver().addAnomalyData(prof_stats);
   }
+}
 
-  m_step_exec_its.clear();
-  m_step_func_exec_its.clear();
-  delete m_counters.flushCounters();
 
-  m_step++;
-  m_step_is_open = false;
+size_t ADsim::nStepExecs(unsigned long step) const{
+  auto mse = m_step_execs.find(step);
+  if(mse == m_step_execs.end()){
+    return 0;
+  }else return mse->second.size();
 }
+