Merge pull request #13898 from nhalliday/Failure

HPCC-24308 Success and Failure contingency items are executed by parallel workflow

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

common/workunit/workflow.hpp

@@ -22,9 +22,17 @@
 #include "workunit.hpp"
 #include "jlog.hpp"
 #include "eclhelper.hpp"
+#include <queue>
+#include <thread>
+#include <atomic>
+
+#ifdef _DEBUG
+//    #define TRACE_WORKFLOW
+#endif
 
 #define WFERR_ExecutingInWaitState      5100
 #define WFERR_ExecutingInBlockedState   5101
+#define WFERR_ExecutingItemMoreThanOnce 5103
 
 class WORKUNIT_API WorkflowException : public IException, public CInterface
 {
@@ -66,6 +74,7 @@ private:
   *  - Support once, stored, persist workflow items.
   *
   */
+class CCloneWorkflowItem;
 class WORKUNIT_API WorkflowMachine : public CInterface
 {
 public:
@@ -121,9 +130,83 @@ protected:
     bool attemptRetry(IRuntimeWorkflowItem & item, unsigned dep, unsigned scheduledWfid);
     void handleFailure(IRuntimeWorkflowItem & item, WorkflowException const * e, bool isDep);
 
+    void addSuccessors();
+    //This function defines the implicit dependencies of the workflow, by creating logical successorships.
+    //It traverses through all the items, by recursing through their dependencies.
+    //It also reverses dependencies, so that items point to their successors.
+    void defineLogicalRelationships(unsigned wfid, CCloneWorkflowItem * logicalPredecessor, bool prevOrdered);
+    CCloneWorkflowItem & queryWorkflowItem(unsigned wfid);
+    //This creates a new node which is inserted as a predecessor to the successor item.
+    //This new runtime item is a logical predecessor - one that may activate the successor.
+    //The logical predecessor can also activate any of the successor's children.
+    //The pointer to the runtime item is returned.
+    CCloneWorkflowItem * insertLogicalPredecessor(unsigned successorWfid);
+
+    void performParallel(IGlobalCodeContext *_ctx, IEclProcess *_process);
+    void processWfItems();
+    void executeItemParallel(unsigned wfid);
+    void doExecuteItemParallel(IRuntimeWorkflowItem & item);
+    void doExecuteConditionExpression(CCloneWorkflowItem & item);
+    void performItemParallel(unsigned wfid);
+    //Returns true if a failure contingency has been queued
+    bool handleFailureParallel(CCloneWorkflowItem & item, WorkflowException * e);
+    //Returns true if a failure contingency has been queued
+    bool activateFailureContingency(CCloneWorkflowItem & item);
+    void checkAbort(CCloneWorkflowItem & item, bool depFailed);
+    void startContingency();
+    void endContingency();
+
+    void processDependentSuccessors(CCloneWorkflowItem &item);
+    void processLogicalSuccessors(CCloneWorkflowItem &item);
+    //when an item fails, this marks dependentSuccessors with the exception belonging to their predecessor
+    void failDependentSuccessors(CCloneWorkflowItem &item);
+
+    void addToItemQueue(unsigned wfid);
+    bool checkIfDone();
+
+    virtual bool getParallelFlag() const = 0;
+    virtual unsigned getThreadNumFlag() const = 0;
+    bool isParallelViable();
+
+
 protected:
     const IContextLogger &logctx;
     Owned<IWorkflowItemArray> workflow;
+    //contains extra workflow items that are created at runtime. These support logical successorships
+    std::vector<Shared<IRuntimeWorkflowItem>> logicalWorkflow;
+    std::queue<unsigned> wfItemQueue;
+    Semaphore wfItemQueueSem;
+    //used to pop/add items to the queue
+    CriticalSection queueCritSec;
+    //optional debug value "parallelThreads" to select number of threads
+    unsigned numThreads = 1U;
+    //the wfid of the parent item. It has no successors, only dependents.
+    unsigned parentWfid = 0U;
+    //If startItem has an item as its logical successor, then that item will be active before the start.
+    //Any items that are active from the start don't need to perform the defineLogicalRelationships algorithm more than once.
+    CCloneWorkflowItem * startItem = nullptr;
+     //flag is set when the "parent" item is reached. There may still be pending contingencies
+    std::atomic<bool> parentReached{false};
+    //flag is set once the workflow is completed
+    std::atomic<bool> done{false};
+    //flag is set when a workflowItem fails and is not successfully recovered
+    std::atomic<bool> abort{false};
+    //This protects against a race condition between activate() and deactivate()
+    CriticalSection activationCritSec;
+    //This protects each item from having its exception set twice, in a race condition
+    CriticalSection exceptionCritSec;
+    //This counts the active contingency clauses (that haven't finished being executed)
+    //This ensures that the query doesn't finish without completing the contingencies.
+    std::atomic<unsigned> activeContingencies{0U};
+    //The number of branches is the number of dependent successors to the failed workflow item.
+    //Each successor then fails its own successors, so the branch count increases.
+    //In order to verify that all possible contingency clauses have been reached, the number of open-ended "branches" that haven't yet reached the parent item must be tracked.
+    //The query is finished when there are zero open-ended branches.
+    std::atomic<unsigned> branchCount{0U};
+    //optional debug value "parallelWorkflow" to select parallel algorithm
+    bool parallel = false;
+    Owned<WorkflowException> runtimeError;
+
     IGlobalCodeContext *ctx;
     IEclProcess *process;
     IntArray wfidStack;
@@ -131,6 +214,9 @@ protected:
     unsigned currentScheduledWfid;
     unsigned itemsWaiting;
     unsigned itemsUnblocked;
+
+    //allows the condition result to be returned from a process in a thread-safe way
+    CriticalSection conditionCritSec;
     unsigned condition;
 };
 

common/workunit/workunit.hpp

@@ -571,8 +571,12 @@ enum WFMode
     WFModeBeginWait = 5,
     WFModeWait = 6,
     WFModeOnce = 7,
-    WFModeSize = 8,
-    WFModeCritical = 9
+    WFModeUnused = 8,
+    WFModeCritical = 9,
+    WFModeOrdered = 10,
+    WFModeConditionExpression = 11,
+    //Size needs to be the last mode
+    WFModeSize = 12
 };
 
 enum WFState

