Relocate GlobalStatisticCollection to workunit.cpp

Prepare dynamic stats tracking by GlobalStatisticCollection Remove unnecessary calls to updateAggregation Track active graph stats for aggregation Improve performance by updating aggregates intermittantly and at end of each graph Load serialized graph stats when resuming job to ensure aggregates are calculated correctly. Only update aggregates if they have changed Signed-off-by: Shamser Ahmed <[email protected]>
hpcc-systems · Oct 12, 2023 · 25bdaa9 · 25bdaa9
1 parent c2e13ef
commit 25bdaa9
Show file tree

Hide file tree

Showing 12 changed files with 393 additions and 170 deletions.
diff --git a/common/workunit/workunit.cpp b/common/workunit/workunit.cpp
diff --git a/common/workunit/workunit.hpp b/common/workunit/workunit.hpp
@@ -1174,7 +1174,6 @@ interface IConstWUScopeIterator : extends IScmIterator
 };
 
 //---------------------------------------------------------------------------------------------------------------------
-
 //! IWorkUnit
 //! Provides high level access to WorkUnit "header" data.
 interface IWorkUnit;
@@ -1204,6 +1203,7 @@ interface IConstWorkUnitInfo : extends IInterface
     virtual IConstWUAppValueIterator & getApplicationValues() const = 0;
 };
 
+class GlobalStatisticCollection;
 interface IConstWorkUnit : extends IConstWorkUnitInfo
 {
     virtual bool aborting() const = 0;
@@ -1302,7 +1302,7 @@ interface IConstWorkUnit : extends IConstWorkUnitInfo
     virtual WUGraphState queryNodeState(const char *graphName, WUGraphIDType nodeId) const = 0;
     virtual void setGraphState(const char *graphName, unsigned wfid, WUGraphState state) const = 0;
     virtual void setNodeState(const char *graphName, WUGraphIDType nodeId, WUGraphState state) const = 0;
-    virtual IWUGraphStats *updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned _wfid, unsigned subgraph, bool merge) const = 0;
+    virtual IWUGraphStats *updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned _wfid, unsigned subgraph, bool merge, GlobalStatisticCollection * stats=nullptr) const = 0;
     virtual void clearGraphProgress() const = 0;
     virtual IStringVal & getAbortBy(IStringVal & str) const = 0;
     virtual unsigned __int64 getAbortTimeStamp() const = 0;
@@ -1725,7 +1725,6 @@ 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);
 extern WORKUNIT_API const char *getTargetClusterComponentName(const char *clustname, const char *processType, StringBuffer &name);
 extern WORKUNIT_API void descheduleWorkunit(char const * wuid);
 #if 0
@@ -1780,4 +1779,22 @@ extern WORKUNIT_API TraceFlags loadTraceFlags(IConstWorkUnit * wu, const std::in
 
 extern WORKUNIT_API bool executeGraphOnLingeringThor(IConstWorkUnit &workunit, unsigned wfid, const char *graphName);
 
+
+class WORKUNIT_API GlobalStatisticCollection : public CInterface
+{
+public:
+    GlobalStatisticCollection();
+
+    void load(IConstWorkUnit &workunit, const char * graphName, bool aggregatesOnly, bool missingScopesOnly);
+    void loadGlobalAggregates(IConstWorkUnit &workunit);
+    IStatisticCollection * getCollectionForUpdate(const StatsScopeId & wfScopeId, const StatsScopeId & graphScopeId, const StatsScopeId & sgScopeId, StatisticCreatorType creatorType, const char * creator, bool clearStats);
+    bool refreshAggregates();
+    IStatisticCollection * queryCollection() { return statsCollection; }
+    void updateAggregates(IWorkUnit *wu);
+private:
+    Owned<IStatisticCollection> statsCollection;
+    std::vector<StatisticKind> aggregateKinds = {StCostFileAccess, StSizeGraphSpill, StSizeSpillFile};
+    StringBuffer wuid;
+};
+
 #endif
diff --git a/common/workunit/workunit.ipp b/common/workunit/workunit.ipp
@@ -232,7 +232,7 @@ public:
     virtual void setGraphState(const char *graphName, unsigned wfid, WUGraphState state) const;
     virtual void setNodeState(const char *graphName, WUGraphIDType nodeId, WUGraphState state) const;
     virtual WUGraphState queryNodeState(const char *graphName, WUGraphIDType nodeId) const;
-    virtual IWUGraphStats *updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned _wfid, unsigned subgraph, bool merge) const override;
+    virtual IWUGraphStats *updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned _wfid, unsigned subgraph, bool merge, GlobalStatisticCollection * stats=nullptr) const override;
     void clearGraphProgress() const;
     virtual void import(IPropertyTree *wuTree, IPropertyTree *graphProgressTree) {}; //No GraphProgressTree in CLocalWorkUnit.
 
