Moves execute*() functions into task files

Author: grussorusso

internal/workflow/choice_task.go

@@ -62,6 +62,60 @@ func (c *ChoiceNode) AddOutput(workflow *Workflow, taskId TaskId) error {
 	return nil
 }
 
+func (workflow *Workflow) executeChoice(progress *Progress, input *PartialData, task *ChoiceNode, r *Request) (*PartialData, *Progress, bool, error) {
+
+	outputData := NewPartialData(ReqId(r.Id), "", task.GetId(), nil) // partial initialization of outputData
+
+	// NOTE: we do not call task.CheckInput() as this task has no signature to match against
+
+	// simply evaluate the Conditions and set the matching one
+	matchedCondition := -1
+	for i, condition := range task.Conditions {
+		ok, err := condition.Evaluate(input.Data)
+		if err != nil {
+			return nil, progress, false, fmt.Errorf("error while testing condition: %v", err)
+		}
+		if ok {
+			matchedCondition = i
+			break
+		}
+	}
+
+	if matchedCondition < 0 {
+		return nil, progress, false, fmt.Errorf("no condition is met")
+	}
+
+	nextTaskId := task.Alternatives[matchedCondition]
+	outputData.ForTask = nextTaskId
+	outputData.Data = input.Data
+
+	nextTask, ok := workflow.Find(nextTaskId)
+	if !ok {
+		return nil, progress, false, fmt.Errorf("node not found while preparing output")
+	}
+	switch typedTask := nextTask.(type) {
+	case *SimpleNode:
+		err := typedTask.MapOutput(outputData.Data)
+		if err != nil {
+			return outputData, progress, false, fmt.Errorf("choice task %s cannot prepare the output for simple task: %v", task.String(), err)
+		}
+	}
+
+	// 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
+	}
+
+	err := progress.CompleteNode(task.GetId())
+	if err != nil {
+		return outputData, progress, false, err
+	}
+
+	return outputData, progress, true, nil
+}
+
 // VisitBranch returns all node ids of a branch under a choice node; branch number starts from 0
 func (c *ChoiceNode) VisitBranch(workflow *Workflow, branch int) []Task {
 	branchNodes := make([]Task, 0)

internal/workflow/end_task.go

@@ -41,6 +41,14 @@ func (e *EndNode) Equals(cmp types.Comparable) bool {
 	return e.Id == e2.Id && e.NodeType == e2.NodeType
 }
 
+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
+	}
+	return partialData, progress, false, nil // false because we want to stop when reaching the end
+}
+
 func (e *EndNode) AddOutput(workflow *Workflow, taskId TaskId) error {
 	return nil // should not do anything. End node cannot be chained to anything
 }

internal/workflow/fail_task.go

@@ -42,6 +42,22 @@ func (f *FailNode) Equals(cmp types.Comparable) bool {
 		f.BranchId == f2.BranchId
 }
 
