In RunStats merge, if one input has accumulate enabled and the other does not, the accumulated value of the one without will be reconstructed from the mean and count

GlobalAnomalyStats function profile now accumulates total runtime for each function over all ranks/threads. Modified unit tests to adapt to this change.

Author: giltirn

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/pserver/AggregateFuncStats.cpp

@@ -9,12 +9,15 @@ using namespace chimbuko;
 
 TEST(TestAggregateFuncStats, AggregationAndJSON){  
   //Generate some stats representing the inclusive and exclusive runtimes collected over 2 different io steps
-  RunStats incl1, incl2, excl1, excl2;
+  RunStats incl1(true), incl2(true), excl1(true), excl2(true); //total runtimes should aggregate
+  RunStats incl1_noac(false), incl2_noac(false), excl1_noac(false), excl2_noac(false); //versions which don't accumulate so we can test that AggregateFuncStats forces accumulate despite input settings
 
   std::mt19937 gen(1234);
   std::normal_distribution<> dexcl(100,20);
   std::normal_distribution<> dincl_add(200,100);
 
+  double tsum_incl = 0, tsum_excl = 0;
+
   for(int i=0;i<100;i++){
     double e1 = dexcl(gen);
     double i1 = e1 + dincl_add(gen);
@@ -27,6 +30,15 @@ TEST(TestAggregateFuncStats, AggregationAndJSON){
 
     incl2.push(i2);
     excl2.push(e2);
+
+    incl1_noac.push(i1);
+    excl1_noac.push(e1);
+    
+    incl2_noac.push(i2);
+    excl2_noac.push(e2);
+
+    tsum_incl += i1 + i2;
+    tsum_excl += e1 + e2;
   }
 
   int pid = 99;
@@ -37,26 +49,55 @@ TEST(TestAggregateFuncStats, AggregationAndJSON){
   int nanom2 = 44; //anoms on second step
 
   AggregateFuncStats fstats(pid,fid,fname);
-  fstats.add(nanom1, incl1, excl1);
-  fstats.add(nanom2, incl2, excl2);
+  fstats.add(nanom1, incl1_noac, excl1_noac);
+  fstats.add(nanom2, incl2_noac, excl2_noac);
+
+  nlohmann::json fstats_j = fstats.get_json();
+  //std::cout << fstats_j.dump(4) << std::endl;
 
   EXPECT_EQ(fstats.get_func(), fname);
   EXPECT_EQ(fstats.get_func_anomaly().count(), 2);
   EXPECT_EQ(fstats.get_func_anomaly().accumulate(), nanom1 + nanom2);
 
   RunStats incl_sum = incl1 + incl2;
   RunStats excl_sum = excl1 + excl2;
-  EXPECT_EQ(fstats.get_inclusive().get_stat_values(), incl_sum.get_stat_values());
-  EXPECT_EQ(fstats.get_exclusive().get_stat_values(), excl_sum.get_stat_values());
+  EXPECT_NEAR(incl_sum.accumulate(), tsum_incl, 1e-8);
+  EXPECT_NEAR(excl_sum.accumulate(), tsum_excl, 1e-8);
+
+  RunStats::RunStatsValues incl_got = fstats.get_inclusive().get_stat_values();
+  RunStats::RunStatsValues incl_expect = incl_sum.get_stat_values();
+  
+#define RR(A) EXPECT_NEAR(incl_got.A, incl_expect.A, 1e-8)
+  RR(count);
+  RR(accumulate);
+  RR(minimum);
+  RR(maximum);
+  RR(mean);
+  RR(stddev);
+  RR(skewness);
+  RR(kurtosis);
+#undef RR
+  
+  RunStats::RunStatsValues excl_got = fstats.get_exclusive().get_stat_values();
+  RunStats::RunStatsValues excl_expect = excl_sum.get_stat_values();
+
+#define RR(A) EXPECT_NEAR(excl_got.A, excl_expect.A, 1e-8)
+  RR(count);
+  RR(accumulate);
+  RR(minimum);
+  RR(maximum);
+  RR(mean);
+  RR(stddev);
+  RR(skewness);
+  RR(kurtosis);
+#undef RR
+
 
-  nlohmann::json fstats_j = fstats.get_json();
   EXPECT_EQ(fstats_j["app"], pid);
   EXPECT_EQ(fstats_j["fid"], fid);
   EXPECT_EQ(fstats_j["name"], fname);
   EXPECT_EQ(fstats_j["stats"]["count"],  2);
   EXPECT_EQ(fstats_j["stats"]["accumulate"],  nanom1+nanom2);
-  EXPECT_EQ(fstats_j["inclusive"], incl_sum.get_json());
-  EXPECT_EQ(fstats_j["exclusive"], excl_sum.get_json());
 }
 
 

test/unit_tests/pserver/GlobalAnomalyStats.cpp

@@ -11,7 +11,7 @@ TEST(TestGlobalAnomalyStats, UpdateFuncStat){
   GlobalAnomalyStats stats;
 
   //Generate some stats representing the inclusive and exclusive runtimes collected over 2 different io steps
-  RunStats incl1, incl2, excl1, excl2;
+  RunStats incl1(true), incl2(true), excl1(true), excl2(true); //accumulate total time over all ranks
 
   std::mt19937 gen(1234);
   std::normal_distribution<> dexcl(100,20);