Merge pull request #13918 from nhalliday/queryWfid

HPCC-24344 Removed calls to workflow->getWorkflowId()

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

common/workunit/workflow.hpp

@@ -69,7 +69,6 @@ public:
     WorkflowMachine();
     WorkflowMachine(const IContextLogger &logctx);
     void perform(IGlobalCodeContext *_ctx, IEclProcess *_process);
-    unsigned queryCurrentWfid() const { return currentWfid; }
     int queryLastFailCode() const;
     char const * queryLastFailMessage() const;
     const char * queryEventName() const;

ecl/eclagent/eclagent.cpp

@@ -2197,7 +2197,7 @@ void EclAgent::runProcess(IEclProcess *process)
 
 unsigned EclAgent::getWorkflowId()
 {
-    return workflow->queryCurrentWfid();
+    throwUnexpected();
 }
 
 //----------------------------------------------------------------
@@ -3153,7 +3153,7 @@ void EclAgent::reportProgress(const char *progress, unsigned flags)
         CriticalBlock block(wusect);
         WorkunitUpdate w = updateWorkUnit();
         w->setState(WUStateAborting);
-        throw new WorkflowException(0, "Workunit abort request received", (workflow ? workflow->queryCurrentWfid() : 0), WorkflowException::ABORT, MSGAUD_user);
+        throw new WorkflowException(0, "Workunit abort request received", 0, WorkflowException::ABORT, MSGAUD_user);
     }
     if (progress)
     {

ecl/eclagent/eclgraph.cpp

@@ -874,7 +874,7 @@ void EclSubGraph::updateProgress()
 {
     if (!isChildGraph && agent->queryRemoteWorkunit())
     {
-        Owned<IWUGraphStats> progress = parent.updateStats(queryStatisticsComponentType(), queryStatisticsComponentName(), agent->getWorkflowId(), id);
+        Owned<IWUGraphStats> progress = parent.updateStats(queryStatisticsComponentType(), queryStatisticsComponentName(), parent.queryWfid(), id);
         IStatisticGatherer & stats = progress->queryStatsBuilder();
         updateProgress(stats);
 
@@ -893,7 +893,7 @@ void EclSubGraph::updateProgress()
                 if (cost)
                 {
                     StringBuffer scope;
-                    scope.append(WorkflowScopePrefix).append(agent->getWorkflowId()).append(":").append(subgraphid);
+                    scope.append(WorkflowScopePrefix).append(parent.queryWfid()).append(":").append(subgraphid);
                     lockedwu->setStatistic(queryStatisticsComponentType(), queryStatisticsComponentName(), SSTsubgraph, scope, StCostExecute, NULL, cost, 1, 0, StatsMergeReplace);
                 }
             }
@@ -1242,7 +1242,7 @@ void EclGraph::execute(const byte * parentExtract)
 
     {
         Owned<IWorkUnit> wu(agent->updateWorkUnit());
-        addTimeStamp(wu, SSTgraph, queryGraphName(), StWhenStarted, agent->getWorkflowId());
+        addTimeStamp(wu, SSTgraph, queryGraphName(), StWhenStarted, wfid);
     }
 
     try
@@ -1265,13 +1265,13 @@ void EclGraph::execute(const byte * parentExtract)
             formatGraphTimerLabel(description, queryGraphName(), 0, 0);
 
             unsigned __int64 elapsedNs = milliToNano(elapsed);
-            updateWorkunitStat(wu, SSTgraph, queryGraphName(), StTimeElapsed, description.str(), elapsedNs, agent->getWorkflowId());
+            updateWorkunitStat(wu, SSTgraph, queryGraphName(), StTimeElapsed, description.str(), elapsedNs, wfid);
 
             const cost_type cost = calcCost(agent->queryAgentMachineCost(), nanoToMilli(elapsedNs));
             if (cost)
             {
                 StringBuffer scope;
-                scope.append(WorkflowScopePrefix).append(agent->getWorkflowId()).append(":").append(queryGraphName());
+                scope.append(WorkflowScopePrefix).append(wfid).append(":").append(queryGraphName());
                 wu->setStatistic(queryStatisticsComponentType(), queryStatisticsComponentName(), SSTgraph, scope, StCostExecute, NULL, cost, 1, 0, StatsMergeReplace);
             }
         }
@@ -1490,7 +1490,7 @@ IWUGraphStats *EclGraph::updateStats(StatisticCreatorType creatorType, const cha
 
 void EclGraph::updateWUStatistic(IWorkUnit *lockedwu, StatisticScopeType scopeType, const char * scope, StatisticKind kind, const char * descr, unsigned __int64 value)
 {
-    updateWorkunitStat(lockedwu, scopeType, scope, kind, descr, value, agent->getWorkflowId());
+    updateWorkunitStat(lockedwu, scopeType, scope, kind, descr, value, wfid);
 }
 
 IThorChildGraph * EclGraph::resolveChildQuery(unsigned subgraphId)

ecl/hqlcpp/hqlcatom.cpp

@@ -382,7 +382,6 @@ IIdAtom * getResultVarStringId;
 IIdAtom * getResultVarUnicodeId;
 IIdAtom * getRootResultId;
 IIdAtom * getStartCyclesId;
-IIdAtom * getWorkflowIdId;
 IIdAtom * getWuidId;
 IIdAtom * groupedDataset2RowsetXId;
 IIdAtom * groupedRowset2DatasetXId;
@@ -1048,7 +1047,6 @@ MODULE_INIT(INIT_PRIORITY_HQLATOM-1)
     MAKEID(getResultVarUnicode);
     MAKEID(getRootResult);
     MAKEID(getStartCycles);
-    MAKEID(getWorkflowId);
     MAKEID(getWuid);
     MAKEID(groupedDataset2RowsetX);
     MAKEID(groupedRowset2DatasetX);

ecl/hqlcpp/hqlcatom.hpp

@@ -380,7 +380,6 @@ extern IIdAtom * getResultVarStringId;
 extern IIdAtom * getResultVarUnicodeId;
 extern IIdAtom * getRootResultId;
 extern IIdAtom * getStartCyclesId;
-extern IIdAtom * getWorkflowIdId;
 extern IIdAtom * getWuidId;
 extern IIdAtom * groupedDataset2RowsetXId;
 extern IIdAtom * groupedRowset2DatasetXId;

ecl/hqlcpp/hqlcppsys.ecl

@@ -695,7 +695,6 @@ const char * cppSystemText[]  = {
     "   setResultVarUnicode(const varstring stepname, unsigned4 sequence, const varunicode value) : ctxmethod,entrypoint='setResultVarUnicode';",
     "   setConditionCode(integer4 stepname) : ctxmethod,entrypoint='setConditionCode';",
     "   const varstring loadResource(unsigned4 id) : ctxmethod,entrypoint='loadResource';",
-    "   unsigned4 getWorkflowId() : gctxmethod,entrypoint='getWorkflowId';",
     "   doNotify(const varstring name, const varstring text) : gctxmethod,entrypoint='doNotify';",
     "   doNotifyTarget(const varstring name, const varstring text, const varstring _target) : gctxmethod,entrypoint='doNotify';",
     "   setWorkflowCondition(boolean value) : gctxmethod,entrypoint='setWorkflowCondition';",

roxie/ccd/ccdcontext.cpp

@@ -1502,7 +1502,7 @@ public:
         if (debugContext)
             debugContext->checkBreakpoint(DebugStateGraphStart, NULL, graphName);
         if (collectingDetailedStatistics())
-            graphStats.setown(workUnit->updateStats(graph->queryName(), SCTroxie, queryStatisticsComponentName(), getWorkflowId(), 0));
+            graphStats.setown(workUnit->updateStats(graph->queryName(), SCTroxie, queryStatisticsComponentName(), graph->queryWorkflowId(), 0));
     }
 
     virtual void endGraph(unsigned __int64 startTimeStamp, cycle_t startCycles, bool aborting)
@@ -1524,10 +1524,10 @@ public:
                     formatGraphTimerLabel(graphDesc, graphName);
                     WorkunitUpdate progressWorkUnit(&workUnit->lock());
                     StringBuffer graphScope;
-                    graphScope.append(WorkflowScopePrefix).append(getWorkflowId()).append(":").append(graphName);
+                    graphScope.append(WorkflowScopePrefix).append(graph->queryWorkflowId()).append(":").append(graphName);
                     progressWorkUnit->setStatistic(queryStatisticsComponentType(), queryStatisticsComponentName(), SSTgraph, graphScope, StWhenStarted, NULL, startTimeStamp, 1, 0, StatsMergeAppend);
                     updateWorkunitStat(progressWorkUnit, SSTgraph, graphScope, StTimeElapsed, graphDesc, elapsedTime);
-                    addTimeStamp(progressWorkUnit, SSTgraph, graphName, StWhenFinished, getWorkflowId());
+                    addTimeStamp(progressWorkUnit, SSTgraph, graphName, StWhenFinished, graph->queryWorkflowId());
                 }
                 graph->reset();
             }
