HPCC-29657 Load serialized graph stats when resuming job

Load serialized graph stats when resuming job to ensure aggregates are calculated correctly. Signed-off-by: Shamser Ahmed <shamser.ahmed@lexisnexis.com>
hpcc-systems · Oct 4, 2023 · c0f357e · c0f357e
1 parent 4930b80
commit c0f357e
Show file tree

Hide file tree

Showing 6 changed files with 23 additions and 24 deletions.
diff --git a/common/workunit/workunit.cpp b/common/workunit/workunit.cpp
@@ -2777,10 +2777,15 @@ class SubGraphUpdaterCollection : public CInterfaceOf<IStatisticCollection>
     }
 };
 
-GlobalStatisticCollection::GlobalStatisticCollection(IPropertyTree * root, StatisticScopeType minScope)
+GlobalStatisticCollection::GlobalStatisticCollection()
 {
     StatsScopeId globalScopeId(SSTglobal, (unsigned)0);
     statsCollection.setown(createStatisticCollection(nullptr, globalScopeId));
+}
+
+void GlobalStatisticCollection::load(const char *wuid, StatisticScopeType minScope)
+{
+    Owned<IPropertyTree> root = getWUGraphProgress(wuid, true);
     if (!root)
         return;
     Owned<IPropertyTreeIterator> iter = root->getElements("*");
@@ -2867,28 +2872,15 @@ class StatisticsAggregatesWriter : implements IStatisticVisitor
     }
 };
 
-void updateAggregates(IWorkUnit *wu, GlobalStatisticCollection *statsCollection)
+void updateAggregates(IWorkUnit *wu, GlobalStatisticCollection & statsCollection)
 {
     // Further improvements:
     // 1) maintain a dirty flag for each subgraph scope so that only modified subgraphs are serialized.
     // 2) Serialize the aggregates into a blob in GraphProgress rather than to global stats
     std::vector<StatisticKind> aggregateKinds = {StCostFileAccess, StSizeGraphSpill, StSizeSpillFile};
     StatisticsAggregatesWriter statsAggregatorWriter(wu, aggregateKinds);
-    if (statsCollection)
-    {
-        statsCollection->refreshAggregates(aggregateKinds);
-        statsCollection->visit(statsAggregatorWriter);
-    }
-    else
-    {
-        Owned<IPropertyTree> root = getWUGraphProgress(wu->queryWuid(), true);
-        if (root)
-        {
-            Owned<GlobalStatisticCollection> stats = new GlobalStatisticCollection(root, SSTsubgraph);
-            stats->refreshAggregates(aggregateKinds);
-            stats->visit(statsAggregatorWriter);
-        }
-    }
+    statsCollection.refreshAggregates(aggregateKinds);
+    statsCollection.visit(statsAggregatorWriter);
 }
 
 //---------------------------------------------------------------------------------------------------------------------

diff --git a/common/workunit/workunit.hpp b/common/workunit/workunit.hpp
@@ -1178,12 +1178,13 @@ interface IConstWUScopeIterator : extends IScmIterator
 class WORKUNIT_API GlobalStatisticCollection : public CInterface
 {
 public:
-    GlobalStatisticCollection(IPropertyTree * root=nullptr, StatisticScopeType minScope=SSTnone);
+    GlobalStatisticCollection();
     bool refreshAggregates(std::vector<StatisticKind> & aggregateKinds);
     void visit(IStatisticVisitor & target) const;
     // getCollection() returns IStatisticCollection for given rootScope
     // if clear==true, clears the stats at this and below scope
     IStatisticCollection * getCollection(const StatsScopeId & wfScope, const StatsScopeId & graphScope, const StatsScopeId & sgScope, StatisticCreatorType creatorType, const char * creator);
+    void load(const char *wuid, StatisticScopeType minScope=SSTnone);
 private:
     Owned<IStatisticCollection> statsCollection;
 };