+func (workflow *Workflow) executeFail(progress *Progress, task *FailNode, r *Request) (*PartialData, *Progress, bool, error) {
+
+	forNode := task.GetNext()[0]
+	output := make(map[string]interface{})
+	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
+	}
+
+	shouldContinueExecution := task.GetNodeType() != Fail && task.GetNodeType() != Succeed
+	return outputData, progress, shouldContinueExecution, nil
+}
+
 func (f *FailNode) AddOutput(workflow *Workflow, taskId TaskId) error {
 	_, ok := workflow.Nodes[taskId].(*EndNode)
 	if !ok {

internal/workflow/fanin_task.go

@@ -26,6 +26,16 @@ func NewFanInNode(fanInDegree int) *FanInNode {
 	return &fanIn
 }
 
+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())
+	if err != nil {
+		return nil, progress, false, err
+	}
+
+	return outputData, progress, true, nil
+}
+
 func (f *FanInNode) Equals(cmp types.Comparable) bool {
 	switch f1 := cmp.(type) {
 	case *FanInNode:

internal/workflow/fanout_task.go

@@ -55,46 +55,57 @@ func (f *FanOutNode) Equals(cmp types.Comparable) bool {
 	}
 }
 
-// Exec splits the output for the next parallel dags
+// executeFanOut splits the output for the next parallel dags
 // Scatter mode can only be used if the value held in the map is of type slice. Subdivides each map entry to a different node
 // Broadcast mode can always be used. Copies the entire map to each of the subsequent nodes
-func (f *FanOutNode) Exec(_ *Request, params ...map[string]interface{}) (map[string]interface{}, error) {
+func (workflow *Workflow) executeFanOut(progress *Progress, input *PartialData, task *FanOutNode, r *Request) (*PartialData, *Progress, bool, error) {
 	output := make(map[string]interface{})
 
 	// TODO: avoid forcing the interface implementation, so that the signature of Exec can be adapted
-	if len(params) != 1 {
-		return nil, fmt.Errorf("failed to get one input for choice node: received %d inputs", len(params))
+	if len(input.Data) != 1 {
+		return nil, progress, false, fmt.Errorf("failed to get one input for choice node: received %d inputs", len(input.Data))
 	}
 
 	// input -> output: map["input":1] -> map["0":map["input":1], "1":map["input":1]]
-	if f.Type == Broadcast {
-		for _, nextNode := range f.GetNext() {
+	if task.Type == Broadcast {
+		for _, nextNode := range task.GetNext() {
 			// TODO: check if a copy of params[0] is needed
-			output[string(nextNode)] = params[0] // simply returns input, that will be copied to each subsequent node
+			output[string(nextNode)] = input.Data // simply returns input, that will be copied to each subsequent node
 		}
-	} else if f.Type == Scatter { // scatter only accepts an array with exactly fanOutDegree elements. However, multiple input values are allowed
-		inputName := maps.Keys(params[0])[0]
-		inputToScatter := params[0][inputName]
+	} else if task.Type == Scatter { // scatter only accepts an array with exactly fanOutDegree elements. However, multiple input values are allowed
+		inputName := maps.Keys(input.Data)[0]
+		inputToScatter := input.Data[inputName]
 		inputArrayToScatter, errNotSlice := utils.ConvertToSlice(inputToScatter)
 		if errNotSlice != nil {
-			return nil, fmt.Errorf("cannot convert input %v to slice", inputToScatter)
+			return nil, progress, false, fmt.Errorf("cannot convert input %v to slice", inputToScatter)
 		}
 
-		if len(inputArrayToScatter) != f.FanOutDegree {
-			return nil, fmt.Errorf("input array size (%d) must be equal to fanOutDegree (%d). Check the previous node output",
-				len(inputArrayToScatter), f.FanOutDegree)
+		if len(inputArrayToScatter) != task.FanOutDegree {
+			return nil, progress, false, fmt.Errorf("input array size (%d) must be equal to fanOutDegree (%d). Check the previous node output",
+				len(inputArrayToScatter), task.FanOutDegree)
 		}
 
-		for i, nextNode := range f.GetNext() {
+		for i, nextNode := range task.GetNext() {
 			iOutput := make(map[string]interface{})
 			iOutput[inputName] = inputArrayToScatter[i]
 			output[string(nextNode)] = iOutput
 		}
 	} else {
-		return nil, fmt.Errorf("invalid fanout mode: %d", f.Type)
+		return nil, progress, false, fmt.Errorf("invalid fanout mode: %d", task.Type)
 	}
 
-	return output, nil
+	/* using forNode = "" in order to create a special input to handle fanout
+	 * case with Data field which contains a map[string]interface{} with the key set
+	 * to nodeId and the value which is also a map[string]interface{} containing the
+	 * effective input for the nth-parallel node */
+	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
+	}
+	return outputData, progress, true, nil
 }
 
 func (f *FanOutNode) AddOutput(workflow *Workflow, taskId TaskId) error {

internal/workflow/simple_task.go

@@ -5,12 +5,12 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/labstack/gommon/log"
+	"github.com/lithammer/shortuuid"
 	"github.com/serverledge-faas/serverledge/internal/function"
 	"github.com/serverledge-faas/serverledge/internal/node"
 	"github.com/serverledge-faas/serverledge/internal/scheduling"
 	"github.com/serverledge-faas/serverledge/internal/types"
-	"github.com/labstack/gommon/log"
-	"github.com/lithammer/shortuuid"
 )
 
 // SimpleNode is a Task that receives one input and sends one result
@@ -30,7 +30,34 @@ func NewSimpleNode(f string) *SimpleNode {
 	}
 }
 
-func (s *SimpleNode) Exec(compRequest *Request, params ...map[string]interface{}) (map[string]interface{}, error) {
+func (workflow *Workflow) executeSimple(progress *Progress, input *PartialData, task *SimpleNode, r *Request) (*PartialData, *Progress, bool, error) {
+
+	err := task.CheckInput(input.Data)
+	if err != nil {
+		return nil, progress, false, err
+	}
+	output, err := task.exec(r, input.Data)
+	if err != nil {
+		return nil, progress, false, err
+	}
+
+	errSend := task.PrepareOutput(workflow, output)
+	if errSend != nil {
+		return nil, progress, false, fmt.Errorf("the node %s cannot send the output2: %v", task.String(), errSend)
+	}
+
+	nextTask := task.GetNext()[0]
+	outputData := NewPartialData(ReqId(r.Id), nextTask, task.Id, output)
+
+	err = progress.CompleteNode(task.Id)
+	if err != nil {
+		return outputData, progress, false, err
+	}
+
+	return outputData, progress, true, nil
+}
+
+func (s *SimpleNode) exec(compRequest *Request, params ...map[string]interface{}) (map[string]interface{}, error) {
 	funct, ok := function.GetFunction(s.Func)
 	if !ok {
 		return nil, fmt.Errorf("SimpleNode.function is null: you must initialize SimpleNode's function to execute it")

internal/workflow/start_task.go

@@ -46,6 +46,15 @@ func (s *StartNode) AddOutput(workflow *Workflow, nodeId TaskId) error {
 	return nil
 }
 
+func (workflow *Workflow) executeStart(progress *Progress, partialData *PartialData, node *StartNode) (*PartialData, *Progress, bool, error) {
+
+	err := progress.CompleteNode(node.GetId())
+	if err != nil {
+		return partialData, progress, false, err
+	}
+	return partialData, progress, true, nil
+}
+
 func (s *StartNode) GetNext() []TaskId {
 	// we only have one output
 	return []TaskId{s.Next}