@@ -3718,7 +3718,11 @@ public:
         addWuExceptionEx(text, code, 2, MSGAUD_user, "user");
     }
 
-    virtual unsigned getWorkflowId() { return workflow->queryCurrentWfid(); }
+    virtual unsigned getWorkflowId()
+    {
+        logctx.CTXLOG("Trying to access WFID from workflow");
+        throwUnexpected();
+    }
 
     void doNotify(char const * name, char const * text)
     {

roxie/ccd/ccdquery.cpp

@@ -308,6 +308,8 @@ QueryOptions::QueryOptions()
     allSortsMaySpill = false; // No global default for this
     failOnLeaks = false;
     collectFactoryStatistics = defaultCollectFactoryStatistics;
+    parallelWorkflow = false;
+    numWorkflowThreads = 1;
 }
 
 QueryOptions::QueryOptions(const QueryOptions &other)
@@ -340,6 +342,9 @@ QueryOptions::QueryOptions(const QueryOptions &other)
     allSortsMaySpill = other.allSortsMaySpill;
     failOnLeaks = other.failOnLeaks;
     collectFactoryStatistics = other.collectFactoryStatistics;
+
+    parallelWorkflow = other.parallelWorkflow;
+    numWorkflowThreads = other.numWorkflowThreads;
 }
 
 void QueryOptions::setFromWorkUnit(IConstWorkUnit &wu, const IPropertyTree *stateInfo)
