Merge pull request #8499 from jakesmith/hpcc-15320

HPCC-15320 Refactor graphs to avoid reconnecting graphs

Reviewed-by: Gavin Halliday <ghalliday@hpccsystems.com>

thorlcr/activities/catch/thcatchslave.cpp

@@ -196,7 +196,7 @@ public:
     {
         global = !container->queryLocalOrGrouped();
     }
-    virtual void init(MemoryBuffer & data, MemoryBuffer &slaveData)
+    virtual void init(MemoryBuffer & data, MemoryBuffer &slaveData) override
     {
         CCatchSlaveActivityBase::init(data, slaveData);
         if (global)
@@ -205,7 +205,7 @@ public:
             barrier.setown(container.queryJobChannel().createBarrier(barrierTag));
         }
     }
-    virtual void start()
+    virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
         CCatchSlaveActivityBase::start();
@@ -257,10 +257,6 @@ public:
         dataLinkIncrement();
         return row.getClear();
     }
-    virtual void stop()
-    {
-        CCatchSlaveActivityBase::stop();
-    }
     virtual void abort()
     {
         CCatchSlaveActivityBase::abort();

thorlcr/activities/fetch/thfetchslave.cpp

@@ -203,8 +203,11 @@ public:
             keyOutStream->stop();
             keyOutStream.clear();
         }
-        distributor->disconnect(true);  
-        distributor->join();
+        if (distributor)
+        {
+            distributor->disconnect(true);
+            distributor->join();
+        }
         stopInput();
     }
     const void *nextRow()
@@ -268,34 +271,31 @@ class CFetchSlaveBase : public CSlaveActivity, implements IFetchHandler
 {
     typedef CSlaveActivity PARENT;
 
-    IRowStream *fetchStreamOut;
-    unsigned maxKeyRecSize;
-    rowcount_t limit;
-    unsigned offsetCount;
-    unsigned offsetMapSz;
+    IRowStream *fetchStreamOut = nullptr;
+    unsigned maxKeyRecSize = 0;
+    rowcount_t limit = 0;
+    unsigned offsetCount = 0;
+    unsigned offsetMapSz = 0;
     MemoryBuffer offsetMapBytes;
     Owned<IExpander> eexp;
     Owned<IEngineRowAllocator> keyRowAllocator;
 
 protected:
     Owned<IThorRowInterfaces> fetchDiskRowIf;
-    IFetchStream *fetchStream;
+    IFetchStream *fetchStream = nullptr;
     IHThorFetchBaseArg *fetchBaseHelper;
     IHThorFetchContext *fetchContext;
-    unsigned files;
+    unsigned files = 0;
     CPartDescriptorArray parts;
-    IRowStream *keyIn;
-    bool indexRowExtractNeeded;
-    mptag_t mptag;
+    IRowStream *keyIn = nullptr;
+    bool indexRowExtractNeeded = false;
+    mptag_t mptag = TAG_NULL;
 
 public:
     IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
 
     CFetchSlaveBase(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
-        fetchStream = NULL;
-        keyIn = NULL;
-        fetchStreamOut = NULL;
         fetchBaseHelper = (IHThorFetchBaseArg *)queryHelper();
         fetchContext = static_cast<IHThorFetchContext *>(fetchBaseHelper->selectInterface(TAIfetchcontext_1));
         reInit = 0 != (fetchContext->getFetchFlags() & (FFvarfilename|FFdynamicfilename));
@@ -326,12 +326,8 @@ public:
         if (!container.queryLocalOrGrouped())
             mptag = container.queryJobChannel().deserializeMPTag(data);
 
-        indexRowExtractNeeded = fetchBaseHelper->transformNeedsRhs();
-
         files = parts.ordinality();
 
-        limit = (rowcount_t)fetchBaseHelper->getRowLimit(); // MORE - if no filtering going on could keyspan to get count
-
         unsigned encryptedKeyLen;
         void *encryptedKey;
         fetchContext->getFileEncryptKey(encryptedKeyLen,encryptedKey);
@@ -343,13 +339,7 @@ public:
             memset(encryptedKey, 0, encryptedKeyLen);
             free(encryptedKey);
         }
-        fetchDiskRowIf.setown(createThorRowInterfaces(queryRowManager(), fetchContext->queryDiskRecordSize(),queryId(),queryCodeContext()));
-        if (fetchBaseHelper->extractAllJoinFields())
-        {
-            IOutputMetaData *keyRowMeta = QUERYINTERFACE(fetchBaseHelper->queryExtractedSize(), IOutputMetaData);
-            assertex(keyRowMeta);
-            keyRowAllocator.setown(getRowAllocator(keyRowMeta));
-        }
+        fetchDiskRowIf.setown(createThorRowInterfaces(queryRowManager(), fetchContext->queryDiskRecordSize(), queryId(), queryCodeContext()));
         appendOutputLinked(this);
     }
 
@@ -363,6 +353,18 @@ public:
         ActivityTimer s(totalCycles, timeActivities);
         PARENT::start();
 
+        if (!keyRowAllocator && fetchBaseHelper->extractAllJoinFields())
+        {
+            IOutputMetaData *keyRowMeta = QUERYINTERFACE(fetchBaseHelper->queryExtractedSize(), IOutputMetaData);
+            assertex(keyRowMeta);
+            keyRowAllocator.setown(getRowAllocator(keyRowMeta));
+        }
+
+        limit = (rowcount_t)fetchBaseHelper->getRowLimit(); // MORE - if no filtering going on could keyspan to get count
+
+        // NB: indexRowExtractNeeded is a member variable, because referenced by callback IFetchHandler::extractFpos()
+        indexRowExtractNeeded = fetchBaseHelper->transformNeedsRhs();
+
         class CKeyFieldExtractBase : public CSimpleInterface, implements IRowStream
         {
         protected:
@@ -461,7 +463,8 @@ public:
     }
     virtual void stop() override
     {
-        fetchStreamOut->stop();
+        if (queryInputStarted(0))
+            fetchStreamOut->stop();
         dataLinkStop();
     }
     virtual void abort()

thorlcr/activities/filter/thfilterslave.cpp

@@ -58,27 +58,40 @@ class CFilterSlaveActivity : public CFilterSlaveActivityBase, public CThorSteppa
 
     IHThorFilterArg *helper;
     unsigned matched;
+
 public:
     CFilterSlaveActivity(CGraphElementBase *container)
         : CFilterSlaveActivityBase(container), CThorSteppable(this)
     {
     }
-    void init(MemoryBuffer &data, MemoryBuffer &slaveData)
+    void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
     {
         PARENT::init(data,slaveData);
         helper = static_cast <IHThorFilterArg *> (queryHelper());
     }
-    void start()
+    virtual void start() override
     {   
         ActivityTimer s(totalCycles, timeActivities);
         matched = 0;
-        abortSoon = !helper->canMatchAny();
-        PARENT::start();
+        if (helper->canMatchAny())
+            PARENT::start();
+        else
+        {
+            dataLinkStart();
+            abortSoon = true;
+            stop();
+        }
+    }
+    virtual void stop() override
+    {
+        if (!queryInputStopped(0))
+            PARENT::stop();
+        // else already stopped
     }
     CATCH_NEXTROW()
     {
         ActivityTimer t(totalCycles, timeActivities);
-        while(!abortSoon)
+        while (!abortSoon)
         {
             OwnedConstThorRow row = inputStream->nextRow();
             if (!row)
@@ -92,7 +105,7 @@ public:
                 if (!row)
                     break;
             }
-            if(helper->isValid(row))
+            if (helper->isValid(row))
             {
                 matched++;
                 anyThisGroup = true;
@@ -100,9 +113,9 @@ public:
                 return row.getClear();
             }
         }
-        return NULL;
+        return nullptr;
     }
-    const void *nextRowGE(const void *seek, unsigned numFields, bool &wasCompleteMatch, const SmartStepExtra &stepExtra)
+    virtual const void *nextRowGE(const void *seek, unsigned numFields, bool &wasCompleteMatch, const SmartStepExtra &stepExtra) override
     {
         try { return nextRowGENoCatch(seek, numFields, wasCompleteMatch, stepExtra); }
         CATCH_NEXTROWX_CATCH;
@@ -140,11 +153,11 @@ public:
         }
         return NULL;
     }
-    bool gatherConjunctions(ISteppedConjunctionCollector &collector) 
+    virtual bool gatherConjunctions(ISteppedConjunctionCollector &collector) override
     { 
         return input->gatherConjunctions(collector); 
     }