@@ -661,7 +661,7 @@ public:
 class WORKUNIT_API CWuGraphStats : public CInterfaceOf<IWUGraphStats>
 {
 public:
-    CWuGraphStats(StatisticCreatorType _creatorType, const char * _creator, unsigned wfid, const char * _rootScope, unsigned _id, bool _merge);
+    CWuGraphStats(StatisticCreatorType _creatorType, const char * _creator, unsigned wfid, const char * _rootScope, unsigned _id, bool _merge, GlobalStatisticCollection * stats);
     virtual void beforeDispose();
     virtual IStatisticGatherer & queryStatsBuilder();
 protected:

diff --git a/ecl/eclagent/eclagent.cpp b/ecl/eclagent/eclagent.cpp
@@ -1988,7 +1988,6 @@ void EclAgent::doProcess()
         const cost_type cost = aggregateCost(w, nullptr, false);
         if (cost)
             w->setStatistic(queryStatisticsComponentType(), queryStatisticsComponentName(), SSTglobal, "", StCostExecute, NULL, cost, 1, 0, StatsMergeReplace);
-        updateAggregates(w);
         addTimings(w);
 
         switch (w->getState())
@@ -2534,7 +2533,6 @@ void EclAgentWorkflowMachine::noteTiming(unsigned wfid, timestamp_type startTime
     const cost_type cost = money2cost_type(calcCost(agent.queryAgentMachineCost(), nanoToMilli(elapsedNs))) + aggregateCost(wu, scope, true);
     if (cost)
         wu->setStatistic(queryStatisticsComponentType(), queryStatisticsComponentName(), SSTworkflow, scope, StCostExecute, NULL, cost, 1, 0, StatsMergeReplace);
-    updateAggregates(wu);
 }
 
 void EclAgentWorkflowMachine::doExecutePersistItem(IRuntimeWorkflowItem & item)

diff --git a/ecl/eclagent/eclagent.ipp b/ecl/eclagent/eclagent.ipp
@@ -1048,14 +1048,15 @@ public:
         graphAgentContext.set(&_agent);
         agent = &graphAgentContext;
         aborted = false;
+        globalStats.load(*wu, graphName, true, true);
     }
 
     void createFromXGMML(ILoadedDllEntry * dll, IPropertyTree * xgmml);
     void execute(const byte * parentExtract);
     void executeLibrary(const byte * parentExtract, IHThorGraphResults * results);
     IWUGraphStats *updateStats(StatisticCreatorType creatorType, const char * creator, unsigned wfid, unsigned subgraph);
     void updateWUStatistic(IWorkUnit* lockedwu, StatisticScopeType scopeType, const char* scope, StatisticKind kind, const char* descr, long long unsigned int value);
-
+    void updateAggregates(IWorkUnit* lockedwu);
     EclSubGraph * idToGraph(unsigned id);
     EclGraphElement * idToActivity(unsigned id);
     const char *queryGraphName() { return graphName; }
@@ -1083,6 +1084,7 @@ protected:
     IProbeManager * probeManager;
     unsigned wfid;
     bool aborted;
+    GlobalStatisticCollection globalStats;
 };
 
 

diff --git a/ecl/eclagent/eclgraph.cpp b/ecl/eclagent/eclgraph.cpp
@@ -879,10 +879,10 @@ void EclSubGraph::updateProgress()
         Owned<IWUGraphStats> progress = parent.updateStats(queryStatisticsComponentType(), queryStatisticsComponentName(), parent.queryWfid(), id);
         IStatisticGatherer & stats = progress->queryStatsBuilder();
         updateProgress(stats);