@@ -382,6 +387,9 @@ void QueryOptions::setFromWorkUnit(IConstWorkUnit &wu, const IPropertyTree *stat
     updateFromWorkUnit(failOnLeaks, wu, "failOnLeaks");
     updateFromWorkUnit(noSeekBuildIndex, wu, "noSeekBuildIndex");
     updateFromWorkUnit(collectFactoryStatistics, wu, "collectFactoryStatistics");
+
+    updateFromWorkUnit(parallelWorkflow, wu, "parallelWorkflow");
+    updateFromWorkUnit(numWorkflowThreads, wu, "numWorkflowthreads");
 }
 
 void QueryOptions::updateFromWorkUnitM(memsize_t &value, IConstWorkUnit &wu, const char *name)
@@ -442,6 +450,9 @@ void QueryOptions::setFromContext(const IPropertyTree *ctx)
         // Note: allSortsMaySpill is not permitted at context level (too late anyway, unless I refactored)
         updateFromContext(failOnLeaks, ctx, "@failOnLeaks", "_FailOnLeaks");
         updateFromContext(collectFactoryStatistics, ctx, "@collectFactoryStatistics", "_CollectFactoryStatistics");
+
+        updateFromContext(parallelWorkflow, ctx, "@parallelWorkflow", "_parallelWorkflow");
+        updateFromContext(numWorkflowThreads, ctx, "@numWorkflowThreads", "_numWorkflowThreads");
     }
 }
 
@@ -897,13 +908,13 @@ protected:
         return NULL;
     }
 