@@ -1738,7 +1739,7 @@ extern WORKUNIT_API void updateWorkunitTimings(IWorkUnit * wu, ITimeReporter *ti
 extern WORKUNIT_API void updateWorkunitTimings(IWorkUnit * wu, StatisticScopeType scopeType, StatisticKind kind, ITimeReporter *timer);
 extern WORKUNIT_API void aggregateStatistic(StatsAggregation & result, IConstWorkUnit * wu, const WuScopeFilter & filter, StatisticKind search);
 extern WORKUNIT_API cost_type aggregateCost(const IConstWorkUnit * wu, const char *scope=nullptr, bool excludehThor=false);
-extern WORKUNIT_API void updateAggregates(IWorkUnit *wu, GlobalStatisticCollection *statsCollection=nullptr);
+extern WORKUNIT_API void updateAggregates(IWorkUnit *wu, GlobalStatisticCollection & statsCollection);
 extern WORKUNIT_API const char *getTargetClusterComponentName(const char *clustname, const char *processType, StringBuffer &name);
 extern WORKUNIT_API void descheduleWorkunit(char const * wuid);
 #if 0

diff --git a/ecl/eclagent/eclgraph.cpp b/ecl/eclagent/eclgraph.cpp
@@ -882,7 +882,7 @@ void EclSubGraph::updateProgress()
         if (startGraphTime || elapsedGraphCycles)
         {
             WorkunitUpdate lockedwu(agent->updateWorkUnit());
-            updateAggregates(lockedwu);
+            parent.updateAggregates(lockedwu);
             StringBuffer subgraphid;
             subgraphid.append(parent.queryGraphName()).append(":").append(SubGraphScopePrefix).append(id);
             if (startGraphTime)
@@ -1494,7 +1494,7 @@ IWUGraphStats *EclGraph::updateStats(StatisticCreatorType creatorType, const cha
 
 void EclGraph::updateAggregates(IWorkUnit* lockedwu)
 {
-    ::updateAggregates(lockedwu, &statsCache);
+    ::updateAggregates(lockedwu, statsCache);
 }
 
 void EclGraph::updateWUStatistic(IWorkUnit *lockedwu, StatisticScopeType scopeType, const char * scope, StatisticKind kind, const char * descr, unsigned __int64 value)

diff --git a/thorlcr/master/thdemonserver.cpp b/thorlcr/master/thdemonserver.cpp
@@ -157,7 +157,7 @@ class DeMonServer : public CSimpleInterface, implements IDeMonServer
                     unsigned startTime = graphStarts.item(g2);
                     reportStatus(wu, graph, startTime, finished, success);
                 }
-                ::updateAggregates(wu, &statsCache);
+                ::updateAggregates(wu, statsCache);
                 queryServerStatus().commitProperties();
             }
             catch (IException *E)
@@ -304,7 +304,11 @@ class DeMonServer : public CSimpleInterface, implements IDeMonServer
     }
     virtual void updateAggregates(IWorkUnit * lockedWu) override
     {
-        ::updateAggregates(lockedWu, &statsCache);
+        ::updateAggregates(lockedWu, statsCache);
+    }
+    virtual void loadStats(const char *wuid) override
+    {
+        statsCache.load(wuid, SSTsubgraph);
     }
 };
 

diff --git a/thorlcr/master/thdemonserver.hpp b/thorlcr/master/thdemonserver.hpp
@@ -31,6 +31,7 @@ interface IDeMonServer : extends IInterface
     virtual void endGraph(CGraphBase *graph, bool success) = 0;
     virtual void endGraphs() = 0;
     virtual void updateAggregates(IWorkUnit * lockedWu) = 0;
+    virtual void loadStats(const char *wuid) = 0;
 };
 
 

diff --git a/thorlcr/master/thgraphmanager.cpp b/thorlcr/master/thgraphmanager.cpp
@@ -1086,7 +1086,8 @@ bool CJobManager::executeGraph(IConstWorkUnit &workunit, const char *graphName,
         }
 
         setWuid(workunit.queryWuid(), workunit.queryClusterName());
-
+        if (job->queryResumed() && globals->getPropBool("@watchdogProgressEnabled"))
+            queryDeMonServer()->loadStats(workunit.queryWuid());
         allDone = job->go();
 
         Owned<IWorkUnit> wu = &workunit.lock();