-    void resetEOF() 
+    virtual void resetEOF() override
     { 
         abortSoon = !helper->canMatchAny();
         anyThisGroup = false;
@@ -171,18 +184,24 @@ public:
         : CFilterSlaveActivityBase(container)
     {
     }
-    void init(MemoryBuffer &data, MemoryBuffer &slaveData)
+    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
     {
         PARENT::init(data,slaveData);
         helper = static_cast <IHThorFilterProjectArg *> (queryHelper());
         allocator.set(queryRowAllocator());
     }
-    void start()
+    virtual void start() override
     {   
         ActivityTimer s(totalCycles, timeActivities);
-        abortSoon = !helper->canMatchAny();
         recordCount = 0;
-        PARENT::start();
+        if (helper->canMatchAny())
+            PARENT::start();
+        else
+        {
+            dataLinkStart();
+            abortSoon = true;
+            stopInput(0);
+        }
     }
     CATCH_NEXTROW()
     {
@@ -260,8 +279,14 @@ public:
     virtual void start() override
     {   
         ActivityTimer s(totalCycles, timeActivities);
-        abortSoon = !helper->canMatchAny();
-        PARENT::start();
+        if (helper->canMatchAny())
+            PARENT::start();
+        else
+        {
+            dataLinkStart();
+            abortSoon = true;
+            stopInput(0);
+        }
     }
     CATCH_NEXTROW()
     {

thorlcr/activities/firstn/thfirstnslave.cpp

@@ -45,17 +45,17 @@ public:
         stopped = true;
         helper = (IHThorFirstNArg *)container.queryHelper();
     }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
+    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
     {
         appendOutputLinked(this);
     }
-    virtual void start()
+    virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
         PARENT::start();
         stopped = false;
     }
-    virtual void stop()
+    virtual void stop() override
     {
         if (!stopped)
         {
@@ -216,6 +216,7 @@ class CFirstNSlaveGlobal : public CFirstNSlaveBase, implements ILookAheadStopNot
     rowcount_t maxres, skipped, totallimit;
     bool firstget;
     ThorDataLinkMetaInfo inputMeta;
+    Owned<IStartableEngineRowStream> firstNLookAhead;
 
 protected:
     virtual void doStop()
@@ -229,21 +230,12 @@ public:
     CFirstNSlaveGlobal(CGraphElementBase *container) : CFirstNSlaveBase(container)
     {
     }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
+    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
     {
         PARENT::init(data, slaveData);
         mpTag = container.queryJobChannel().deserializeMPTag(data);
     }
-    virtual void setInputStream(unsigned index, CThorInput &_input, bool consumerOrdered) override
-    {
-        PARENT::setInputStream(index, _input, consumerOrdered);
-        totallimit = (rowcount_t)helper->getLimit();
-        rowcount_t _skipCount = validRC(helper->numToSkip()); // max
-        rowcount_t maxRead = (totallimit>(RCUNBOUND-_skipCount))?RCUNBOUND:totallimit+_skipCount;
-        setLookAhead(0, createRowStreamLookAhead(this, inputStream, queryRowInterfaces(input), FIRSTN_SMART_BUFFER_SIZE, isSmartBufferSpillNeeded(this), false,
-                                          maxRead, this, &container.queryJob().queryIDiskUsage())); // if a very large limit don't bother truncating
-    }
-    virtual void start()
+    virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
         PARENT::start(); // adds to totalTime (common to local and global firstn)
@@ -252,6 +244,18 @@ public:
         skipped = 0;
         firstget = true;
         input->getMetaInfo(inputMeta);
+        totallimit = (rowcount_t)helper->getLimit();
+        rowcount_t _skipCount = validRC(helper->numToSkip()); // max
+        rowcount_t maxRead = (totallimit>(RCUNBOUND-_skipCount))?RCUNBOUND:totallimit+_skipCount;
+        firstNLookAhead.setown(createRowStreamLookAhead(this, inputStream, queryRowInterfaces(input), FIRSTN_SMART_BUFFER_SIZE, isSmartBufferSpillNeeded(this), false,
+                                          maxRead, this, &container.queryJob().queryIDiskUsage())); // if a very large limit don't bother truncating
+        firstNLookAhead->start();
+    }
+    virtual void stop() override
+    {
+        if (firstNLookAhead)
+            firstNLookAhead->stop(); // will call input stop()
+        dataLinkStop();
     }
     virtual void abort()
     {
@@ -333,7 +337,7 @@ public:
                     return NULL;
                 while (skipped<skipCount)
                 {
-                    OwnedConstThorRow row = inputStream->ungroupedNextRow();
+                    OwnedConstThorRow row = firstNLookAhead->ungroupedNextRow();
                     if (!row)
                     {
                         stop();
@@ -344,7 +348,7 @@ public:
             }
             if (getDataLinkCount() < limit)
             {
-                OwnedConstThorRow row = inputStream->ungroupedNextRow();
+                OwnedConstThorRow row = firstNLookAhead->ungroupedNextRow();
                 if (row)
                 {
                     dataLinkIncrement();
@@ -357,13 +361,13 @@ public:
     }
 
     virtual bool isGrouped() const override { return false; } // need to do different if is!
-    virtual void getMetaInfo(ThorDataLinkMetaInfo &info)
+    virtual void getMetaInfo(ThorDataLinkMetaInfo &info) override
     {
         PARENT::getMetaInfo(info);
         info.canBufferInput = true;
     }
 // ILookAheadStopNotify
-    virtual void onInputFinished(rowcount_t count)  // count is the total read from input (including skipped)
+    virtual void onInputFinished(rowcount_t count) override // count is the total read from input (including skipped)
     {
         // sneaky short circuit
         {

thorlcr/activities/funnel/thfunnelslave.cpp

@@ -288,7 +288,7 @@ public:
     {
         if (eog) delete [] eog;
     }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
+    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
     {
         IHThorFunnelArg *helper = (IHThorFunnelArg *)queryHelper();
         parallel = !container.queryGrouped() && !helper->isOrdered() && getOptBool(THOROPT_PARALLEL_FUNNEL, true);
@@ -296,7 +296,7 @@ public:
         appendOutputLinked(this);
         ActPrintLog("FUNNEL mode = %s, grouped=%s", parallel?"PARALLEL":"ORDERED", grouped?"GROUPED":"UNGROUPED");
     }
-    virtual void start()
+    virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
         if (!grouped && parallel)
@@ -333,7 +333,7 @@ public:
         }
         dataLinkStart();
     }
-    virtual void stop()
+    virtual void stop() override
     {
         if (parallelOutput)
         {

thorlcr/activities/hashdistrib/thhashdistribslave.cpp

@@ -2385,7 +2385,7 @@ public:
         if (lookup)
             delete lookup;
     }
-    void init(MemoryBuffer &data, MemoryBuffer &slaveData)
+    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
     {
         HashDistributeSlaveBase::init(data, slaveData);
 
@@ -2396,9 +2396,8 @@ public:
         Owned<IFileIO> iFileIO = createIFileI((size32_t)tlkSz, data.readDirect((size32_t)tlkSz));
 
         // NB: this TLK is an in-memory TLK serialized from the master - the name is for tracing by the key code only
-        OwnedRoxieString indexFileName(helper->getIndexFileName());
-        StringBuffer name(indexFileName);
-        name.append("_tlk");
+        VStringBuffer name("index");
+        name.append(queryId()).append("_tlk");
         lookup = new CKeyLookup(*this, helper, createKeyIndex(name.str(), 0, *iFileIO, true, false)); // MORE - crc is not 0...
         ihash = lookup;
     }
@@ -3423,8 +3422,6 @@ bool CBucketHandler::addRow(const void *row)
 class LocalHashDedupSlaveActivity : public HashDedupSlaveActivityBase
 {
 public:
-    IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
-
     LocalHashDedupSlaveActivity(CGraphElementBase *container)
         : HashDedupSlaveActivityBase(container, true)
     {
@@ -3449,8 +3446,6 @@ class GlobalHashDedupSlaveActivity : public HashDedupSlaveActivityBase, implemen
     Owned<IRowStream> instrm;
 
 public:
-    IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
-
     GlobalHashDedupSlaveActivity(CGraphElementBase *container)
         : HashDedupSlaveActivityBase(container, false)
     {
@@ -3478,7 +3473,7 @@ public:
         mptag = container.queryJobChannel().deserializeMPTag(data);
         distributor = createHashDistributor(this, queryJobChannel().queryJobComm(), mptag, true, this);
     }
-    virtual void start()
+    virtual void start() override
     {
         HashDedupSlaveActivityBase::start();
         ActivityTimer s(totalCycles, timeActivities);
@@ -3486,7 +3481,7 @@ public:
         instrm.setown(distributor->connect(myRowIf, distInput, iHash, iCompare));
         distInput = instrm.get();
     }
-    virtual void stop()
+    virtual void stop() override
     {
         ActPrintLog("stopping");
         if (instrm)
@@ -3494,8 +3489,11 @@ public:
             instrm->stop();
             instrm.clear();
         }
-        distributor->disconnect(true);
-        distributor->join();
+        if (distributor)
+        {
+            distributor->disconnect(true);
+            distributor->join();
+        }
         stopInput();
     }
     virtual void abort()
@@ -3541,9 +3539,6 @@ class HashJoinSlaveActivity : public CSlaveActivity, implements IStopInput
     Owned<IHashDistributor> lhsDistributor, rhsDistributor;
 
 public:
-
-    IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
-
     HashJoinSlaveActivity(CGraphElementBase *_container)
         : CSlaveActivity(_container)
     {
@@ -3654,8 +3649,11 @@ public:
         ActPrintLog("HASHJOIN: stopping");
         stopInputL();
         stopInputR();
-        lhsProgressCount = joinhelper->getLhsProgress();
-        rhsProgressCount = joinhelper->getRhsProgress();
+        if (joinhelper)
+        {
+            lhsProgressCount = joinhelper->getLhsProgress();
+            rhsProgressCount = joinhelper->getRhsProgress();
+        }
         strmL.clear();
         strmR.clear();
         {
@@ -3881,8 +3879,6 @@ class CHashAggregateSlave : public CSlaveActivity, implements IHThorRowAggregato
     }
 
 public:
-    IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
-
     CHashAggregateSlave(CGraphElementBase *_container)
         : CSlaveActivity(_container)
     {
@@ -3919,7 +3915,8 @@ public:
     virtual void stop()
     {
         ActPrintLog("HASHAGGREGATE: stopping");
-        localAggTable->reset();
+        if (localAggTable)
+            localAggTable->reset();
         PARENT::stop();
     }
     virtual void abort()
@@ -3974,8 +3971,6 @@ class CHashDistributeSlavedActivity : public CSlaveActivity
     unsigned myNode, nodes;
 
 public:
-    IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
-
     CHashDistributeSlavedActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
         IHThorHashDistributeArg *distribargs = (IHThorHashDistributeArg *)queryHelper();

thorlcr/activities/indexread/thindexread.cpp

@@ -191,7 +191,7 @@ public:
         inputProgress.setown(new ProgressInfo(queryJob()));
         reInit = 0 != (indexBaseHelper->getFlags() & (TIRvarfilename|TIRdynamicfilename));
     }
-    virtual void init()
+    virtual void init() override
     {
         CMasterActivity::init();
         nofilter = false;
@@ -318,7 +318,7 @@ public:
     {
         helper = (IHThorIndexReadArg *)queryHelper();
     }
-    virtual void init()
+    virtual void init() override
     {
         CIndexReadBase::init();
         if (!container.queryLocalOrGrouped())
@@ -353,6 +353,10 @@ public:
         helper = (IHThorIndexCountArg *)queryHelper();
         totalCount = 0;
         totalCountKnown = false;
+    }
+    virtual void init() override
+    {
+        CIndexReadBase::init();
         if (!container.queryLocalOrGrouped())
         {
             if (helper->canMatchAny())

thorlcr/activities/indexread/thindexreadslave.cpp

@@ -560,8 +560,8 @@ public:
         keyedLimitCount = RCMAX;
         keyedProcessed = 0;
         helper = (IHThorIndexReadArg *)queryContainer().queryHelper();
-        stopAfter = (rowcount_t)helper->getChooseNLimit();
-        needTransform = helper->needTransform();
+        stopAfter = 0;
+        needTransform = false;
         rawMeta = helper->queryRawSteppingMeta();
         projectedMeta = helper->queryProjectedSteppingMeta();
         steppedExtra = static_cast<IHThorSteppedSourceExtra *>(helper->selectInterface(TAIsteppedsourceextra_1));
@@ -578,11 +578,6 @@ public:
             seekSizes.append(fields[0].size);
             for (unsigned i=1; i < maxFields; i++)
                 seekSizes.append(seekSizes.item(i-1) + fields[i].size);
-            bool hasPostFilter = helper->transformMayFilter() && optimizeSteppedPostFilter;
-            if (projectedMeta)
-                steppingMeta.init(projectedMeta, hasPostFilter);
-            else
-                steppingMeta.init(rawMeta, hasPostFilter);
         }
     }
     ~CIndexReadSlaveActivity()
@@ -608,6 +603,25 @@ public:
     {
         CIndexReadSlaveBase::init(data, slaveData);
 
+        if (rawMeta)
+        {
+            bool hasPostFilter = helper->transformMayFilter() && optimizeSteppedPostFilter;
+            if (projectedMeta)
+                steppingMeta.init(projectedMeta, hasPostFilter);
+            else
+                steppingMeta.init(rawMeta, hasPostFilter);
+        }
+        appendOutputLinked(this);
+    }
+
+// IThorDataLink
+    virtual void start()
+    {
+        ActivityTimer s(totalCycles, timeActivities);
+        PARENT::start();
+
+        stopAfter = (rowcount_t)helper->getChooseNLimit();
+        needTransform = helper->needTransform();
         helperKeyedLimit = (rowcount_t)helper->getKeyedLimit();
         rowLimit = (rowcount_t)helper->getRowLimit(); // MORE - if no filtering going on could keyspan to get count
         if (0 != (TIRlimitskips & helper->getFlags()))
@@ -619,14 +633,7 @@ public:
             if (TIRkeyedlimitskips & helper->getFlags())
                 keyedLimitSkips = true;
         }
-        appendOutputLinked(this);
-    }
 
-// IThorDataLink
-    virtual void start()
-    {
-        ActivityTimer s(totalCycles, timeActivities);
-        PARENT::start();
         first = true;
         eoi = false;
         keyedLimit = helperKeyedLimit;
@@ -770,7 +777,7 @@ class CIndexGroupAggregateSlaveActivity : public CIndexReadSlaveBase, implements
     Owned<IHashDistributor> distributor;
 
 public:
-    IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
+    IMPLEMENT_IINTERFACE_USING(PARENT);
 
     CIndexGroupAggregateSlaveActivity(CGraphElementBase *_container) : CIndexReadSlaveBase(_container)
     {
@@ -874,24 +881,20 @@ class CIndexCountSlaveActivity : public CIndexReadSlaveBase
 
     bool eoi;
     IHThorIndexCountArg *helper;
-    rowcount_t choosenLimit;
-    rowcount_t preknownTotalCount;
-    bool totalCountKnown;
+    rowcount_t choosenLimit = 0;
+    rowcount_t preknownTotalCount = 0;
+    bool totalCountKnown = false;
 
 public:
     CIndexCountSlaveActivity(CGraphElementBase *_container) : CIndexReadSlaveBase(_container)
     {
         helper = static_cast <IHThorIndexCountArg *> (container.queryHelper());
-        preknownTotalCount = 0;
-        totalCountKnown = false;
-        preknownTotalCount = 0;
     }
 
 // IThorSlaveActivity
     virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
     {
         CIndexReadSlaveBase::init(data, slaveData);
-        choosenLimit = (rowcount_t)helper->getChooseNLimit();
         appendOutputLinked(this);
     }
 
@@ -907,6 +910,7 @@ public:
     {
         ActivityTimer s(totalCycles, timeActivities);
         PARENT::start();
+        choosenLimit = (rowcount_t)helper->getChooseNLimit();
         eoi = false;
         if (!helper->canMatchAny())
         {

thorlcr/activities/indexwrite/thindexwriteslave.cpp

@@ -80,7 +80,7 @@ class IndexWriteSlaveActivity : public ProcessSlaveActivity, public ILookAheadSt
         receivingTag2 = false;
     }
 public:
-    IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
+    IMPLEMENT_IINTERFACE_USING(PARENT);
 
     IndexWriteSlaveActivity(CGraphElementBase *_container) : ProcessSlaveActivity(_container)
     {
@@ -119,10 +119,8 @@ public:
             {
                 if (buildTlk)
                     tlkDesc.setown(deserializePartFileDescriptor(data));
-                else if (!isLocal) // exising tlk then..
+                else if (!isLocal) // existing tlk then..
                 {
-                    OwnedRoxieString diName(helper->getDistributeIndexName());
-                    assertex(diName.get());
                     tlkDesc.setown(deserializePartFileDescriptor(data));
                     unsigned c;
                     data.read(c);
@@ -138,7 +136,7 @@ public:
                         }
                     }
                     if (!existingTlkIFile)
-                        throw MakeThorException(TE_FileNotFound, "Top level key part does not exist, for key: %s", diName.get());
+                        throw MakeActivityException(this, TE_FileNotFound, "Top level key part does not exist, for key");
                 }
             }
         }

thorlcr/activities/join/thjoinslave.cpp

@@ -140,8 +140,6 @@ class JoinSlaveActivity : public CSlaveActivity, implements ILookAheadStopNotify
     } compareReverse, compareReverseUpper;
 
 public:
-    IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
-
     JoinSlaveActivity(CGraphElementBase *_container, bool local)
         : CSlaveActivity(_container), spillStats(spillStatistics)
     {
@@ -163,7 +161,7 @@ public:
             freePort(portbase,NUMSLAVEPORTS);
     }
 
-    void init(MemoryBuffer &data, MemoryBuffer &slaveData)
+    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
     {
         if (!islocal)
         {
@@ -223,7 +221,7 @@ public:
         if (1 == index)
             rightInputStream = queryInputStream(1);
     }
-    virtual void onInputFinished(rowcount_t count)
+    virtual void onInputFinished(rowcount_t count) override
     {
         ActPrintLog("JOIN: %s input finished, %" RCPF "d rows read", rightpartition?"LHS":"RHS", count);
     }
@@ -249,7 +247,6 @@ public:
             }
         }
     }
-
     void startSecondaryInput()
     {
         try
@@ -262,7 +259,7 @@ public:
         }
 
     }
-    virtual void start()
+    virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
 
@@ -371,38 +368,48 @@ public:
         else
             stopRightInput();
     }
-    virtual void abort()
+    virtual void abort() override
     {
         CSlaveActivity::abort();
         if (joinhelper)
             joinhelper->stop();
     }
-    virtual void stop()
+    virtual void stop() override
     {
         stopLeftInput();
         stopRightInput();
-        lhsProgressCount = joinhelper->getLhsProgress();
-        rhsProgressCount = joinhelper->getRhsProgress();
-        {
-            CriticalBlock b(joinHelperCrit);
-            joinhelper.clear();
-        }
-        ActPrintLog("SortJoinSlaveActivity::stop");
-        rightStream.clear();
-        if (!islocal) {
-            unsigned bn=noSortPartitionSide()?2:4;
-            ActPrintLog("JOIN waiting barrier.%d",bn);
-            barrier->wait(false);
-            ActPrintLog("JOIN barrier.%d raised",bn);
-            sorter->stopMerge();
+        /* need to if input started, because activity might never have been started, if conditional
+         * activities downstream stopped their inactive inputs.
+         * stop()'s are chained like this, so that upstream splitters can be stopped as quickly as possible
+         * in order to reduce buffering.
+         */
+        if (queryInputStarted(0))
+        {
+            lhsProgressCount = joinhelper->getLhsProgress();
+            rhsProgressCount = joinhelper->getRhsProgress();
+            {
+                CriticalBlock b(joinHelperCrit);
+                joinhelper.clear();
+            }
+            ActPrintLog("SortJoinSlaveActivity::stop");
+            rightStream.clear();
+            if (!islocal)
+            {
+                unsigned bn=noSortPartitionSide()?2:4;
+                ActPrintLog("JOIN waiting barrier.%d",bn);
+                barrier->wait(false);
+                ActPrintLog("JOIN barrier.%d raised",bn);
+                sorter->stopMerge();
+            }
+            leftStream.clear();
+            dataLinkStop();
+            leftInput.clear();
+            rightInput.clear();
         }
-        leftStream.clear();
-        dataLinkStop();
-        leftInput.clear();
-        rightInput.clear();
     }
-    virtual void reset()
+    virtual void reset() override
     {
+        PARENT::reset();
         if (sorter) return; // JCSMORE loop - shouldn't have to recreate sorter between loop iterations
         if (!islocal && TAG_NULL != mpTagRPC)
             sorter.setown(CreateThorSorter(this, server,&container.queryJob().queryIDiskUsage(),&queryJobChannel().queryJobComm(),mpTagRPC));
@@ -429,7 +436,7 @@ public:
         return NULL;
     }
     virtual bool isGrouped() const override { return false; }
-    virtual void getMetaInfo(ThorDataLinkMetaInfo &info)
+    virtual void getMetaInfo(ThorDataLinkMetaInfo &info) override
     {
         initMetaInfo(info);
         info.unknownRowsOutput = true;
@@ -603,7 +610,7 @@ public:
         }
         return true;
     }
-    virtual void serializeStats(MemoryBuffer &mb)
+    virtual void serializeStats(MemoryBuffer &mb) override
     {
         CSlaveActivity::serializeStats(mb);
         CriticalBlock b(joinHelperCrit);
@@ -627,6 +634,8 @@ public:
 
 class CMergeJoinSlaveBaseActivity : public CThorNarySlaveActivity, public CThorSteppable
 {
+    typedef CThorNarySlaveActivity PARENT;
+
     IHThorNWayMergeJoinArg *helper;
     Owned<IEngineRowAllocator> inputAllocator, outputAllocator;
 
@@ -637,7 +646,7 @@ protected:
     void beforeProcessing();
 
 public:
-    IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
+    IMPLEMENT_IINTERFACE_USING(PARENT);
 
     CMergeJoinSlaveBaseActivity(CGraphElementBase *container, CMergeJoinProcessor &_processor) : CThorNarySlaveActivity(container), CThorSteppable(this), processor(_processor)
     {

thorlcr/activities/keydiff/thkeydiffslave.cpp

@@ -73,6 +73,11 @@ public:
                 patchTlkPart.setown(deserializePartFileDescriptor(data));
             }
         }
+    }
+    virtual void process()
+    {
+        processed = THORDATALINK_STARTED;
+        if (abortSoon) return;
 
         StringBuffer originalFilePart, updatedFilePart;
         OwnedRoxieString origName(helper->getOriginalName());
@@ -100,11 +105,6 @@ public:
                 tlkDiffGenerator.setown(createKeyDiffGenerator(originalFilePart.str(), updatedFilePart.str(), tmp.str(), 0, true, COMPRESS_METHOD_LZMA));
             }
         }
-    }
-    virtual void process()
-    {
-        processed = THORDATALINK_STARTED;
-        if (abortSoon) return;
         try
         {
             diffGenerator->run();

thorlcr/activities/keyedjoin/thkeyedjoinslave.cpp

@@ -1797,29 +1797,17 @@ public:
         parallelLookups = (unsigned)container.queryJob().getWorkUnitValueInt("parallelKJLookups", DEFAULTMAXRESULTPULLPOOL);
         freeQSize = (unsigned)container.queryJob().getWorkUnitValueInt("freeQSize", DEFAULTFREEQSIZE);
         joinFlags = helper->getJoinFlags();
-        keepLimit = helper->getKeepLimit();
-        atMost = helper->getJoinLimit();
-        if (atMost == 0)
-        {
-            if (JFleftonly == (joinFlags & JFleftonly))
-                keepLimit = 1; // don't waste time and memory collating and returning record which will be discarded.
-            atMostProvided = false;
-            atMost = (unsigned)-1;
-        }
-        else
-            atMostProvided = true;
-        abortLimit = helper->getMatchAbortLimit();
-        if (abortLimit == 0) abortLimit = (unsigned)-1;
-        if (keepLimit == 0) keepLimit = (unsigned)-1;
-        if (abortLimit < atMost)
-            atMost = abortLimit;
-        rowLimit = (rowcount_t)helper->getRowLimit();
         additionalStats = 5; // (seeks, scans, accepted, prefiltered, postfiltered)
         needsDiskRead = helper->diskAccessRequired();
         globalFPosToNodeMap = NULL;
         localFPosToNodeMap = NULL;
         fetchHandler = NULL;
         filePartTotal = 0;
+        keepLimit = 0;
+        atMost = 0;
+        atMostProvided = false;
+        abortLimit = 0;
+        rowLimit = 0;
 
         if (needsDiskRead)
             additionalStats += 3; // (diskSeeks, diskAccepted, diskRejected)
@@ -2035,12 +2023,30 @@ public:
             resultDistStream->stop();
         pendingGroupSem.signal();
     }
-    virtual void start()
+    virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
         assertex(inputs.ordinality() == 1);
         PARENT::start();
 
+        keepLimit = helper->getKeepLimit();
+        atMost = helper->getJoinLimit();
+        if (atMost == 0)
+        {
+            if (JFleftonly == (joinFlags & JFleftonly))
+                keepLimit = 1; // don't waste time and memory collating and returning record which will be discarded.
+            atMostProvided = false;
+            atMost = (unsigned)-1;
+        }
+        else
+            atMostProvided = true;
+        abortLimit = helper->getMatchAbortLimit();
+        if (abortLimit == 0) abortLimit = (unsigned)-1;
+        if (keepLimit == 0) keepLimit = (unsigned)-1;
+        if (abortLimit < atMost)
+            atMost = abortLimit;
+        rowLimit = (rowcount_t)helper->getRowLimit();
+
         eos = false;
         inputHelper = LINK(input->queryFromActivity()->queryContainer().queryHelper());
         inputStopped = false;

thorlcr/activities/keypatch/thkeypatchslave.cpp

@@ -73,6 +73,11 @@ public:
                     copyTlk = true;
             }
         }
+    }
+    virtual void process()
+    {
+        processed = THORDATALINK_STARTED;
+        if (abortSoon) return;
 
         StringBuffer originalFilePart, patchFilePart;
         OwnedRoxieString originalName(helper->getOriginalName());
@@ -101,11 +106,6 @@ public:
                 tlkPatchApplicator.setown(createKeyDiffApplicator(patchFilePart.str(), originalFilePart.str(), tmp.str(), NULL, true, true));
             }
         }
-    }
-    virtual void process()
-    {
-        processed = THORDATALINK_STARTED;
-        if (abortSoon) return;
         try
         {
             patchApplictor->run();

thorlcr/activities/loop/thloopslave.cpp

@@ -210,14 +210,14 @@ class CLoopSlaveActivity : public CLoopSlaveActivityBase
                     exception.setown(e);
             }
         }
-        virtual const void *nextRow()
+        virtual const void *nextRow() override
         {
             OwnedConstThorRow row = smartbuf->nextRow();
             if (exception)
                 throw exception.getClear();
             return row.getClear();
         }