-    virtual IRoxieServerActivityFactory *getRoxieServerActivityFactory(unsigned id) const
+    virtual IRoxieServerActivityFactory *getRoxieServerActivityFactory(unsigned id) const override
     {
         checkSuspended();
         return LINK(QUERYINTERFACE(findActivity(id), IRoxieServerActivityFactory));
     }
 
-    virtual ISlaveActivityFactory *getSlaveActivityFactory(unsigned id) const
+    virtual ISlaveActivityFactory *getSlaveActivityFactory(unsigned id) const override
     {
         checkSuspended();
         IActivityFactory *f = findActivity(id);
@@ -913,7 +924,7 @@ protected:
     ActivityArray *loadChildGraph(IPropertyTree &graph)
     {
         // MORE - this is starting to look very much like loadGraph (on Roxie server side)
-        ActivityArray *activities = new ActivityArray(true, graph.getPropBool("@delayed"), graph.getPropBool("@library"), graph.getPropBool("@sequential"));
+        ActivityArray *activities = new ActivityArray(true, graph.getPropBool("@delayed"), graph.getPropBool("@library"), graph.getPropBool("@sequential"), graph.getPropBool("@wfid"));
         unsigned subgraphId = graph.getPropInt("@id");
         try
         {
@@ -1101,12 +1112,12 @@ public:
         package.Release();
     }
 
-    virtual IQueryFactory *lookupLibrary(const char *libraryName, unsigned expectedInterfaceHash, const IRoxieContextLogger &logctx) const
+    virtual IQueryFactory *lookupLibrary(const char *libraryName, unsigned expectedInterfaceHash, const IRoxieContextLogger &logctx) const override
     {
         return globalPackageSetManager->lookupLibrary(libraryName, expectedInterfaceHash, logctx);
     }
 
-    virtual void beforeDispose()
+    virtual void beforeDispose() override
     {
         // NOTE: it's theoretically possible for the final release to happen after a replacement has been inserted into hash table. 
         // So only remove from hash table if what we find there matches the item that is being deleted.
@@ -1263,28 +1274,28 @@ public:
         queryMap.setValue(hv, this);
     }
 
-    virtual unsigned queryChannel() const
+    virtual unsigned queryChannel() const override
     {
         return channelNo;
     }
 
-    virtual hash64_t queryHash() const
+    virtual hash64_t queryHash() const override
     {
         return hashValue;
     }
 
-    virtual ISharedOnceContext *querySharedOnceContext() const
+    virtual ISharedOnceContext *querySharedOnceContext() const override
     {
         return sharedOnceContext;
     }
 
-    virtual IDeserializedResultStore &queryOnceResultStore() const
+    virtual IDeserializedResultStore &queryOnceResultStore() const override
     {
         assertex(sharedOnceContext);
         return sharedOnceContext->queryOnceResultStore();
     }
 
-    virtual IPropertyTree &queryOnceContext(const IRoxieContextLogger &logctx) const
+    virtual IPropertyTree &queryOnceContext(const IRoxieContextLogger &logctx) const override
     {
         assertex(sharedOnceContext);
         return sharedOnceContext->queryOnceContext(this, logctx);
@@ -1295,12 +1306,12 @@ public:
         return (const char *) queryDll()->getResource(id);
     }
 
-    virtual ActivityArray *lookupGraphActivities(const char *name) const
+    virtual ActivityArray *lookupGraphActivities(const char *name) const override
     {
         return *graphMap.getValue(name);
     }
 
-    virtual IActivityGraph *lookupGraph(IRoxieSlaveContext *ctx, const char *name, IProbeManager *probeManager, const IRoxieContextLogger &logctx, IRoxieServerActivity *parentActivity) const
+    virtual IActivityGraph *lookupGraph(IRoxieSlaveContext *ctx, const char *name, IProbeManager *probeManager, const IRoxieContextLogger &logctx, IRoxieServerActivity *parentActivity) const override
     {
         assertex(name && *name);
         ActivityArrayPtr *graph = graphMap.getValue(name);
@@ -1336,7 +1347,7 @@ public:
         toXML(graph, reply);
     }
 
-    virtual IPropertyTree* cloneQueryXGMML() const
+    virtual IPropertyTree* cloneQueryXGMML() const override
     {
         assertex(dll && dll->queryWorkUnit());
         Owned<IPropertyTree> tree = createPTree("Query", ipt_lowmem);
@@ -1354,7 +1365,7 @@ public:
         return tree.getClear();
     }
 
-    virtual void getStats(StringBuffer &reply, const char *graphName) const
+    virtual void getStats(StringBuffer &reply, const char *graphName) const override
     {
         if (dll)
         {
@@ -1376,7 +1387,7 @@ public:
             }
         }
     }
-    virtual void getActivityMetrics(StringBuffer &reply) const
+    virtual void getActivityMetrics(StringBuffer &reply) const override
     {
         HashIterator i(allActivities);
         StringBuffer myReply;
@@ -1392,7 +1403,7 @@ public:
             }
         }
     }