ecl/eclagent/eclagent.cpp

@@ -2212,6 +2212,14 @@ void EclAgentWorkflowMachine::begin()
     if(agent.queryWorkUnit()->getDebugValueBool("prelockpersists", false))
         prelockPersists();
 }
+bool EclAgentWorkflowMachine::getParallelFlag() const
+{
+    return agent.queryWorkUnit()->getDebugValueBool("parallelWorkflow", false);
+}
+unsigned EclAgentWorkflowMachine::getThreadNumFlag() const
+{
+    return agent.queryWorkUnit()->getDebugValueInt("numWorkflowThreads", 4);
+}
 
 void EclAgentWorkflowMachine::prelockPersists()
 {

ecl/eclagent/eclagent.ipp

@@ -282,6 +282,10 @@ public:
 
 protected:
     virtual void begin();
+
+    virtual bool getParallelFlag() const;
+    virtual unsigned getThreadNumFlag() const;
+
     virtual void end();
     virtual void schedulingStart();
     virtual bool schedulingPull();

roxie/ccd/ccdcontext.cpp

@@ -215,12 +215,14 @@ class CRoxieWorkflowMachine : public WorkflowMachine
     };
 
 public:
-    CRoxieWorkflowMachine(IPropertyTree *_workflowInfo, IConstWorkUnit *_wu, bool _doOnce, const IRoxieContextLogger &_logctx)
+    CRoxieWorkflowMachine(IPropertyTree *_workflowInfo, IConstWorkUnit *_wu, bool _doOnce, bool _parallelWorkflow, unsigned _numWorkflowThreads, const IRoxieContextLogger &_logctx)
     : WorkflowMachine(_logctx)
     {
         workunit = _wu;
         workflowInfo = _workflowInfo;
         doOnce = _doOnce;
+        parallelWorkflow = _parallelWorkflow;
+        numWorkflowThreads = _numWorkflowThreads;
     }
     void returnPersistVersion(char const * logicalName, unsigned eclCRC, unsigned __int64 allCRC, bool isFile)
     {
@@ -251,6 +253,14 @@ protected:
             }
         }
     }
+    virtual bool getParallelFlag() const override
+    {
+        return parallelWorkflow;
+    }
+    virtual unsigned getThreadNumFlag() const override
+    {
+        return numWorkflowThreads;
+    }
     virtual void end()
     {
         if (workunit)
@@ -775,11 +785,13 @@ private:
     Owned<PersistVersion> persist;
     IArray persistReadLocks;
     bool doOnce;
+    bool parallelWorkflow;
+    unsigned numWorkflowThreads;
 };
 