-        virtual void stop()
+        virtual void stop() override
         {
             /* NB: signals wants to stop and discards further rows coming out of loop,
              * but reader thread keeps looping, until finishedLooping=true.
@@ -423,9 +423,10 @@ public:
     {
         return nextRowFeeder->nextRow();
     }
-    virtual void stop()
+    virtual void stop() override
     {
-        nextRowFeeder->stop(); // NB: This will block if this slave's loop hasn't hit eof, it will continue looping until 'finishedLooping'
+        if (nextRowFeeder)
+            nextRowFeeder->stop(); // NB: This will block if this slave's loop hasn't hit eof, it will continue looping until 'finishedLooping'
     }
 };
 
@@ -790,65 +791,177 @@ class CConditionalActivity : public CSlaveActivity
 {
     typedef CSlaveActivity PARENT;
 
-    IThorDataLink *selectedInput = NULL;
-    IEngineRowStream *selectInputStream = NULL;
+    bool grouped = false;
+    bool hasGrouped = false;
+    IEngineRowStream *selectInputStream = nullptr;
+    IHThorIfArg *helper;
+
+protected:
+    unsigned branch = (unsigned)-1;
+
 public:
     CConditionalActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
     }
-    void init(MemoryBuffer &data, MemoryBuffer &slaveData)
+    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
     {
         appendOutputLinked(this);
     }
-    virtual void start()
+    virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
-        selectedInput = container.whichBranch>=inputs.ordinality() ? NULL : queryInput(container.whichBranch);
-        selectInputStream = NULL;
-        if (selectedInput)
+        ForEachItemIn(i, inputs)
+        {
+            if (i != branch)
+                stopInput(i);
+        }
+        if (queryInput(branch))
         {
-            startInput(container.whichBranch);
-            selectInputStream = queryInputStream(container.whichBranch);
+            startInput(branch);
+            selectInputStream = inputs.item(branch).stream;
         }
         dataLinkStart();
     }
-    virtual void stop()
+    virtual void stop() override
     {
-        if (selectInputStream)
-            stopInput(container.whichBranch);
+        if ((branch>0) && queryInput(branch)) // branch 0 stopped by PARENT::stop
+            stopInput(branch);
+        selectInputStream = NULL;
         abortSoon = true;
-        dataLinkStop();
+        PARENT::stop();
     }
     CATCH_NEXTROW()
     {
         ActivityTimer t(totalCycles, timeActivities);
         if (abortSoon)
-            return NULL;
-        if (!selectedInput)
-            return NULL;
-
+            return nullptr;
+        if (!selectInputStream)
+            return nullptr;
         OwnedConstThorRow ret = selectInputStream->nextRow();
         if (ret)
             dataLinkIncrement();
         return ret.getClear();
     }
-    virtual bool isGrouped() const override { return selectedInput?selectedInput->isGrouped():false; }
-    virtual void getMetaInfo(ThorDataLinkMetaInfo &info)
+    virtual bool isGrouped() const override { return grouped; }
+    virtual void getMetaInfo(ThorDataLinkMetaInfo &info) override
     {
         initMetaInfo(info);
     }
+    virtual void setInputStream(unsigned index, CThorInput &_input, bool consumerOrdered) override
+    {
+        PARENT::setInputStream(index, _input, consumerOrdered);
+        if (_input.itdl)
+        {
+            bool thisInputGrouped = _input.itdl->isGrouped();
+            if (!hasGrouped)
+            {
+                hasGrouped = true;
+                grouped = thisInputGrouped;
+            }
+            else
+                assertex(grouped == thisInputGrouped);
+        }
+    }
+};
+
+class CIfConditionalActivity : public CConditionalActivity
+{
+    typedef CConditionalActivity PARENT;
+
+    IHThorIfArg *helper;
+public:
+    CIfConditionalActivity(CGraphElementBase *_container) : CConditionalActivity(_container)
+    {
+        helper = (IHThorIfArg *)baseHelper.get();
+    }
+    virtual void start() override
+    {
+        ActivityTimer s(totalCycles, timeActivities);
+        branch = helper->getCondition() ? 0 : 1;
+        PARENT::start();
+    }
 };
 
 activityslaves_decl CActivityBase *createIfSlave(CGraphElementBase *container)
 {
-    return new CConditionalActivity(container);
+    return new CIfConditionalActivity(container);
 }
 
+class CCaseConditionalActivity : public CConditionalActivity
+{
+    typedef CConditionalActivity PARENT;
+
+    IHThorCaseArg *helper;
+public:
+    CCaseConditionalActivity(CGraphElementBase *_container) : CConditionalActivity(_container)
+    {
+        helper = (IHThorCaseArg *)baseHelper.get();
+    }
+    virtual void start() override
+    {
+        ActivityTimer s(totalCycles, timeActivities);
+        branch = helper->getBranch();
+        if (branch >= queryNumInputs())
+            branch = queryNumInputs() - 1;
+        PARENT::start();
+    }
+};
+
 activityslaves_decl CActivityBase *createCaseSlave(CGraphElementBase *container)
 {
-    return new CConditionalActivity(container);
+    return new CCaseConditionalActivity(container);
 }
 
+class CIfActionActivity : public ProcessSlaveActivity
+{
+    typedef ProcessSlaveActivity PARENT;
+
+    bool cond = false;
+    IHThorIfArg *helper;
+
+public:
+    CIfActionActivity(CGraphElementBase *_container) : ProcessSlaveActivity(_container)
+    {
+        helper = (IHThorIfArg *)baseHelper.get();
+    }
+    // IThorSlaveProcess overloaded methods
+    virtual void process() override
+    {
+        processed = THORDATALINK_STARTED;
+        ActivityTimer t(totalCycles, timeActivities);
+        cond = helper->getCondition();
+        if (cond)
+        {
+            if (inputs.item(1).itdl)
+                stopInput(1); // Note: stopping unused branches early helps us avoid buffering splits too long.
+            startInput(0);
+        }
+        else
+        {
+            stopInput(0); // Note: stopping unused branches early helps us avoid buffering splits too long.
+            if (inputs.item(1).itdl)
+                startInput(1);
+        }
+    }
+    virtual void endProcess() override
+    {
+        if (processed & THORDATALINK_STARTED)
+        {
+            if (cond)
+                stopInput(0);
+            else
+                stopInput(1);
+            processed |= THORDATALINK_STOPPED;
+        }
+    }
+};
+
+activityslaves_decl CActivityBase *createIfActionSlave(CGraphElementBase *container)
+{
+    return new CIfActionActivity(container);
+}
+
+
 
 //////////// NewChild acts - move somewhere else..
 
@@ -993,7 +1106,7 @@ public:
         allocator.set(queryRowAllocator());
         appendOutputLinked(this);
     }
-    virtual void start()
+    virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
         PARENT::start();
@@ -1002,10 +1115,6 @@ public:
         aggregated.setown(new RowAggregator(*helper, *helper));
         aggregated->start(queryRowAllocator());
     }
-    virtual void stop()
-    {
-        PARENT::stop();
-    }
     CATCH_NEXTROW()
     {
         if (eos)

thorlcr/activities/merge/thmergeslave.cpp

@@ -184,7 +184,8 @@ public:
         }
         void stop() 
         {
-            if (!stopped) {
+            if (!stopped)
+            {
                 stopped = true;
                 parent->queryJobChannel().queryJobComm().cancel(RANK_ALL, tag);
                 join();
@@ -420,7 +421,7 @@ public:
     LocalMergeSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container) { }
 
 // IThorSlaveActivity overloaded methods
-    void init(MemoryBuffer &data, MemoryBuffer &slaveData)
+    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
     {
         helper = (IHThorMergeArg *)queryHelper();
         appendOutputLinked(this);
@@ -434,7 +435,7 @@ public:
 
 
 // IThorDataLink
-    virtual void start()
+    virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
         ForEachItemIn(i, inputs)
@@ -460,9 +461,10 @@ public:
         dataLinkStart();
     }
 
-    virtual void stop()
+    virtual void stop() override
     {
-        out->stop();
+        if (out)
+            out->stop();
         dataLinkStop();
     }
 
@@ -526,6 +528,8 @@ public:
 
 class CNWayMergeActivity : public CThorNarySlaveActivity, public CThorSteppable
 {
+    typedef CThorNarySlaveActivity PARENT;
+
     IHThorNWayMergeArg *helper;
     CThorStreamMerger merger;
     CSteppingMeta meta;
@@ -534,7 +538,7 @@ class CNWayMergeActivity : public CThorNarySlaveActivity, public CThorSteppable
     PointerArrayOf<IEngineRowStream> expandedInputStreams;
 
 public:
-    IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
+    IMPLEMENT_IINTERFACE_USING(PARENT);
 
     CNWayMergeActivity(CGraphElementBase *container) : CThorNarySlaveActivity(container), CThorSteppable(this)
     {
@@ -560,7 +564,7 @@ public:
         merger.done();
         CThorNarySlaveActivity::stop();
     }
-    virtual void reset()
+    virtual void reset() override
     {
         CThorNarySlaveActivity::reset();
         initializedMeta = false;

thorlcr/activities/msort/thmsortslave.cpp

@@ -154,16 +154,23 @@ public:
             output->stop();
             output.clear();
         }
-        ActPrintLog("SORT waiting barrier.2");
-        barrier->wait(false);
-        ActPrintLog("SORT barrier.2 raised");
+        if (queryInputStarted(0))
+        {
+            ActPrintLog("SORT waiting barrier.2");
+            barrier->wait(false);
+            ActPrintLog("SORT barrier.2 raised");
+        }
         PARENT::stop();
-        sorter->stopMerge();
-        ActPrintLog("SORT waiting for merge");
+        if (queryInputStarted(0))
+        {
+            ActPrintLog("SORT waiting for merge");
+            sorter->stopMerge();
+        }
         dataLinkStop();
     }
-    virtual void reset()
+    virtual void reset() override
     {
+        PARENT::reset();
         if (sorter) return; // JCSMORE loop - shouldn't have to recreate sorter between loop iterations
         sorter.setown(CreateThorSorter(this, server,&container.queryJob().queryIDiskUsage(),&queryJobChannel().queryJobComm(),mpTagRPC));
     }

thorlcr/activities/nsplitter/thnsplitterslave.cpp

@@ -25,27 +25,49 @@
 interface ISharedSmartBuffer;
 class NSplitterSlaveActivity;
 
-class CSplitterOutputBase : public CSimpleInterfaceOf<IStartableEngineRowStream>, public COutputTiming
-{
-public:
-// IEngineRowStream
-    virtual void resetEOF() { throwUnexpected(); }
-};
-
-class CSplitterOutput : public CSplitterOutputBase
+class CSplitterOutput : public CSimpleInterfaceOf<IStartableEngineRowStream>, public CEdgeProgress, public COutputTiming, implements IThorDataLink
 {
     NSplitterSlaveActivity &activity;
     Semaphore writeBlockSem;
+    bool started = false, stopped = false;
 
-    unsigned output;
-    rowcount_t rec, max;
+    unsigned activeOutput;
+    rowcount_t rec = 0, max = 0;
 
 public:
-    CSplitterOutput(NSplitterSlaveActivity &_activity, unsigned output);
+    IMPLEMENT_IINTERFACE_USING(CSimpleInterfaceOf<IStartableEngineRowStream>);
+
+    CSplitterOutput(NSplitterSlaveActivity &_activity, unsigned activeOutput);
+
+    void reset()
+    {
+        started = stopped = false;
+        rec = max = 0;
+    }
+    inline bool isStopped() const { return stopped; }
+
+// IThorDataLink impl.
+    virtual CSlaveActivity *queryFromActivity() override;
+    virtual void getMetaInfo(ThorDataLinkMetaInfo &info) override;
+    virtual void dataLinkSerialize(MemoryBuffer &mb) const override { CEdgeProgress::dataLinkSerialize(mb); }
+    virtual rowcount_t getProgressCount() const override { return CEdgeProgress::getCount(); }
+    virtual bool isGrouped() const override;
+    virtual IOutputMetaData * queryOutputMeta() const override;
+    virtual bool isInputOrdered(bool consumerOrdered) const override;
+    virtual void setOutputStream(unsigned index, IEngineRowStream *stream) override;
+    virtual IStrandJunction *getOutputStreams(CActivityBase &ctx, unsigned idx, PointerArrayOf<IEngineRowStream> &streams, const CThorStrandOptions * consumerOptions, bool consumerOrdered, IOrderedCallbackCollection * orderedCallbacks) override;
+    virtual unsigned __int64 queryTotalCycles() const override { return COutputTiming::queryTotalCycles(); }
+    virtual unsigned __int64 queryEndCycles() const { return COutputTiming::queryEndCycles(); }
+    virtual void debugRequest(MemoryBuffer &mb) override;
+// Stepping methods
+    virtual IInputSteppingMeta *querySteppingMeta() { return nullptr; }
+    virtual bool gatherConjunctions(ISteppedConjunctionCollector & collector) { return false; }
 
+// IStartableEngineRowStream
     virtual void start() override;
     virtual void stop() override;
     virtual const void *nextRow() override;
+    virtual void resetEOF() { throwUnexpected(); }
 };
 
 
@@ -57,20 +79,18 @@ class NSplitterSlaveActivity : public CSlaveActivity, implements ISharedSmartBuf
 {
     typedef CSlaveActivity PARENT;
 
-    bool spill;
-    bool eofHit;
-    bool inputsConfigured;
-    bool writeBlocked, pagedOut;
+    bool spill = false;
+    bool eofHit = false;
+    bool writeBlocked = false, pagedOut = false;
     CriticalSection startLock, writeAheadCrit;
     PointerArrayOf<Semaphore> stalledWriters;
-    unsigned nstopped;
-    rowcount_t recsReady;
-    Owned<IException> startException, writeAheadException;
+    unsigned stoppedOutputs = 0;
+    unsigned activeOutputs = 0;
+    rowcount_t recsReady = 0;
+    Owned<IException> writeAheadException;
     Owned<ISharedSmartBuffer> smartBuf;
     bool inputPrepared = false;
     bool inputConnected = false;
-    IPointerArrayOf<IThorDataLinkExt> delayInputsList;
-
 
     // NB: CWriter only used by 'balanced' splitter, which blocks write when too far ahead
     class CWriter : public CSimpleInterface, IThreaded
@@ -107,105 +127,6 @@ class NSplitterSlaveActivity : public CSlaveActivity, implements ISharedSmartBuf
             }
         }
     } writer;
-    class CNullInput : public CSplitterOutputBase
-    {
-    public:
-        virtual void start() override { throwUnexpected(); }
-        virtual const void *nextRow() override { throwUnexpected(); return NULL; }
-        virtual void stop() override { throwUnexpected(); }
-    };
-    class CInputWrapper : public CSplitterOutputBase
-    {
-        IRowStream *inputStream = nullptr;
-        NSplitterSlaveActivity &activity;
-
-    public:
-        CInputWrapper(NSplitterSlaveActivity &_activity) : activity(_activity) { }
-        virtual void start() override
-        {
-            activity.start();
-            inputStream = activity.inputStream;
-        }
-        virtual const void *nextRow() override
-        {
-            ActivityTimer t(totalCycles, activity.queryTimeActivities());
-            return inputStream->nextRow();
-        }
-        virtual void stop() override { activity.stop(); }
-    };
-    class CDelayedInput : public CSimpleInterfaceOf<IThorDataLinkExt>, public CEdgeProgress, implements IEngineRowStream
-    {
-        Owned<CSplitterOutputBase> inputStream;
-        Linked<NSplitterSlaveActivity> activity;
-        mutable SpinLock processActiveLock;
-        unsigned outputIdx = 0;
-
-    public:
-        IMPLEMENT_IINTERFACE_USING(CSimpleInterfaceOf<IThorDataLinkExt>);
-
-        CDelayedInput(NSplitterSlaveActivity &_activity) : CEdgeProgress(&_activity), activity(&_activity) { }
-        void setInput(CSplitterOutputBase *_inputStream)
-        {
-            SpinBlock b(processActiveLock);
-            inputStream.setown(_inputStream);
-        }
-        const void *nextRow()
-        {
-            OwnedConstThorRow row = inputStream->nextRow();
-            if (row)
-                dataLinkIncrement();
-            return row.getClear();
-        }
-        void stop()
-        {
-            inputStream->stop();
-            dataLinkStop();
-        }
-        void resetEOF()
-        {
-            inputStream->resetEOF();
-        }
-    // IThorDataLink impl.
-        virtual void start()
-        {
-            activity->ensureInputsConfigured();
-            inputStream->start();
-            dataLinkStart();
-        }
-        virtual CSlaveActivity *queryFromActivity() override { return activity; }
-        virtual void getMetaInfo(ThorDataLinkMetaInfo &info) override { activity->getMetaInfo(info); }
-        virtual void dataLinkSerialize(MemoryBuffer &mb) const { CEdgeProgress::dataLinkSerialize(mb); }
-        virtual bool isGrouped() const { return activity->isGrouped(); }
-        virtual IOutputMetaData * queryOutputMeta() const { return activity->queryOutputMeta(); }
-        virtual unsigned queryOutputIdx() const { return outputIdx; }
-        virtual bool isInputOrdered(bool consumerOrdered) const { return activity->isInputOrdered(consumerOrdered); }
-        virtual void setOutputStream(unsigned index, IEngineRowStream *stream) { activity->setOutputStream(index, stream); }
-        virtual IStrandJunction *getOutputStreams(CActivityBase &ctx, unsigned idx, PointerArrayOf<IEngineRowStream> &streams, const CThorStrandOptions * consumerOptions, bool consumerOrdered, IOrderedCallbackCollection * orderedCallbacks)
-        {
-            activity->connectInput(consumerOrdered);
-            streams.append(this);
-            return NULL;
-        }
-        virtual unsigned __int64 queryTotalCycles() const override
-        {
-            SpinBlock b(processActiveLock);
-            if (!inputStream)
-                return 0;
-            return inputStream->queryTotalCycles();
-        }
-        virtual unsigned __int64 queryEndCycles() const
-        {
-            SpinBlock b(processActiveLock);
-            return inputStream->queryEndCycles();
-        }
-        virtual void debugRequest(MemoryBuffer &mb) { activity->debugRequest(mb); }
-    // Stepping methods
-        virtual IInputSteppingMeta *querySteppingMeta() { return NULL; }
-        virtual bool gatherConjunctions(ISteppedConjunctionCollector & collector) { return false; }
-    // IThorDataLinkExt
-        virtual void setOutputIdx(unsigned idx) override { outputIdx = idx; }
-    };
-
     void connectInput(bool consumerOrdered)
     {
         CriticalBlock block(startLock);
@@ -217,86 +138,53 @@ class NSplitterSlaveActivity : public CSlaveActivity, implements ISharedSmartBuf
         }
     }
 public:
-    IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
-
     NSplitterSlaveActivity(CGraphElementBase *container) : CSlaveActivity(container), writer(*this)
     {
-        spill = false;
-        nstopped = 0;
-        eofHit = inputsConfigured = writeBlocked = pagedOut = false;
-        recsReady = 0;
     }
-    virtual ~NSplitterSlaveActivity()
+    virtual void reset() override
     {
-        delayInputsList.kill();
+        PARENT::reset();
+        stoppedOutputs = 0;
+        eofHit = false;
+        inputPrepared = false;
+        recsReady = 0;
+        writeBlocked = false;
+        stalledWriters.kill();
+        ForEachItemIn(o, outputs)
+        {
+            CSplitterOutput *output = (CSplitterOutput *)outputs.item(o);
+            if (output)
+                output->reset();
+        }
     }
-    void ensureInputsConfigured()
+    void init(MemoryBuffer &data, MemoryBuffer &slaveData)
     {
-        CriticalBlock block(startLock);
-        if (inputsConfigured)
-            return;
-        inputsConfigured = true;
-        unsigned noutputs = container.connectedOutputs.getCount();
-        ActPrintLog("Number of connected outputs: %d", noutputs);
-        if (1 == noutputs)
+        ForEachItemIn(o, container.outputs)
         {
-            CIOConnection *io = NULL;
-            ForEachItemIn(o, container.connectedOutputs)
-            {
-                io = container.connectedOutputs.item(o);
-                if (io)
-                    break;
-            }
-            assertex(io);
-            ForEachItemIn(o2, delayInputsList)
+            if (nullptr != container.connectedOutputs.queryItem(o))
+                ++activeOutputs;
+        }
+        ActPrintLog("Number of connected outputs: %u", activeOutputs);
+        if (activeOutputs <= 1)
+        {
+            ForEachItemIn(o2, container.outputs)
             {
-                CDelayedInput *delayedInput = (CDelayedInput *)delayInputsList.item(o2);
                 if (o2 == o)
-                    delayedInput->setInput(new CInputWrapper(*this));
+                    appendOutputLinked(this);
                 else
-                    delayedInput->setInput(new CNullInput());
+                    appendOutput(nullptr);
             }
         }
         else
         {
-            ForEachItemIn(o, delayInputsList)
+            unsigned activeOutput = 0;
+            ForEachItemIn(o, container.outputs)
             {
-                CDelayedInput *delayedInput = (CDelayedInput *)delayInputsList.item(o);
-                if (NULL != container.connectedOutputs.queryItem(o))
-                    delayedInput->setInput(new CSplitterOutput(*this, o));
+                if (nullptr != container.connectedOutputs.queryItem(o))
+                    appendOutput(new CSplitterOutput(*this, activeOutput++));
                 else
-                    delayedInput->setInput(new CNullInput());
-            }
-        }
-    }
-    void reset()
-    {
-        CSlaveActivity::reset();
-        nstopped = 0;
-        eofHit = false;
-        inputPrepared = false;
-        inputConnected = false;
-        recsReady = 0;
-        writeBlocked = false;
-        stalledWriters.kill();
-        if (inputsConfigured)
-        {
-            // ensure old inputs cleared, to avoid being reused before re-setup on subsequent executions
-            ForEachItemIn(o, delayInputsList)
-            {
-                CDelayedInput *delayedInput = (CDelayedInput *)delayInputsList.item(o);
-                delayedInput->setInput(NULL);
+                    appendOutput(nullptr);
             }
-            inputsConfigured = false;
-        }
-    }
-    void init(MemoryBuffer &data, MemoryBuffer &slaveData)
-    {
-        ForEachItemIn(o, container.outputs)
-        {
-            Owned<CDelayedInput> delayedInput = new CDelayedInput(*this);
-            delayInputsList.append(delayedInput.getLink());
-            appendOutput(delayedInput.getClear());
         }
         IHThorSplitArg *helper = (IHThorSplitArg *)queryHelper();
         int dV = getOptInt(THOROPT_SPLITTER_SPILL, -1);
@@ -305,52 +193,57 @@ public:
         else
             spill = dV>0;
     }
-    void prepareInput(unsigned output)
+    bool prepareInput()
     {
         CriticalBlock block(startLock);
         if (!inputPrepared)
         {
             inputPrepared = true;
-            try
+            PARENT::start();
+            unsigned remainingOutputs = activeOutputs;
+            ForEachItemIn(o, outputs)
             {
-                PARENT::start();
-                nstopped = container.connectedOutputs.getCount();
-                if (smartBuf)
-                    smartBuf->reset();
+                CSplitterOutput *output = (CSplitterOutput *)outputs.item(o);
+                if (output && output->isStopped())
+                    --remainingOutputs;
+            }
+            assertex(remainingOutputs); // must be >=1, as this output (activeOutput) has invoked prepareInput
+            if (1 == remainingOutputs)
+                return false;
+            if (smartBuf)
+                smartBuf->reset();
+            else
+            {
+                if (spill)
+                {
+                    StringBuffer tempname;
+                    GetTempName(tempname, "nsplit", true); // use alt temp dir
+                    smartBuf.setown(createSharedSmartDiskBuffer(this, tempname.str(), activeOutputs, queryRowInterfaces(input), &container.queryJob().queryIDiskUsage()));
+                    ActPrintLog("Using temp spill file: %s", tempname.str());
+                }
                 else
                 {
-                    if (spill)
-                    {
-                        StringBuffer tempname;
-                        GetTempName(tempname,"nsplit",true); // use alt temp dir
-                        smartBuf.setown(createSharedSmartDiskBuffer(this, tempname.str(), outputs.ordinality(), queryRowInterfaces(input), &container.queryJob().queryIDiskUsage()));
-                        ActPrintLog("Using temp spill file: %s", tempname.str());
-                    }
-                    else
-                    {
-                        ActPrintLog("Spill is 'balanced'");
-                        smartBuf.setown(createSharedSmartMemBuffer(this, outputs.ordinality(), queryRowInterfaces(input), NSPLITTER_SPILL_BUFFER_SIZE));
-                    }
-                    // mark any unconnected outputs of smartBuf as already stopped.
-                    ForEachItemIn(o, outputs)
-                    {
-                        IThorDataLink *delayedInput = outputs.item(o);
-                        if (NULL == container.connectedOutputs.queryItem(o))
-                            smartBuf->queryOutput(o)->stop();
-                    }
+                    ActPrintLog("Spill is 'balanced'");
+                    smartBuf.setown(createSharedSmartMemBuffer(this, activeOutputs, queryRowInterfaces(input), NSPLITTER_SPILL_BUFFER_SIZE));
+                }
+                // mark any outputs already stopped
+                ForEachItemIn(o, outputs)
+                {
+                    CSplitterOutput *output = (CSplitterOutput *)outputs.item(o);
+                    if (output && output->isStopped())
+                        smartBuf->queryOutput(o)->stop();
                 }
-                if (!spill)
-                    writer.start(); // writer keeps writing ahead as much as possible, the readahead impl. will block when has too much
-            }
-            catch (IException *e)
-            {
-                startException.setown(e); 
             }
+            if (!spill)
+                writer.start(); // writer keeps writing ahead as much as possible, the readahead impl. will block when has too much
         }
+        return true;
     }
-    inline const void *nextRow(unsigned output)
+    inline const void *nextRow(unsigned activeOutput)
     {
-        OwnedConstThorRow row = smartBuf->queryOutput(output)->nextRow(); // will block until available
+        if (!smartBuf) // will be true, if only 1 input connect, or only 1 input was active (others stopped) when it started reading
+            return inputStream->nextRow();
+        OwnedConstThorRow row = smartBuf->queryOutput(activeOutput)->nextRow(); // will block until available
         if (writeAheadException)
             throw LINK(writeAheadException);
         return row.getClear();
@@ -376,6 +269,10 @@ public:
                 break;
         }
         ActivityTimer t(totalCycles, queryTimeActivities());
+        if (!prepareInput()) // returns true, if
+        {
+            return RCMAX; // signals to requester that you are the only output
+        }
         pagedOut = false;
         OwnedConstThorRow row;
         loop
@@ -390,7 +287,7 @@ public:
                     row.setown(inputStream->nextRow());
                     if (row)
                     {
-                        smartBuf->putRow(NULL, this); // may call blocked() (see ISharedSmartBufferCallback impl. below)
+                        smartBuf->putRow(nullptr, this); // may call blocked() (see ISharedSmartBufferCallback impl. below)
                         ++recsReady;
                     }
                 }
@@ -408,9 +305,18 @@ public:
         }
         return recsReady;
     }
-    void inputStopped()
+    void inputStopped(unsigned activeOutput)
     {
-        if (nstopped && --nstopped==0) 
+        CriticalBlock block(startLock);
+        if (smartBuf)
+        {
+            /* If no output has started reading (nextRow()), then it will not have been prepared
+             * If only 1 output is left, it will bypass the smart buffer when it starts.
+             */
+            smartBuf->queryOutput(activeOutput)->stop();
+        }
+        ++stoppedOutputs;
+        if (stoppedOutputs == activeOutputs)
         {
             writer.stop();
             PARENT::stop();
@@ -441,20 +347,24 @@ public:
         }
     }
 
