HPCC-9577 - Handle sequential/parallel and refactor ifaction

With this commit ifaction/sequential/parallel can be generated
into the same graph as their actions.
This changes the way sinks are handled, to handle these 'sink'
activities and conditionally prepare/execute their inputs.

Signed-off-by: Jake Smith <jake.smith@lexisnexis.com>

ecl/hqlcpp/hqlcpp.cpp

@@ -1563,7 +1563,7 @@ void HqlCppTranslator::cacheOptions()
         DebugOption(options.resourceSequential,"resourceSequential", false),
         DebugOption(options.workunitTemporaries,"workunitTemporaries", true),
         DebugOption(options.resourceConditionalActions,"resourceConditionalActions", false),  //targetRoxie() ??
-        DebugOption(options.actionLinkInNewGraph,"actionLinkInNewGraph",true),
+        DebugOption(options.actionLinkInNewGraph,"actionLinkInNewGraph", false),
         DebugOption(options.minimizeWorkunitTemporaries, "<exception>", false),
         DebugOption(options.pickBestEngine,"pickBestEngine", true),
         DebugOption(options.groupedChildIterators,"groupedChildIterators", false),

ecl/hqlcpp/hqlresource.cpp

@@ -1865,7 +1865,7 @@ EclResourcer::EclResourcer(IErrorReceiver * _errors, IConstWorkUnit * _wu, Clust
     options.minimiseSpills = _translatorOptions.minimiseSpills;
     spillMultiCondition = _translatorOptions.spillMultiCondition;
     spotThroughAggregate = _translatorOptions.spotThroughAggregate && (targetClusterType != RoxieCluster) && (targetClusterType != ThorLCRCluster);
-    options.noConditionalLinks = (targetClusterType == RoxieCluster);
+    options.noConditionalLinks = (targetClusterType != HThorCluster);
     options.hoistResourced = _translatorOptions.hoistResourced;
     options.useGraphResults = false;        // modified by later call
     options.groupedChildIterators = _translatorOptions.groupedChildIterators;
@@ -1878,7 +1878,7 @@ EclResourcer::EclResourcer(IErrorReceiver * _errors, IConstWorkUnit * _wu, Clust
     options.createSpillAsDataset = _translatorOptions.optimizeSpillProject && (targetClusterType != HThorCluster);
     options.combineSiblings = _translatorOptions.combineSiblingGraphs && (targetClusterType != HThorCluster) && (targetClusterType != RoxieCluster);
     options.optimizeSharedInputs = _translatorOptions.optimizeSharedGraphInputs && options.combineSiblings;
-    options.actionLinkInNewGraph = _translatorOptions.actionLinkInNewGraph  || (targetClusterType == HThorCluster);
+    options.actionLinkInNewGraph = _translatorOptions.actionLinkInNewGraph || (targetClusterType == HThorCluster);
     options.convertCompoundToExecuteWhen = false;
 }
 

testing/ecl/ifaction.ecl

@@ -17,9 +17,6 @@
 
 #option ('resourceConditionalActions', true)
 
-//skip type==thorlcr TBD
-//nothor
-
 r := { string x{maxlength(256)}; };
 
 trueSimple1 := true : stored('trueSimple1');
@@ -30,9 +27,9 @@ falseSimple1 := false : stored('falseSimple1');
 falseSimple2 := false : stored('falseSimple2');
 falseSimple3 := false : stored('falseSimple3');
 
-complex1 := dedup(nofold(dataset([{'1'},{'b'},{'c'}], r)), x, all);
-complex2 := dedup(nofold(dataset([{'2'},{'b'},{'c'}], r)), x, all);
-complex3 := dedup(nofold(dataset([{'3'},{'b'},{'c'}], r)), x, all);
+complex1 := sort(dedup(nofold(dataset([{'1'},{'b'},{'c'}], r)), x, all), x);
+complex2 := sort(dedup(nofold(dataset([{'2'},{'b'},{'c'}], r)), x, all), x);
+complex3 := sort(dedup(nofold(dataset([{'3'},{'b'},{'c'}], r)), x, all), x);
 
 trueComplex1 := count(dedup(nofold(dataset([{'1'},{'b'},{'c'}], r)), x, all)) < 4;
 trueComplex2 := count(dedup(nofold(dataset([{'2'},{'b'},{'c'}], r)), x, all)) < 4;

thorlcr/activities/nsplitter/thnsplitterslave.cpp

@@ -306,15 +306,7 @@ public:
         IHThorSplitArg *helper = (IHThorSplitArg *)queryHelper();
         int dV = getOptInt(THOROPT_SPLITTER_SPILL, -1);
         if (-1 == dV)
-        {
             spill = !helper->isBalanced();
-            bool forcedUnbalanced = queryContainer().queryXGMML().getPropBool("@unbalanced", false);
-            if (!spill && forcedUnbalanced)
-            {
-                ActPrintLog("Was marked balanced, but forced unbalanced due to UPDATE changes.");
-                spill = true;
-            }
-        }
         else
             spill = dV>0;
     }