-
         if (startGraphTime || elapsedGraphCycles)
         {
             WorkunitUpdate lockedwu(agent->updateWorkUnit());
+            parent.updateAggregates(lockedwu);
             StringBuffer subgraphid;
             subgraphid.append(parent.queryGraphName()).append(":").append(SubGraphScopePrefix).append(id);
             if (startGraphTime)
@@ -1278,8 +1278,6 @@ void EclGraph::execute(const byte * parentExtract)
             const cost_type cost = money2cost_type(calcCost(agent->queryAgentMachineCost(), elapsed));
             if (cost)
                 wu->setStatistic(queryStatisticsComponentType(), queryStatisticsComponentName(), SSTgraph, scope, StCostExecute, NULL, cost, 1, 0, StatsMergeReplace);
-
-            updateAggregates(wu);
         }
 
         if (agent->queryRemoteWorkunit())
@@ -1348,7 +1346,7 @@ void EclGraph::updateLibraryProgress()
     {
         EclSubGraph & cur = graphs.item(idx);
         unsigned wfid = cur.parent.queryWfid();
-        Owned<IWUGraphStats> progress = wu->updateStats(queryGraphName(), queryStatisticsComponentType(), queryStatisticsComponentName(), wfid, cur.id, false);
+        Owned<IWUGraphStats> progress = wu->updateStats(queryGraphName(), queryStatisticsComponentType(), queryStatisticsComponentName(), wfid, cur.id, false, &globalStats);
         cur.updateProgress(progress->queryStatsBuilder());
     }
 }
@@ -1491,7 +1489,12 @@ void GraphResults::setResult(unsigned id, IHThorGraphResult * result)
 
 IWUGraphStats *EclGraph::updateStats(StatisticCreatorType creatorType, const char * creator, unsigned activeWfid, unsigned subgraph)
 {
-    return wu->updateStats (queryGraphName(), creatorType, creator, activeWfid, subgraph, false);
+    return wu->updateStats (queryGraphName(), creatorType, creator, activeWfid, subgraph, false, &globalStats);
+}
+
+void EclGraph::updateAggregates(IWorkUnit* lockedwu)
+{
+    globalStats.updateAggregates(lockedwu);
 }
 
 void EclGraph::updateWUStatistic(IWorkUnit *lockedwu, StatisticScopeType scopeType, const char * scope, StatisticKind kind, const char * descr, unsigned __int64 value)

diff --git a/plugins/cassandra/cassandrawu.cpp b/plugins/cassandra/cassandrawu.cpp
@@ -2743,8 +2743,8 @@ class CCassandraWorkUnit : public CPersistedWorkUnit
     class CCassandraWuGraphStats : public CWuGraphStats
     {
     public:
-        CCassandraWuGraphStats(const CCassandraWorkUnit *_parent, StatisticCreatorType _creatorType, const char * _creator, unsigned _wfid, const char * _rootScope, unsigned _id, bool _merge)
-        : CWuGraphStats(_creatorType, _creator, _wfid, _rootScope, _id, _merge),
+        CCassandraWuGraphStats(const CCassandraWorkUnit *_parent, StatisticCreatorType _creatorType, const char * _creator, unsigned _wfid, const char * _rootScope, unsigned _id, bool _merge, GlobalStatisticCollection * stats)
+        : CWuGraphStats(_creatorType, _creator, _wfid, _rootScope, _id, _merge, stats),
           progress(createPTree(_rootScope)), parent(_parent)
         {
         }
@@ -2764,9 +2764,9 @@ class CCassandraWorkUnit : public CPersistedWorkUnit
         StringAttr wuid;
     };
 
-    IWUGraphStats *updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned wfid, unsigned subgraph, bool merge) const override
+    IWUGraphStats *updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned wfid, unsigned subgraph, bool merge, GlobalStatisticCollection * stats) const override
     {
-        return new CCassandraWuGraphStats(this, creatorType, creator, wfid, graphName, subgraph, merge);
+        return new CCassandraWuGraphStats(this, creatorType, creator, wfid, graphName, subgraph, merge, stats);
     }