Merge pull request #9027 from jakesmith/hpcc-16051.master

HPCC-16051 Fix problems with child act. initialization

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

testing/regress/ecl/key/loopvar.xml

@@ -0,0 +1,14 @@
+<Dataset name='Result 1'>
+</Dataset>
+<Dataset name='Result 2'>
+</Dataset>
+<Dataset name='Result 3'>
+</Dataset>
+<Dataset name='Result 4'>
+</Dataset>
+<Dataset name='dsLoop'>
+ <Row><id>100</id><val>val from dsVal2=100</val></Row>
+</Dataset>
+<Dataset name='dsLoop2'>
+ <Row><id>100</id><val>val from dsVal2=100</val></Row>
+</Dataset>

testing/regress/ecl/loopvar.ecl

@@ -0,0 +1,73 @@
+/*##############################################################################
+
+    HPCC SYSTEMS software Copyright (C) 2012 HPCC Systems®.
+
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+############################################################################## */
+
+//noroxie
+
+import Std;
+
+filename1 := '~someflatfile1';
+filename2 := '~someflatfile2';
+keyname1 := '~somekey' : STORED('keyname1');
+keyname2 := '~someotherkey' : STORED('keyname2');
+
+dummy := 0;
+keynamecalc := IF(Std.System.thorlib.Daliserver() = 'dummy', '~somekey' , '~someotherkey');
+
+lVal := RECORD
+ INTEGER id := 0;
+ STRING val := '';
+END;
+
+lValKey := RECORD
+ lVal;
+ Unsigned8 fpos {virtual(fileposition)} ;
+END;
+
+dsVal1 := DATASET(10000, TRANSFORM(lVal, SELF.id := COUNTER; SELF.val := 'val from dsVal1=' + COUNTER;), DISTRIBUTED);
+dsVal2 := DATASET(10000, TRANSFORM(lVal, SELF.id := COUNTER; SELF.val := 'val from dsVal2=' + COUNTER;), DISTRIBUTED);
+
+saveit1 := OUTPUT(SORT(DISTRIBUTE(dsVal1, id), id, LOCAL), , filename1, OVERWRITE);
+saveit2 := OUTPUT(SORT(DISTRIBUTE(dsVal2, id), id, LOCAL), , filename2, OVERWRITE);
+
+dsKey1 := DATASET(filename1, lValKey, THOR);
+dsKey2 := DATASET(filename2, lValKey, THOR);
+
+buildit1 := BUILDINDEX(INDEX(dsKey1,{id}, {val, fpos}, keyname1), SORTED, OVERWRITE);
+buildit2 := BUILDINDEX(INDEX(dsKey2,{id}, {val, fpos}, keyname2), SORTED, OVERWRITE);
+
+dsQry := DATASET([{100, ''}], lVal);
+
+lVal loopBody(dataset(lVal) inVal, unsigned4 c) := FUNCTION
+ keyname := IF(c%2=1, keyname1, keyname2);
+ valKey := INDEX(DATASET([], lValKey), {id}, {val, fpos}, keyname);
+ RETURN JOIN(inVal, valKey, LEFT.id = RIGHT.id, TRANSFORM(lVal, SELF := RIGHT));
+END;
+
+dsLoop := LOOP(dsQry, 2, loopBody(rows(left), counter));
+
+lVal loopBody2(dataset(lVal) inVal, unsigned4 c) := FUNCTION
+ valKey := INDEX(DATASET([], lValKey), {id}, {val, fpos}, keynamecalc);
+ RETURN JOIN(inVal, valKey, LEFT.id = RIGHT.id, TRANSFORM(lVal, SELF := RIGHT));
+END;
+
+dsLoop2 := LOOP(dsQry, 2, loopBody2(rows(left), counter));
+
+SEQUENTIAL(
+ saveit1, saveit2, buildit1, buildit2,
+ OUTPUT(dsLoop, named('dsLoop'));
+ OUTPUT(dsLoop2, named('dsLoop2'));
+);

thorlcr/activities/aggregate/thaggregateslave.cpp