thorlcr/activities/when/thwhen.cpp

@@ -105,11 +105,3 @@ public:
     }
 };
 
-CActivityBase *createIfActionActivityMaster(CMasterGraphElement *container)
-{
-    if (container->queryLocalOrGrouped())
-        return new CMasterActivity(container);
-    else
-        return new CIfActionMaster(container);
-}
-

thorlcr/activities/when/thwhen.ipp

@@ -22,7 +22,6 @@
 
 
 CActivityBase *createWhenActivityMaster(CMasterGraphElement *info);
-CActivityBase *createIfActionActivityMaster(CMasterGraphElement *info);
 
 
 #endif

thorlcr/activities/when/thwhenslave.cpp

@@ -132,50 +132,3 @@ CActivityBase *createWhenSlave(CGraphElementBase *container)
 {
     return new CWhenSlaveActivity(container);
 }
-
-///////////
-
-class CIfActionSlaveActivity : public ProcessSlaveActivity, public CDependencyExecutorSlaveActivity
-{
-public:
-    IMPLEMENT_IINTERFACE_USING(CDependencyExecutorSlaveActivity);
-
-    CIfActionSlaveActivity(CGraphElementBase *_container) : ProcessSlaveActivity(_container), CDependencyExecutorSlaveActivity(this)
-    {
-    }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
-    {
-        CDependencyExecutorSlaveActivity::init(data, slaveData);
-    }
-    void preStart(size32_t parentExtractSz, const byte *parentExtract)
-    {
-        CDependencyExecutorSlaveActivity::preStart(parentExtractSz, parentExtract);
-    }
-    virtual void process()
-    {
-        processed = THORDATALINK_STARTED;
-        IHThorIfArg *helper = (IHThorIfArg *)queryHelper();
-        int controlId = helper->getCondition() ? 1 : 2;
-        if (!executeDependencies(controlId))
-            abortSoon = true;
-    }
-    virtual void endProcess()
-    {
-        if (processed & THORDATALINK_STARTED)
-            processed |= THORDATALINK_STOPPED;
-    }
-    virtual void abort()
-    {
-        ProcessSlaveActivity::abort();
-        if (global)
-            barrier->cancel();
-    }
-};
-
-////////////////////
-
-CActivityBase *createIfActionSlave(CGraphElementBase *container)
-{
-    return new CIfActionSlaveActivity(container);
-}
-

thorlcr/graph/thgraph.cpp