-// IThorDataLink (for output 0)
-    virtual void getMetaInfo(ThorDataLinkMetaInfo &info) override
+// IEngineRowStream
+    virtual const void *nextRow() override
     {
-        calcMetaInfoSize(info, queryInput(0));
+        ActivityTimer t(totalCycles, queryTimeActivities());
+        return inputStream->nextRow();
     }
-    virtual unsigned __int64 queryTotalCycles() const override
+    virtual void stop() override{ inputStream->stop(); }
+
+// IThorDataLink (if single output connected)
+    virtual IStrandJunction *getOutputStreams(CActivityBase &ctx, unsigned idx, PointerArrayOf<IEngineRowStream> &streams, const CThorStrandOptions * consumerOptions, bool consumerOrdered, IOrderedCallbackCollection * orderedCallbacks) override
     {
-        unsigned __int64 _totalCycles = PARENT::queryTotalCycles(); // more() time
-        ForEachItemIn(o, outputs)
-        {
-            IThorDataLink *delayedInput = outputs.item(o);
-            _totalCycles += delayedInput->queryTotalCycles();
-        }
-        return _totalCycles;
+        connectInput(consumerOrdered);
+        streams.append(this);
+        return nullptr;
+    }
+    virtual void getMetaInfo(ThorDataLinkMetaInfo &info) override
+    {
+        calcMetaInfoSize(info, queryInput(0));
     }
 
 friend class CInputWrapper;
@@ -465,36 +375,80 @@ friend class CWriter;
 //
 // CSplitterOutput
 //
-CSplitterOutput::CSplitterOutput(NSplitterSlaveActivity &_activity, unsigned _output)
-   : activity(_activity), output(_output)
+
+CSlaveActivity *CSplitterOutput::queryFromActivity()
+{
+    return &activity;
+}
+
+void CSplitterOutput::getMetaInfo(ThorDataLinkMetaInfo &info)
+{
+    activity.getMetaInfo(info);
+}
+
+bool CSplitterOutput::isGrouped() const
+{
+    return activity.isGrouped();
+}
+
+IOutputMetaData *CSplitterOutput::queryOutputMeta() const
+{
+    return activity.queryOutputMeta();
+}
+bool CSplitterOutput::isInputOrdered(bool consumerOrdered) const
+{
+    return activity.isInputOrdered(consumerOrdered);
+
+}
+void CSplitterOutput::setOutputStream(unsigned index, IEngineRowStream *stream)
+{
+    activity.setOutputStream(index, stream);
+}
+
+IStrandJunction *CSplitterOutput::getOutputStreams(CActivityBase &ctx, unsigned idx, PointerArrayOf<IEngineRowStream> &streams, const CThorStrandOptions * consumerOptions, bool consumerOrdered, IOrderedCallbackCollection * orderedCallbacks)
+{
+    activity.connectInput(consumerOrdered);
+    streams.append(this);
+    return nullptr;
+}
+
+void CSplitterOutput::debugRequest(MemoryBuffer &mb)
+{
+    activity.debugRequest(mb);
+}
+
+
+CSplitterOutput::CSplitterOutput(NSplitterSlaveActivity &_activity, unsigned _activeOutput)
+   : CEdgeProgress(_activity), activity(_activity), activeOutput(_activeOutput)
 {
-    rec = max = 0;
 }
 
 // IStartableEngineRowStream
 void CSplitterOutput::start()
 {
     ActivityTimer s(totalCycles, activity.queryTimeActivities());
-    rec = max = 0;
-    activity.prepareInput(output);
-    if (activity.startException)
-        throw LINK(activity.startException);
+    started = true;
+    dataLinkStart();
 }
 
 // IEngineRowStream
 void CSplitterOutput::stop()
 { 
     CriticalBlock block(activity.startLock);
-    activity.smartBuf->queryOutput(output)->stop();
-    activity.inputStopped();
+    stopped = true;
+    activity.inputStopped(activeOutput);
+    dataLinkStop();
 }
 
 const void *CSplitterOutput::nextRow()
 {
     if (rec == max)
+    {
         max = activity.writeahead(max, activity.queryAbortSoon(), writeBlockSem);
+        // NB: if this is sole input that actually started, writeahead will have returned RCMAX and calls to activity.nextRow will go directly to splitter input
+    }
     ActivityTimer t(totalCycles, activity.queryTimeActivities());
-    const void *row = activity.nextRow(output); // pass ptr to max if need more
+    const void *row = activity.nextRow(activeOutput); // pass ptr to max if need more
     ++rec;
     return row;
 }

thorlcr/activities/parse/thparseslave.cpp

@@ -43,8 +43,6 @@ class CParseSlaveActivity : public CSlaveActivity, implements IMatchedAction
     Owned<IEngineRowAllocator> allocator;
 
 public:
-    IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
-
     CParseSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
         anyThisGroup = false;
@@ -64,13 +62,13 @@ public:
         algorithm.setown(createThorParser(queryCodeContext(), *helper));
         parser.setown(algorithm->createParser(queryCodeContext(), (unsigned)container.queryId(), helper->queryHelper(), helper));
         rowIter = parser->queryResultIter();
-        rowIter->first();
         allocator.set(queryRowAllocator());
     } 
     virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
         PARENT::start();
+        rowIter->first();
     }
     void processRecord(const void * in)
     {

thorlcr/activities/piperead/thprslave.cpp

@@ -233,22 +233,17 @@ public:
         flags = helper->getPipeFlags();
         needTransform = false;
 
-        IThorRowInterfaces *_inrowif;
         if (needTransform)
-        {
-            inrowif.setown(createThorRowInterfaces(queryRowManager(), helper->queryDiskRecordSize(),queryId(),queryCodeContext()));
-            _inrowif = inrowif;
-        }
-        else
-            _inrowif = this;
-        OwnedRoxieString xmlIteratorPath(helper->getXmlIteratorPath());
-        readTransformer.setown(createReadRowStream(_inrowif->queryRowAllocator(), _inrowif->queryRowDeserializer(), helper->queryXmlTransformer(), helper->queryCsvTransformer(), xmlIteratorPath, flags));
+            inrowif.setown(createThorRowInterfaces(queryRowManager(), helper->queryDiskRecordSize(), queryId(), queryCodeContext()));
         appendOutputLinked(this);
     }
     virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
         PARENT::start();
+        OwnedRoxieString xmlIteratorPath(helper->getXmlIteratorPath());
+        IThorRowInterfaces *_inrowif = needTransform ? inrowif.get() : this;
+        readTransformer.setown(createReadRowStream(_inrowif->queryRowAllocator(), _inrowif->queryRowDeserializer(), helper->queryXmlTransformer(), helper->queryCsvTransformer(), xmlIteratorPath, flags));
         eof = false;
         OwnedRoxieString pipeProgram(helper->getPipeProgram());
         openPipe(pipeProgram, "PIPEREAD");
@@ -356,16 +351,16 @@ public:
         recreate = helper->recreateEachRow();
         grouped = 0 != (flags & TPFgroupeachrow);
 
-        OwnedRoxieString xmlIterator(helper->getXmlIteratorPath());
-        readTransformer.setown(createReadRowStream(queryRowAllocator(), queryRowDeserializer(), helper->queryXmlTransformer(), helper->queryCsvTransformer(), xmlIterator, flags));
-        readTransformer->setStream(pipeStream); // NB the pipe process stream is provided to pipeStream after pipe->run()
-
         appendOutputLinked(this);
     }
     virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
         PARENT::start();
+        OwnedRoxieString xmlIterator(helper->getXmlIteratorPath());
+        readTransformer.setown(createReadRowStream(queryRowAllocator(), queryRowDeserializer(), helper->queryXmlTransformer(), helper->queryCsvTransformer(), xmlIterator, flags));
+        readTransformer->setStream(pipeStream); // NB the pipe process stream is provided to pipeStream after pipe->run()
+
         eof = anyThisGroup = inputExhausted = false;
         firstRead = true;
 

thorlcr/activities/project/thprojectslave.cpp

@@ -118,7 +118,7 @@ class CPrefetchProjectSlaveActivity : public CSlaveActivity
     rowcount_t numProcessedLastGroup;
     bool eof;
     Owned<IEngineRowAllocator> allocator;
-    IThorChildGraph *child;
+    IThorChildGraph *child = nullptr;
     bool parallel;
     unsigned preload;
 
@@ -255,7 +255,6 @@ public:
         preload = helper->getLookahead();
         if (!preload)
             preload = 10; // default
