Merge pull request #2319 from jakesmith/loop-sync-optimization

Loop subgraph init optimzation

Reviewed-By: Gavin Halliday <gavin.halliday@lexisnexis.com>
Reviewed-By: Richard Chapman <rchapman@hpccsystems.com>

thorlcr/graph/thgraph.cpp

@@ -715,20 +715,12 @@ void CGraphElementBase::initActivity()
     if (activity)
         return;
     activity.setown(factory());
-    switch (getKind())
+    if (isLoopActivity(*this))
     {
-        case TAKlooprow:
-        case TAKloopcount:
-        case TAKloopdataset:
-        case TAKgraphloop:
-        case TAKparallelgraphloop:
-        {
-            unsigned loopId = queryXGMML().getPropInt("att[@name=\"_loopid\"]/@value");
-            Owned<CGraphBase> childGraph = owner->getChildGraph(loopId);
-            Owned<IThorBoundLoopGraph> boundLoopGraph = createBoundLoopGraph(childGraph, baseHelper->queryOutputMeta(), queryId());
-            setBoundGraph(boundLoopGraph);
-            break;
-        }
+        unsigned loopId = queryXGMML().getPropInt("att[@name=\"_loopid\"]/@value");
+        Owned<CGraphBase> childGraph = owner->getChildGraph(loopId);
+        Owned<IThorBoundLoopGraph> boundLoopGraph = createBoundLoopGraph(childGraph, baseHelper->queryOutputMeta(), queryId());
+        setBoundGraph(boundLoopGraph);
     }
 }
 
@@ -1004,6 +996,19 @@ bool isGlobalActivity(CGraphElementBase &container)
     return false;
 }
 
+bool isLoopActivity(CGraphElementBase &container)
+{
+    switch (container.getKind())
+    {
+        case TAKlooprow:
+        case TAKloopcount:
+        case TAKloopdataset:
+        case TAKgraphloop:
+        case TAKparallelgraphloop:
+            return true;
+    }
+    return false;
+}
 /////
 
 CGraphBase::CGraphBase(CJobBase &_job) : job(_job)
@@ -1344,6 +1349,15 @@ void CGraphBase::done()
     }
 }
 
+bool CGraphBase::syncInitData()
+{
+    CGraphElementBase *parentElement = queryOwner() ? queryOwner()->queryElement(queryParentActivityId()) : NULL;
+    if (parentElement && isLoopActivity(*parentElement))
+        return parentElement->queryLoopGraph()->queryGraph()->isGlobal();
+    else
+        return !isLocalChild();
+}
+
 void CGraphBase::end()
 {
 // always called, any final action clear up
@@ -1667,22 +1681,10 @@ void CGraphBase::createFromXGMML(IPropertyTree *_node, CGraphBase *_owner, CGrap
     {
         CGraphElementBase *parentElement = owner->queryElement(parentActivityId);
         parentElement->addAssociatedChildGraph(this);
-        switch (parentElement->getKind())
-        {
-            case TAKlooprow:
-            case TAKloopcount:
-            case TAKloopdataset:
-            case TAKgraphloop:
-            case TAKparallelgraphloop:
-            {
-                if (parentElement->queryOwner().isLocalChild())
-                    localChild = true;
-                break;
-            }
-            default:
-                localChild = true;
-                break;
-        }
+        if (isLoopActivity(*parentElement))
+            localChild = parentElement->queryOwner().isLocalChild();
+        else
+            localChild = true;
     }
 
     Owned<IPropertyTreeIterator> nodes = xgmml->getElements("node");
@@ -2491,18 +2493,10 @@ void CJobBase::addDependencies(IPropertyTree *xgmml, bool failIfMissing)
         if (subGraph.queryOwner() && subGraph.queryParentActivityId())
         {
             CGraphElementBase *parentElement = subGraph.queryOwner()->queryElement(subGraph.queryParentActivityId());
-            switch (parentElement->getKind())
+            if (isLoopActivity(*parentElement))
             {
-                case TAKlooprow:
-                case TAKloopcount:
-                case TAKloopdataset:
-                case TAKgraphloop:
-                case TAKparallelgraphloop:
-                {
-                    if (!parentElement->queryOwner().isLocalChild() && !subGraph.isLocalOnly())
-                        subGraph.setGlobal(true);
-                    break;
-                }
+                if (!parentElement->queryOwner().isLocalChild() && !subGraph.isLocalOnly())
+                    subGraph.setGlobal(true);
             }
         }
         bool log = queryForceLogging(subGraph.queryGraphId(), subGraph.isGlobal());

thorlcr/graph/thgraph.hpp

@@ -287,7 +287,6 @@ public:
     IThorGraphDependencyIterator *getDependsIterator() const;
     void ActPrintLog(const char *format, ...)  __attribute__((format(printf, 2, 3)));
     void ActPrintLog(IException *e, const char *format, ...) __attribute__((format(printf, 3, 4)));
-
     void setBoundGraph(IThorBoundLoopGraph *graph) { loopGraph.set(graph); }
     IThorBoundLoopGraph *queryLoopGraph() { return loopGraph; }
     bool executeDependencies(size32_t parentExtractSz, const byte *parentExtract, int controlId, bool async);
@@ -619,6 +618,7 @@ public:
     IGraphTempHandler *queryTempHandler() const { assertex(tmpHandler.get()); return tmpHandler; }
     CGraphBase *queryOwner() { return owner; }
     CGraphBase *queryParent() { return parent?parent:this; }
+    bool syncInitData();
     bool isComplete() const { return complete; }
     bool isPrepared() const { return prepared; }
     bool isGlobal() const { return global; }
@@ -1135,6 +1135,7 @@ extern graph_decl void registerCreateFunc(CreateFunc func);
 extern graph_decl CGraphElementBase *createGraphElement(IPropertyTree &node, CGraphBase &owner, CGraphBase *resultsGraph);
 extern graph_decl IThorBoundLoopGraph *createBoundLoopGraph(CGraphBase *graph, IOutputMetaData *resultMeta, unsigned activityId);
 extern graph_decl bool isDiskInput(ThorActivityKind kind);
+extern graph_decl bool isLoopActivity(CGraphElementBase &container);
 
 
 #endif

thorlcr/graph/thgraphmaster.cpp

@@ -2333,7 +2333,8 @@ bool CMasterGraph::preStart(size32_t parentExtractSz, const byte *parentExtract)
             throw;
         }
     }
-    if (!queryOwner() || isGlobal())
+
+    if (syncInitData())
         sendActivityInitData(); // has to be done at least once
     CGraphBase::preStart(parentExtractSz, parentExtract);
     if (isGlobal())

thorlcr/graph/thgraphslave.cpp

@@ -383,7 +383,8 @@ bool CSlaveGraph::recvActivityInitData()
         CMessageBuffer actInitRtnData;
         actInitRtnData.append(false);
         CMessageBuffer msg;
-        if (!queryOwner() || isGlobal())
+
+        if (syncInitData())
         {
             if (!job.queryJobComm().recv(msg, 0, mpTag, NULL, LONGTIMEOUT))
                 throw MakeStringException(0, "Error receiving actinit data for graph: %"GIDPF"d", graphId);