-    virtual void getQueryInfo(StringBuffer &reply, bool full, IArrayOf<IQueryFactory> *slaveQueries, const IRoxieContextLogger &logctx) const
+    virtual void getQueryInfo(StringBuffer &reply, bool full, IArrayOf<IQueryFactory> *slaveQueries, const IRoxieContextLogger &logctx) const override
     {
         Owned<IPropertyTree> xref = createPTree("Query", ipt_fast);
         xref->setProp("@id", id);
@@ -1423,7 +1434,7 @@ public:
         }
         toXML(xref, reply, 1, XML_Embed|XML_LineBreak|XML_SortTags);
     }
-    virtual void resetQueryTimings()
+    virtual void resetQueryTimings() override
     {
         HashIterator i(allActivities);
         ForEach(i)
@@ -1432,56 +1443,56 @@ public:
             f->resetNodeProgressInfo();
         }
     }
-    virtual const char *queryErrorMessage() const
+    virtual const char *queryErrorMessage() const override
     {
         return errorMessage.str();
     }
-    virtual const char *queryQueryName() const
+    virtual const char *queryQueryName() const override
     {
         return id;
     }
-    virtual bool isQueryLibrary() const 
+    virtual bool isQueryLibrary() const override
     {
         return libraryInterfaceHash != 0; 
     }
-    virtual unsigned getQueryLibraryInterfaceHash() const 
+    virtual unsigned getQueryLibraryInterfaceHash() const override
     {
         return libraryInterfaceHash;
     }
-    virtual void suspend(const char* errMsg)
+    virtual void suspend(const char* errMsg) override
     {
         isSuspended = true;
         isLoadFailed = true;
         errorMessage.append(errMsg);
     }
 
-    virtual bool loadFailed() const
+    virtual bool loadFailed() const override
     {
         return isLoadFailed;
     }
-    virtual bool suspended() const
+    virtual bool suspended() const override
     {
         return isSuspended;
     }
-    virtual const QueryOptions &queryOptions() const
+    virtual const QueryOptions &queryOptions() const override
     {
         return options;
     }
-    virtual ILoadedDllEntry *queryDll() const 
+    virtual ILoadedDllEntry *queryDll() const override
     {
         assertex(dll);
         return dll->queryDll();
     }
-    virtual IConstWorkUnit *queryWorkUnit() const
+    virtual IConstWorkUnit *queryWorkUnit() const override
     {
         assertex(dll);
         return dll->queryWorkUnit();
     }
-    virtual const IRoxiePackage &queryPackage() const
+    virtual const IRoxiePackage &queryPackage() const override
     {
         return package;
     }
-    virtual CRoxieWorkflowMachine *createWorkflowMachine(IConstWorkUnit *wu, bool isOnce, const IRoxieContextLogger &logctx) const
+    virtual CRoxieWorkflowMachine *createWorkflowMachine(IConstWorkUnit *wu, bool isOnce, const IRoxieContextLogger &logctx) const override
     {
         throwUnexpected();  // only on server...
     }
@@ -1504,28 +1515,28 @@ public:
         return strdup(result ? result : defaultValue);
     }
 
