Introduces ReadyToExecute in Progress to replace group

Author: grussorusso

internal/test/progress_test.go

@@ -15,7 +15,7 @@ func simpleProgress(t *testing.T) (*workflow.Progress, *workflow.Workflow) {
 	u.AssertNil(t, err)
 	wflow, err := CreateSequenceWorkflow(py, py)
 	u.AssertNil(t, err)
-	return workflow.InitProgressRecursive("simple", wflow), wflow
+	return workflow.InitProgress("simple", wflow), wflow
 }
 
 func choiceProgress(t *testing.T, condition workflow.Condition) (*workflow.Progress, *workflow.Workflow) {
@@ -34,7 +34,7 @@ func choiceProgress(t *testing.T, condition workflow.Condition) (*workflow.Progr
 		EndChoiceAndBuild()
 	u.AssertNil(t, err)
 
-	return workflow.InitProgressRecursive("abc", wflow), wflow
+	return workflow.InitProgress("abc", wflow), wflow
 }
 
 func parallelProgress(t *testing.T) (*workflow.Progress, *workflow.Workflow) {
@@ -50,7 +50,7 @@ func parallelProgress(t *testing.T) (*workflow.Progress, *workflow.Workflow) {
 		Build()
 	u.AssertNil(t, err)
 
-	return workflow.InitProgressRecursive("abc", wflow), wflow
+	return workflow.InitProgress("abc", wflow), wflow
 }
 
 func complexProgress(t *testing.T, condition workflow.Condition) (*workflow.Progress, *workflow.Workflow) {
@@ -74,7 +74,7 @@ func complexProgress(t *testing.T, condition workflow.Condition) (*workflow.Prog
 		EndChoiceAndBuild()
 	u.AssertNil(t, err)
 
-	return workflow.InitProgressRecursive("abc", wflow), wflow
+	return workflow.InitProgress("abc", wflow), wflow
 }
 
 func TestProgressMarshaling(t *testing.T) {
@@ -127,10 +127,8 @@ func TestProgressCache(t *testing.T) {
 		u.AssertTrueMsg(t, found, "progress not found")
 		u.AssertTrueMsg(t, progress.Equals(retrievedProgress), "progresses don't match")
 
-		err = progress.CompleteNode(wflow.Start.Id)
-		u.AssertNilMsg(t, err, "failed to update progress")
-		err = progress.CompleteNode(wflow.Start.Next)
-		u.AssertNilMsg(t, err, "failed to update progress")
+		progress.CompleteNode(wflow.Start.Id)
+		progress.CompleteNode(wflow.Start.Next)
 
 		err = workflow.SaveProgress(progress, cache.Persist)
 		u.AssertNilMsg(t, err, "failed to save after update")
@@ -148,236 +146,3 @@ func TestProgressCache(t *testing.T) {
 		u.AssertFalseMsg(t, found, "progress should have been deleted")
 	}
 }
-
-// TestProgressSequence tests a sequence workflow with 2 simple node
-func TestProgressSequence(t *testing.T) {
-	progress, workflow := simpleProgress(t)
-
-	// Start node
-	checkAndCompleteNext(t, progress, workflow)
-
-	// Simple Node 1
-	checkAndCompleteNext(t, progress, workflow)
-
-	// Simple Node 2
-	checkAndCompleteNext(t, progress, workflow)
-
-	// End node
-	checkAndCompleteNext(t, progress, workflow)
-
-	// End of workflow
-	finishProgress(t, progress)
-}
-
-// TestProgressChoice1 tests the left branch
-func TestProgressChoice1(t *testing.T) {
-	condition := workflow.NewPredicate().And(
-		workflow.NewEqCondition(1, 3),
-		workflow.NewGreaterCondition(1, 3),
-	).Build()
-	progress, wflow := choiceProgress(t, condition)
-
-	// Start node
-	checkAndCompleteNext(t, progress, wflow)
-
-	// Choice node
-	choice := checkAndCompleteNext(t, progress, wflow).(*workflow.ChoiceNode)
-
-	// Simple node (left) // suppose the left condition is true
-	checkAndCompleteChoice(t, progress, choice, wflow, 0)
-
-	// End
-	checkAndCompleteNext(t, progress, wflow)
-
-	// End of workflow
-	finishProgress(t, progress)
-}
-
-// TestProgressChoice2 tests the right branch
-func TestProgressChoice2(t *testing.T) {
-	condition := workflow.NewPredicate().And(
-		workflow.NewEqCondition(1, 1),
-		workflow.NewGreaterCondition(5, 3),
-	).Build()
-	progress, wflow := choiceProgress(t, condition)
-
-	// Start node
-	checkAndCompleteNext(t, progress, wflow)
-
-	// Choice node
-	choice := checkAndCompleteNext(t, progress, wflow).(*workflow.ChoiceNode)
-
-	// Simple Node left is skipped, right is executed
-	expectedMatch := 1
-	checkAndCompleteChoice(t, progress, choice, wflow, expectedMatch)
-
-	// Simple Node right 2
-	checkAndCompleteNext(t, progress, wflow)
-
-	// End node
-	checkAndCompleteNext(t, progress, wflow)
-
-	// End of workflow
-	finishProgress(t, progress)
-}
-
-func TestParallelProgress(t *testing.T) {
-	progress, wflow := parallelProgress(t)
-
-	// Start node
-	checkAndCompleteNext(t, progress, wflow)
-
-	// FanOut node
-	checkAndCompleteNext(t, progress, wflow)
-
-	// 3 Simple Nodes in parallel
-	checkAndCompleteMultiple(t, progress, wflow)
-	// simpleNode1 := fanOut.GetNext()[0]
-	// simpleNode2 := fanOut.GetNext()[1]
-	// simpleNode3 := fanOut.GetNext()[2]
-
-	// 2 Simple Nodes in parallel // here should get two nodes
-	checkAndCompleteMultiple(t, progress, wflow)
-	// nextNode = progress.NextNodes()
-	// simpleNodeCentral2 := simpleNode2.GetNext()[0]
-	// u.AssertEquals(t, nextNode[0], simpleNodeCentral2.GetId())
-	// u.AssertEquals(t, 3, progress.GetGroup(nextNode[0]))
-	// err = progress.CompleteNode(nextNode[0])
-	// u.AssertNil(t, err)
-	// simpleNodeCentral3 := simpleNode3.GetNext()[0]
-	// u.AssertEquals(t, nextNode[1], simpleNodeCentral3.GetId())
-	// u.AssertEquals(t, 3, progress.GetGroup(nextNode[1]))
-	// err = progress.CompleteNode(nextNode[1])
-	// u.AssertNil(t, err)
-
-	// 1 Simple node (parallel) right, bottom
-	checkAndCompleteMultiple(t, progress, wflow)
-	// nextNode = progress.NextNodes()
-	// simpleNodeBottom3 := simpleNodeCentral3.GetNext()[0]
-	// u.AssertEquals(t, nextNode[0], simpleNodeBottom3.GetId())
-	// u.AssertEquals(t, 4, progress.GetGroup(nextNode[0]))
-	// err = progress.CompleteNode(nextNode[0])
-	// u.AssertNil(t, err)
-
-	// Fan in
-	checkAndCompleteNext(t, progress, wflow)
-
-	// End node
-	checkAndCompleteNext(t, progress, wflow)
-
-	// End of workflow
-	finishProgress(t, progress)
-}
-
-func TestComplexProgress(t *testing.T) {
-	condition := workflow.NewPredicate().And(
-		workflow.NewEqCondition(1, 3),
-		workflow.NewGreaterCondition(1, 3),
-	).Build()
-	progress, wflow := complexProgress(t, condition)
-
-	// Start node
-	checkAndCompleteNext(t, progress, wflow)
-
-	// SimpleNode
-	checkAndCompleteNext(t, progress, wflow)
-
-	// Choice
-	choice := checkAndCompleteNext(t, progress, wflow).(*workflow.ChoiceNode)
-
-	// Simple Node, FanOut
-	checkAndCompleteChoice(t, progress, choice, wflow, 0)
-
-	// End node
-	checkAndCompleteNext(t, progress, wflow)
-
-	// End of workflow
-	finishProgress(t, progress)
-}
-
-func TestComplexProgress2(t *testing.T) {
-	condition := workflow.NewPredicate().And(
-		workflow.NewEqCondition(1, 1),
-		workflow.NewGreaterCondition(4, 3),
-	).Build()
-	progress, wflow := complexProgress(t, condition)
-
-	// Start node
-	checkAndCompleteNext(t, progress, wflow)
-
-	// Simple Node
-	checkAndCompleteNext(t, progress, wflow)
-
-	// Choice
-	choice := checkAndCompleteNext(t, progress, wflow).(*workflow.ChoiceNode)
-
-	// Simple Node, FanOut // suppose the fanout node at the right and all its children are skipped
-	checkAndCompleteChoice(t, progress, choice, wflow, 1)
-
-	// 3 Simple Nodes in parallel
-	checkAndCompleteMultiple(t, progress, wflow)
-
-	// 3 other Simple Nodes
-	checkAndCompleteMultiple(t, progress, wflow)
-
-	// Fan in
-	checkAndCompleteNext(t, progress, wflow)
-
-	// End node
-	checkAndCompleteNext(t, progress, wflow)
-
-	// End of workflow
-	finishProgress(t, progress)
-}
-
-func checkAndCompleteNext(t *testing.T, progress *workflow.Progress, workflow *workflow.Workflow) workflow.Task {
-	nextNode, err := progress.NextNodes()
-	u.AssertNil(t, err)
-	nodeId := nextNode[0]
-	node, ok := workflow.Find(nodeId)
-	u.AssertTrue(t, ok)
-	u.AssertEquals(t, nodeId, node.GetId())
-	u.AssertEquals(t, progress.NextGroup, progress.GetGroup(nodeId))
-	err = progress.CompleteNode(nodeId)
-	u.AssertNil(t, err)
-	return node
-}
-
-func checkAndCompleteChoice(t *testing.T, progress *workflow.Progress, choice *workflow.ChoiceNode, workflow *workflow.Workflow, conditionToMatch int) {
-	nextNode, err := progress.NextNodes() // Simple1, Simple2
-	u.AssertNil(t, err)
-	simpleNodeLeft := choice.Alternatives[0]
-	fanOut := choice.Alternatives[1]
-	u.AssertEquals(t, nextNode[0], simpleNodeLeft)
-	u.AssertEquals(t, nextNode[1], fanOut)
-	u.AssertEquals(t, progress.NextGroup, progress.GetGroup(nextNode[0]))
-	u.AssertEquals(t, progress.NextGroup, progress.GetGroup(nextNode[1]))
-
-	err = progress.CompleteNode(nextNode[conditionToMatch])
-	u.AssertNil(t, err)
-	nodeToSkip := choice.GetNodesToSkip(workflow, conditionToMatch)
-	err = progress.SkipAll(nodeToSkip)
-	u.AssertNil(t, err)
-}
-
-func checkAndCompleteMultiple(t *testing.T, progress *workflow.Progress, wflow *workflow.Workflow) []workflow.Task {
-	nextNode, err := progress.NextNodes()
-	completedNodes := make([]workflow.Task, 0)
-	u.AssertNil(t, err)
-	for _, nodeId := range nextNode {
-		node, ok := wflow.Find(nodeId)
-		u.AssertTrue(t, ok)
-		u.AssertEquals(t, nodeId, node.GetId())
-		u.AssertEquals(t, progress.NextGroup, progress.GetGroup(nodeId))
-		err = progress.CompleteNode(nodeId)
-		u.AssertNil(t, err)
-		completedNodes = append(completedNodes, node)
-	}
-	return completedNodes
-}
-
-func finishProgress(t *testing.T, progress *workflow.Progress) {
-	nothing, err := progress.NextNodes()
-	u.AssertNil(t, err)
-	u.AssertEmptySlice(t, nothing)
-}

internal/workflow/choice_task.go

@@ -90,16 +90,15 @@ func (workflow *Workflow) executeChoice(progress *Progress, input *PartialData,
 
 	// for task node, we skip all branch that will not be executed
 	nodesToSkip := task.GetNodesToSkip(workflow, matchedCondition)
-	errSkip := progress.SkipAll(nodesToSkip)
-	if errSkip != nil {
-		return outputData, progress, false, errSkip
+	for _, node := range nodesToSkip {
+		progress.SkipNode(node.GetId())
 	}
 
-	err := progress.CompleteNode(task.GetId())
+	progress.CompleteNode(task.GetId())
+	err := progress.AddReadyTask(nextTaskId)
 	if err != nil {
-		return outputData, progress, false, err
+		return nil, progress, false, err
 	}
-
 	return outputData, progress, true, nil
 }
 

internal/workflow/end_task.go

@@ -42,10 +42,7 @@ func (e *EndNode) Equals(cmp types.Comparable) bool {
 }
 
 func (workflow *Workflow) executeEnd(progress *Progress, partialData *PartialData, node *EndNode, r *Request) (*PartialData, *Progress, bool, error) {
-	err := progress.CompleteNode(node.Id)
-	if err != nil {
-		return partialData, progress, false, err
-	}
+	progress.CompleteNode(node.Id)
 	return partialData, progress, false, nil // false because we want to stop when reaching the end
 }
 

internal/workflow/fail_task.go

@@ -47,10 +47,7 @@ func (workflow *Workflow) executeFail(progress *Progress, task *FailNode, r *Req
 	output[task.Error] = task.Cause
 	outputData := NewPartialData(ReqId(r.Id), forNode, task.GetId(), output)
 
-	err := progress.CompleteNode(task.GetId())
-	if err != nil {
-		return outputData, progress, false, err
-	}
+	progress.CompleteNode(task.GetId())
 
 	shouldContinueExecution := task.GetNodeType() != Fail && task.GetNodeType() != Succeed
 	return outputData, progress, shouldContinueExecution, nil

internal/workflow/fanin_task.go

@@ -27,7 +27,8 @@ func NewFanInNode(fanInDegree int) *FanInNode {
 
 func (workflow *Workflow) executeFanIn(progress *Progress, input *PartialData, task *FanInNode, r *Request) (*PartialData, *Progress, bool, error) {
 	outputData := NewPartialData(ReqId(r.Id), task.GetNext()[0], task.GetId(), input.Data)
-	err := progress.CompleteNode(task.GetId())
+	progress.CompleteNode(task.GetId())
+	err := progress.AddReadyTask(task.GetNext()[0])
 	if err != nil {
 		return nil, progress, false, err
 	}

internal/workflow/fanout_task.go

@@ -62,8 +62,7 @@ func (workflow *Workflow) executeFanOut(progress *Progress, input *PartialData,
 	// input -> output: map["input":1] -> map["0":map["input":1], "1":map["input":1]]
 	if task.Type == Broadcast {
 		for _, nextNode := range task.GetNext() {
-			// TODO: check if a copy of params[0] is needed
-			output[string(nextNode)] = input.Data // simply returns input, that will be copied to each subsequent node
+			output[string(nextNode)] = maps.Clone(input.Data) // simply returns input, that will be copied to each subsequent node
 		}
 	} else if task.Type == Scatter { // scatter only accepts an array with exactly fanOutDegree elements. However, multiple input values are allowed
 		if len(input.Data) != 1 {
@@ -97,10 +96,14 @@ func (workflow *Workflow) executeFanOut(progress *Progress, input *PartialData,
 	outputData := NewPartialData(ReqId(r.Id), "", task.GetId(), output)
 	//newOutputDataMap := make(map[string]interface{})        // TODO: consider using a map of PartialData rather than a single PartialData object
 
-	err := progress.CompleteNode(task.GetId())
-	if err != nil {
-		return nil, progress, false, err
+	progress.CompleteNode(task.GetId())
+	for _, nextNode := range task.GetNext() {
+		err := progress.AddReadyTask(nextNode)
+		if err != nil {
+			return nil, progress, false, err
+		}
 	}
+
 	return outputData, progress, true, nil
 }