Merge pull request #3721 from jakesmith/apply-sequential

HPCC-3309 Ensure child graphs are executed sequentially

Reviewed-By: Richard Chapman <rchapman@hpccsystems.com>

thorlcr/graph/thgraph.cpp

@@ -1049,6 +1049,7 @@ void CGraphBase::clean()
     ::Release(doneBarrier);
     localResults.clear();
     graphLoopResults.clear();
+    childGraphsTable.kill();
     childGraphs.kill();
     disconnectActivities();
     containers.kill();
@@ -1113,7 +1114,7 @@ void CGraphBase::reset()
     setCompleteEx(false);
     if (0 == containers.count())
     {
-        SuperHashIteratorOf<CGraphBase> iter(childGraphs);
+        SuperHashIteratorOf<CGraphBase> iter(childGraphsTable);
         ForEach(iter)
             iter.query().reset();
     }
@@ -1136,14 +1137,15 @@ void CGraphBase::reset()
 void CGraphBase::addChildGraph(CGraphBase &graph)
 {
     CriticalBlock b(crit);
-    childGraphs.replace(graph);
+    childGraphs.append(graph);
+    childGraphsTable.replace(graph);
     job.associateGraph(graph);
 }
 
 IThorGraphIterator *CGraphBase::getChildGraphs() const
 {
     CriticalBlock b(crit);
-    return new CGraphTableIterator(childGraphs);
+    return new CGraphArrayCopyIterator(childGraphs);
 }
 
 bool CGraphBase::fireException(IException *e)
@@ -1660,6 +1662,7 @@ void CGraphBase::createFromXGMML(IPropertyTree *_node, CGraphBase *_owner, CGrap
     node.setown(createPTreeFromIPT(_node));
     xgmml = node->queryPropTree("att/graph");
     sink = xgmml->getPropBool("att[@name=\"rootGraph\"]/@value", false);
+    sequential = xgmml->getPropBool("@sequential");
     graphId = node->getPropInt("@id");
     global = false;
     localOnly = -1; // unset
@@ -1735,6 +1738,8 @@ void CGraphBase::createFromXGMML(IPropertyTree *_node, CGraphBase *_owner, CGrap
 
 void CGraphBase::executeChildGraphs(size32_t parentExtractSz, const byte *parentExtract)
 {
+    // JCSMORE - would need to respect codegen 'sequential' flag, if these child graphs
+    // could be executed in parallel.
     Owned<IThorGraphIterator> iter = getChildGraphs();
     ForEach(*iter)
     {

thorlcr/graph/thgraph.hpp

@@ -241,6 +241,7 @@ typedef CIArrayOf<CGraphBase> CGraphArray;
 typedef CopyCIArrayOf<CGraphBase> CGraphArrayCopy;
 typedef IIteratorOf<CGraphBase> IThorGraphIterator;
 typedef ArrayIIteratorOf<const CGraphArray, CGraphBase, IThorGraphIterator> CGraphArrayIterator;
+typedef ArrayIIteratorOf<const CGraphArrayCopy, CGraphBase, IThorGraphIterator> CGraphArrayCopyIterator;
 
 class CJobBase;
 class graph_decl CGraphElementBase : public CInterface, implements IInterface
@@ -425,7 +426,8 @@ class graph_decl CGraphBase : public CInterface, implements ILocalGraph, impleme
     mutable int localOnly;
     activity_id parentActivityId;
     IPropertyTree *xgmml;
-    CGraphTable childGraphs;
+    CGraphTable childGraphsTable;
+    CGraphArrayCopy childGraphs;
     Owned<IGraphTempHandler> tmpHandler;
 
     void clean();
@@ -522,7 +524,7 @@ protected:
     Owned<IPropertyTree> node;
     IBarrier *startBarrier, *waitBarrier, *doneBarrier;
     mptag_t mpTag, startBarrierTag, waitBarrierTag, doneBarrierTag;
-    bool created, connected, started, aborted, graphDone, prepared;
+    bool created, connected, started, aborted, graphDone, prepared, sequential;
     bool reinit, sentInitData, sentStartCtx;
     CJobBase &job;
     graph_id graphId;
@@ -698,11 +700,11 @@ public:
     const activity_id &queryParentActivityId() const { return parentActivityId; }
     const graph_id &queryGraphId() const { return graphId; }
     void addChildGraph(CGraphBase &graph);
-    unsigned queryChildGraphCount() { return childGraphs.count(); }
+    unsigned queryChildGraphCount() { return childGraphs.ordinality(); }
     CGraphBase *getChildGraph(graph_id gid)
     {
         CriticalBlock b(crit);
-        return LINK(childGraphs.find(gid));
+        return LINK(childGraphsTable.find(gid));
     }
     IThorGraphIterator *getChildGraphs() const;