@@ -507,7 +507,6 @@ void CGraphElementBase::addAssociatedChildGraph(CGraphBase *childGraph)
 void CGraphElementBase::serializeCreateContext(MemoryBuffer &mb)
 {
     if (!onCreateCalled) return;
-    mb.append(queryId());
     DelayedSizeMarker sizeMark(mb);
     queryHelper()->serializeCreateContext(mb);
     sizeMark.write();
@@ -515,7 +514,7 @@ void CGraphElementBase::serializeCreateContext(MemoryBuffer &mb)
 
 void CGraphElementBase::serializeStartContext(MemoryBuffer &mb)
 {
-    assertex(onStartCalled);
+    if (!onStartCalled) return;
     DelayedSizeMarker sizeMark(mb);
     queryHelper()->serializeStartContext(mb);
     sizeMark.write();
@@ -625,12 +624,22 @@ bool CGraphElementBase::prepareContext(size32_t parentExtractSz, const byte *par
                 owner->ifs.append(*this);
                 // fall through
             case TAKif:
+            case TAKifaction:
             {
                 if (_shortCircuit) return true;
                 onCreate();
                 onStart(parentExtractSz, parentExtract);
                 IHThorIfArg *helper = (IHThorIfArg *)baseHelper.get();
-                whichBranch = helper->getCondition() ? 0 : 1;       // True argument preceeds false...
+                whichBranch = helper->getCondition() ? 0 : 1;       // True argument precedes false...
+                /* NB - The executeDependencies code below is only needed if actionLinkInNewGraph=true, which is no longer the default
+                 * It should be removed, once we are positive there are no issues with in-line conditional actions
+                 */
+                if (TAKifaction == getKind())
+                {
+                    if (!executeDependencies(parentExtractSz, parentExtract, whichBranch+1, async)) //NB whenId 1 based
+                        return false;
+                }
+
                 if (inputs.queryItem(whichBranch))
                 {
                     if (!whichBranchBitSet->testSet(whichBranch)) // if not set, new
@@ -678,6 +687,19 @@ bool CGraphElementBase::prepareContext(size32_t parentExtractSz, const byte *par
                     return true;
                 break;
             }
+            case TAKsequential:
+            case TAKparallel:
+            {
+                /* NB - The executeDependencies code below is only needed if actionLinkInNewGraph=true, which is no longer the default
+                 * It should be removed, once we are positive there are no issues with in-line sequential/parallel activities
+                 */
+                for (unsigned s=1; s<=dependsOn.ordinality(); s++)
+                {
+                    if (!executeDependencies(parentExtractSz, parentExtract, s, async))
+                        return false;
+                }
+                break;
+            }
         }
         ForEachItemIn(i, inputs)
         {
@@ -763,6 +785,26 @@ void CGraphElementBase::createActivity(size32_t parentExtractSz, const byte *par
                         factorySet(TAKnull);
                 }
                 break;
+            case TAKifaction:
+                if (inputs.queryItem(whichBranch))
+                {
+                    CGraphElementBase *input = inputs.item(whichBranch)->activity;
+                    input->createActivity(parentExtractSz, parentExtract);
+                }
+                break;
+            case TAKsequential:
+            case TAKparallel:
+            {
+                ForEachItemIn(i, inputs)
+                {
+                    if (inputs.queryItem(i))
+                    {
+                        CGraphElementBase *input = inputs.item(i)->activity;
+                        input->createActivity(parentExtractSz, parentExtract);
+                    }
+                }
+                break;
+            }
             default:
                 if (!isEof)
                 {
@@ -1060,16 +1102,21 @@ void CGraphBase::clean()
     disconnectActivities();
     containers.kill();
     sinks.kill();
+    connectedSinks.kill();
 }
 
 void CGraphBase::serializeCreateContexts(MemoryBuffer &mb)
 {
     DelayedSizeMarker sizeMark(mb);
-    Owned<IThorActivityIterator> iter = (queryOwner() && !isGlobal()) ? getIterator() : getTraverseIterator(true); // all if non-global-child, or graph with conditionals
+    Owned<IThorActivityIterator> iter = getIterator();
     ForEach (*iter)
     {
         CGraphElementBase &element = iter->query();
-        element.serializeCreateContext(mb);
+        if (element.isOnCreated())
+        {
+            mb.append(element.queryId());
+            element.serializeCreateContext(mb);
+        }
     }
     mb.append((activity_id)0);
     sizeMark.write();
@@ -1078,12 +1125,15 @@ void CGraphBase::serializeCreateContexts(MemoryBuffer &mb)
 void CGraphBase::serializeStartContexts(MemoryBuffer &mb)
 {
     DelayedSizeMarker sizeMark(mb);
-    Owned<IThorActivityIterator> iter = getTraverseIterator();
+    Owned<IThorActivityIterator> iter = getIterator();
     ForEach (*iter)
     {
         CGraphElementBase &element = iter->query();
-        mb.append(element.queryId());
-        element.serializeStartContext(mb);
+        if (element.isOnStarted())
+        {
+            mb.append(element.queryId());
+            element.serializeStartContext(mb);
+        }
     }
     mb.append((activity_id)0);
     sizeMark.write();
@@ -1127,10 +1177,10 @@ void CGraphBase::reset()
     }
     else
     {
-        CGraphElementIterator iterC(containers);
-        ForEach(iterC)
+        Owned<IThorActivityIterator> iter = getIterator();
+        ForEach(*iter)
         {
-            CGraphElementBase &element = iterC.query();
+            CGraphElementBase &element = iter->query();
             element.reset();
         }
         dependentSubGraphs.kill();
@@ -1162,7 +1212,7 @@ bool CGraphBase::fireException(IException *e)
 
 bool CGraphBase::preStart(size32_t parentExtractSz, const byte *parentExtract)
 {
-    Owned<IThorActivityIterator> iter = getTraverseIterator();
+    Owned<IThorActivityIterator> iter = getConnectedIterator();
     ForEach(*iter)
     {
         CGraphElementBase &element = iter->query();
@@ -1270,7 +1320,7 @@ void CGraphBase::doExecute(size32_t parentExtractSz, const byte *parentExtract,
     {
         if (started)
             reset();
-        Owned<IThorActivityIterator> iter = getTraverseIterator();
+        Owned<IThorActivityIterator> iter = getConnectedIterator();
         ForEach(*iter)
         {
             CGraphElementBase &element = iter->query();
@@ -1349,10 +1399,10 @@ bool CGraphBase::prepare(size32_t parentExtractSz, const byte *parentExtract, bo
 
 void CGraphBase::create(size32_t parentExtractSz, const byte *parentExtract)
 {
-    CGraphElementIterator iterC(containers);
-    ForEach(iterC)
+    Owned<IThorActivityIterator> iter = getIterator();
+    ForEach(*iter)
     {
-        CGraphElementBase &element = iterC.query();
+        CGraphElementBase &element = iter->query();
         element.clearConnections();
     }
     ForEachItemIn(s, sinks)
@@ -1360,13 +1410,20 @@ void CGraphBase::create(size32_t parentExtractSz, const byte *parentExtract)
         CGraphElementBase &sink = sinks.item(s);
         sink.createActivity(parentExtractSz, parentExtract);
     }
+    connectedSinks.kill();
+    ForEach(*iter)
+    {
+        CGraphElementBase &element = iter->query();
+        if (element.queryActivity() && 0 == element.connectedOutputs.ordinality())
+            connectedSinks.append(*LINK(&element));
+    }
     created = true;
 }
 
 void CGraphBase::done()
 {
     if (aborted) return; // activity done methods only called on success
-    Owned<IThorActivityIterator> iter = getTraverseIterator();
+    Owned<IThorActivityIterator> iter = getConnectedIterator();
     ForEach (*iter)
     {
         CGraphElementBase &element = iter->query();
@@ -1515,6 +1572,7 @@ public:
             switch (cur->getKind())
             {
                 case TAKif:
+                case TAKifaction:
                 case TAKchildif:
                 case TAKchildcase:
                 case TAKcase:
@@ -1552,12 +1610,9 @@ public:
     }
 };
 
-IThorActivityIterator *CGraphBase::getTraverseIterator(bool all)
+IThorActivityIterator *CGraphBase::getConnectedIterator()
 {
-    if (all)
-        return new CGraphTraverseIterator(*this); // all sinks + conditionals
-    else
-        return new CGraphTraverseConnectedIterator(*this);
+    return new CGraphTraverseConnectedIterator(*this);
 }
 
 bool CGraphBase::wait(unsigned timeout)
@@ -1585,7 +1640,7 @@ bool CGraphBase::wait(unsigned timeout)
             throw MakeGraphException(graph, 0, "Timed out waiting for graph to end");
         }
     } waitException(this);
-    Owned<IThorActivityIterator> iter = getTraverseIterator();
+    Owned<IThorActivityIterator> iter = getConnectedIterator();
     ForEach (*iter)
     {
         CGraphElementBase &element = iter->query();
@@ -1647,7 +1702,7 @@ void CGraphBase::abort(IException *e)
     }
     if (started && !graphDone)
     {
-        Owned<IThorActivityIterator> iter = getTraverseIterator();
+        Owned<IThorActivityIterator> iter = getConnectedIterator();
         ForEach (*iter)
         {
             iter->query().abort(e); // JCSMORE - could do in parallel, they can take some time to timeout
@@ -1684,9 +1739,9 @@ void CGraphBase::GraphPrintLog(IException *e)
 
 void CGraphBase::setLogging(bool tf)
 {
-    CGraphElementIterator iterC(containers);
-    ForEach(iterC)
-        iterC.query().setLogging(tf);
+    Owned<IThorActivityIterator> iter = getIterator();
+    ForEach(*iter)
+        iter->query().setLogging(tf);
 }
 
 void CGraphBase::createFromXGMML(IPropertyTree *_node, CGraphBase *_owner, CGraphBase *_parent, CGraphBase *resultsGraph)
@@ -1767,6 +1822,20 @@ void CGraphBase::createFromXGMML(IPropertyTree *_node, CGraphBase *_owner, CGrap
         CGraphElementBase *target = queryElement(edge.getPropInt("@target"));
         target->addInput(targetInput, source, sourceOutput);
     }
+    Owned<IThorActivityIterator> iter = getIterator();
+    ForEach(*iter)
+    {
+        CGraphElementBase &element = iter->query();
+        if (0 == element.getOutputs())
+        {
+            /* JCSMORE - Making some outputs conditional, will require:
+             * a) Pass through information as to which dependent graph causes this graph (and this sink) to execute)
+             * b) Allow the subgraph to re-executed by other dependent subgraphs and avoid re-executing completed sinks
+             * c) Keep common points (splitters) around (preferably in memory), re-execution of graph will need them
+             */
+            sinks.append(*LINK(&element));
+        }
+    }
     init();
 }
 
@@ -2538,10 +2607,22 @@ static void getGlobalDeps(CGraphBase &graph, CopyCIArrayOf<CGraphDependency> &de
     }
 }
 
+static void noteDependency(CGraphElementBase *targetActivity, CGraphElementBase *sourceActivity, CGraphBase *targetGraph, CGraphBase *sourceGraph, unsigned controlId)
+{
+    targetActivity->addDependsOn(sourceGraph, controlId);
+    // NB: record dependency in source graph, serialized to slaves, used to decided if should run dependency sinks or not
+    Owned<IPropertyTree> dependencyFor = createPTree();
+    dependencyFor->setPropInt("@id", sourceActivity->queryId());
+    dependencyFor->setPropInt("@graphId", targetGraph->queryGraphId());
+    if (controlId)
+        dependencyFor->setPropInt("@conditionalId", controlId);
+    sourceGraph->queryXGMML().addPropTree("Dependency", dependencyFor.getClear());
+}
+
 void CJobBase::addDependencies(IPropertyTree *xgmml, bool failIfMissing)
 {
     CGraphArrayCopy childGraphs;
-    CGraphElementArrayCopy targetActivities;
+    CGraphElementArrayCopy targetActivities, sourceActivities;
 
     Owned<IPropertyTreeIterator> iter = xgmml->getElements("edge");
     ForEach(*iter)
@@ -2577,7 +2658,7 @@ void CJobBase::addDependencies(IPropertyTree *xgmml, bool failIfMissing)
                 targetActivity = targetGraph->queryElement(targetGraphContext);
             }
             assertex(targetActivity && sourceActivity);
-            targetActivity->addDependsOn(source, controlId);
+            noteDependency(targetActivity, sourceActivity, target, source, controlId);
         }
         else if (edge.getPropBool("att[@name=\"_conditionSource\"]/@value", false))
         { /* Ignore it */ }
@@ -2586,18 +2667,20 @@ void CJobBase::addDependencies(IPropertyTree *xgmml, bool failIfMissing)
             // NB: any dependencies of the child acts. are dependencies of this act.
             childGraphs.append(*source);
             targetActivities.append(*targetActivity);
+            sourceActivities.append(*sourceActivity);
         }
         else
         {
             if (!edge.getPropBool("att[@name=\"_childGraph\"]/@value", false)) // JCSMORE - not sure if necess. roxie seem to do.
                 controlId = edge.getPropInt("att[@name=\"_when\"]/@value", 0);
-            targetActivity->addDependsOn(source, controlId);
+            noteDependency(targetActivity, sourceActivity, target, source, controlId);
         }
     }
     ForEachItemIn(c, childGraphs)
     {
         CGraphBase &childGraph = childGraphs.item(c);
         CGraphElementBase &targetActivity = targetActivities.item(c);
+        CGraphElementBase &sourceActivity = sourceActivities.item(c);
         if (!childGraph.isGlobal())
         {
             CopyCIArrayOf<CGraphDependency> globalChildGraphDeps;
@@ -2605,7 +2688,7 @@ void CJobBase::addDependencies(IPropertyTree *xgmml, bool failIfMissing)
             ForEachItemIn(gcd, globalChildGraphDeps)
             {
                 CGraphDependency &globalDep = globalChildGraphDeps.item(gcd);
-                targetActivity.addDependsOn(globalDep.graph, globalDep.controlId);
+                noteDependency(&targetActivity, &sourceActivity, globalDep.graph, &childGraph, globalDep.controlId);
             }
         }
     }

thorlcr/graph/thgraph.hpp

@@ -303,8 +303,10 @@ public:
     void onCreate();
     void abort(IException *e);
     virtual void preStart(size32_t parentExtractSz, const byte *parentExtract);
-    const bool &isOnCreated() const { return onCreateCalled; }
-    const bool &isPrepared() const { return prepared; }
+    bool isOnCreated() const { return onCreateCalled; }
+    bool isOnStarted() const { return onStartCalled; }
+    bool isPrepared() const { return prepared; }
+
     CGraphBase &queryOwner() const { return *owner; }
     CGraphBase *queryResultsGraph() const { return resultsGraph; }
     IThorGraphIterator *getAssociatedChildGraphs() const;
@@ -426,7 +428,7 @@ class graph_decl CGraphBase : public CInterface, implements IEclGraphResults, im
     mutable CriticalSection crit;
     CriticalSection evaluateCrit;
     CGraphElementTable containers;
-    CGraphElementArray sinks;
+    CGraphElementArray sinks, connectedSinks;
     bool sink, complete, global, localChild;
     mutable int localOnly;
     activity_id parentActivityId;
@@ -546,56 +548,6 @@ protected:
     unsigned counter;
     CReplyCancelHandler graphCancelHandler;
 
-    class CGraphGraphActElementIterator : public CInterface, implements IThorActivityIterator
-    {
-    protected:
-        CGraphBase &graph;
-        IPropertyTree &xgmml;
-        Owned<IPropertyTreeIterator> iter;
-        CGraphElementBase *current;
-    public:
-        IMPLEMENT_IINTERFACE;
-
-        CGraphGraphActElementIterator(CGraphBase &_graph, IPropertyTree &_xgmml) : graph(_graph), xgmml(_xgmml)
-        {
-            iter.setown(xgmml.getElements("node"));
-        }
-        virtual bool first()
-        {
-            if (iter->first())
-            {
-                IPropertyTree &node = iter->query();
-                current = graph.queryElement(node.getPropInt("@id"));
-                if (current)
-                    return true;
-                else if (next())
-                    return true;
-            }
-            current = NULL;
-            return false;
-        }
-        virtual bool next()
-        {
-            loop
-            {
-                if (!iter->next())
-                    break;
-                IPropertyTree &node = iter->query();
-                current = graph.queryElement(node.getPropInt("@id"));
-                if (current)
-                    return true;
-            }
-            current = NULL;
-            return false;
-        }
-        virtual bool isValid() { return NULL!=current; }
-        virtual CGraphElementBase & query()
-        {
-            return *current;
-        }
-        CGraphElementBase & get() { CGraphElementBase &c = query(); c.Link(); return c; }
-    };
-
 public:
     IMPLEMENT_IINTERFACE;
 
@@ -604,16 +556,15 @@ public:
 
     CGraphBase(CJobBase &job);
     ~CGraphBase();
-    
+
     const void *queryFindParam() const { return &queryGraphId(); } // for SimpleHashTableOf
 
     virtual void init() { }
-    IThorActivityIterator *getTraverseIterator(bool all=false); // all traverses and includes conditionals, others traverses connected nodes only
     void GraphPrintLog(const char *msg, ...) __attribute__((format(printf, 2, 3)));
     void GraphPrintLog(IException *e, const char *msg, ...) __attribute__((format(printf, 3, 4)));
     void GraphPrintLog(IException *e);
     void createFromXGMML(IPropertyTree *node, CGraphBase *owner, CGraphBase *parent, CGraphBase *resultsGraph);
-    const bool &queryAborted() const { return aborted; }
+    bool queryAborted() const { return aborted; }
     CJobBase &queryJob() const { return job; }
     IGraphTempHandler *queryTempHandler() const { assertex(tmpHandler.get()); return tmpHandler; }
     CGraphBase *queryOwner() { return owner; }
@@ -659,11 +610,12 @@ public:
     virtual void execute(size32_t parentExtractSz, const byte *parentExtract, bool checkDependencies, bool async);
     IThorActivityIterator *getIterator()
     {
-        return new CGraphGraphActElementIterator(*this, *xgmml);
+        return new CGraphElementIterator(containers);
     }
+    IThorActivityIterator *getConnectedIterator();
     IThorActivityIterator *getSinkIterator() const
     {
-        return new CGraphElementArrayIterator(sinks);
+        return new CGraphElementArrayIterator(connectedSinks);
     }
     IPropertyTree &queryXGMML() const { return *xgmml; }
     void addActivity(CGraphElementBase *element)
@@ -673,16 +625,7 @@ public:
             element->Release();
             return;
         }
-
         containers.replace(*element);
-        if (element->isSink())
-            sinks.append(*LINK(element));
-    }
-    bool removeActivity(CGraphElementBase *element)
-    {
-        bool res = containers.removeExact(element);
-        sinks.zap(* element);
-        return res;
     }
     unsigned activityCount() const
     {
@@ -847,7 +790,7 @@ public:
     void init();
     void setXGMML(IPropertyTree *_xgmml) { xgmml.set(_xgmml); }
     IPropertyTree *queryXGMML() { return xgmml; }
-    const bool &queryAborted() const { return aborted; }
+    bool queryAborted() const { return aborted; }
     const char *queryKey() const { return key; }
     const char *queryGraphName() const { return graphName; }
     bool queryForceLogging(graph_id graphId, bool def) const;
@@ -888,8 +831,8 @@ public:
     IThorResult *getOwnedResult(graph_id gid, activity_id ownerId, unsigned resultId);
     IThorAllocator *queryThorAllocator() const { return thorAllocator; }
     bool queryUseCheckpoints() const;
-    const bool &queryPausing() const { return pausing; }
-    const bool &queryResumed() const { return resumed; }
+    bool queryPausing() const { return pausing; }
+    bool queryResumed() const { return resumed; }
     IGraphTempHandler *queryTempHandler() const { return tmpHandler; }
     ILoadedDllEntry &queryDllEntry() const { return *querySo; }
     ICodeContext &queryCodeContext() const;
@@ -910,7 +853,7 @@ public:
     unsigned querySlaves() const { return slaveGroup->ordinality(); }
     ICommunicator &queryJobComm() const { return *jobComm; }
     IGroup &queryJobGroup() const { return *jobGroup; }
-    const bool &queryTimeActivities() const { return timeActivities; }
+    bool queryTimeActivities() const { return timeActivities; }
     unsigned queryMaxDefaultActivityCores() const { return maxActivityCores; }
     IGroup &querySlaveGroup() const { return *slaveGroup; }
     const rank_t &queryMyRank() const { return myrank; }
@@ -972,10 +915,10 @@ public:
     CJobBase &queryJob() const { return container.queryJob(); }
     CGraphBase &queryGraph() const { return container.queryOwner(); }
     inline const mptag_t queryMpTag() const { return mpTag; }
-    inline const bool &queryAbortSoon() const { return abortSoon; }
+    inline bool queryAbortSoon() const { return abortSoon; }
     inline IHThorArg *queryHelper() const { return baseHelper; }
     inline bool needReInit() const { return reInit; }
-    inline const bool &queryTimeActivities() const { return timeActivities; } 
+    inline bool queryTimeActivities() const { return timeActivities; }
     void onStart(size32_t _parentExtractSz, const byte *_parentExtract) { parentExtractSz = _parentExtractSz; parentExtract = _parentExtract; }
     bool receiveMsg(CMessageBuffer &mb, const rank_t rank, const mptag_t mpTag, rank_t *sender=NULL, unsigned timeout=MP_WAIT_FOREVER);
     void cancelReceiveMsg(const rank_t rank, const mptag_t mpTag);

thorlcr/graph/thgraphmaster.cpp

@@ -2117,7 +2117,7 @@ void CMasterGraph::abort(IException *e)
 void CMasterGraph::serializeCreateContexts(MemoryBuffer &mb)
 {
     CGraphBase::serializeCreateContexts(mb);
-    Owned<IThorActivityIterator> iter = (queryOwner() && !isGlobal()) ? getIterator() : getTraverseIterator();
+    Owned<IThorActivityIterator> iter = getIterator();
     ForEach (*iter)
     {
         CMasterGraphElement &element = (CMasterGraphElement &)iter->query();
@@ -2234,7 +2234,7 @@ void CMasterGraph::create(size32_t parentExtractSz, const byte *parentExtract)
 
 void CMasterGraph::start()
 {
-    Owned<IThorActivityIterator> iter = getTraverseIterator();
+    Owned<IThorActivityIterator> iter = getConnectedIterator();
     ForEach (*iter)
         iter->query().queryActivity()->startProcess();
 }
@@ -2250,7 +2250,7 @@ void CMasterGraph::sendActivityInitData()
     for (; w<queryJob().querySlaves(); w++)
     {
         unsigned needActInit = 0;
-        Owned<IThorActivityIterator> iter = getTraverseIterator();
+        Owned<IThorActivityIterator> iter = getConnectedIterator();
         ForEach(*iter)
         {
             CGraphElementBase &element = iter->query();
@@ -2265,7 +2265,7 @@ void CMasterGraph::sendActivityInitData()
             try
             {
                 msg.rewrite(pos);
-                Owned<IThorActivityIterator> iter = getTraverseIterator();
+                Owned<IThorActivityIterator> iter = getConnectedIterator();
                 serializeActivityInitData(w, msg, *iter);
             }
             catch (IException *e)

thorlcr/graph/thgraphslave.cpp

@@ -392,7 +392,7 @@ bool CSlaveGraph::recvActivityInitData(size32_t parentExtractSz, const byte *par
 {
     bool ret = true;
     unsigned needActInit = 0;
-    Owned<IThorActivityIterator> iter = getTraverseIterator();
+    Owned<IThorActivityIterator> iter = getConnectedIterator();
     ForEach(*iter)
     {
         CGraphElementBase &element = (CGraphElementBase &)iter->query();
@@ -426,7 +426,7 @@ bool CSlaveGraph::recvActivityInitData(size32_t parentExtractSz, const byte *par
             assertex(!parentExtractSz || NULL!=parentExtract);
             msg.append(parentExtractSz);
             msg.append(parentExtractSz, parentExtract);
-            Owned<IThorActivityIterator> iter = getTraverseIterator();
+            Owned<IThorActivityIterator> iter = getConnectedIterator();
             ForEach(*iter)
             {
                 CSlaveGraphElement &element = (CSlaveGraphElement &)iter->query();
@@ -452,7 +452,7 @@ bool CSlaveGraph::recvActivityInitData(size32_t parentExtractSz, const byte *par
             if (queryOwner() && !isGlobal())
             {
                 // initialize any for which no data was sent
-                Owned<IThorActivityIterator> iter = getTraverseIterator();
+                Owned<IThorActivityIterator> iter = getConnectedIterator();
                 ForEach(*iter)
                 {
                     CSlaveGraphElement &element = (CSlaveGraphElement &)iter->query();
@@ -530,7 +530,7 @@ void CSlaveGraph::start()
 void CSlaveGraph::connect()
 {
     CriticalBlock b(progressCrit);
-    Owned<IThorActivityIterator> iter = getTraverseIterator();
+    Owned<IThorActivityIterator> iter = getConnectedIterator();
     ForEach(*iter)
         iter->query().doconnect();
 }
@@ -708,7 +708,7 @@ bool CSlaveGraph::serializeStats(MemoryBuffer &mb)
         if (collect)
         {
             unsigned sPos = mb.length();
-            Owned<IThorActivityIterator> iter = getTraverseIterator();
+            Owned<IThorActivityIterator> iter = getConnectedIterator();
             ForEach (*iter)
             {
                 CGraphElementBase &element = iter->query();
@@ -749,7 +749,7 @@ void CSlaveGraph::serializeDone(MemoryBuffer &mb)
     unsigned cPos = mb.length();
     unsigned count=0;
     mb.append(count);
-    Owned<IThorActivityIterator> iter = getTraverseIterator();
+    Owned<IThorActivityIterator> iter = getConnectedIterator();
     ForEach (*iter)
     {
         CGraphElementBase &element = iter->query();

thorlcr/master/thactivitymaster.cpp

@@ -383,13 +383,11 @@ public:
                 UNIMPLEMENTED;
             case TAKcase:           // gen. time.
             case TAKif:
+            case TAKifaction:
                 throwUnexpected();
             case TAKwhen_dataset:
                 ret = createWhenActivityMaster(this);
                 break;
-            case TAKifaction:
-                ret = createIfActionActivityMaster(this);
-                break;
             default:
                 throw MakeActivityException(this, TE_UnsupportedActivityKind, "Unsupported activity kind: %s", activityKindStr(kind));
         }

thorlcr/master/thdemonserver.cpp

@@ -41,7 +41,11 @@ private:
     
     void doReportGraph(IWUGraphProgress *progress, CGraphBase *graph, bool finished)
     {
-        Owned<IThorActivityIterator> iter = (graph->queryOwner() && !graph->isGlobal()) ? graph->getIterator() : graph->getTraverseIterator();
+        Owned<IThorActivityIterator> iter;
+        if (graph->queryOwner() && !graph->isGlobal())
+            iter.setown(graph->getIterator()); // Local child graphs still send progress, but aren't connected in master
+        else
+            iter.setown(graph->getConnectedIterator());
         ForEach (*iter)
         {
             CMasterGraphElement &container = (CMasterGraphElement &)iter->query();

thorlcr/slave/slave.cpp

@@ -255,7 +255,6 @@ CActivityBase *createChildAggregateSlave(CGraphElementBase *container);
 CActivityBase *createChildGroupAggregateSlave(CGraphElementBase *container);
 CActivityBase *createChildThroughNormalizeSlave(CGraphElementBase *container);
 CActivityBase *createWhenSlave(CGraphElementBase *container);
-CActivityBase *createIfActionSlave(CGraphElementBase *container);
 CActivityBase *createDictionaryWorkunitWriteSlave(CGraphElementBase *container);
 CActivityBase *createDictionaryResultWriteSlave(CGraphElementBase *container);
 
@@ -621,6 +620,7 @@ public:
                 break;
             case TAKcase:
             case TAKif:
+            case TAKifaction:
                 throwUnexpected();
                 break;
             case TAKwhen_dataset:
@@ -743,9 +743,6 @@ public:
             case TAKstreamediterator:
                 ret = createStreamedIteratorSlave(this);
                 break;
-            case TAKifaction:
-                ret = createIfActionSlave(this);
-                break;
             default:
                 throw MakeStringException(TE_UnsupportedActivityKind, "Unsupported activity kind: %s", activityKindStr(kind));
         }