-CRoxieWorkflowMachine *createRoxieWorkflowMachine(IPropertyTree *_workflowInfo, IConstWorkUnit *_wu, bool _doOnce, const IRoxieContextLogger &_logctx)
+CRoxieWorkflowMachine *createRoxieWorkflowMachine(IPropertyTree *_workflowInfo, IConstWorkUnit *_wu, bool _doOnce, bool _parallelWorkflow, unsigned _numWorkflowThreads, const IRoxieContextLogger &_logctx)
 {
-    return new CRoxieWorkflowMachine(_workflowInfo, _wu, _doOnce, _logctx);
+    return new CRoxieWorkflowMachine(_workflowInfo, _wu, _doOnce, _parallelWorkflow, _numWorkflowThreads, _logctx);
 }
 
 //=======================================================================================================================
@@ -2644,7 +2656,7 @@ public:
     {
         init();
         rowManager->setMemoryLimit(options.memoryLimit);
-        workflow.setown(_factory->createWorkflowMachine(workUnit, true, logctx));
+        workflow.setown(_factory->createWorkflowMachine(workUnit, true, logctx, options));
         context.setown(createPTree(ipt_caseInsensitive|ipt_fast));
     }
 
@@ -2654,7 +2666,7 @@ public:
         init();
         workUnit.set(_workUnit);
         rowManager->setMemoryLimit(options.memoryLimit);
-        workflow.setown(_factory->createWorkflowMachine(workUnit, false, logctx));
+        workflow.setown(_factory->createWorkflowMachine(workUnit, false, logctx, options));
         context.setown(createPTree(ipt_caseInsensitive|ipt_fast));
 
         //MORE: Use various debug settings to override settings:
@@ -2706,7 +2718,7 @@ public:
         rowManager->setActivityTracking(context->getPropBool("_TraceMemory", false));
         rowManager->setMemoryLimit(options.memoryLimit);
 
-        workflow.setown(_factory->createWorkflowMachine(workUnit, false, logctx));
+        workflow.setown(_factory->createWorkflowMachine(workUnit, false, logctx, options));
     }
 
     virtual roxiemem::IRowManager &queryRowManager()

roxie/ccd/ccdcontext.hpp