-    virtual IRoxieSlaveContext *createSlaveContext(const SlaveContextLogger &logctx, IRoxieQueryPacket *packet, bool hasChildren) const
+    virtual IRoxieSlaveContext *createSlaveContext(const SlaveContextLogger &logctx, IRoxieQueryPacket *packet, bool hasChildren) const override
     {
         throwUnexpected();   // only implemented in derived slave class
     }
 
-    virtual IRoxieServerContext *createContext(IPropertyTree *xml, IHpccProtocolResponse *protocol, unsigned flags, const ContextLogger &_logctx, PTreeReaderOptions xmlReadFlags, const char *querySetName) const
+    virtual IRoxieServerContext *createContext(IPropertyTree *xml, IHpccProtocolResponse *protocol, unsigned flags, const ContextLogger &_logctx, PTreeReaderOptions xmlReadFlags, const char *querySetName) const override
     {
         throwUnexpected();   // only implemented in derived server class
     }
-    virtual IRoxieServerContext *createContext(IConstWorkUnit *wu, const ContextLogger &_logctx) const
+    virtual IRoxieServerContext *createContext(IConstWorkUnit *wu, const ContextLogger &_logctx) const override
     {
         throwUnexpected();   // only implemented in derived server class
     }
-    virtual void noteQuery(time_t startTime, bool failed, unsigned elapsed, unsigned memused, unsigned slavesReplyLen, unsigned bytesOut)
+    virtual void noteQuery(time_t startTime, bool failed, unsigned elapsed, unsigned memused, unsigned slavesReplyLen, unsigned bytesOut) override
     {
         throwUnexpected();   // only implemented in derived server class
     }
-    virtual IPropertyTree *getQueryStats(time_t from, time_t to)
+    virtual IPropertyTree *getQueryStats(time_t from, time_t to) override
     {
         throwUnexpected();   // only implemented in derived server class
     }
-    virtual void getGraphNames(StringArray &ret) const
+    virtual void getGraphNames(StringArray &ret) const override
     {
         Owned<IConstWUGraphIterator> graphs = &dll->queryWorkUnit()->getGraphs(GraphTypeActivities);
         ForEach(*graphs)
@@ -1536,7 +1547,7 @@ public:
         }
     }
 
-    virtual bool isDynamic() const
+    virtual bool isDynamic() const override
     {
         return dynamic;
     }
@@ -1557,7 +1568,7 @@ protected:
             return false;
     }
 