@@ -119,6 +119,8 @@ public:
     {
         hadElement = inputStopped = false;
         appendOutputLinked(this);
+        if (container.queryLocal())
+            setRequireInitData(false);
     }
     virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
     {

thorlcr/activities/aggregate/thgroupaggregateslave.cpp

@@ -29,13 +29,9 @@ public:
         : CSlaveActivity(_container)
     { 
         helper = static_cast <IHThorAggregateArg *> (queryHelper());
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
-
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
-    {
-    }
-
     virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);

thorlcr/activities/apply/thapplyslave.cpp

@@ -23,10 +23,11 @@ class CApplySlaveActivity : public ProcessSlaveActivity
     IHThorApplyArg *helper;
 
 public:
-    CApplySlaveActivity(CGraphElementBase *container) 
-        : ProcessSlaveActivity(container)
+    CApplySlaveActivity(CGraphElementBase *_container)
+        : ProcessSlaveActivity(_container)
     { 
         helper = static_cast <IHThorApplyArg *> (queryHelper());
+        setRequireInitData(false);
     }
 // IThorSlaveProcess overloaded methods
     virtual void process()

thorlcr/activities/catch/thcatchslave.cpp

@@ -33,6 +33,7 @@ public:
     CCatchSlaveActivityBase(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
         helper = static_cast <IHThorCatchArg *> (queryHelper());
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     virtual void start() override
@@ -187,10 +188,12 @@ class CSkipCatchSlaveActivity : public CCatchSlaveActivityBase
     }
 
 public:
-    CSkipCatchSlaveActivity(CGraphElementBase *container) 
-        : CCatchSlaveActivityBase(container)
+    CSkipCatchSlaveActivity(CGraphElementBase *_container)
+        : CCatchSlaveActivityBase(_container)
     {
-        global = !container->queryLocalOrGrouped();
+        global = !container.queryLocalOrGrouped();
+        if (!global)
+            setRequireInitData(false);
     }
     virtual void init(MemoryBuffer & data, MemoryBuffer &slaveData) override
     {

thorlcr/activities/choosesets/thchoosesetsslave.cpp

@@ -260,6 +260,8 @@ public:
         totalCounts = NULL;
         limits = NULL;
         inputCounter.setown(new CInputCounter(*this));
+        if (container.queryLocalOrGrouped())
+            setRequireInitData(false);
         appendOutputLinked(this);
     }
     ~ChooseSetsPlusActivity()

thorlcr/activities/countproject/thcountprojectslave.cpp

@@ -47,8 +47,9 @@ class LocalCountProjectActivity : public BaseCountProjectActivity
     bool anyThisGroup;
 
 public:
-    LocalCountProjectActivity(CGraphElementBase *container) : BaseCountProjectActivity(container)
+    LocalCountProjectActivity(CGraphElementBase *_container) : BaseCountProjectActivity(_container)
     {
+        setRequireInitData(false);
     }
     virtual void start()
     {

thorlcr/activities/degroup/thdegroupslave.cpp

@@ -27,6 +27,7 @@ public:
     CDegroupSlaveActivity(CGraphElementBase *_container) 
         : CSlaveActivity(_container), CThorSteppable(this)
     { 
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     virtual void start() override

thorlcr/activities/enth/thenthslave.cpp

@@ -129,10 +129,11 @@ class CLocalEnthSlaveActivity : public BaseEnthActivity
 
     bool localCountReq;
 public:
-    CLocalEnthSlaveActivity(CGraphElementBase *container) : BaseEnthActivity(container)
+    CLocalEnthSlaveActivity(CGraphElementBase *_container) : BaseEnthActivity(_container)
     {
         actStr.append("LOCALENTH");
         localCountReq = false;
+        setRequireInitData(false);
     }
     virtual void start()
     {

thorlcr/activities/fetch/thfetchslave.cpp

@@ -516,10 +516,6 @@ class CFetchSlaveActivity : public CFetchSlaveBase
 {
 public:
     CFetchSlaveActivity(CGraphElementBase *container) : CFetchSlaveBase(container) { }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
-    {
-        CFetchSlaveBase::init(data, slaveData);
-    }
     virtual size32_t fetch(ARowBuilder & rowBuilder, const void *keyRow, unsigned filePartIndex, unsigned __int64 localFpos, unsigned __int64 fpos)
     {
         Owned<ISerialStream> stream = createFileSerialStream(fetchStream->queryPartIO(filePartIndex), localFpos);

thorlcr/activities/filter/thfilterslave.cpp

@@ -28,6 +28,7 @@ public:
     explicit CFilterSlaveActivityBase(CGraphElementBase *_container)
         : CSlaveActivity(_container)
     {
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     virtual void start() override
@@ -177,10 +178,6 @@ public:
         : CFilterSlaveActivityBase(container)
     {
         helper = static_cast <IHThorFilterProjectArg *> (queryHelper());
-    }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
-    {
-        PARENT::init(data,slaveData);
         allocator.set(queryRowAllocator());
     }
     virtual void start() override

thorlcr/activities/firstn/thfirstnslave.cpp

@@ -72,8 +72,9 @@ class CFirstNSlaveLocal : public CFirstNSlaveBase
     bool firstget;
     rowcount_t skipped;
 public:
-    CFirstNSlaveLocal(CGraphElementBase *container) : CFirstNSlaveBase(container)
+    CFirstNSlaveLocal(CGraphElementBase *_container) : CFirstNSlaveBase(_container)
     {
+        setRequireInitData(false);
     }
 
 // IRowStream overrides
@@ -126,8 +127,9 @@ class CFirstNSlaveGrouped : public CFirstNSlaveBase
 
     unsigned countThisGroup;
 public:
-    CFirstNSlaveGrouped(CGraphElementBase *container) : CFirstNSlaveBase(container)
+    CFirstNSlaveGrouped(CGraphElementBase *_container) : CFirstNSlaveBase(_container)
     {
+        setRequireInitData(false);
     }
 
 // IRowStream overrides

thorlcr/activities/funnel/thfunnelslave.cpp

@@ -279,27 +279,23 @@ class FunnelSlaveActivity : public CSlaveActivity
 public:
     FunnelSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
-        grouped = false;
         eog = NULL;
         current = NULL;
         currentMarker = 0;
         eogNext = false;
         readThisInput = 0;
         stopped = true;
-        parallel = false;
+        IHThorFunnelArg *helper = (IHThorFunnelArg *)queryHelper();
+        parallel = !container.queryGrouped() && !helper->isOrdered() && getOptBool(THOROPT_PARALLEL_FUNNEL, true);
+        grouped = container.queryGrouped();
+        ActPrintLog("FUNNEL mode = %s, grouped=%s", parallel?"PARALLEL":"ORDERED", grouped?"GROUPED":"UNGROUPED");
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     ~FunnelSlaveActivity()
     {
         if (eog) delete [] eog;
     }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
-    {
-        IHThorFunnelArg *helper = (IHThorFunnelArg *)queryHelper();
-        parallel = !container.queryGrouped() && !helper->isOrdered() && getOptBool(THOROPT_PARALLEL_FUNNEL, true);
-        grouped = container.queryGrouped();
-        ActPrintLog("FUNNEL mode = %s, grouped=%s", parallel?"PARALLEL":"ORDERED", grouped?"GROUPED":"UNGROUPED");
-    }
     virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
@@ -481,6 +477,7 @@ public:
     {
         grouped = container.queryGrouped();
         helper = (IHThorCombineArg *) queryHelper();
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     virtual void start() override
@@ -580,6 +577,7 @@ public:
     {
         grouped = container.queryGrouped();
         helper = (IHThorRegroupArg *) queryHelper();
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     virtual void start() override
@@ -686,6 +684,8 @@ public:
         sendReceiving = false;
         masterMpTag = TAG_NULL;
         appendOutputLinked(this);
+        if (container.queryLocalOrGrouped())
+            setRequireInitData(false);
     }
 
 // IThorSlaveActivity overloaded methods
@@ -780,6 +780,7 @@ public:
     CNWaySelectActivity(CGraphElementBase *_container) : CSlaveActivity(_container), CThorSteppable(this)
     {
         helper = (IHThorNWaySelectArg *)queryHelper();
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     virtual void start() override
@@ -903,10 +904,11 @@ class CThorNWayInputSlaveActivity : public CSlaveActivity, implements IThorNWayI
 public:
     IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
 
-    CThorNWayInputSlaveActivity(CGraphElementBase *container) : CSlaveActivity(container)
+    CThorNWayInputSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
         helper = (IHThorNWayInputArg *)queryHelper();
         grouped = helper->queryOutputMeta()->isGrouped(); // JCSMORE should match graph info, i.e. container.queryGrouped()
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     virtual void start() override

thorlcr/activities/group/thgroupslave.cpp

@@ -61,6 +61,8 @@ public:
         numGroups = 0;
         numGroupMax = 0;
         startLastGroup = 0;
+        if (container.queryLocalOrGrouped())
+            setRequireInitData(false);
         appendOutputLinked(this);
     }
     virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)

thorlcr/activities/hashdistrib/thhashdistribslave.cpp

@@ -1962,6 +1962,8 @@ public:
     HashDistributeSlaveBase(CGraphElementBase *_container)
         : CSlaveActivity(_container)
     {
+        IHThorHashDistributeArg *distribargs = (IHThorHashDistributeArg *)queryHelper();
+        ihash = distribargs->queryHash();
         appendOutputLinked(this);
     }
     ~HashDistributeSlaveBase()
@@ -2069,12 +2071,8 @@ public:
 class HashDistributeSlaveActivity : public HashDistributeSlaveBase
 {
 public:
-    HashDistributeSlaveActivity(CGraphElementBase *container) : HashDistributeSlaveBase(container) { }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
+    HashDistributeSlaveActivity(CGraphElementBase *container) : HashDistributeSlaveBase(container)
     {
-        HashDistributeSlaveBase::init(data, slaveData);
-        IHThorHashDistributeArg *distribargs = (IHThorHashDistributeArg *)queryHelper();
-        ihash = distribargs->queryHash();
     }
 };
 
@@ -2083,15 +2081,11 @@ public:
 class HashDistributeMergeSlaveActivity : public HashDistributeSlaveBase
 {
 public:
-    HashDistributeMergeSlaveActivity(CGraphElementBase *container) : HashDistributeSlaveBase(container) { }
-    void init(MemoryBuffer &data, MemoryBuffer &slaveData)
+    HashDistributeMergeSlaveActivity(CGraphElementBase *container) : HashDistributeSlaveBase(container)
     {
         IHThorHashDistributeArg *distribargs = (IHThorHashDistributeArg *)queryHelper();
         mergecmp = distribargs->queryMergeCompare();
-        HashDistributeSlaveBase::init(data, slaveData);
-        ihash = distribargs->queryHash();
     }
-
 };
 
 //===========================================================================
@@ -3900,6 +3894,8 @@ public:
         helper = static_cast <IHThorHashAggregateArg *> (queryHelper());
         mptag = TAG_NULL;
         eos = true;
+        if (container.queryLocalOrGrouped())
+            setRequireInitData(false);
         appendOutputLinked(this);
     }
     virtual void init(MemoryBuffer & data, MemoryBuffer &slaveData)
@@ -3977,7 +3973,7 @@ public:
 #endif
 
 
-class CHashDistributeSlavedActivity : public CSlaveActivity
+class CHashDistributedSlaveActivity : public CSlaveActivity
 {
     typedef CSlaveActivity PARENT;
 
@@ -3985,12 +3981,13 @@ class CHashDistributeSlavedActivity : public CSlaveActivity
     unsigned myNode, nodes;
 
 public:
-    CHashDistributeSlavedActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
+    CHashDistributedSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
         IHThorHashDistributeArg *distribargs = (IHThorHashDistributeArg *)queryHelper();
         ihash = distribargs->queryHash();
         myNode = queryJobChannel().queryMyRank()-1;
         nodes = container.queryJob().querySlaves();
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     virtual void start() override
@@ -4084,6 +4081,6 @@ CActivityBase *createReDistributeSlave(CGraphElementBase *container)
 
 CActivityBase *createHashDistributedSlave(CGraphElementBase *container)
 {
-    return new CHashDistributeSlavedActivity(container);
+    return new CHashDistributedSlaveActivity(container);
 }
 

thorlcr/activities/iterate/thgroupiterateslave.cpp

@@ -35,6 +35,7 @@ public:
     GroupIterateSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
         helper = static_cast <IHThorGroupIterateArg *> (queryHelper());
+        setRequireInitData(false);
         appendOutputLinked(this);   // adding 'me' to outputs array
     }
     virtual void start() override
@@ -117,12 +118,9 @@ public:
     GroupProcessSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
         helper = static_cast <IHThorProcessArg *> (queryHelper());
-        appendOutputLinked(this);   // adding 'me' to outputs array
-    }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
-    {
         rightrowif.setown(createThorRowInterfaces(queryRowManager(), helper->queryRightRecordSize(),queryId(),queryCodeContext()));
         rightAllocator.set(rightrowif->queryRowAllocator());
+        appendOutputLinked(this);   // adding 'me' to outputs array
     }
     virtual void start() override
     {

thorlcr/activities/iterate/thiterateslave.cpp

@@ -36,6 +36,8 @@ public:
     IterateSlaveActivityBase(CGraphElementBase *_container, bool _global) : CSlaveActivity(_container)
     {
         global = _global;
+        if (!global)
+            setRequireInitData(false);
         appendOutputLinked(this);   // adding 'me' to outputs array
     }
     virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
@@ -189,11 +191,7 @@ public:
         : IterateSlaveActivityBase(_container,_global)
     {
         helper = static_cast <IHThorProcessArg *> (queryHelper());
-    }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
-    {
         rightRowAllocator.setown(getRowAllocator(helper->queryRightRecordSize()));
-        IterateSlaveActivityBase::init(data,slaveData);
     }
     CATCH_NEXTROW()
     {
@@ -275,6 +273,7 @@ public:
     CChildIteratorSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
         helper = static_cast <IHThorChildIteratorArg *> (queryHelper());
+        setRequireInitData(false);
         appendOutputLinked(this);   // adding 'me' to outputs array
     }
     virtual void start() override
@@ -325,11 +324,9 @@ public:
         : CSlaveActivity(_container)
     {
         helper = static_cast <IHThorLinkedRawIteratorArg *> (queryHelper());
-        appendOutputLinked(this);   // adding 'me' to outputs array
-    }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
-    {
         grouped = helper->queryOutputMeta()->isGrouped();
+        setRequireInitData(false);
+        appendOutputLinked(this);   // adding 'me' to outputs array
     }
     virtual void start() override
     {
@@ -384,6 +381,7 @@ public:
         : CSlaveActivity(_container)
     {
         helper = static_cast <IHThorStreamedIteratorArg *> (queryHelper());
+        setRequireInitData(false);
         appendOutputLinked(this);   // adding 'me' to outputs array
     }
     virtual void start() override

thorlcr/activities/join/thjoinslave.cpp

@@ -165,6 +165,8 @@ public:
         leftKeySerializer = helper->querySerializeLeft();
         rightKeySerializer = helper->querySerializeRight();
         rightpartition = (container.getKind()==TAKjoin)&&((helper->getJoinFlags()&JFpartitionright)!=0);
+        if (islocal)
+            setRequireInitData(false);
         appendOutputLinked(this);
     }
 
@@ -636,9 +638,10 @@ protected:
 public:
     IMPLEMENT_IINTERFACE_USING(PARENT);
 
-    CMergeJoinSlaveBaseActivity(CGraphElementBase *container, CMergeJoinProcessor &_processor) : CThorNarySlaveActivity(container), CThorSteppable(this), processor(_processor)
+    CMergeJoinSlaveBaseActivity(CGraphElementBase *_container, CMergeJoinProcessor &_processor) : CThorNarySlaveActivity(_container), CThorSteppable(this), processor(_processor)
     {
         helper = (IHThorNWayMergeJoinArg *)queryHelper();
+        setRequireInitData(false);
         inputAllocator.setown(getRowAllocator(helper->queryInputMeta()));
         outputAllocator.setown(getRowAllocator(helper->queryOutputMeta()));
         appendOutputLinked(this);

thorlcr/activities/keyedjoin/thkeyedjoin.cpp

@@ -58,7 +58,7 @@ public:
         localKey = false;
         numTags = 0;
         tags[0] = tags[1] = tags[2] = tags[3] = TAG_NULL;
-        reInit = 0 != (helper->getFetchFlags() & (FFvarfilename|FFdynamicfilename));
+        reInit = 0 != (helper->getFetchFlags() & (FFvarfilename|FFdynamicfilename)) || (helper->getJoinFlags() & JFvarindexfilename);
         remoteDataFiles = false;
     }
     ~CKeyedJoinMaster()

thorlcr/activities/keyedjoin/thkeyedjoinslave.cpp

@@ -1598,7 +1598,7 @@ public:
         lastTick = 0;
 #endif
         helper = (IHThorKeyedJoinArg *)queryHelper();
-        reInit = 0 != (helper->getFetchFlags() & (FFvarfilename|FFdynamicfilename));
+        reInit = 0 != (helper->getFetchFlags() & (FFvarfilename|FFdynamicfilename)) || (helper->getJoinFlags() & JFvarindexfilename);
         appendOutputLinked(this);
     }
     ~CKeyedJoinSlave()

thorlcr/activities/limit/thlimitslave.cpp

@@ -49,6 +49,8 @@ public:
         resultSent = true; // unless started suppress result send
         eos = stopped = anyThisGroup = eogNext = false;
         rowLimit = RCMAX;
+        if (container.queryLocal())
+            setRequireInitData(false);
         appendOutputLinked(this);
     }
     virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
@@ -231,6 +233,8 @@ public:
         limitChecked = eof = false;
         rowTransform = _rowTransform;
         helperex = NULL;
+        if (rowTransform)
+            helperex = static_cast<IHThorLimitTransformExtra *>(queryHelper()->selectInterface(TAIlimittransformextra_1));
     }
     void abort()
     {
@@ -238,12 +242,6 @@ public:
             cancelReceiveMsg(0, mpTag);
         CLimitSlaveActivityBase::abort();
     }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
-    {
-        CLimitSlaveActivityBase::init(data,slaveData);
-        if (rowTransform)
-            helperex = static_cast<IHThorLimitTransformExtra *>(queryHelper()->selectInterface(TAIlimittransformextra_1));
-    }
     virtual void start() override
     {
         CLimitSlaveActivityBase::start();

thorlcr/activities/lookupjoin/thlookupjoinslave.cpp

@@ -1352,6 +1352,8 @@ public:
             rightThorAllocator = queryJobChannel().queryThorAllocator();
         rightRowManager = rightThorAllocator->queryRowManager();
         broadcastLock = NULL;
+        if (!isGlobal())
+            setRequireInitData(false);
         appendOutputLinked(this);
     }
     ~CInMemJoinBase()

thorlcr/activities/loop/thloopslave.cpp

@@ -76,10 +76,12 @@ protected:
         sendLoopingCount(0, 0);
     }
 public:
-    CLoopSlaveActivityBase(CGraphElementBase *container) : CSlaveActivity(container)
+    CLoopSlaveActivityBase(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
         mpTag = TAG_NULL;
         maxEmptyLoopIterations = getOptUInt(THOROPT_LOOP_MAX_EMPTY, 1000);
+        if (container.queryLocalOrGrouped())
+            setRequireInitData(false);
         appendOutputLinked(this);
     }
     void init(MemoryBuffer &data, MemoryBuffer &slaveData)
@@ -528,6 +530,7 @@ public:
         helper = (IHThorLocalResultReadArg *)queryHelper();
         curRow = 0;
         replyTag = queryMPServer().createReplyTag();
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     void init(MemoryBuffer &data, MemoryBuffer &slaveData)
@@ -704,8 +707,9 @@ public:
 class CLocalResultWriteActivity : public CLocalResultWriteActivityBase
 {
 public:
-    CLocalResultWriteActivity(CGraphElementBase *container) : CLocalResultWriteActivityBase(container)
+    CLocalResultWriteActivity(CGraphElementBase *_container) : CLocalResultWriteActivityBase(_container)
     {
+        setRequireInitData(false);
     }
     virtual IThorResult *createResult()
     {
@@ -733,6 +737,7 @@ public:
     CDictionaryResultWriteActivity(CGraphElementBase *_container) : ProcessSlaveActivity(_container)
     {
         helper = (IHThorDictionaryResultWriteArg *)queryHelper();
+        setRequireInitData(false);
     }
     virtual void process()
     {
@@ -800,6 +805,7 @@ protected:
 public:
     CConditionalActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     virtual void start() override
@@ -905,6 +911,7 @@ public:
     CIfActionActivity(CGraphElementBase *_container) : ProcessSlaveActivity(_container)
     {
         helper = (IHThorIfArg *)baseHelper.get();
+        setRequireInitData(false);
     }
     // IThorSlaveProcess overloaded methods
     virtual void process() override
@@ -959,11 +966,9 @@ public:
     CChildNormalizeSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
         helper = (IHThorChildNormalizeArg *)queryHelper();
-        appendOutputLinked(this);
-    }
-    void init(MemoryBuffer &data, MemoryBuffer &slaveData)
-    {
         allocator.set(queryRowAllocator());
+        setRequireInitData(false);
+        appendOutputLinked(this);
     }
     virtual void start()
     {
@@ -1031,6 +1036,7 @@ public:
     CChildAggregateSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
         helper = (IHThorChildAggregateArg *)queryHelper();
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     virtual void start()
@@ -1079,11 +1085,9 @@ public:
     CChildGroupAggregateActivitySlave(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
         helper = (IHThorChildGroupAggregateArg *)queryHelper();
-        appendOutputLinked(this);
-    }
-    void init(MemoryBuffer &data, MemoryBuffer &slaveData)
-    {
+        setRequireInitData(false);
         allocator.set(queryRowAllocator());
+        appendOutputLinked(this);
     }
     virtual void start() override
     {
@@ -1148,12 +1152,10 @@ public:
     CChildThroughNormalizeSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container), nextOutput(NULL)
     {
         helper = (IHThorChildThroughNormalizeArg *)queryHelper();
-        appendOutputLinked(this);
-    }
-    void init(MemoryBuffer &data, MemoryBuffer &slaveData)
-    {
         allocator.set(queryRowAllocator());
         nextOutput.setAllocator(allocator);
+        setRequireInitData(false);
+        appendOutputLinked(this);
     }
     virtual void start()
     {
@@ -1224,9 +1226,10 @@ class CGraphLoopResultReadSlaveActivity : public CSlaveActivity
     Owned<IRowStream> resultStream;
 
 public:
-    CGraphLoopResultReadSlaveActivity(CGraphElementBase *container) : CSlaveActivity(container)
+    CGraphLoopResultReadSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
         helper = (IHThorGraphLoopResultReadArg *)queryHelper();
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     virtual void kill()
@@ -1361,8 +1364,9 @@ activityslaves_decl CActivityBase *createGraphLoopResultReadSlave(CGraphElementB
 class CGraphLoopResultWriteSlaveActivity : public CLocalResultWriteActivityBase
 {
 public:
-    CGraphLoopResultWriteSlaveActivity(CGraphElementBase *container) : CLocalResultWriteActivityBase(container)
+    CGraphLoopResultWriteSlaveActivity(CGraphElementBase *_container) : CLocalResultWriteActivityBase(_container)
     {
+        setRequireInitData(false);
     }
     virtual IThorResult *createResult()
     {

thorlcr/activities/merge/thmergeslave.cpp

@@ -421,6 +421,7 @@ public:
     LocalMergeSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
         helper = (IHThorMergeArg *)queryHelper();
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     void abort()

thorlcr/activities/msort/thgroupsortslave.cpp

@@ -46,14 +46,12 @@ public:
     CLocalSortSlaveActivity(CGraphElementBase *_container)
         : CSlaveActivity(_container), spillStats(spillStatistics)
     {
-        appendOutputLinked(this);
-    }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
-    {
         helper = (IHThorSortArg *)queryHelper();
         iCompare = helper->queryCompare();
         IHThorAlgorithm * algo = helper?(static_cast<IHThorAlgorithm *>(helper->selectInterface(TAIalgorithm_1))):NULL;
         unstable = (algo&&(algo->getAlgorithmFlags()&TAFunstable));
+        setRequireInitData(false);
+        appendOutputLinked(this);
     }
     virtual void start()
     {
@@ -138,13 +136,9 @@ public:
     {
         helper = (IHThorSortedArg *)queryHelper();
         icompare = helper->queryCompare();
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
-    {
-        helper = (IHThorSortedArg *)queryHelper();
-        icompare = helper->queryCompare();
-    }
     virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);

thorlcr/activities/normalize/thnormalizeslave.cpp

@@ -43,11 +43,9 @@ public:
         : CSlaveActivity(_container)
     {
         helper = static_cast <IHThorNormalizeArg *> (queryHelper());
-        appendOutputLinked(this);
-    }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
-    {
         allocator.set(queryRowAllocator());
+        setRequireInitData(false);
+        appendOutputLinked(this);
     }
     virtual void start() override
     { 
@@ -117,14 +115,12 @@ public:
         : CSlaveActivity(_container)
     { 
         helper = static_cast <IHThorNormalizeChildArg *> (queryHelper());
-        appendOutputLinked(this);
-    }
-    virtual bool isGrouped() const override { return queryInput(0)->isGrouped(); }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
-    {
         cursor = helper->queryIterator();
         allocator.set(queryRowAllocator());
+        setRequireInitData(false);
+        appendOutputLinked(this);
     }
+    virtual bool isGrouped() const override { return queryInput(0)->isGrouped(); }
     virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
@@ -209,6 +205,7 @@ public:
         : CSlaveActivity(_container)
     { 
         helper = static_cast <IHThorNormalizeLinkedChildArg *> (queryHelper());
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     virtual bool isGrouped() const override { return queryInput(0)->isGrouped(); }

thorlcr/activities/nsplitter/thnsplitterslave.cpp

@@ -144,6 +144,13 @@ public:
     {
         activeOutputs = container.getOutputs();
         ActPrintLog("Number of connected outputs: %u", activeOutputs);
+        setRequireInitData(false);
+        IHThorSplitArg *helper = (IHThorSplitArg *)queryHelper();
+        int dV = getOptInt(THOROPT_SPLITTER_SPILL, -1);
+        if (-1 == dV)
+            spill = !helper->isBalanced();
+        else
+            spill = dV>0;
         ForEachItemIn(o, container.outputs)
             appendOutput(new CSplitterOutput(*this, o));
     }
@@ -164,15 +171,6 @@ public:
                 output->reset();
         }
     }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
-    {
-        IHThorSplitArg *helper = (IHThorSplitArg *)queryHelper();
-        int dV = getOptInt(THOROPT_SPLITTER_SPILL, -1);
-        if (-1 == dV)
-            spill = !helper->isBalanced();
-        else
-            spill = dV>0;
-    }
     bool prepareInput()
     {
         // NB: called from writeahead by outputs

thorlcr/activities/null/thnullslave.cpp

@@ -21,13 +21,11 @@
 class CNullSinkSlaveActivity : public ProcessSlaveActivity
 {
 public:
-    CNullSinkSlaveActivity(CGraphElementBase *container) : ProcessSlaveActivity(container)
+    CNullSinkSlaveActivity(CGraphElementBase *_container) : ProcessSlaveActivity(_container)
     {
+        setRequireInitData(false);
     }
 // IThorSlaveActivity
-    virtual void init(MemoryBuffer & data, MemoryBuffer &slaveData)
-    {       
-    }
     virtual void process() override
     {
         start();
@@ -84,13 +82,9 @@ class CThroughSlaveActivity : public CSlaveActivity
 public:
     CThroughSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
-// IThorSlaveActivity
-    virtual void init(MemoryBuffer & data, MemoryBuffer &slaveData)
-    {       
-    }
-
 // IThorDataLink
     virtual void start() override
     {

thorlcr/activities/nullaction/thnullactionslave.cpp

@@ -31,6 +31,7 @@ class CNullActionSlaveActivity : public CSlaveActivity
 public:
     CNullActionSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     ~CNullActionSlaveActivity()

thorlcr/activities/parse/thparseslave.cpp

@@ -49,6 +49,11 @@ public:
         anyThisGroup = false;
         curSearchTextLen = 0;
         curSearchText = NULL;
+        algorithm.setown(createThorParser(queryCodeContext(), *helper));
+        parser.setown(algorithm->createParser(queryCodeContext(), (unsigned)container.queryId(), helper->queryHelper(), helper));
+        rowIter = parser->queryResultIter();
+        allocator.set(queryRowAllocator());
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     ~CParseSlaveActivity()
@@ -56,13 +61,6 @@ public:
         if (helper->searchTextNeedsFree())
             rtlFree(curSearchText);
     }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
-    {
-        algorithm.setown(createThorParser(queryCodeContext(), *helper));
-        parser.setown(algorithm->createParser(queryCodeContext(), (unsigned)container.queryId(), helper->queryHelper(), helper));
-        rowIter = parser->queryResultIter();
-        allocator.set(queryRowAllocator());
-    } 
     virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);

thorlcr/activities/piperead/thprslave.cpp

@@ -190,6 +190,12 @@ public:
         : CPipeSlaveBase(_container)
     {
         helper = static_cast <IHThorPipeReadArg *> (queryHelper());
+        flags = helper->getPipeFlags();
+        needTransform = false;
+
+        if (needTransform)
+            inrowif.setown(createThorRowInterfaces(queryRowManager(), helper->queryDiskRecordSize(), queryId(), queryCodeContext()));
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     CATCH_NEXTROW()
@@ -229,14 +235,6 @@ public:
         eof = true;
         return NULL;
     }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
-    {
-        flags = helper->getPipeFlags();
-        needTransform = false;
-
-        if (needTransform)
-            inrowif.setown(createThorRowInterfaces(queryRowManager(), helper->queryDiskRecordSize(), queryId(), queryCodeContext()));
-    }
     virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
@@ -343,18 +341,16 @@ public:
         helper = static_cast <IHThorPipeThroughArg *> (queryHelper());
         pipeWriter = NULL;
         grouped = false;
+        flags = helper->getPipeFlags();
+        recreate = helper->recreateEachRow();
+        grouped = 0 != (flags & TPFgroupeachrow);
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     ~CPipeThroughSlaveActivity()
     {
         ::Release(pipeWriter);
     }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
-    {
-        flags = helper->getPipeFlags();
-        recreate = helper->recreateEachRow();
-        grouped = 0 != (flags & TPFgroupeachrow);
-    }
     virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);

thorlcr/activities/pipewrite/thpwslave.cpp

@@ -37,18 +37,13 @@ private:
     bool pipeOpen;
 
 public:
-    CPipeWriteSlaveActivity(CGraphElementBase *container) : ProcessSlaveActivity(container)
+    CPipeWriteSlaveActivity(CGraphElementBase *_container) : ProcessSlaveActivity(_container)
     {
         helper = static_cast <IHThorPipeWriteArg *> (queryHelper());
         pipe.setown(createPipeProcess(globals->queryProp("@allowedPipePrograms")));
         pipeOpen = false;
-    }
-    ~CPipeWriteSlaveActivity()
-    {
-    }
-    void init(MemoryBuffer &data, MemoryBuffer &slaveData)
-    {
         recreate = helper->recreateEachRow();
+        setRequireInitData(false);
     }
     void open()
     {

thorlcr/activities/project/thprojectslave.cpp

@@ -84,6 +84,7 @@ public:
     explicit CProjectSlaveActivity(CGraphElementBase *_container) : CThorStrandedActivity(_container)
     {
         helper = static_cast <IHThorProjectArg *> (queryHelper());
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
 
@@ -248,11 +249,9 @@ public:
     {
         helper = (IHThorPrefetchProjectArg *) queryHelper();
         parallel = 0 != (helper->getFlags() & PPFparallel);
-        appendOutputLinked(this);
-    }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
-    {
+        setRequireInitData(false);
         allocator.set(queryRowAllocator());
+        appendOutputLinked(this);
     }
     virtual void start() override
     {

thorlcr/activities/pull/thpullslave.cpp

@@ -28,6 +28,7 @@ class PullSlaveActivity : public CSlaveActivity
 public:
     PullSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
 

thorlcr/activities/rollup/throllupslave.cpp

@@ -170,12 +170,14 @@ protected:
 
     unsigned numKept; // not used by rollup
 public:
-    CDedupRollupBaseActivity(CGraphElementBase *container, bool _rollup, bool _global, bool _groupOp) 
-        : CSlaveActivity(container)
+    CDedupRollupBaseActivity(CGraphElementBase *_container, bool _rollup, bool _global, bool _groupOp)
+        : CSlaveActivity(_container)
     {
         rollup = _rollup;
         global = _global;
         groupOp = _groupOp;
+        if (!global)
+            setRequireInitData(false);
     }
     virtual void stopInput()
     {
@@ -295,6 +297,7 @@ public:
         : CDedupRollupBaseActivity(_container, false, global, groupOp)
     {
         ddhelper = static_cast <IHThorDedupArg *>(queryHelper());
+        setRequireInitData(false);
         appendOutputLinked(this);   // adding 'me' to outputs array
     }
     virtual void start() override
@@ -402,10 +405,6 @@ public:
     {
         lastEog = false;
     }
-    void init(MemoryBuffer &data, MemoryBuffer &slaveData)
-    {
-        CDedupBaseSlaveActivity::init(data, slaveData);
-    }
     virtual void start()
     {
         ActivityTimer s(totalCycles, timeActivities);
@@ -555,11 +554,9 @@ public:
     {
         helper = (IHThorRollupGroupArg *)queryHelper();
         eoi = false;
-        appendOutputLinked(this);   // adding 'me' to outputs array
-    }
-    void init(MemoryBuffer &data, MemoryBuffer &slaveData)
-    {
         groupLoader.setown(createThorRowLoader(*this, NULL, stableSort_none, rc_allMem));
+        setRequireInitData(false);
+        appendOutputLinked(this);   // adding 'me' to outputs array
     }
     virtual void start()
     {

thorlcr/activities/sample/thsampleslave.cpp

@@ -31,6 +31,7 @@ public:
     SampleSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
         helper = static_cast <IHThorSampleArg *> (queryHelper());
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     virtual void start() override

thorlcr/activities/selectnth/thselectnthslave.cpp

@@ -69,6 +69,8 @@ public:
         helper = static_cast <IHThorSelectNArg *> (queryHelper());
         isLocal = _isLocal;
         createDefaultIfFail = isLocal || lastNode();
+        if (container.queryLocalOrGrouped())
+            setRequireInitData(false);
         appendOutputLinked(this);
     }
 

thorlcr/activities/selfjoin/thselfjoinslave.cpp

@@ -111,6 +111,8 @@ public:
         keyserializer = NULL;
         inputStopped = false;
         mpTagRPC = TAG_NULL;
+        if (isLocal)
+            setRequireInitData(false);
         appendOutputLinked(this);
     }
 
@@ -123,7 +125,7 @@ public:
 // IThorSlaveActivity
     virtual void init(MemoryBuffer & data, MemoryBuffer &slaveData) override
     {       
-        if(!isLocal)
+        if (!isLocal)
         {
             mpTagRPC = container.queryJobChannel().deserializeMPTag(data);
             mptag_t barrierTag = container.queryJobChannel().deserializeMPTag(data);

thorlcr/activities/soapcall/thsoapcallslave.cpp

@@ -45,14 +45,11 @@ public:
 
     CWscRowCallSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
+        buildAuthToken(queryJob().queryUserDescriptor(), authToken);
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
 
-    // IThorSlaveActivity overloaded methods
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
-    {
-        buildAuthToken(queryJob().queryUserDescriptor(), authToken);
-    }
     // IThorDataLink methods
     virtual void start()
     {
@@ -140,6 +137,7 @@ public:
 
     SoapDatasetCallSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
 
@@ -223,7 +221,10 @@ class SoapRowActionSlaveActivity : public ProcessSlaveActivity, implements IWSCR
 public:
     IMPLEMENT_IINTERFACE_USING(PARENT);
 
-    SoapRowActionSlaveActivity(CGraphElementBase *container) : ProcessSlaveActivity(container) { }
+    SoapRowActionSlaveActivity(CGraphElementBase *_container) : ProcessSlaveActivity(_container)
+    {
+        setRequireInitData(false);
+    }
 
     // IThorSlaveActivity overloaded methods
     virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
@@ -276,7 +277,10 @@ class SoapDatasetActionSlaveActivity : public ProcessSlaveActivity, implements I
 public:
     IMPLEMENT_IINTERFACE_USING(PARENT);
 
-    SoapDatasetActionSlaveActivity(CGraphElementBase *container) : ProcessSlaveActivity(container) { }
+    SoapDatasetActionSlaveActivity(CGraphElementBase *_container) : ProcessSlaveActivity(_container)
+    {
+        setRequireInitData(false);
+    }
 
     // IThorSlaveActivity overloaded methods
     virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)

thorlcr/activities/temptable/thtmptableslave.cpp

@@ -46,6 +46,7 @@ public:
         startRow = 0;
         currentRow = 0;
         maxRow = 0;
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
     virtual bool isGrouped() const override { return false; }

thorlcr/activities/thdiskbaseslave.ipp

@@ -133,7 +133,7 @@ protected:
 
 public:
     CDiskWriteSlaveActivityBase(CGraphElementBase *container);
-    void init(MemoryBuffer &data, MemoryBuffer &slaveData);
+    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData);
     virtual void abort();
     virtual void serializeStats(MemoryBuffer &mb);
 

thorlcr/activities/topn/thtopnslave.cpp

@@ -87,6 +87,8 @@ public:
         assertex(!(global && grouped));
         helper = (IHThorTopNArg *) queryHelper();
         eog = eos = false;
+        if (container.queryLocalOrGrouped())
+            setRequireInitData(false);
         appendOutputLinked(this);
     }
     ~TopNSlaveActivity()

thorlcr/activities/trace/thtraceslave.cpp

@@ -39,11 +39,8 @@ public:
           keepLimit(0), skip(0), sample(0), traceEnabled(false)
     {
         helper = (IHThorTraceArg *) queryHelper();
-        appendOutputLinked(this);
-    }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
-    {
         traceEnabled = getOptBool(THOROPT_TRACE_ENABLED, false);
+        appendOutputLinked(this);
     }
     virtual void start() override
     {

thorlcr/activities/when/thwhenslave.cpp

@@ -34,6 +34,8 @@ public:
     CDependencyExecutorSlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
         global = !queryContainer().queryOwner().queryOwner() || queryContainer().queryOwner().isGlobal();
+        if (!global)
+            setRequireInitData(false);
     }
     virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
     {

thorlcr/activities/wuidread/thwuidreadslave.cpp

@@ -45,12 +45,10 @@ public:
         replyTag = queryMPServer().createReplyTag();
         replyStream.setown(createMemoryBufferSerialStream(masterReplyMsg));
         rowSource.setStream(replyStream);
+        grouped = helper->queryOutputMeta()->isGrouped();
+        setRequireInitData(false);
         appendOutputLinked(this);
     }
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
-    {
-        grouped = helper->queryOutputMeta()->isGrouped();
-    } 
     virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);

thorlcr/graph/thgraph.cpp

@@ -468,7 +468,6 @@ void CGraphElementBase::reset()
 {
     alreadyUpdated = false;
     onStartCalled = false;
-//  prepared = false;
     if (activity)
         activity->reset();
 }
@@ -579,7 +578,7 @@ void CGraphElementBase::deserializeStartContext(MemoryBuffer &mb)
 {
     size32_t startCtxLen;
     mb.read(startCtxLen);
-    startCtxMb.append(startCtxLen, mb.readDirect(startCtxLen));
+    startCtxMb.clear().append(startCtxLen, mb.readDirect(startCtxLen));
     haveStartCtx = true;
     onStartCalled = false; // allow to be called again
 }
@@ -600,6 +599,13 @@ void CGraphElementBase::onCreate()
         }
         else
             baseHelper->onCreate(queryCodeContext(), NULL, haveCreateCtx?&createCtxMb:NULL);
+        if (isLoopActivity(*this))
+        {
+            unsigned loopId = queryXGMML().getPropInt("att[@name=\"_loopid\"]/@value");
+            Owned<CGraphBase> childGraph = owner->getChildGraph(loopId);
+            Owned<IThorBoundLoopGraph> boundLoopGraph = createBoundLoopGraph(childGraph, baseHelper->queryOutputMeta(), queryId());
+            setBoundGraph(boundLoopGraph);
+        }
     }
 }
 
@@ -785,8 +791,9 @@ bool CGraphElementBase::prepareContext(size32_t parentExtractSz, const byte *par
         }
         if (create)
         {
-            if (activity) // no need to recreate
+            if (prepared) // no need to recreate
                 return true;
+            prepared = true;
             ForEachItemIn(i2, inputs)
             {
                 CIOConnection *inputIO = inputs.item(i2);
@@ -794,6 +801,7 @@ bool CGraphElementBase::prepareContext(size32_t parentExtractSz, const byte *par
             }
             if (isSink())
                 owner->addActiveSink(*this);
+            assertex(!activity);
             activity.setown(factory());
         }
         return true;
@@ -822,15 +830,9 @@ void CGraphElementBase::preStart(size32_t parentExtractSz, const byte *parentExt
 
 void CGraphElementBase::initActivity()
 {
-    if (!activity)
-        activity.setown(factory());
-    if (isLoopActivity(*this))
-    {
-        unsigned loopId = queryXGMML().getPropInt("att[@name=\"_loopid\"]/@value");
-        Owned<CGraphBase> childGraph = owner->getChildGraph(loopId);
-        Owned<IThorBoundLoopGraph> boundLoopGraph = createBoundLoopGraph(childGraph, baseHelper->queryOutputMeta(), queryId());
-        setBoundGraph(boundLoopGraph);
-    }
+    if (activity)
+        return;
+    activity.setown(factory());
 }
 
 ICodeContext *CGraphElementBase::queryCodeContext()
@@ -1056,7 +1058,7 @@ CGraphBase::CGraphBase(CJobChannel &_jobChannel) : jobChannel(_jobChannel), job(
     graphId = 0;
     complete = false;
     parentActivityId = 0;
-    connected = started = graphDone = aborted = prepared = false;
+    connected = started = graphDone = aborted = false;
     startBarrier = waitBarrier = doneBarrier = NULL;
     mpTag = waitBarrierTag = startBarrierTag = doneBarrierTag = TAG_NULL;
     executeReplyTag = TAG_NULL;
@@ -1243,7 +1245,6 @@ void CGraphBase::onCreate()
     {
         CGraphElementBase &element = iter->query();
         element.onCreate();
-        element.initActivity();
     }
 }
 
@@ -1283,6 +1284,7 @@ void CGraphBase::doExecute(size32_t parentExtractSz, const byte *parentExtract,
         {
             CGraphElementBase &element = iter->query();
             element.onStart(parentExtractSz, parentExtract);
+            element.initActivity();
         }
         if (!preStart(parentExtractSz, parentExtract)) return;
         start();
@@ -1349,7 +1351,6 @@ bool CGraphBase::prepare(size32_t parentExtractSz, const byte *parentExtract, bo
         if (sink.prepareContext(parentExtractSz, parentExtract, checkDependencies, shortCircuit, async, false))
             needToExecute = true;
     }
-//  prepared = true;
     onCreate();
     return needToExecute;
 }
@@ -2876,7 +2877,7 @@ IThorResource &queryThor()
 CActivityBase::CActivityBase(CGraphElementBase *_container) : container(*_container), timeActivities(_container->queryJob().queryTimeActivities())
 {
     mpTag = TAG_NULL;
-    abortSoon = receiving = cancelledReceive = reInit = false;
+    abortSoon = receiving = cancelledReceive = initialized = reInit = false;
     baseHelper.set(container.queryHelper());
     parentExtractSz = 0;
     parentExtract = NULL;

thorlcr/graph/thgraph.hpp

@@ -548,7 +548,7 @@ protected:
     Owned<IPropertyTree> node;
     IBarrier *startBarrier, *waitBarrier, *doneBarrier;
     mptag_t mpTag, startBarrierTag, waitBarrierTag, doneBarrierTag;
-    bool connected, started, aborted, graphDone, prepared, sequential;
+    bool connected, started, aborted, graphDone, sequential;
     CJobBase &job;
     CJobChannel &jobChannel;
     graph_id graphId;
@@ -587,7 +587,6 @@ public:
     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 isStarted() const { return started; }
     bool isLocalOnly() const; // this graph and all upstream dependencies
@@ -1008,7 +1007,7 @@ protected:
     bool timeActivities; // purely for access efficiency
     size32_t parentExtractSz;
     const byte *parentExtract;
-    bool receiving, cancelledReceive, reInit;
+    bool receiving, cancelledReceive, initialized, reInit;
     Owned<IThorGraphResults> ownedResults; // NB: probably only to be used by loop results
 
 public:
@@ -1026,6 +1025,8 @@ public:
     inline bool queryAbortSoon() const { return abortSoon; }
     inline IHThorArg *queryHelper() const { return baseHelper; }
     inline bool needReInit() const { return reInit; }
+    inline bool queryInitialized() const { return initialized; }
+    inline void setInitialized(bool tf) { initialized = tf; }
     inline bool queryTimeActivities() const { return timeActivities; }
     void onStart(size32_t _parentExtractSz, const byte *_parentExtract) { parentExtractSz = _parentExtractSz; parentExtract = _parentExtract; }
     bool receiveMsg(ICommunicator &comm, CMessageBuffer &mb, const rank_t rank, const mptag_t mpTag, rank_t *sender=NULL, unsigned timeout=MP_WAIT_FOREVER);

thorlcr/graph/thgraphmaster.cpp

@@ -234,7 +234,7 @@ void CSlaveMessageHandler::main()
                         try
                         {
                             element->reset();
-                            element->doCreateActivity(parentExtractSz, parentExtract);
+                            element->doCreateActivity(parentExtractSz, parentExtract, &msg);
                         }
                         catch (IException *e)
                         {
@@ -617,37 +617,12 @@ void CMasterGraphElement::initActivity()
     owner->setInitialized();
 }
 
-void CMasterGraphElement::doCreateActivity(size32_t parentExtractSz, const byte *parentExtract)
+void CMasterGraphElement::doCreateActivity(size32_t parentExtractSz, const byte *parentExtract, MemoryBuffer *startCtx)
 {
-    bool ok=false;
-    switch (getKind())
-    {
-        case TAKspill:
-        case TAKdiskwrite:
-        case TAKfetch:
-        case TAKkeyedjoin:
-        case TAKkeyeddenormalize:
-        case TAKkeyeddenormalizegroup:
-        case TAKworkunitwrite:
-        case TAKworkunitread:
-        case TAKdictionaryworkunitwrite:
-        case TAKdictionaryresultwrite:
-            ok = true;
-            break;
-        default:
-        {
-            if (isDiskInput(getKind()))
-                ok = true;
-            else if (!queryLocalOrGrouped())
-                ok = true;
-            break;
-        }
-    }
-    if (!ok)
-        return;
     onCreate();
-    if (isDiskInput(getKind()))
-       onStart(parentExtractSz, parentExtract);
+    if (startCtx)
+        deserializeStartContext(*startCtx);
+    onStart(parentExtractSz, parentExtract);
     initActivity();
 }
 
@@ -2429,11 +2404,6 @@ void CMasterGraph::executeSubGraph(size32_t parentExtractSz, const byte *parentE
             }
         }
     }
-    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);
@@ -2496,6 +2466,11 @@ void CMasterGraph::sendGraph()
 bool CMasterGraph::preStart(size32_t parentExtractSz, const byte *parentExtract)
 {
     GraphPrintLog("Processing graph");
+    if (syncInitData())
+    {
+        sendActivityInitData(); // has to be done at least once
+        // NB: At this point, on the slaves, the graphs will start
+    }
     CGraphBase::preStart(parentExtractSz, parentExtract);
     if (isGlobal())
     {

thorlcr/graph/thgraphmaster.ipp

@@ -327,7 +327,7 @@ class graphmaster_decl CMasterGraphElement : public CGraphElementBase
     bool initialized = false;
 public:
     CMasterGraphElement(CGraphBase &owner, IPropertyTree &xgmml);
-    void doCreateActivity(size32_t parentExtractSz=0, const byte *parentExtract=NULL);
+    void doCreateActivity(size32_t parentExtractSz=0, const byte *parentExtract=NULL, MemoryBuffer *startCtx=nullptr);
     virtual bool checkUpdate();
 
     virtual void initActivity() override;

thorlcr/graph/thgraphslave.cpp

@@ -769,10 +769,6 @@ void CSlaveGraph::init(MemoryBuffer &mb)
 
 void CSlaveGraph::initWithActData(MemoryBuffer &in, MemoryBuffer &out)
 {
-    CriticalBlock b(progressCrit);
-    initialized = true;
-    if (0 == in.length())
-        return;
     activity_id id;
     loop
     {
@@ -787,11 +783,9 @@ void CSlaveGraph::initWithActData(MemoryBuffer &in, MemoryBuffer &out)
         in.read(sz);
         unsigned aread = in.getPos();
         CSlaveActivity *activity = (CSlaveActivity *)element->queryActivity();
-        if (activity)
-        {
-            element->sentActInitData->set(0);
-            activity->init(in, out);
-        }
+        assertex(activity);
+        element->sentActInitData->set(0);
+        activity->init(in, out);
         aread = in.getPos()-aread;
         if (aread<sz)
         {
@@ -801,6 +795,7 @@ void CSlaveGraph::initWithActData(MemoryBuffer &in, MemoryBuffer &out)
         }
         else if (aread>sz)
             throw MakeActivityException(element, TE_SeriailzationError, "Serialization error - activity read beyond serialized data (%d byte(s))", aread-sz);
+        activity->setInitialized(true);
         size32_t dl = out.length() - l;
         if (dl)
             out.writeDirect(l-sizeof(size32_t), sizeof(size32_t), &dl);
@@ -814,32 +809,43 @@ bool CSlaveGraph::recvActivityInitData(size32_t parentExtractSz, const byte *par
 {
     bool ret = true;
     unsigned needActInit = 0;
+    unsigned uninitialized = 0;
     Owned<IThorActivityIterator> iter = getConnectedIterator();
     ForEach(*iter)
     {
         CGraphElementBase &element = (CGraphElementBase &)iter->query();
         CActivityBase *activity = element.queryActivity();
-        if (activity && activity->needReInit())
-            element.sentActInitData->set(0, false); // force act init to be resent
-        if (!element.sentActInitData->test(0))
-            ++needActInit;
-    }
-    if (needActInit)
-    {
-        mptag_t replyTag = TAG_NULL;
-        size32_t len;
-        CMessageBuffer actInitRtnData;
-        actInitRtnData.append(false);
-        CMessageBuffer msg;
-
-        if (syncInitData())
+        if (activity)
         {
-            if (!graphCancelHandler.recv(queryJobChannel().queryJobComm(), msg, 0, mpTag, NULL, LONGTIMEOUT))
-                throw MakeStringException(0, "Error receiving actinit data for graph: %" GIDPF "d", graphId);
-            replyTag = msg.getReplyTag();
-            msg.read(len);
+            if (activity->needReInit())
+            {
+                element.sentActInitData->set(0, false); // force act init to be resent
+                activity->setInitialized(false);
+            }
+            if (!element.sentActInitData->test(0))
+                ++needActInit;
+            if (!activity->queryInitialized())
+                ++uninitialized;
         }
-        else
+    }
+    if (0 == uninitialized)
+        return true;
+    mptag_t replyTag = TAG_NULL;
+    size32_t len = 0;
+    CMessageBuffer actInitRtnData;
+    actInitRtnData.append(false);
+    CMessageBuffer msg;
+
+    if (syncInitData())
+    {
+        if (!graphCancelHandler.recv(queryJobChannel().queryJobComm(), msg, 0, mpTag, NULL, LONGTIMEOUT))
+            throw MakeStringException(0, "Error receiving actinit data for graph: %" GIDPF "d", graphId);
+        replyTag = msg.getReplyTag();
+        msg.read(len);
+    }
+    else
+    {
+        if (needActInit)
         {
             // initialize any for which no data was sent
             msg.append(smt_initActDataReq); // may cause graph to be created at master
@@ -848,14 +854,20 @@ bool CSlaveGraph::recvActivityInitData(size32_t parentExtractSz, const byte *par
             assertex(!parentExtractSz || NULL!=parentExtract);
             msg.append(parentExtractSz);
             msg.append(parentExtractSz, parentExtract);
-            Owned<IThorActivityIterator> iter = getConnectedIterator();
+
+            // NB: will only request activities that need initializaton data (those that override CSlaveActivity::init())
             ForEach(*iter)
             {
                 CSlaveGraphElement &element = (CSlaveGraphElement &)iter->query();
-                if (!element.sentActInitData->test(0))
+                CActivityBase *activity = element.queryActivity();
+                if (activity)
                 {
-                    msg.append(element.queryId());
-//                    element.serializeStartContext(msg);
+                    if (!element.sentActInitData->test(0))
+                    {
+                        msg.append(element.queryId());
+                        // JCSMORE -> GH - do you always generate a start context serializer?
+                        element.serializeStartContext(msg);
+                    }
                 }
             }
             msg.append((activity_id)0);
@@ -875,33 +887,16 @@ bool CSlaveGraph::recvActivityInitData(size32_t parentExtractSz, const byte *par
             }
             msg.read(len);
         }
+    }
+    if (len)
+    {
         try
         {
             MemoryBuffer actInitData;
-            if (len)
-                actInitData.append(len, msg.readDirect(len));
+            actInitData.append(len, msg.readDirect(len));
+            CriticalBlock b(progressCrit);
+            initialized = true;
             initWithActData(actInitData, actInitRtnData);
-
-            if (queryOwner() && !isGlobal())
-            {
-                // initialize any for which no data was sent
-                Owned<IThorActivityIterator> iter = getConnectedIterator();
-                ForEach(*iter)
-                {
-                    CSlaveGraphElement &element = (CSlaveGraphElement &)iter->query();
-                    if (!element.sentActInitData->test(0))
-                    {
-                        element.sentActInitData->set(0);
-                        CSlaveActivity *activity = (CSlaveActivity *)element.queryActivity();
-                        if (activity)
-                        {
-                            MemoryBuffer in, out;
-                            activity->init(in, out);
-                            assertex(0 == out.length());
-                        }
-                    }
-                }
-            }
         }
         catch (IException *e)
         {
@@ -911,14 +906,33 @@ bool CSlaveGraph::recvActivityInitData(size32_t parentExtractSz, const byte *par
             e->Release();
             ret = false;
         }
+    }
+    if (syncInitData() || needActInit)
+    {
         if (!queryJobChannel().queryJobComm().send(actInitRtnData, 0, replyTag, LONGTIMEOUT))
             throw MakeStringException(0, "Timeout sending init data back to master");
     }
+    // initialize any for which no data was sent
+    ForEach(*iter)
+    {
+        CSlaveGraphElement &element = (CSlaveGraphElement &)iter->query();
+        CSlaveActivity *activity = (CSlaveActivity *)element.queryActivity();
+        if (activity && !activity->queryInitialized())
+        {
+            activity->setInitialized(true);
+            element.sentActInitData->set(0);
+            MemoryBuffer in, out;
+            activity->init(in, out);
+            assertex(0 == out.length());
+        }
+    }
     return ret;
 }
 
 bool CSlaveGraph::preStart(size32_t parentExtractSz, const byte *parentExtract)
 {
+    if (!recvActivityInitData(parentExtractSz, parentExtract))
+        throw MakeThorException(0, "preStart failure");
     CGraphBase::preStart(parentExtractSz, parentExtract);
     if (isGlobal())
     {
@@ -1024,8 +1038,6 @@ void CSlaveGraph::executeSubGraph(size32_t parentExtractSz, const byte *parentEx
             }
             connect(); // only now do slave acts. have all their outputs prepared.
         }
-        if (!recvActivityInitData(parentExtractSz, parentExtract))
-            throw MakeThorException(0, "preStart failure");
         CGraphBase::executeSubGraph(parentExtractSz, parentExtract);
     }
     catch (IException *e)

thorlcr/graph/thgraphslave.hpp

@@ -159,6 +159,11 @@ public:
 
     CSlaveActivity(CGraphElementBase *container);
     ~CSlaveActivity();
+    void setRequireInitData(bool tf)
+    {
+        // If not required sets sentActInitdata to true, to prevent it being request at graph initialization time.
+        container.sentActInitData->set(0, !tf);
+    }
     virtual void clearConnections();
     virtual void releaseIOs();
     virtual MemoryBuffer &queryInitializationData(unsigned slave) const;
@@ -218,7 +223,7 @@ public:
     virtual void resetEOF() override { throwUnexpected(); }
 
 // IThorSlaveActivity
-    virtual void init(MemoryBuffer &in, MemoryBuffer &out) override { }
+    virtual void init(MemoryBuffer &in, MemoryBuffer &out) { }
     virtual void setInputStream(unsigned index, CThorInput &input, bool consumerOrdered) override;
     virtual void processDone(MemoryBuffer &mb) override { };
     virtual void reset() override;

thorlcr/slave/slave.ipp

@@ -98,8 +98,9 @@ protected:
     PointerArrayOf<IStrandJunction> expandedJunctions;
 
 public:
-    CThorNarySlaveActivity(CGraphElementBase *container) : CSlaveActivity(container)
+    CThorNarySlaveActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
+        setRequireInitData(false);
     }
     virtual void start() override
     {