-        child = helper->queryChild();
     }
     virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
     {
@@ -266,7 +265,7 @@ public:
     {
         ActivityTimer s(totalCycles, timeActivities);
         PARENT::start();
-
+        child = helper->queryChild();
         numProcessedLastGroup = getDataLinkGlobalCount();
         eof = !helper->canMatchAny();
         if (parallel)

thorlcr/activities/rollup/throllupslave.cpp

@@ -286,8 +286,8 @@ class CDedupBaseSlaveActivity : public CDedupRollupBaseActivity
 {
 protected:
     IHThorDedupArg *ddhelper;
-    bool keepLeft;
-    unsigned numToKeep;
+    bool keepLeft = 0;
+    unsigned numToKeep = 0;
 
 public:
     CDedupBaseSlaveActivity(CGraphElementBase *_container, bool global, bool groupOp)
@@ -299,14 +299,14 @@ public:
         CDedupRollupBaseActivity::init(data, slaveData);
         appendOutputLinked(this);   // adding 'me' to outputs array
         ddhelper = static_cast <IHThorDedupArg *>(queryHelper());
-        keepLeft = ddhelper->keepLeft();
-        numToKeep = ddhelper->numToKeep();
-        assertex(keepLeft || numToKeep == 1);
     }
     virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
         CDedupRollupBaseActivity::start();
+        keepLeft = ddhelper->keepLeft();
+        numToKeep = ddhelper->numToKeep();
+        assertex(keepLeft || numToKeep == 1);
     }
     virtual bool isGrouped() const override { return groupOp; }
     virtual void getMetaInfo(ThorDataLinkMetaInfo &info) override
@@ -320,8 +320,6 @@ public:
 class CDedupSlaveActivity : public CDedupBaseSlaveActivity
 {
 public:
-    IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
-
     CDedupSlaveActivity(CGraphElementBase *_container, bool global, bool groupOp)
         : CDedupBaseSlaveActivity(_container, global, groupOp)
     {
@@ -410,12 +408,12 @@ public:
     void init(MemoryBuffer &data, MemoryBuffer &slaveData)
     {
         CDedupBaseSlaveActivity::init(data, slaveData);
-        assertex(1 == numToKeep);
     }
     virtual void start()
     {
         ActivityTimer s(totalCycles, timeActivities);
         CDedupBaseSlaveActivity::start();
+        assertex(1 == numToKeep);
 
         lastEog = false;
         assertex(!global);      // dedup(),local,all only supported

thorlcr/activities/selectnth/thselectnthslave.cpp

@@ -63,8 +63,6 @@ class CSelectNthSlaveActivity : public CSlaveActivity, implements ILookAheadStop
     }
 
 public:
-    IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
-
     CSelectNthSlaveActivity(CGraphElementBase *_container, bool _isLocal) : CSlaveActivity(_container)
     {
         isLocal = _isLocal;

thorlcr/activities/selfjoin/thselfjoinslave.cpp

@@ -79,12 +79,13 @@ private:
 #endif
         sorter->Gather(::queryRowInterfaces(input), inputStream, compare, NULL, NULL, keyserializer, NULL, false, isUnstable(), abortSoon, NULL);
         PARENT::stop();
-        if(abortSoon)
+        if (abortSoon)
         {
             barrier->cancel();
             return NULL;
         }
-        if (!barrier->wait(false)) {
+        if (!barrier->wait(false))
+        {
             Sleep(1000); // let original error through
             throw MakeThorException(TE_BarrierAborted,"SELFJOIN: Barrier Aborted");
         }
@@ -141,9 +142,9 @@ public:
         else
             ActPrintLog("SELFJOIN: GLOBAL");
     }
-    virtual void reset()
+    virtual void reset() override
     {
-        CSlaveActivity::reset();
+        PARENT::reset();
         if (sorter) return; // JCSMORE loop - shouldn't have to recreate sorter between loop iterations
         if (!isLocal && TAG_NULL != mpTagRPC)
             sorter.setown(CreateThorSorter(this, server,&container.queryJob().queryIDiskUsage(),&queryJobChannel().queryJobComm(),mpTagRPC));
@@ -187,7 +188,7 @@ public:
     }
     virtual void stop() override
     {
-        if(!isLocal)
+        if (!isLocal)
         {
             barrier->wait(false);
             sorter->stopMerge();
@@ -196,8 +197,11 @@ public:
             CriticalBlock b(joinHelperCrit);
             joinhelper.clear();
         }
-        strm->stop();
-        strm.clear();
+        if (strm)
+        {
+            strm->stop();
+            strm.clear();
+        }
         PARENT::stop();
     }
     

thorlcr/activities/soapcall/thsoapcallslave.cpp

@@ -212,10 +212,12 @@ public:
 
 class SoapRowActionSlaveActivity : public ProcessSlaveActivity, implements IWSCRowProvider
 {
+    typedef ProcessSlaveActivity PARENT;
+
     Owned<IWSCHelper> wscHelper;
 
 public:
-    IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
+    IMPLEMENT_IINTERFACE_USING(PARENT);
 
     SoapRowActionSlaveActivity(CGraphElementBase *container) : ProcessSlaveActivity(container) { }
 
@@ -262,11 +264,13 @@ public:
 
 class SoapDatasetActionSlaveActivity : public ProcessSlaveActivity, implements IWSCRowProvider
 {
+    typedef ProcessSlaveActivity PARENT;
+
     Owned<IWSCHelper> wscHelper;
     CriticalSection crit;
 
 public:
-    IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
+    IMPLEMENT_IINTERFACE_USING(PARENT);
 
     SoapDatasetActionSlaveActivity(CGraphElementBase *container) : ProcessSlaveActivity(container) { }
 

thorlcr/activities/thactivityutil.cpp

@@ -234,6 +234,7 @@ public:
         ActPrintLog(&activity, "CRowStreamLookAhead start %x",(unsigned)(memsize_t)this);
 #endif
         stopped = false;
+        running = true;
         thread.start();
         startSem.wait();
     }

thorlcr/activities/thdiskbaseslave.cpp

@@ -228,20 +228,7 @@ void CDiskReadSlaveActivityBase::init(MemoryBuffer &data, MemoryBuffer &slaveDat
     unsigned parts;
     data.read(parts);
     if (parts)
-    {
         deserializePartFileDescriptors(data, partDescs);
-        unsigned encryptedKeyLen;
-        void *encryptedKey;
-        helper->getEncryptKey(encryptedKeyLen, encryptedKey);
-        if (0 != encryptedKeyLen) 
-        {
-            bool dfsEncrypted = partDescs.item(0).queryOwner().queryProperties().getPropBool("@encrypted");
-            if (dfsEncrypted) // otherwise ignore (warning issued by master)
-                eexp.setown(createAESExpander256(encryptedKeyLen, encryptedKey));
-            memset(encryptedKey, 0, encryptedKeyLen);
-            free(encryptedKey);
-        }
-    }
 }
 
 const char *CDiskReadSlaveActivityBase::queryLogicalFilename(unsigned index)
@@ -254,6 +241,17 @@ void CDiskReadSlaveActivityBase::start()
     PARENT::start();
     markStart = true;
     diskProgress = 0;
+    unsigned encryptedKeyLen;
+    void *encryptedKey;
+    helper->getEncryptKey(encryptedKeyLen, encryptedKey);
+    if (0 != encryptedKeyLen)
+    {
+        bool dfsEncrypted = partDescs.item(0).queryOwner().queryProperties().getPropBool("@encrypted");
+        if (dfsEncrypted) // otherwise ignore (warning issued by master)
+            eexp.setown(createAESExpander256(encryptedKeyLen, encryptedKey));
+        memset(encryptedKey, 0, encryptedKeyLen);
+        free(encryptedKey);
+    }
 }
 
 void CDiskReadSlaveActivityBase::kill()
@@ -485,16 +483,6 @@ void CDiskWriteSlaveActivityBase::init(MemoryBuffer &data, MemoryBuffer &slaveDa
     }
     if (0 != (diskHelperBase->getFlags() & TDXgrouped))
         grouped = true;
-    compress = partDesc->queryOwner().isCompressed();
-    void *ekey;
-    size32_t ekeylen;
-    diskHelperBase->getEncryptKey(ekeylen,ekey);
-    if (ekeylen!=0) {
-        ecomp.setown(createAESCompressor256(ekeylen,ekey));
-        memset(ekey,0,ekeylen);
-        free(ekey);
-        compress = true;
-    }
 }
 
 void CDiskWriteSlaveActivityBase::abort()
@@ -533,6 +521,17 @@ void CDiskWriteSlaveActivityBase::kill()
 
 void CDiskWriteSlaveActivityBase::process()
 {
+    compress = partDesc->queryOwner().isCompressed();
+    void *ekey;
+    size32_t ekeylen;
+    diskHelperBase->getEncryptKey(ekeylen,ekey);
+    if (ekeylen!=0)
+    {
+        ecomp.setown(createAESCompressor256(ekeylen,ekey));
+        memset(ekey,0,ekeylen);
+        free(ekey);
+        compress = true;
+    }
     calcFileCrc = false;
     uncompressedBytesWritten = 0;
     replicateDone = 0;

thorlcr/activities/when/thwhenslave.cpp

@@ -81,20 +81,14 @@ public:
         CDependencyExecutorSlaveActivity::init(data, slaveData);
         appendOutputLinked(this);
     }
-    void preStart(size32_t parentExtractSz, const byte *parentExtract)
-    {
-        CDependencyExecutorSlaveActivity::preStart(parentExtractSz, parentExtract);
-    }
-    virtual void start() override
-    {
-        ActivityTimer s(totalCycles, timeActivities);
-        PARENT::start();
-    }
     virtual void stop() override
     {
         PARENT::stop();
-        if (!executeDependencies(abortSoon ? WhenFailureId : WhenSuccessId))
-            abortSoon = true;
+        if (queryInputStarted(0))
+        {
+            if (!executeDependencies(abortSoon ? WhenFailureId : WhenSuccessId))
+                abortSoon = true;
+        }
     }
     virtual bool isGrouped() const override { return input->isGrouped(); }
     CATCH_NEXTROW()

thorlcr/activities/wuidread/thwuidread.cpp

@@ -79,6 +79,7 @@ CActivityBase *createWorkUnitActivityMaster(CMasterGraphElement *container)
 {
     StringBuffer diskFilename;
     IHThorWorkunitReadArg *wuReadHelper = (IHThorWorkunitReadArg *)container->queryHelper();
+    wuReadHelper->onCreate(container->queryCodeContext(), NULL, NULL);
     OwnedRoxieString fromWuid(wuReadHelper->getWUID());
     if (getWorkunitResultFilename(*container, diskFilename, fromWuid, wuReadHelper->queryName(), wuReadHelper->querySequence()))
     {

thorlcr/graph/thgraph.cpp

@@ -326,7 +326,7 @@ bool isDiskInput(ThorActivityKind kind)
 
 void CIOConnection::connect(unsigned which, CActivityBase *destActivity)
 {
-    destActivity->setInput(which, activity->queryActivity(true), index);
+    destActivity->setInput(which, activity->queryActivity(), index);
 }
 
 /////////////////////////////////// 
@@ -371,11 +371,9 @@ CGraphElementBase::CGraphElementBase(CGraphBase &_owner, IPropertyTree &_xgmml)
         throw makeOsExceptionV(GetLastError(), "Failed to load helper factory method: %s (dll handle = %p)", helperName.str(), queryJob().queryDllEntry().getInstance());
     alreadyUpdated = false;
     whichBranch = (unsigned)-1;
-    whichBranchBitSet.setown(createThreadSafeBitSet());
-    newWhichBranch = false;
-    hasNullInput = false;
     log = true;
     sentActInitData.setown(createThreadSafeBitSet());
+    baseHelper.setown(helperFactory());
 }
 
 CGraphElementBase::~CGraphElementBase()
@@ -584,7 +582,6 @@ void CGraphElementBase::onCreate()
     if (onCreateCalled)
         return;
     onCreateCalled = true;
-    baseHelper.setown(helperFactory());
     if (!nullAct)
     {
         CGraphElementBase *ownerActivity = owner->queryOwner() ? owner->queryOwner()->queryElement(ownerId) : NULL;
@@ -629,140 +626,168 @@ bool CGraphElementBase::executeDependencies(size32_t parentExtractSz, const byte
     return true;
 }
 
-bool CGraphElementBase::prepareContext(size32_t parentExtractSz, const byte *parentExtract, bool checkDependencies, bool shortCircuit, bool async)
+bool CGraphElementBase::prepareContext(size32_t parentExtractSz, const byte *parentExtract, bool checkDependencies, bool shortCircuit, bool async, bool connectOnly)
 {
     try
     {
-        bool _shortCircuit = shortCircuit;
-        Owned<IThorGraphDependencyIterator> deps = getDependsIterator();
-        bool depsDone = true;
-        ForEach(*deps)
+        bool create = true;
+        if (connectOnly)
         {
-            CGraphDependency &dep = deps->query();
-            if (0 == dep.controlId && NotFound == owner->dependentSubGraphs.find(*dep.graph))
+            if (activity)
+                return true;
+            ForEachItemIn(i, inputs)
             {
-                owner->dependentSubGraphs.append(*dep.graph);
-                if (!dep.graph->isComplete())
-                    depsDone = false;
+                if (!queryInput(i)->prepareContext(parentExtractSz, parentExtract, false, false, async, true))
+                    return false;
             }
         }
-        if (depsDone) _shortCircuit = false;
-        if (!depsDone && checkDependencies)
+        else
         {
-            if (!executeDependencies(parentExtractSz, parentExtract, 0, async))
-                return false;
-        }
-        whichBranch = (unsigned)-1;
-        hasNullInput = false;
-        alreadyUpdated = false;
-        switch (getKind())
-        {
-            case TAKindexwrite:
-            case TAKdiskwrite:
-            case TAKcsvwrite:
-            case TAKxmlwrite:
-            case TAKjsonwrite:
-                if (_shortCircuit) return true;
-                onCreate();
-                alreadyUpdated = checkUpdate();
-                if (alreadyUpdated)
-                    return false;
-                break;
-            case TAKchildif:
-                owner->ifs.append(*this);
-                // fall through
-            case TAKif:
-            case TAKifaction:
+            bool _shortCircuit = shortCircuit;
+            Owned<IThorGraphDependencyIterator> deps = getDependsIterator();
+            bool depsDone = true;
+            ForEach(*deps)
             {
-                if (_shortCircuit) return true;
-                onCreate();
-                onStart(parentExtractSz, parentExtract);
-                IHThorIfArg *helper = (IHThorIfArg *)baseHelper.get();
-                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))
+                CGraphDependency &dep = deps->query();
+                if (0 == dep.controlId && NotFound == owner->dependentSubGraphs.find(*dep.graph))
                 {
-                    if (!whichBranchBitSet->testSet(whichBranch)) // if not set, new
-                        newWhichBranch = true;
-                    return inputs.item(whichBranch)->activity->prepareContext(parentExtractSz, parentExtract, checkDependencies, false, async);
+                    owner->dependentSubGraphs.append(*dep.graph);
+                    if (!dep.graph->isComplete())
+                        depsDone = false;
                 }
-                return true;
             }
-            case TAKchildcase:
-                owner->ifs.append(*this);
-                // fall through
-            case TAKcase:
+            if (depsDone) _shortCircuit = false;
+            if (!depsDone && checkDependencies)
             {
-                if (_shortCircuit) return true;
-                onCreate();
-                onStart(parentExtractSz, parentExtract);
-                IHThorCaseArg *helper = (IHThorCaseArg *)baseHelper.get();
-                whichBranch = helper->getBranch();
-                if (whichBranch >= inputs.ordinality())
-                    whichBranch = inputs.ordinality()-1;
-                if (inputs.queryItem(whichBranch))
-                    return inputs.item(whichBranch)->activity->prepareContext(parentExtractSz, parentExtract, checkDependencies, false, async);
-                return true;
+                if (!executeDependencies(parentExtractSz, parentExtract, 0, async))
+                    return false;
             }
-            case TAKfilter:
-            case TAKfiltergroup:
-            case TAKfilterproject:
+            whichBranch = (unsigned)-1;
+            alreadyUpdated = false;
+            switch (getKind())
             {
-                if (_shortCircuit) return true;
-                onCreate();
-                onStart(parentExtractSz, parentExtract);
-                switch (getKind())
+                case TAKindexwrite:
+                case TAKdiskwrite:
+                case TAKcsvwrite:
+                case TAKxmlwrite:
+                case TAKjsonwrite:
+                    if (_shortCircuit) return true;
+                    onCreate();
+                    alreadyUpdated = checkUpdate();
+                    if (alreadyUpdated)
+                        return false;
+                    break;
+                case TAKchildif:
+                case TAKif:
+                case TAKifaction:
                 {
-                    case TAKfilter:
-                        hasNullInput = !((IHThorFilterArg *)baseHelper.get())->canMatchAny();
-                        break;
-                    case TAKfiltergroup:
-                        hasNullInput = !((IHThorFilterGroupArg *)baseHelper.get())->canMatchAny();
-                        break;
-                    case TAKfilterproject:
-                        hasNullInput = !((IHThorFilterProjectArg *)baseHelper.get())->canMatchAny();
-                        break;
+                    if (_shortCircuit) return true;
+                    onCreate();
+                    onStart(parentExtractSz, parentExtract);
+                    IHThorIfArg *helper = (IHThorIfArg *)baseHelper.get();
+                    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;
+                        create = false;
+                    }
+                    break;
+                }
+                case TAKchildcase:
+                case TAKcase:
+                {
+                    if (_shortCircuit) return true;
+                    onCreate();
+                    onStart(parentExtractSz, parentExtract);
+                    IHThorCaseArg *helper = (IHThorCaseArg *)baseHelper.get();
+                    whichBranch = helper->getBranch();
+                    if (whichBranch >= inputs.ordinality())
+                        whichBranch = inputs.ordinality()-1;
+                    break;
+                }
+                case TAKfilter:
+                case TAKfiltergroup:
+                case TAKfilterproject:
+                {
+                    if (_shortCircuit) return true;
+                    onCreate();
+                    onStart(parentExtractSz, parentExtract);
+                    switch (getKind())
+                    {
+                        case TAKfilter:
+                            whichBranch = ((IHThorFilterArg *)baseHelper.get())->canMatchAny() ? 0 : 1;
+                            break;
+                        case TAKfiltergroup:
+                            whichBranch = ((IHThorFilterGroupArg *)baseHelper.get())->canMatchAny() ? 0 : 1;
+                            break;
+                        case TAKfilterproject:
+                            whichBranch = ((IHThorFilterProjectArg *)baseHelper.get())->canMatchAny() ? 0 : 1;
+                            break;
+                    }
+                    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++)
+                        executeDependencies(parentExtractSz, parentExtract, s, async);
+                    create = false;
+                    break;
+                }
+                case TAKwhen_dataset:
+                case TAKwhen_action:
+                {
+                    if (!executeDependencies(parentExtractSz, parentExtract, WhenBeforeId, async))
+                        return false;
+                    if (!executeDependencies(parentExtractSz, parentExtract, WhenParallelId, async))
+                        return false;
+                    break;
                 }
-                if (hasNullInput)
-                    return true;
-                break;
             }
-            case TAKsequential:
-            case TAKparallel:
+            if (checkDependencies && ((unsigned)-1 != whichBranch))
             {
-                /* 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 (inputs.queryItem(whichBranch))
                 {
-                    if (!executeDependencies(parentExtractSz, parentExtract, s, async))
+                    if (!queryInput(whichBranch)->prepareContext(parentExtractSz, parentExtract, true, false, async, connectOnly))
                         return false;
                 }
-                break;
+                ForEachItemIn(i, inputs)
+                {
+                    if (i != whichBranch)
+                    {
+                        if (!queryInput(i)->prepareContext(parentExtractSz, parentExtract, false, false, async, true))
+                            return false;
+                    }
+                }
             }
-            case TAKwhen_dataset:
-            case TAKwhen_action:
+            else
             {
-                if (!executeDependencies(parentExtractSz, parentExtract, WhenBeforeId, async))
-                    return false;
-                if (!executeDependencies(parentExtractSz, parentExtract, WhenParallelId, async))
-                    return false;
-                break;
+                ForEachItemIn(i, inputs)
+                {
+                    if (!queryInput(i)->prepareContext(parentExtractSz, parentExtract, checkDependencies, false, async, connectOnly))
+                        return false;
+                }
             }
         }
-        ForEachItemIn(i, inputs)
+        if (create)
         {
-            CGraphElementBase *input = inputs.item(i)->activity;
-            if (!input->prepareContext(parentExtractSz, parentExtract, checkDependencies, shortCircuit, async))
-                return false;
+            if (activity) // no need to recreate
+                return true;
+            ForEachItemIn(i2, inputs)
+            {
+                CIOConnection *inputIO = inputs.item(i2);
+                connectInput(i2, inputIO->activity, inputIO->index);
+            }
+            if (isSink())
+                owner->addActiveSink(*this);
+            activity.setown(factory());
         }
         return true;
     }
@@ -790,12 +815,8 @@ void CGraphElementBase::preStart(size32_t parentExtractSz, const byte *parentExt
 
 void CGraphElementBase::initActivity()
 {
-    CriticalBlock b(crit);
-    if (isSink())
-        owner->addActiveSink(*this);
-    if (activity) // no need to recreate
-        return;
-    activity.setown(factory());
+    if (!activity)
+        activity.setown(factory());
     if (isLoopActivity(*this))
     {
         unsigned loopId = queryXGMML().getPropInt("att[@name=\"_loopid\"]/@value");
@@ -805,113 +826,6 @@ void CGraphElementBase::initActivity()
     }
 }
 
-void CGraphElementBase::createActivity(size32_t parentExtractSz, const byte *parentExtract)
-{
-    if (connectedInputs.ordinality()) // ensure not traversed twice (e.g. via splitter)
-        return;
-    try
-    {
-        switch (getKind())
-        {
-            case TAKchildif:
-            case TAKchildcase:
-            {
-                if (inputs.queryItem(whichBranch))
-                {
-                    CGraphElementBase *input = inputs.item(whichBranch)->activity;
-                    input->createActivity(parentExtractSz, parentExtract);
-                }
-                onCreate();
-                initActivity();
-                if (inputs.queryItem(whichBranch))
-                {
-                    CIOConnection *inputIO = inputs.item(whichBranch);
-                    connectInput(whichBranch, inputIO->activity, inputIO->index);
-                }
-                break;
-            }
-            case TAKif:
-            case TAKcase:
-                if (inputs.queryItem(whichBranch))
-                {
-                    CGraphElementBase *input = inputs.item(whichBranch)->activity;
-                    input->createActivity(parentExtractSz, parentExtract);
-                }
-                else
-                {
-                    onCreate();
-                    if (!activity)
-                        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 (!hasNullInput)
-                {
-                    ForEachItemIn(i, inputs)
-                    {
-                        CGraphElementBase *input = inputs.item(i)->activity;
-                        input->createActivity(parentExtractSz, parentExtract);
-                    }
-                    onCreate();
-                    if (isDiskInput(getKind()))
-                        onStart(parentExtractSz, parentExtract);
-                    ForEachItemIn(i2, inputs)
-                    {
-                        CIOConnection *inputIO = inputs.item(i2);
-                        loop
-                        {
-                            CGraphElementBase *input = inputIO->activity;
-                            switch (input->getKind())
-                            {
-                                case TAKif:
-                                case TAKcase:
-                                {
-                                    if (input->whichBranch >= input->getInputs()) // if, will have TAKnull activity, made at create time.
-                                    {
-                                        input = NULL;
-                                        break;
-                                    }
-                                    inputIO = input->inputs.item(input->whichBranch);
-                                    assertex(inputIO);
-                                    break;
-                                }
-                                default:
-                                    input = NULL;
-                                    break;
-                            }
-                            if (!input)
-                                break;
-                        }
-                        connectInput(i2, inputIO->activity, inputIO->index);
-                    }
-                }
-                initActivity();
-                break;
-        }
-    }
-    catch (IException *e) { ActPrintLog(e); activity.clear(); throw; }
-}
-
 ICodeContext *CGraphElementBase::queryCodeContext()
 {
     return queryOwner().queryCodeContext();
@@ -1134,13 +1048,8 @@ CGraphBase::CGraphBase(CJobChannel &_jobChannel) : jobChannel(_jobChannel), job(
     parent = owner = NULL;
     graphId = 0;
     complete = false;
-    reinit = false; // should graph reinitialize each time it is called (e.g. in loop graphs)
-                    // This is currently for 'init' (Create time) info and onStart into
-    sentInitData = false;
-//  sentStartCtx = false;
-    sentStartCtx = true; // JCSMORE - disable for now
     parentActivityId = 0;
-    created = connected = started = graphDone = aborted = prepared = false;
+    connected = started = graphDone = aborted = prepared = false;
     startBarrier = waitBarrier = doneBarrier = NULL;
     mpTag = waitBarrierTag = startBarrierTag = doneBarrierTag = TAG_NULL;
     executeReplyTag = TAG_NULL;
@@ -1275,6 +1184,7 @@ bool CGraphBase::fireException(IException *e)
 
 bool CGraphBase::preStart(size32_t parentExtractSz, const byte *parentExtract)
 {
+    started = true; // causes reset() to be called on all subsequent executions of this subgraph.
     Owned<IThorActivityIterator> iter = getConnectedIterator();
     ForEach(*iter)
     {
@@ -1292,27 +1202,11 @@ void CGraphBase::executeSubGraph(size32_t parentExtractSz, const byte *parentExt
     Owned<IException> exception;
     try
     {
-        if (!prepare(parentExtractSz, parentExtract, false, false, false))
+        if (!queryOwner())
         {
-            setCompleteEx();
-            return;
-        }
-        try
-        {
-            if (!queryOwner())
-            {
-                StringBuffer s;
-                toXML(&queryXGMML(), s, 2);
-                GraphPrintLog("Running graph [%s] : %s", isGlobal()?"global":"local", s.str());
-            }
-            create(parentExtractSz, parentExtract);
-        }
-        catch (IException *e)
-        {
-            Owned<IThorException> e2 = MakeGraphException(this, e);
-            e2->setAction(tea_abort);
-            queryJobChannel().fireException(e2);
-            throw;
+            StringBuffer s;
+            toXML(&queryXGMML(), s, 2);
+            GraphPrintLog("Running graph [%s] : %s", isGlobal()?"global":"local", s.str());
         }
         if (localResults)
             localResults->clear();
@@ -1334,21 +1228,31 @@ void CGraphBase::executeSubGraph(size32_t parentExtractSz, const byte *parentExt
         throw exception.getClear();
 }
 
+void CGraphBase::onCreate()
+{
+    Owned<IThorActivityIterator> iter = getConnectedIterator();
+    ForEach(*iter)
+    {
+        CGraphElementBase &element = iter->query();
+        element.onCreate();
+        element.initActivity();
+    }
+}
+
 void CGraphBase::execute(size32_t _parentExtractSz, const byte *parentExtract, bool checkDependencies, bool async)
 {
-    parentExtractSz = _parentExtractSz;
     if (isComplete())
         return;
     if (async)
-        queryJobChannel().startGraph(*this, queryJobChannel(), checkDependencies, parentExtractSz, parentExtract); // may block if enough running
+        queryJobChannel().startGraph(*this, checkDependencies, _parentExtractSz, parentExtract); // may block if enough running
     else
     {
-        if (!prepare(parentExtractSz, parentExtract, checkDependencies, async, async))
+        if (!prepare(_parentExtractSz, parentExtract, checkDependencies, false, false))
         {
             setComplete();
             return;
         }
-        executeSubGraph(parentExtractSz, parentExtract);
+        executeSubGraph(_parentExtractSz, parentExtract);
     }
 }
 
@@ -1359,18 +1263,7 @@ void CGraphBase::doExecute(size32_t parentExtractSz, const byte *parentExtract,
     {
         if (abortException)
             throw abortException.getLink();
-        throw MakeGraphException(this, 0, "subgraph aborted(1)");
-    }
-    if (!prepare(parentExtractSz, parentExtract, checkDependencies, false, false))
-    {
-        setComplete();
-        return;
-    }
-    if (queryAborted())
-    {
-        if (abortException)
-            throw abortException.getLink();
-        throw MakeGraphException(this, 0, "subgraph aborted(2)");
+        throw MakeGraphException(this, 0, "subgraph aborted");
     }
     Owned<IException> exception;
     try
@@ -1442,34 +1335,17 @@ bool CGraphBase::prepare(size32_t parentExtractSz, const byte *parentExtract, bo
 {
     if (isComplete()) return false;
     bool needToExecute = false;
-    ifs.kill();
     ForEachItemIn(s, sinks)
     {
         CGraphElementBase &sink = sinks.item(s);
-        if (sink.prepareContext(parentExtractSz, parentExtract, checkDependencies, shortCircuit, async))
+        if (sink.prepareContext(parentExtractSz, parentExtract, checkDependencies, shortCircuit, async, false))
             needToExecute = true;
     }
 //  prepared = true;
+    onCreate();
     return needToExecute;
 }
 
-void CGraphBase::create(size32_t parentExtractSz, const byte *parentExtract)
-{
-    Owned<IThorActivityIterator> iter = getIterator();
-    ForEach(*iter)
-    {
-        CGraphElementBase &element = iter->query();
-        element.clearConnections();
-    }
-    activeSinks.kill(); // NB: activeSinks are added to during activity creation
-    ForEachItemIn(s, sinks)
-    {
-        CGraphElementBase &sink = sinks.item(s);
-        sink.createActivity(parentExtractSz, parentExtract);
-    }
-    created = true;
-}
-
 void CGraphBase::done()
 {
     if (aborted) return; // activity done methods only called on success
@@ -1539,21 +1415,23 @@ protected:
         cur.setown(&others.popGet());
         return cur;
     }
-    CGraphElementBase *setNext(CIOConnectionArray &inputs, unsigned whichInput=((unsigned)-1))
+    void setNext(bool branchOnConditional)
     {
-        cur.clear();
-        unsigned n = inputs.ordinality();
-        if (((unsigned)-1) != whichInput)
+        if (branchOnConditional && ((unsigned)-1) != cur->whichBranch)
         {
-            CIOConnection *io = inputs.queryItem(whichInput);
+            CIOConnection *io = cur->connectedInputs.queryItem(cur->whichBranch);
             if (io)
                 cur.set(io->activity);
+            else
+                cur.clear();
         }
         else
         {
+            CIOConnectionArray &inputs = cur->connectedInputs;
+            cur.clear();
+            unsigned n = inputs.ordinality();
             bool first = true;
-            unsigned i=0;
-            for (; i<n; i++)
+            for (unsigned i=0; i<n; i++)
             {
                 CIOConnection *io = inputs.queryItem(i);
                 if (io)
@@ -1571,7 +1449,7 @@ protected:
         if (!cur)
         {
             if (!popNext())
-                return NULL;
+                return;
         }
         // check haven't been here before
         loop
@@ -1587,9 +1465,8 @@ protected:
                 }
             }
             if (!popNext())
-                return NULL;
+                return;
         }
-        return cur.get();
     }
 public:
     IMPLEMENT_IINTERFACE;
@@ -1623,28 +1500,19 @@ public:
 
 class CGraphTraverseConnectedIterator : public CGraphTraverseIteratorBase
 {
+    bool branchOnConditional;
 public:
-    CGraphTraverseConnectedIterator(CGraphBase &graph) : CGraphTraverseIteratorBase(graph) { }
+    CGraphTraverseConnectedIterator(CGraphBase &graph, bool _branchOnConditional) : CGraphTraverseIteratorBase(graph), branchOnConditional(_branchOnConditional) { }
     virtual bool next()
     {
-        if (cur->hasNullInput)
-        {
-            do
-            {
-                if (!popNext())
-                    return false;
-            }
-            while (cur->hasNullInput);
-        }
-        else
-            setNext(cur->connectedInputs);
+        setNext(branchOnConditional);
         return NULL!=cur.get();
     }
 };
 
-IThorActivityIterator *CGraphBase::getConnectedIterator()
+IThorActivityIterator *CGraphBase::getConnectedIterator(bool branchOnConditional)
 {
-    return new CGraphTraverseConnectedIterator(*this);
+    return new CGraphTraverseConnectedIterator(*this, branchOnConditional);
 }
 
 bool CGraphBase::wait(unsigned timeout)
@@ -2910,9 +2778,9 @@ void CJobChannel::clean()
     subGraphs.kill();
 }
 
-void CJobChannel::startGraph(CGraphBase &graph, IGraphCallback &callback, bool checkDependencies, size32_t parentExtractSize, const byte *parentExtract)
+void CJobChannel::startGraph(CGraphBase &graph, bool checkDependencies, size32_t parentExtractSize, const byte *parentExtract)
 {
-    graphExecutor->add(&graph, callback, checkDependencies, parentExtractSize, parentExtract);
+    graphExecutor->add(&graph, *this, checkDependencies, parentExtractSize, parentExtract);
 }
 
 IThorResult *CJobChannel::getOwnedResult(graph_id gid, activity_id ownerId, unsigned resultId)

thorlcr/graph/thgraph.hpp

@@ -262,14 +262,13 @@ public:
 
     const void *queryFindParam() const { return &queryId(); } // for SimpleHashTableOf
 
-    bool alreadyUpdated, hasNullInput, newWhichBranch;
+    bool alreadyUpdated;
     EclHelperFactory helperFactory;
 
     CIOConnectionArray inputs, outputs, connectedInputs, connectedOutputs;
 
     CGraphArray associatedChildGraphs;
     unsigned whichBranch;
-    Owned<IBitSet> whichBranchBitSet;
     Owned<IBitSet> sentActInitData;
 
     CGraphElementBase(CGraphBase &_owner, IPropertyTree &_xgmml);
@@ -340,7 +339,6 @@ public:
 
     IPropertyTree &queryXGMML() const { return *xgmml; }
     const activity_id &queryOwnerId() const { return ownerId; }
-    void createActivity(size32_t parentExtractSz, const byte *parentExtract);
 //
     const ThorActivityKind getKind() const { return kind; }
     const activity_id &queryId() const { return id; }
@@ -349,9 +347,9 @@ public:
         dst.append(eclText.get());
         return dst;
     }
-    virtual bool prepareContext(size32_t parentExtractSz, const byte *parentExtract, bool checkDependencies, bool shortCircuit, bool async);
+    virtual bool prepareContext(size32_t parentExtractSz, const byte *parentExtract, bool checkDependencies, bool shortCircuit, bool async, bool connectOnly);
+    CActivityBase *queryActivity() { return activity; }
 //
-    virtual CActivityBase *queryActivity(bool checkNull=false) { return activity; }
     virtual void initActivity();
     virtual CActivityBase *factory(ThorActivityKind kind) { assertex(false); return NULL; }
     virtual CActivityBase *factory() { return factory(getKind()); }
@@ -435,6 +433,7 @@ class graph_decl CGraphBase : public CInterface, implements IEclGraphResults, im
     CGraphTable childGraphsTable;
     CGraphArrayCopy childGraphs;
     Owned<IGraphTempHandler> tmpHandler;
+    bool initialized = false;
 
     void clean();
 
@@ -540,12 +539,10 @@ protected:
     Owned<IThorGraphResults> localResults, graphLoopResults;
     CGraphBase *owner, *parent;
     Owned<IException> abortException;
-    CGraphElementArrayCopy ifs;
     Owned<IPropertyTree> node;
     IBarrier *startBarrier, *waitBarrier, *doneBarrier;
     mptag_t mpTag, startBarrierTag, waitBarrierTag, doneBarrierTag;
-    bool created, connected, started, aborted, graphDone, prepared, sequential;
-    bool reinit, sentInitData, sentStartCtx;
+    bool connected, started, aborted, graphDone, prepared, sequential;
     CJobBase &job;
     CJobChannel &jobChannel;
     graph_id graphId;
@@ -566,6 +563,7 @@ public:
     const void *queryFindParam() const { return &queryGraphId(); } // for SimpleHashTableOf
 
     virtual void init() { }
+    void onCreate();
     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);
@@ -579,10 +577,11 @@ public:
     CGraphBase *queryParent() { return parent?parent:this; }
     IMPServer &queryMPServer() const;
     bool syncInitData();
+    inline void setInitialized() { initialized = true; }
+    inline bool isInitialized() const { return initialized; }
     bool isComplete() const { return complete; }
     bool isPrepared() const { return prepared; }
     bool isGlobal() const { return global; }
-    bool isCreated() const { return created; }
     bool isStarted() const { return started; }
     bool isLocalOnly() const; // this graph and all upstream dependencies
     bool isLocalChild() const { return localChild; }
@@ -620,7 +619,7 @@ public:
     {
         return new CGraphElementIterator(containers);
     }
-    IThorActivityIterator *getConnectedIterator();
+    IThorActivityIterator *getConnectedIterator(bool branchOnConditional=true);
     IThorActivityIterator *getSinkIterator() const
     {
         return new CGraphElementArrayIterator(activeSinks);
@@ -690,7 +689,6 @@ public:
     virtual void executeChild(size32_t parentExtractSz, const byte *parentExtract);
     virtual bool serializeStats(MemoryBuffer &mb) { return false; }
     virtual bool prepare(size32_t parentExtractSz, const byte *parentExtract, bool checkDependencies, bool shortCircuit, bool async);
-    virtual void create(size32_t parentExtractSz, const byte *parentExtract);
     virtual bool preStart(size32_t parentExtractSz, const byte *parentExtract);
     virtual void start() = 0;
     virtual bool wait(unsigned timeout);
@@ -935,7 +933,7 @@ public:
     void wait();
     ITimeReporter &queryTimeReporter() { return *timeReporter; }
     virtual CGraphBase *createGraph() = 0;
-    void startGraph(CGraphBase &graph, IGraphCallback &callback, bool checkDependencies, size32_t parentExtractSize, const byte *parentExtract);
+    void startGraph(CGraphBase &graph, bool checkDependencies, size32_t parentExtractSize, const byte *parentExtract);
     INode *queryMyNode();
     unsigned queryChannel() const { return channel; }
     bool isPrimary() const { return 0 == channel; }

thorlcr/graph/thgraphmaster.cpp

@@ -233,7 +233,6 @@ void CSlaveMessageHandler::main()
                         assertex(element);
                         try
                         {
-                            element->deserializeStartContext(msg);
                             element->doCreateActivity(parentExtractSz, parentExtract);
                         }
                         catch (IException *e)
@@ -548,7 +547,6 @@ void CMasterActivity::done()
 
 CMasterGraphElement::CMasterGraphElement(CGraphBase &_owner, IPropertyTree &_xgmml) : CGraphElementBase(_owner, _xgmml)
 {
-    sentCreateCtx = false;
 }
 
 bool CMasterGraphElement::checkUpdate()
@@ -609,11 +607,13 @@ bool CMasterGraphElement::checkUpdate()
 
 void CMasterGraphElement::initActivity()
 {
-    CriticalBlock b(crit);
-    bool first = (NULL == queryActivity());
     CGraphElementBase::initActivity();
-    if (first || queryActivity()->needReInit())
+    if (!initialized || queryActivity()->needReInit())
+    {
         ((CMasterActivity *)queryActivity())->init();
+        initialized = true;
+    }
+    owner->setInitialized();
 }
 
 void CMasterGraphElement::doCreateActivity(size32_t parentExtractSz, const byte *parentExtract)
@@ -2210,18 +2210,6 @@ void CMasterGraph::abort(IException *e)
     }
 }
 
-void CMasterGraph::serializeCreateContexts(MemoryBuffer &mb)
-{
-    CGraphBase::serializeCreateContexts(mb);
-    Owned<IThorActivityIterator> iter = getIterator();
-    ForEach (*iter)
-    {
-        CMasterGraphElement &element = (CMasterGraphElement &)iter->query();
-        if (reinit || !element.sentCreateCtx)
-            element.sentCreateCtx = true;
-    }
-}
-
 bool CMasterGraph::serializeActivityInitData(unsigned slave, MemoryBuffer &mb, IThorActivityIterator &iter)
 {
     CriticalBlock b(createdCrit);
@@ -2272,64 +2260,22 @@ bool CMasterGraph::prepare(size32_t parentExtractSz, const byte *parentExtract,
     return true;
 }
 
-void CMasterGraph::create(size32_t parentExtractSz, const byte *parentExtract)
+void CMasterGraph::execute(size32_t _parentExtractSz, const byte *parentExtract, bool checkDependencies, bool async)
 {
-    {
-        CriticalBlock b(createdCrit);
-        CGraphBase::create(parentExtractSz, parentExtract);
-    }
-    if (!aborted)
-    {
-        if (!queryOwner()) // owning graph sends query+child graphs
-        {
-            jobM->sendQuery(); // if not previously sent
-            if (globals->getPropBool("@watchdogProgressEnabled"))
-                queryJobManager().queryDeMonServer()->startGraph(this);
-            sendGraph(); // sends child graphs at same time
-        }
-        else
-        {
-            if (isGlobal())
-            {
-                ForEachItemIn(i, ifs)
-                {
-                    CGraphElementBase &ifElem = ifs.item(i);
-                    if (ifElem.newWhichBranch)
-                    {
-                        ifElem.newWhichBranch = false;
-                        sentInitData = false; // force re-request of create data.
-                        break;
-                    }
-                }
-                CMessageBuffer msg;
-                if (reinit || !sentInitData)
-                {
-                    sentInitData = true;
-                    serializeCreateContexts(msg);
-                }
-                else
-                    msg.append((unsigned)0);
-                try
-                {
-                    jobM->broadcast(queryJobChannel().queryJobComm(), msg, mpTag, LONGTIMEOUT, "serializeCreateContexts", &bcastMsgHandler);
-                }
-                catch (IException *e)
-                {
-                    GraphPrintLog(e, "Aborting graph create(2)");
-                    if (abortException)
-                    {
-                        e->Release();
-                        throw LINK(abortException);
-                    }
-                    throw;
-                }
-            }
-        }
-    }
+    if (isComplete())
+        return;
+    if (!queryOwner()) // owning graph sends query+child graphs
+        jobM->sendQuery(); // if not previously sent
+    CGraphBase::execute(parentExtractSz, parentExtract, checkDependencies, async);
 }
 
 void CMasterGraph::start()
 {
+    if (!queryOwner())
+    {
+        if (globals->getPropBool("@watchdogProgressEnabled"))
+            queryJobManager().queryDeMonServer()->startGraph(this);
+    }
     Owned<IThorActivityIterator> iter = getConnectedIterator();
     ForEach (*iter)
         iter->query().queryActivity()->startProcess();
@@ -2346,7 +2292,7 @@ void CMasterGraph::sendActivityInitData()
     for (; w<queryJob().querySlaves(); w++)
     {
         unsigned needActInit = 0;
-        Owned<IThorActivityIterator> iter = getConnectedIterator();
+        Owned<IThorActivityIterator> iter = getConnectedIterator(false);
         ForEach(*iter)
         {
             CGraphElementBase &element = iter->query();
@@ -2361,7 +2307,7 @@ void CMasterGraph::sendActivityInitData()
             try
             {
                 msg.rewrite(pos);
-                Owned<IThorActivityIterator> iter = getConnectedIterator();
+                Owned<IThorActivityIterator> iter = getConnectedIterator(false);
                 serializeActivityInitData(w, msg, *iter);
             }
             catch (IException *e)
@@ -2415,7 +2361,6 @@ void CMasterGraph::sendActivityInitData()
 void CMasterGraph::serializeGraphInit(MemoryBuffer &mb)
 {
     mb.append(graphId);
-    mb.append(reinit);
     serializeMPtag(mb, mpTag);
     mb.append((int)startBarrierTag);
     mb.append((int)waitBarrierTag);
@@ -2447,6 +2392,39 @@ void CMasterGraph::executeSubGraph(size32_t parentExtractSz, const byte *parentE
     }
     if (isComplete())
         return;
+    if (!sentGlobalInit)
+    {
+        sentGlobalInit = true;
+        if (!queryOwner())
+            sendGraph();
+        else
+        {
+            if (isGlobal())
+            {
+                CMessageBuffer msg;
+                serializeCreateContexts(msg);
+                try
+                {
+                    jobM->broadcast(queryJobChannel().queryJobComm(), msg, mpTag, LONGTIMEOUT, "serializeCreateContexts", &bcastMsgHandler);
+                }
+                catch (IException *e)
+                {
+                    GraphPrintLog(e, "Aborting graph create(2)");
+                    if (abortException)
+                    {
+                        e->Release();
+                        throw LINK(abortException);
+                    }
+                    throw;
+                }
+            }
+        }
+        if (syncInitData())
+        {
+            sendActivityInitData(); // has to be done at least once
+            // NB: At this point, on the slaves, the graphs will start
+        }
+    }
     fatalHandler.clear();
     fatalHandler.setown(new CFatalHandler(globals->getPropInt("@fatal_timeout", FATAL_TIMEOUT)));
     CGraphBase::executeSubGraph(parentExtractSz, parentExtract);
@@ -2483,7 +2461,6 @@ void CMasterGraph::sendGraph()
     CMessageBuffer msg;
     msg.append(GraphInit);
     msg.append(job.queryKey());
-    msg.append(queryGraphId());
     if (TAG_NULL == executeReplyTag)
         executeReplyTag = jobM->allocateMPTag();
     serializeMPtag(msg, executeReplyTag);
@@ -2509,31 +2486,7 @@ void CMasterGraph::sendGraph()
 
 bool CMasterGraph::preStart(size32_t parentExtractSz, const byte *parentExtract)
 {
-    started = true;
     GraphPrintLog("Processing graph");
-    if (!sentStartCtx || reinit)
-    {
-        sentStartCtx = true;
-        CMessageBuffer msg;
-        serializeStartContexts(msg);
-        try
-        {
-            jobM->broadcast(queryJobChannel().queryJobComm(), msg, mpTag, LONGTIMEOUT, "startCtx", NULL, true);
-        }
-        catch (IException *e)
-        {
-            GraphPrintLog(e, "Aborting preStart");
-            if (abortException)
-            {
-                e->Release();
-                throw LINK(abortException);
-            }
-            throw;
-        }
-    }
-
-    if (syncInitData())
-        sendActivityInitData(); // has to be done at least once
     CGraphBase::preStart(parentExtractSz, parentExtract);
     if (isGlobal())
     {
@@ -2738,7 +2691,6 @@ bool CMasterGraph::deserializeStats(unsigned node, MemoryBuffer &mb)
                     element->onCreate();
                     element->initActivity();
                     activity = (CMasterActivity *)element->queryActivity();
-                    created = true; // means some activities created within this graph
                 }
                 catch (IException *_e)
                 {

thorlcr/graph/thgraphmaster.ipp

@@ -43,6 +43,7 @@ class graphmaster_decl CMasterGraph : public CGraphBase
     CriticalSection createdCrit;
     Owned<IFatalHandler> fatalHandler;
     CriticalSection exceptCrit;
+    bool sentGlobalInit = false;
 
     CReplyCancelHandler activityInitMsgHandler, bcastMsgHandler, executeReplyMsgHandler;
 
@@ -65,19 +66,18 @@ public:
     virtual void executeSubGraph(size32_t parentExtractSz, const byte *parentExtract);
     CriticalSection &queryCreateLock() { return createdCrit; }
     void handleSlaveDone(unsigned node, MemoryBuffer &mb);
-    void serializeCreateContexts(MemoryBuffer &mb);
     bool serializeActivityInitData(unsigned slave, MemoryBuffer &mb, IThorActivityIterator &iter);
     void readActivityInitData(MemoryBuffer &mb, unsigned slave);
     bool deserializeStats(unsigned node, MemoryBuffer &mb);
     virtual void setComplete(bool tf=true);
-    virtual bool prepare(size32_t parentExtractSz, const byte *parentExtract, bool checkDependencies, bool shortCircuit, bool async);
-    virtual void create(size32_t parentExtractSz, const byte *parentExtract);
-
-    virtual bool preStart(size32_t parentExtractSz, const byte *parentExtract);
-    virtual void start();
-    virtual void done();
-    virtual void reset();
-    virtual void abort(IException *e);
+    virtual bool prepare(size32_t parentExtractSz, const byte *parentExtract, bool checkDependencies, bool shortCircuit, bool async) override;
+    virtual void execute(size32_t _parentExtractSz, const byte *parentExtract, bool checkDependencies, bool async) override;
+
+    virtual bool preStart(size32_t parentExtractSz, const byte *parentExtract) override;
+    virtual void start() override;
+    virtual void done() override;
+    virtual void reset() override;
+    virtual void abort(IException *e) override;
     IThorResult *createResult(CActivityBase &activity, unsigned id, IThorGraphResults *results, IThorRowInterfaces *rowIf, bool distributed, unsigned spillPriority=SPILL_PRIORITY_RESULT);
     IThorResult *createResult(CActivityBase &activity, unsigned id, IThorRowInterfaces *rowIf, bool distributed, unsigned spillPriority=SPILL_PRIORITY_RESULT);
     IThorResult *createGraphLoopResult(CActivityBase &activity, IThorRowInterfaces *rowIf, bool distributed, unsigned spillPriority=SPILL_PRIORITY_RESULT);
@@ -328,16 +328,13 @@ public:
 
 class graphmaster_decl CMasterGraphElement : public CGraphElementBase
 {
+    bool initialized = false;
 public:
-    IMPLEMENT_IINTERFACE;
-    
-    bool sentCreateCtx;
-
     CMasterGraphElement(CGraphBase &owner, IPropertyTree &xgmml);
     void doCreateActivity(size32_t parentExtractSz=0, const byte *parentExtract=NULL);
     virtual bool checkUpdate();
 
-    virtual void initActivity();
+    virtual void initActivity() override;
     virtual void slaveDone(size32_t slaveIdx, MemoryBuffer &mb);
 };
 

thorlcr/graph/thgraphslave.cpp

@@ -152,7 +152,7 @@ void CSlaveActivity::setInput(unsigned index, CActivityBase *inputActivity, unsi
         inputs.append(* new CThorInput());
     CThorInput &newInput = inputs.item(index);
     newInput.set(outLink, inputOutIdx);
-    if (!input)
+    if (0 == index && !input)
     {
         input = outLink;
         inputSourceIdx = inputOutIdx;
@@ -171,7 +171,7 @@ void CSlaveActivity::connectInputStreams(bool consumerOrdered)
 
 void CSlaveActivity::setInputStream(unsigned index, CThorInput &_input, bool consumerOrdered)
 {
-    if (input) // will be none if source act.
+    if (_input.itdl)
     {
         Owned<IStrandJunction> junction;
         IEngineRowStream *_inputStream = connectSingleStream(*this, _input.itdl, _input.sourceIdx, junction, _input.itdl->isInputOrdered(consumerOrdered));
@@ -202,7 +202,6 @@ void CSlaveActivity::setLookAhead(unsigned index, IStartableEngineRowStream *loo
 IStrandJunction *CSlaveActivity::getOutputStreams(CActivityBase &ctx, unsigned idx, PointerArrayOf<IEngineRowStream> &streams, const CThorStrandOptions * consumerOptions, bool consumerOrdered, IOrderedCallbackCollection * orderedCallbacks)
 {
     // Default non-stranded implementation, expects activity to have 1 output.
-    assertex(!idx);
     // By default, activities are assumed NOT to support streams
     bool inputOrdered = isInputOrdered(consumerOrdered);
     connectInputStreams(inputOrdered);
@@ -214,16 +213,13 @@ IStrandJunction *CSlaveActivity::getOutputStreams(CActivityBase &ctx, unsigned i
 
 void CSlaveActivity::appendOutput(IThorDataLink *itdl)
 {
-    IThorDataLinkExt *itdlExt = QUERYINTERFACE(itdl, IThorDataLinkExt);
-    dbgassertex(itdlExt);
-    unsigned outputNum = outputs.ordinality();
-    itdlExt->setOutputIdx(outputNum);
     outputs.append(itdl);
 }
 
 void CSlaveActivity::appendOutputLinked(IThorDataLink *itdl)
 {
-    itdl->Link();
+    if (itdl)
+        itdl->Link();
     appendOutput(itdl);
 }
 
@@ -285,6 +281,7 @@ void CSlaveActivity::startInput(unsigned index, const char *extra)
         if (_input.lookAhead)
             _input.lookAhead->start();
         _input.stopped = false;
+        _input.started = true;
         if (0 == index)
             inputStopped = false;
 #ifdef TRACE_STARTSTOP_EXCEPTIONS
@@ -346,38 +343,20 @@ void CSlaveActivity::reset()
 {
     CActivityBase::reset();
     ForEachItemIn(i, inputs)
-        resetJunction(inputs.item(i).junction);
-    input = nullptr;
-    inputStream = nullptr;
-    inputStopped = true;
-}
-
-void CSlaveActivity::abort()
-{
-    CActivityBase::abort();
-    CriticalBlock b(crit);
-    ForEachItemIn(o, outputs)
-    {
-        StringBuffer msg("-------->  ");
-        msg.append("GraphId = ").append(container.queryOwner().queryGraphId());
-        msg.append(" ActivityId = ").append(container.queryId());
-        msg.append("  OutputId = ").append(o);
-        MemoryBuffer mb;
-        outputs.item(o)->dataLinkSerialize(mb); // JCSMORE should add direct method
-        rowcount_t count;
-        mb.read(count);
-        msg.append(": Count = ").append(count);
-    }
+        inputs.item(i).reset();
+    inputStopped = false;
 }
 
 void CSlaveActivity::releaseIOs()
 {
 //  inputs.kill(); // don't want inputs to die before this dies (release in deconstructor) // JCSMORE not sure why care particularly.
     outputs.kill(); // outputs tend to be self-references, this clears them explicitly, otherwise end up leaking with circular references.
+    outputStreams.kill();
 }
 
 void CSlaveActivity::clearConnections()
 {
+    outputStreams.kill();
     inputs.kill();
 }
 
@@ -415,7 +394,10 @@ unsigned __int64 CSlaveActivity::queryLocalCycles() const
     {
         switch (container.getKind())
         {
+            case TAKif:
             case TAKchildif:
+            case TAKifaction:
+            case TAKcase:
             case TAKchildcase:
                 if (inputs.ordinality() && (((unsigned)-1) != container.whichBranch))
                 {
@@ -444,7 +426,13 @@ void CSlaveActivity::serializeStats(MemoryBuffer &mb)
     CriticalBlock b(crit);
     mb.append((unsigned __int64)cycle_to_nanosec(queryLocalCycles()));
     ForEachItemIn(i, outputs)
-        outputs.item(i)->dataLinkSerialize(mb);
+    {
+        IThorDataLink *output = queryOutput(i);
+        if (output)
+            outputs.item(i)->dataLinkSerialize(mb);
+        else
+            serializeNullItdl(mb);
+    }
 }
 
 void CSlaveActivity::debugRequest(unsigned edgeIdx, MemoryBuffer &msg)
@@ -470,6 +458,11 @@ void CSlaveActivity::dataLinkSerialize(MemoryBuffer &mb) const
     CEdgeProgress::dataLinkSerialize(mb);
 }
 
+rowcount_t CSlaveActivity::getProgressCount() const
+{
+    return CEdgeProgress::getCount();
+}
+
 void CSlaveActivity::debugRequest(MemoryBuffer &msg)
 {
 }
@@ -549,7 +542,6 @@ void CThorStrandedActivity::reset()
     assertex(active==0);
     ForEachItemIn(idx, strands)
         strands.item(idx).reset();
-    strands.kill();
     resetJunction(splitter);
     CSlaveActivity::reset();
     resetJunction(sourceJunction);
@@ -674,17 +666,60 @@ unsigned __int64 CThorStrandedActivity::queryTotalCycles() const
 
 void CThorStrandedActivity::dataLinkSerialize(MemoryBuffer &mb) const
 {
+    mb.append(getProgressCount());
+}
+
+rowcount_t CThorStrandedActivity::getProgressCount() const
+{
     rowcount_t totalCount = getCount();
     ForEachItemIn(i, strands)
     {
         CThorStrandProcessor &strand = strands.item(i);
         totalCount += strand.getCount();
     }
-    mb.append(totalCount);
+    return totalCount;
 }
 
+// CSlaveLateStartActivity
 
+void CSlaveLateStartActivity::lateStart(bool any)
+{
+    prefiltered = !any;
+    if (!prefiltered)
+        startInput(0);
+    else
+        stopInput(0);
+}
 
+void CSlaveLateStartActivity::start()
+{
+    Linked<CThorInput> savedInput = &inputs.item(0);
+    if (!nullInput)
+    {
+        nullInput.setown(new CThorInput);
+        nullInput->sourceIdx = savedInput->sourceIdx; // probably not needed
+    }
+    inputs.replace(* nullInput.getLink(), 0);
+    input = NULL;
+    CSlaveActivity::start();
+    inputs.replace(* savedInput.getClear(), 0);
+    input = inputs.item(0).itdl;
+}
+
+void CSlaveLateStartActivity::stop()
+{
+    if (!prefiltered)
+    {
+        stopInput(0);
+        dataLinkStop();
+    }
+}
+
+void CSlaveLateStartActivity::reset()
+{
+    CSlaveActivity::reset();
+    prefiltered = false;
+}
 
 // CSlaveGraph
 
@@ -695,7 +730,6 @@ CSlaveGraph::CSlaveGraph(CJobChannel &jobChannel) : CGraphBase(jobChannel)
 
 void CSlaveGraph::init(MemoryBuffer &mb)
 {
-    mb.read(reinit);
     mpTag = queryJobChannel().deserializeMPTag(mb);
     startBarrierTag = queryJobChannel().deserializeMPTag(mb);
     waitBarrierTag = queryJobChannel().deserializeMPTag(mb);
@@ -760,24 +794,11 @@ void CSlaveGraph::initWithActData(MemoryBuffer &in, MemoryBuffer &out)
     out.append((activity_id)0);
 }
 
-void CSlaveGraph::recvStartCtx()
-{
-    if (!sentStartCtx)
-    {
-        sentStartCtx = true;
-        CMessageBuffer msg;
-
-        if (!graphCancelHandler.recv(queryJobChannel().queryJobComm(), msg, 0, mpTag, NULL, LONGTIMEOUT))
-            throw MakeStringException(0, "Error receiving startCtx data for graph: %" GIDPF "d", graphId);
-        deserializeStartContexts(msg);
-    }
-}
-
 bool CSlaveGraph::recvActivityInitData(size32_t parentExtractSz, const byte *parentExtract)
 {
     bool ret = true;
     unsigned needActInit = 0;
-    Owned<IThorActivityIterator> iter = getConnectedIterator();
+    Owned<IThorActivityIterator> iter = getConnectedIterator(false);
     ForEach(*iter)
     {
         CGraphElementBase &element = (CGraphElementBase &)iter->query();
@@ -811,14 +832,14 @@ bool CSlaveGraph::recvActivityInitData(size32_t parentExtractSz, const byte *par
             assertex(!parentExtractSz || NULL!=parentExtract);
             msg.append(parentExtractSz);
             msg.append(parentExtractSz, parentExtract);
-            Owned<IThorActivityIterator> iter = getConnectedIterator();
+            Owned<IThorActivityIterator> iter = getConnectedIterator(false);
             ForEach(*iter)
             {
                 CSlaveGraphElement &element = (CSlaveGraphElement &)iter->query();
                 if (!element.sentActInitData->test(0))
                 {
                     msg.append(element.queryId());
-                    element.serializeStartContext(msg);
+//                    element.serializeStartContext(msg);
                 }
             }
             msg.append((activity_id)0);
@@ -848,7 +869,7 @@ bool CSlaveGraph::recvActivityInitData(size32_t parentExtractSz, const byte *par
             if (queryOwner() && !isGlobal())
             {
                 // initialize any for which no data was sent
-                Owned<IThorActivityIterator> iter = getConnectedIterator();
+                Owned<IThorActivityIterator> iter = getConnectedIterator(false);
                 ForEach(*iter)
                 {
                     CSlaveGraphElement &element = (CSlaveGraphElement &)iter->query();
@@ -882,13 +903,7 @@ bool CSlaveGraph::recvActivityInitData(size32_t parentExtractSz, const byte *par
 
 bool CSlaveGraph::preStart(size32_t parentExtractSz, const byte *parentExtract)
 {
-    started = true;
-    recvStartCtx();
     CGraphBase::preStart(parentExtractSz, parentExtract);
-
-    if (!recvActivityInitData(parentExtractSz, parentExtract))
-        return false;
-    connect(); // only now do slave acts. have all their outputs prepared.
     if (isGlobal())
     {
         if (!startBarrier->wait(false))
@@ -926,7 +941,7 @@ void CSlaveGraph::start()
 void CSlaveGraph::connect()
 {
     CriticalBlock b(progressCrit);
-    Owned<IThorActivityIterator> iter = getConnectedIterator();
+    Owned<IThorActivityIterator> iter = getConnectedIterator(false);
     ForEach(*iter)
         iter->query().doconnect();
     iter.setown(getSinkIterator());
@@ -945,6 +960,56 @@ void CSlaveGraph::executeSubGraph(size32_t parentExtractSz, const byte *parentEx
     Owned<IException> exception;
     try
     {
+        if (!doneInit)
+        {
+            doneInit = true;
+            if (queryOwner())
+            {
+                if (isGlobal())
+                {
+                    CMessageBuffer msg;
+                    if (!graphCancelHandler.recv(queryJobChannel().queryJobComm(), msg, 0, mpTag, NULL, LONGTIMEOUT))
+                        throw MakeStringException(0, "Error receiving createctx data for graph: %" GIDPF "d", graphId);
+                    try
+                    {
+                        size32_t len;
+                        msg.read(len);
+                        if (len)
+                        {
+                            MemoryBuffer initData;
+                            initData.append(len, msg.readDirect(len));
+                            deserializeCreateContexts(initData);
+                        }
+                        msg.clear();
+                        msg.append(false);
+                    }
+                    catch (IException *e)
+                    {
+                        msg.clear();
+                        msg.append(true);
+                        serializeThorException(e, msg);
+                    }
+                    if (!queryJobChannel().queryJobComm().send(msg, 0, msg.getReplyTag(), LONGTIMEOUT))
+                        throw MakeStringException(0, "Timeout sending init data back to master");
+                }
+                else
+                {
+                    CMessageBuffer msg;
+                    msg.append(smt_initGraphReq);
+                    msg.append(graphId);
+                    if (!queryJobChannel().queryJobComm().sendRecv(msg, 0, queryJob().querySlaveMpTag(), LONGTIMEOUT))
+                        throwUnexpected();
+                    size32_t len;
+                    msg.read(len);
+                    if (len)
+                        deserializeCreateContexts(msg);
+        // could still request 1 off, onCreate serialization from master 1st.
+                }
+            }
+            if (!recvActivityInitData(parentExtractSz, parentExtract))
+                throw MakeThorException(0, "preStart failure");
+            connect(); // only now do slave acts. have all their outputs prepared.
+        }
         CGraphBase::executeSubGraph(parentExtractSz, parentExtract);
     }
     catch (IException *e)
@@ -968,73 +1033,6 @@ void CSlaveGraph::executeSubGraph(size32_t parentExtractSz, const byte *parentEx
         throw exception.getClear();
 }
 
-void CSlaveGraph::create(size32_t parentExtractSz, const byte *parentExtract)
-{
-    CriticalBlock b(progressCrit);
-    if (queryOwner())
-    {
-        if (isGlobal())
-        {
-            CMessageBuffer msg;
-            // nothing changed if rerunning, unless conditional branches different
-            if (!graphCancelHandler.recv(queryJobChannel().queryJobComm(), msg, 0, mpTag, NULL, LONGTIMEOUT))
-                throw MakeStringException(0, "Error receiving createctx data for graph: %" GIDPF "d", graphId);
-            try
-            {
-                size32_t len;
-                msg.read(len);
-                if (len)
-                {
-                    MemoryBuffer initData;
-                    initData.append(len, msg.readDirect(len));
-                    deserializeCreateContexts(initData);
-                }
-                msg.clear();
-                msg.append(false);
-            }
-            catch (IException *e)
-            {
-                msg.clear();
-                msg.append(true);
-                serializeThorException(e, msg);
-            }
-            if (!queryJobChannel().queryJobComm().send(msg, 0, msg.getReplyTag(), LONGTIMEOUT))
-                throw MakeStringException(0, "Timeout sending init data back to master");
-        }
-        else
-        {
-            ForEachItemIn(i, ifs)
-            {
-                CGraphElementBase &ifElem = ifs.item(i);
-                if (ifElem.newWhichBranch)
-                {
-                    ifElem.newWhichBranch = false;
-                    sentInitData = false; // force re-request of create data.
-                    break;
-                }
-            }
-            if ((reinit || !sentInitData))
-            {
-                sentInitData = true;
-                CMessageBuffer msg;
-                msg.append(smt_initGraphReq);
-                msg.append(graphId);
-                if (!queryJobChannel().queryJobComm().sendRecv(msg, 0, queryJob().querySlaveMpTag(), LONGTIMEOUT))
-                    throwUnexpected();
-                size32_t len;
-                msg.read(len);
-                if (len)
-                    deserializeCreateContexts(msg);
-
-// could still request 1 off, onCreate serialization from master 1st.
-                CGraphBase::create(parentExtractSz, parentExtract);
-                return;
-            }
-        }
-    }
-    CGraphBase::create(parentExtractSz, parentExtract);
-}
-
 void CSlaveGraph::abort(IException *e)
 {
     if (!graphDone) // set pre done(), no need to abort if got that far.

thorlcr/graph/thgraphslave.hpp

@@ -78,9 +78,6 @@ public:
 
     inline void dataLinkStop()
     {
-#ifdef _TESTING
-        assertex(hasStarted());        // ITDL stopped without being started
-#endif
         count |= THORDATALINK_STOPPED;
 #ifdef _TESTING
         owner.ActPrintLog("ITDL output %d stopped, count was %" RCPF "d", outputId, getDataLinkCount());
@@ -119,15 +116,23 @@ public:
     Linked<IThorDebug> tracingStream;
     Linked<IEngineRowStream> stream;
     Linked<IStrandJunction> junction;
-    bool stopped = true;
+    bool stopped = false;
+    bool started = false;
 
     explicit CThorInput() { }
     void set(IThorDataLink *_itdl, unsigned idx) { itdl.set(_itdl); sourceIdx = idx; }
+    void reset()
+    {
+        started = stopped = false;
+        resetJunction(junction);
+    }
+    bool isStopped() const { return stopped; }
+    bool isStarted() const { return started; }
 };
 typedef IArrayOf<CThorInput> CThorInputArray;
 
 class CSlaveGraphElement;
-class graphslave_decl CSlaveActivity : public CActivityBase, public CEdgeProgress, public COutputTiming, implements IThorDataLinkExt, implements IEngineRowStream, implements IThorSlaveActivity
+class graphslave_decl CSlaveActivity : public CActivityBase, public CEdgeProgress, public COutputTiming, implements IThorDataLink, implements IEngineRowStream, implements IThorSlaveActivity
 {
     mutable MemoryBuffer *data;
     mutable CriticalSection crit;
@@ -137,14 +142,13 @@ protected:
     IPointerArrayOf<IThorDataLink> outputs;
     IPointerArrayOf<IEngineRowStream> outputStreams;
     IThorDataLink *input = nullptr;
-    bool inputStopped = true;
+    bool inputStopped = false;
     unsigned inputSourceIdx = 0;
     IEngineRowStream *inputStream = nullptr;
     MemoryBuffer startCtx;
     bool optStableInput = true; // is the input forced to ordered?
     bool optUnstableInput = false;  // is the input forced to unordered?
     bool optUnordered = false; // is the output specified as unordered?
-    unsigned outputIdx = 0; // for IThorDataLinkExt
 
 protected:
     unsigned __int64 queryLocalCycles() const;
@@ -166,6 +170,8 @@ public:
     IThorDataLink *queryInput(unsigned index) const;
     IEngineRowStream *queryInputStream(unsigned index) const;
     IEngineRowStream *queryOutputStream(unsigned index) const;
+    inline bool queryInputStarted(unsigned input) const { return inputs.item(input).isStarted(); }
+    inline bool queryInputStopped(unsigned input) const { return inputs.item(input).isStopped(); }
     unsigned queryInputOutputIndex(unsigned inputIndex) const { return inputs.item(inputIndex).sourceIdx; }
     unsigned queryNumInputs() const { return inputs.ordinality(); }
     void appendOutput(IThorDataLink *itdl);
@@ -184,8 +190,8 @@ public:
     virtual void getMetaInfo(ThorDataLinkMetaInfo &info) override { }
     virtual bool isGrouped() const override;
     virtual IOutputMetaData * queryOutputMeta() const;
-    virtual unsigned queryOutputIdx() const override { return outputIdx; }
     virtual void dataLinkSerialize(MemoryBuffer &mb) const override;
+    virtual rowcount_t getProgressCount() const override;
     virtual bool isInputOrdered(bool consumerOrdered) const override
     {
         if (optStableInput)
@@ -202,8 +208,6 @@ public:
 
 // IThorDataLink
     virtual void start() override;
-// IThorDataLinkExt
-    virtual void setOutputIdx(unsigned idx) override { outputIdx = idx; }
 
 // IEngineRowStream
     virtual const void *nextRow() override { throwUnexpected(); }
@@ -215,10 +219,26 @@ public:
     virtual void setInputStream(unsigned index, CThorInput &input, bool consumerOrdered) override;
     virtual void processDone(MemoryBuffer &mb) override { };
     virtual void reset() override;
-    virtual void abort() override;
 };
 
 
+class graphslave_decl CSlaveLateStartActivity : public CSlaveActivity
+{
+    bool prefiltered = false;
+    Owned<CThorInput> nullInput;
+
+protected:
+    void lateStart(bool any);
+
+public:
+    CSlaveLateStartActivity(CGraphElementBase *container) : CSlaveActivity(container)
+    {
+    }
+    virtual void start() override;
+    virtual void stop() override;
+    virtual void reset() override;
+};
+
 graphslave_decl IEngineRowStream *connectSingleStream(CActivityBase &activity, IThorDataLink *input, unsigned idx, Owned<IStrandJunction> &junction, bool consumerOrdered);
 graphslave_decl IEngineRowStream *connectSingleStream(CActivityBase &activity, IThorDataLink *input, unsigned idx, bool consumerOrdered);
 
@@ -314,10 +334,10 @@ public:
     virtual IStrandJunction *getOutputStreams(CActivityBase &_ctx, unsigned idx, PointerArrayOf<IEngineRowStream> &streams, const CThorStrandOptions * consumerOptions, bool consumerOrdered, IOrderedCallbackCollection * orderedCallbacks) override;
     virtual unsigned __int64 queryTotalCycles() const override;
     virtual void dataLinkSerialize(MemoryBuffer &mb) const override;
+    virtual rowcount_t getProgressCount() const override;
 };
 
 
-
 class graphslave_decl CSlaveGraphElement : public CGraphElementBase
 {
 public:
@@ -334,6 +354,7 @@ class graphslave_decl CSlaveGraph : public CGraphBase
     bool initialized, progressActive, progressToCollect;
     CriticalSection progressCrit;
     SpinLock progressActiveLock;
+    bool doneInit = false;
 
 public:
 
@@ -342,7 +363,6 @@ public:
 
     void connect();
     void init(MemoryBuffer &mb);
-    void recvStartCtx();
     bool recvActivityInitData(size32_t parentExtractSz, const byte *parentExtract);
     void setExecuteReplyTag(mptag_t _executeReplyTag) { executeReplyTag = _executeReplyTag; }
     void initWithActData(MemoryBuffer &in, MemoryBuffer &out);
@@ -350,14 +370,13 @@ public:
     void serializeDone(MemoryBuffer &mb);
     IThorResult *getGlobalResult(CActivityBase &activity, IThorRowInterfaces *rowIf, activity_id ownerId, unsigned id);
 
-    virtual void executeSubGraph(size32_t parentExtractSz, const byte *parentExtract);
+    virtual void executeSubGraph(size32_t parentExtractSz, const byte *parentExtract) override;
     virtual bool serializeStats(MemoryBuffer &mb);
-    virtual bool preStart(size32_t parentExtractSz, const byte *parentExtract);
-    virtual void start();
-    virtual void create(size32_t parentExtractSz, const byte *parentExtract);
-    virtual void abort(IException *e);
-    virtual void done();
-    virtual void end();
+    virtual bool preStart(size32_t parentExtractSz, const byte *parentExtract) override;
+    virtual void start() override;
+    virtual void abort(IException *e) override;
+    virtual void done() override;
+    virtual void end() override;
     virtual IThorGraphResults *createThorGraphResults(unsigned num);
 
 // IExceptionHandler

thorlcr/master/thactivitymaster.cpp

@@ -108,7 +108,9 @@ public:
         {
             case TAKfiltergroup:
             case TAKlocalresultread:
+            case TAKif:
             case TAKchildif:
+            case TAKcase:
             case TAKchildcase:
             case TAKdegroup:
             case TAKsplit:
@@ -384,8 +386,6 @@ public:
                 break;
             case TAKchilddataset:
                 UNIMPLEMENTED;
-            case TAKcase:           // gen. time.
-            case TAKif:
             case TAKifaction:
                 throwUnexpected();
             case TAKwhen_dataset:

thorlcr/master/thdemonserver.cpp

@@ -72,7 +72,7 @@ private:
         try
         {
             StatsSubgraphScope subgraph(stats, graph->queryGraphId());
-            if (graph->isCreated())
+            if (graph->isInitialized())
                 doReportGraph(stats, graph, finished);
             Owned<IThorGraphIterator> graphIter = graph->getChildGraphs();
             ForEach (*graphIter)

thorlcr/slave/slave.cpp

@@ -268,6 +268,7 @@ CActivityBase *createWhenSlave(CGraphElementBase *container);
 CActivityBase *createDictionaryWorkunitWriteSlave(CGraphElementBase *container);
 CActivityBase *createDictionaryResultWriteSlave(CGraphElementBase *container);
 CActivityBase *createTraceSlave(CGraphElementBase *container);
+CActivityBase *createIfActionSlave(CGraphElementBase *container);
 
 
 class CGenericSlaveGraphElement : public CSlaveGraphElement
@@ -300,18 +301,6 @@ public:
         }
         haveCreateCtx = true;
     }
-    virtual CActivityBase *queryActivity(bool checkNull)
-    {
-        if (checkNull && hasNullInput)
-        {
-            CriticalBlock b(nullActivityCs);
-            if (!nullActivity)
-                nullActivity.setown(createNullSlave(this));
-            return nullActivity;
-        }
-        else
-            return activity;
-    }
     virtual CActivityBase *factory(ThorActivityKind kind)
     {
         CActivityBase *ret = NULL;
@@ -516,6 +505,9 @@ public:
             case TAKapply:
                 ret = createApplySlave(this);
                 break;
+            case TAKifaction:
+                ret = createIfActionSlave(this);
+                break;
             case TAKinlinetable:
                 ret = createInlineTableSlave(this);
                 break;
@@ -647,11 +639,6 @@ public:
             case TAKthroughaggregate:
                 ret = createThroughAggregateSlave(this);
                 break;
-            case TAKcase:
-            case TAKif:
-            case TAKifaction:
-                throwUnexpected();
-                break;
             case TAKwhen_dataset:
                 ret = createWhenSlave(this);
                 break;
@@ -743,9 +730,11 @@ public:
             case TAKlocalresultspill:
                 ret = createLocalResultSpillSlave(this);
                 break;
+            case TAKif:
             case TAKchildif:
                 ret = createIfSlave(this);
                 break;
+            case TAKcase:
             case TAKchildcase:
                 ret = createCaseSlave(this);
                 break;

thorlcr/slave/slave.hpp

@@ -94,12 +94,12 @@ interface IThorDataLink : extends IInterface
     virtual void getMetaInfo(ThorDataLinkMetaInfo &info) = 0;
     virtual bool isGrouped() const { return false; }
     virtual IOutputMetaData * queryOutputMeta() const = 0;
-    virtual unsigned queryOutputIdx() const = 0;
     virtual bool isInputOrdered(bool consumerOrdered) const = 0;
     virtual IStrandJunction *getOutputStreams(CActivityBase &_ctx, unsigned idx, PointerArrayOf<IEngineRowStream> &streams, const CThorStrandOptions * consumerOptions, bool consumerOrdered, IOrderedCallbackCollection * orderedCallbacks) = 0;
     virtual void setOutputStream(unsigned index, IEngineRowStream *stream) = 0;
 // progress methods
     virtual void dataLinkSerialize(MemoryBuffer &mb) const = 0;
+    virtual rowcount_t getProgressCount() const = 0;
 // timing methods
     virtual unsigned __int64 queryTotalCycles() const = 0;
     virtual unsigned __int64 queryEndCycles() const = 0;
@@ -110,11 +110,7 @@ interface IThorDataLink : extends IInterface
     virtual bool gatherConjunctions(ISteppedConjunctionCollector & collector) { return false; }
 };
 
-// helper interface. Used by maintainer of output links
-interface IThorDataLinkExt : extends IThorDataLink
-{
-    virtual void setOutputIdx(unsigned idx) = 0;
-};
+inline void serializeNullItdl(MemoryBuffer &mb) { mb.append((rowcount_t)0); }
 
 class CThorInput;
 interface IThorSlaveActivity

thorlcr/slave/slave.ipp

@@ -42,8 +42,6 @@ protected:
     virtual void process() { }
 
 public:
-    IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
-
     ProcessSlaveActivity(CGraphElementBase *container);
     virtual void beforeDispose();
 
@@ -95,7 +93,7 @@ class CThorNarySlaveActivity : public CSlaveActivity
 protected:
     PointerArrayOf<IThorDataLink> expandedInputs;
     Owned<IStrandJunction> *expandedJunctions = nullptr;
-    IPointerArrayOf<IEngineRowStream> expandedStreams;
+    PointerArrayOf<IEngineRowStream> expandedStreams;
 
 public:
     CThorNarySlaveActivity(CGraphElementBase *container) : CSlaveActivity(container)
@@ -126,7 +124,7 @@ public:
         }
         ForEachItemIn(ei, expandedInputs)
             expandedInputs.item(ei)->start();
-        expandedJunctions = new Owned<IStrandJunction> [expandedInputs.length()];
+        expandedJunctions = new Owned<IStrandJunction> [expandedInputs.ordinality()];
         ForEachItemIn(idx, expandedInputs)
         {
             expandedStreams.append(connectSingleStream(*this, expandedInputs.item(idx), 0, expandedJunctions[idx], true));  // MORE - is the index 0 right?

thorlcr/slave/slavmain.cpp

@@ -389,19 +389,18 @@ public:
                         if (!job)
                             throw MakeStringException(0, "Job not found: %s", jobKey.get());
 
-                        graph_id subGraphId = 0;
-                        msg.read(subGraphId);
-
-                        VStringBuffer xpath("node[@id='%" GIDPF "u']", subGraphId);
-                        Owned<IPropertyTree> graphNode = job->queryGraphXGMML()->getPropTree(xpath.str());
                         mptag_t executeReplyTag = job->deserializeMPTag(msg);
                         size32_t len;
                         msg.read(len);
                         MemoryBuffer createInitData;
                         createInitData.append(len, msg.readDirect(len));
-                        graph_id gid;
-                        msg.read(gid);
+
+                        graph_id subGraphId;
+                        msg.read(subGraphId);
                         unsigned graphInitDataPos = msg.getPos();
+
+                        VStringBuffer xpath("node[@id='%" GIDPF "u']", subGraphId);
+                        Owned<IPropertyTree> graphNode = job->queryGraphXGMML()->getPropTree(xpath.str());
                         job->addSubGraph(*graphNode);
 
                         /* JCSMORE - should improve, create 1st graph with create context/init data and clone
@@ -409,9 +408,9 @@ public:
                          */
                         for (unsigned c=0; c<job->queryJobChannels(); c++)
                         {
-                            PROGLOG("GraphInit: %s, graphId=%" GIDPF "d, slaveChannel=%d", jobKey.get(), gid, c);
+                            PROGLOG("GraphInit: %s, graphId=%" GIDPF "d, slaveChannel=%d", jobKey.get(), subGraphId, c);
                             CJobChannel &jobChannel = job->queryJobChannel(c);
-                            Owned<CSlaveGraph> subGraph = (CSlaveGraph *)jobChannel.getGraph(gid);
+                            Owned<CSlaveGraph> subGraph = (CSlaveGraph *)jobChannel.getGraph(subGraphId);
                             subGraph->setExecuteReplyTag(executeReplyTag);
 
                             createInitData.reset(0);
@@ -421,11 +420,18 @@ public:
                             subGraph->init(msg);
 
                             jobChannel.addDependencies(job->queryXGMML(), false);
-
-                            subGraph->execute(0, NULL, true, true);
                         }
                         msg.clear();
                         msg.append(false);
+                        queryNodeComm().reply(msg); // reply to sendGraph()
+
+                        for (unsigned c=0; c<job->queryJobChannels(); c++)
+                        {
+                            CJobChannel &jobChannel = job->queryJobChannel(c);
+                            Owned<CSlaveGraph> subGraph = (CSlaveGraph *)jobChannel.getGraph(subGraphId);
+
+                            jobChannel.startGraph(*subGraph, true, 0, NULL);
+                        }
 
                         break;
                     }