-    virtual void checkSuspended() const
+    virtual void checkSuspended() const override
     {
         if (isSuspended)
         {
@@ -1604,7 +1615,7 @@ public:
         queryGlobalQueryStatsAggregator()->noteQuery(startTime, failed, elapsed, memused, slavesReplyLen, bytesOut);
     }
 
-    virtual void addDependency(unsigned sourceIdx, unsigned sourceId, unsigned targetId, int controlId, const char *edgeId, ActivityArray * activities)
+    virtual void addDependency(unsigned sourceIdx, unsigned sourceId, unsigned targetId, int controlId, const char *edgeId, ActivityArray * activities) override
     {
         // addDependency is expected to fail occasionally on slave, but never on Roxie server
         if (!doAddDependency(sourceIdx, sourceId, targetId, controlId, edgeId, activities))
@@ -1615,7 +1626,8 @@ public:
     {
         bool isLibraryGraph = graph.getPropBool("@library");
         bool isSequential = graph.getPropBool("@sequential");
-        ActivityArray *activities = new ActivityArray(isLibraryGraph, false, isLibraryGraph, isSequential);
+        unsigned wfid = graph.getPropInt("@wfid");
+        ActivityArray *activities = new ActivityArray(isLibraryGraph, false, isLibraryGraph, isSequential, wfid);
         if (isLibraryGraph)
             activities->setLibraryGraphId(graph.getPropInt("node/@id"));
         try
@@ -1650,7 +1662,7 @@ public:
         return createWorkUnitServerContext(wu, this, _logctx);
     }
 
-    virtual CRoxieWorkflowMachine *createWorkflowMachine(IConstWorkUnit *wu, bool isOnce, const IRoxieContextLogger &logctx) const
+    virtual CRoxieWorkflowMachine *createWorkflowMachine(IConstWorkUnit *wu, bool isOnce, const IRoxieContextLogger &logctx) const override
     {
         IPropertyTree *workflow = queryWorkflowTree();
         if (workflow)
@@ -1929,7 +1941,8 @@ public:
         // MORE: common up with loadGraph for the Roxie server..
         bool isLibraryGraph = graph.getPropBool("@library");
         bool isSequential = graph.getPropBool("@sequential");
-        ActivityArray *activities = new ActivityArray(isLibraryGraph, false, isLibraryGraph, isSequential);
+        unsigned wfid = graph.getPropInt("@wfid");
+        ActivityArray *activities = new ActivityArray(isLibraryGraph, false, isLibraryGraph, isSequential, wfid);
         if (isLibraryGraph)
             activities->setLibraryGraphId(graph.getPropInt("node/@id"));
         try

roxie/ccd/ccdquery.hpp

@@ -51,6 +51,7 @@ interface IActivityGraph : extends IInterface
     virtual const char *queryName() const = 0;
     virtual void gatherStatistics(IStatisticGatherer * statsBuilder) const = 0;
     virtual void setPrefix(const char *prefix) = 0;
+    virtual unsigned queryWorkflowId() const = 0;
 };
 
 interface IRoxiePackage;
@@ -108,6 +109,8 @@ public:
     bool failOnLeaks;
     bool collectFactoryStatistics;
     bool noSeekBuildIndex;
+    bool parallelWorkflow;
+    unsigned numWorkflowThreads;
 
 private:
     static const char *findProp(const IPropertyTree *ctx, const char *name1, const char *name2);
@@ -176,10 +179,11 @@ class ActivityArray : public CInterface
     bool library;
     bool sequential;
     unsigned libraryGraphId;
+    unsigned wfid;
 
 public:
-    ActivityArray(bool _multiInstance, bool _delayed, bool _library, bool _sequential)
-     : multiInstance(_multiInstance), delayed(_delayed), library(_library), sequential(_sequential)
+    ActivityArray(bool _multiInstance, bool _delayed, bool _library, bool _sequential, unsigned _wfid)
+     : multiInstance(_multiInstance), delayed(_delayed), library(_library), sequential(_sequential), wfid(_wfid)
     {
         libraryGraphId = 0;
     }
@@ -197,6 +201,7 @@ public:
     inline bool isLibrary() const { return library; }
     inline bool isSequential() const { return sequential; }
     inline unsigned getLibraryGraphId() const { return libraryGraphId; }
+    inline unsigned queryWorkflowId() const { return wfid; }
 };
 typedef CopyReferenceArrayOf<ActivityArray> ActivityArrayArray;
 

roxie/ccd/ccdserver.cpp

@@ -27640,7 +27640,10 @@ public:
         if (exception)
             throw exception.getLink();
     }
-
+    unsigned queryWorkflowId() const
+    {
+        return graphDefinition.queryWorkflowId();
+    }
     virtual void execute()
     {
         results.setown(new CGraphResults);
@@ -27986,6 +27989,10 @@ public:
     virtual void setPrefix(const char *) override
     {
     }
+    virtual unsigned queryWorkflowId() const override
+    {
+        return graphDefinition.queryWorkflowId();
+    }
 protected:
     IRoxieSlaveContext *ctx;
     IRoxieServerActivity *parentActivity;
@@ -28197,7 +28204,10 @@ public:
     {
         prefix.set(pfx);
     }
-
+    virtual unsigned queryWorkflowId() const override
+    {
+        return graphDefinition.queryWorkflowId();
+    }
     virtual IRoxieServerChildGraph * createGraphLoopInstance(IRoxieSlaveContext *ctx, unsigned loopCounter, unsigned parentExtractSize, const byte * parentExtract, const IRoxieContextLogger &logctx) override
     {
         Owned<CIterationActivityGraph> ret = new CIterationActivityGraph(graphName, id, parentActivity, graphDefinition, probeManager, loopCounter, ctx, colocalParent, parentExtractSize, parentExtract, logctx);