@@ -111,6 +111,6 @@ extern IRoxieAgentContext *createAgentContext(const IQueryFactory *factory, cons
 extern IRoxieServerContext *createRoxieServerContext(IPropertyTree *context, IHpccProtocolResponse *protocol, const IQueryFactory *factory, unsigned flags, const ContextLogger &logctx, PTreeReaderOptions xmlReadFlags, const char *querySetName);
 extern IRoxieServerContext *createOnceServerContext(const IQueryFactory *factory, const IRoxieContextLogger &_logctx);
 extern IRoxieServerContext *createWorkUnitServerContext(IConstWorkUnit *wu, const IQueryFactory *factory, const ContextLogger &logctx);
-extern CRoxieWorkflowMachine *createRoxieWorkflowMachine(IPropertyTree *_workflowInfo, IConstWorkUnit *wu, bool doOnce, const IRoxieContextLogger &_logctx);
+extern CRoxieWorkflowMachine *createRoxieWorkflowMachine(IPropertyTree *_workflowInfo, IConstWorkUnit *wu, bool doOnce, bool _parallelWorkflow, unsigned _numWorkflowThreads, const IRoxieContextLogger &_logctx);
 
 #endif

roxie/ccd/ccdquery.cpp

@@ -1587,7 +1587,7 @@ static hash64_t getQueryHash(const char *id, const IQueryDll *dll, const IRoxieP
     {
         return package;
     }
-    virtual CRoxieWorkflowMachine *createWorkflowMachine(IConstWorkUnit *wu, bool isOnce, const IRoxieContextLogger &logctx) const override
+    virtual CRoxieWorkflowMachine *createWorkflowMachine(IConstWorkUnit *wu, bool isOnce, const IRoxieContextLogger &logctx, const QueryOptions & options) const override
     {
         throwUnexpected();  // only on server...
     }
@@ -1757,12 +1757,12 @@ public:
         return createWorkUnitServerContext(wu, this, _logctx);
     }
 
-    virtual CRoxieWorkflowMachine *createWorkflowMachine(IConstWorkUnit *wu, bool isOnce, const IRoxieContextLogger &logctx) const override
+    virtual CRoxieWorkflowMachine *createWorkflowMachine(IConstWorkUnit *wu, bool isOnce, const IRoxieContextLogger &logctx, const QueryOptions & options) const
     {
         IPropertyTree *workflow = queryWorkflowTree();
         if (workflow)
         {
-            return ::createRoxieWorkflowMachine(workflow, wu, isOnce, logctx);
+            return ::createRoxieWorkflowMachine(workflow, wu, isOnce, options.parallelWorkflow, options.numWorkflowThreads, logctx);
         }
         else
             return NULL;

roxie/ccd/ccdquery.hpp

@@ -154,7 +154,7 @@ interface IQueryFactory : extends IInterface
     virtual void getActivityMetrics(StringBuffer &reply) const = 0;
 
     virtual IPropertyTree *cloneQueryXGMML() const = 0;
-    virtual CRoxieWorkflowMachine *createWorkflowMachine(IConstWorkUnit *wu, bool isOnce, const IRoxieContextLogger &logctx) const = 0;
+    virtual CRoxieWorkflowMachine *createWorkflowMachine(IConstWorkUnit *wu, bool isOnce, const IRoxieContextLogger &logctx, const QueryOptions & options) const = 0;
     virtual char *getEnv(const char *name, const char *defaultValue) const = 0;
 
     virtual IRoxieServerContext *createContext(IPropertyTree *xml, IHpccProtocolResponse *protocol, unsigned flags, const ContextLogger &_logctx, PTreeReaderOptions xmlReadFlags, const char *querySetName) const = 0;

testing/regress/ecl/workflow_1.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_10.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_11.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_12.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor,noroxie
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);
@@ -23,7 +23,7 @@ optParallel := #IFDEFINED(root.parallel, false);
 #option ('parallelWorkflow', optParallel);
 #option('numWorkflowThreads', 5);
 
-//graphs are going to run concurrently in eclagent/roxie
+//graphs are going to run concurrently in eclagent/roxie. Roxie can be enabled once multi threading issues are resolved
 MyRec := RECORD
     STRING1 Value1;
     STRING1 Value2;

testing/regress/ecl/workflow_13.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_14.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_15.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_16.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_2.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_3.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_4.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_5.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_6.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_7.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_8.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_9a.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_9b.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_contingency_1.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_contingency_2.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_contingency_3.ecl

@@ -15,8 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
-
+//version parallel=true,nothor
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);
 
@@ -24,7 +23,8 @@ optParallel := #IFDEFINED(root.parallel, false);
 #option('numWorkflowThreads', 5);
 #onwarning(5102, ignore);
 //#option('reportFailureToFirstDependant', false);
-//This tests the Failure flag i.e. reportFailureToFirstDependant in the workflow engine
+
+//This verifies that the parallel engine doesn't execute Failure clauses when the corresponding item is inactive
 display(String thisString) := FUNCTION
   ds := dataset([thisString], {String text});
   RETURN Output(ds, NAMED('logging'), EXTEND);
@@ -42,4 +42,3 @@ e := SEQUENTIAL(b,c);
 
 e;
 //expect a, failure ... for b
-//Once the flag is actually read, expect a, failure ... for b, failure ... for c.

testing/regress/ecl/workflow_contingency_4.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);
@@ -31,6 +31,7 @@ display(String thisString) := FUNCTION
   RETURN Output(ds, NAMED('logging'), EXTEND);
 END;
 
+//this tests that items are aborted once the workflow fails
 b := SEQUENTIAL(display('b'), FAIL(5103)) : independent;
 
 c0 := sleep(2000) : independent;
@@ -40,6 +41,5 @@ c3 := sleep(2003) : independent;
 
 c := SEQUENTIAL(c0, c1, c2, c3, display('c'));
 
-//to get a consistent order in output
-SEQUENTIAL(b,c);
-//IF(optParallel, PARALLEL(b, c), SEQUENTIAL(b,c));
+//Note: the sequential engine behaves differently if Parallel is used, because the workflow output from the code generator is inconsistent
+IF(optParallel, PARALLEL(b, c), SEQUENTIAL(b,c));

testing/regress/ecl/workflow_contingency_5.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_contingency_6.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_contingency_7.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);

testing/regress/ecl/workflow_contingency_8.ecl

@@ -15,7 +15,7 @@
     limitations under the License.
 ############################################################################## */
 //version parallel=false
-//version parallel=true
+//version parallel=true,nothor
 
 import ^ as root;
 optParallel := #IFDEFINED(root.parallel, false);