Avoids recursive DAG visit

Author: grussorusso

internal/test/workflow_test.go

@@ -448,17 +448,16 @@ func TestVisit(t *testing.T) {
 
 	choice := startNext.GetNext()[0]
 
-	nodeList := make([]workflow.Task, 0)
-	visitedNodes := workflow.Visit(complexWorkflow, complexWorkflow.Start.Id, nodeList, false)
+	visitedNodes := workflow.Visit(complexWorkflow, complexWorkflow.Start.Id, false)
 	u.AssertEquals(t, len(complexWorkflow.Tasks), len(visitedNodes))
 
-	visitedNodes = workflow.Visit(complexWorkflow, complexWorkflow.Start.Id, nodeList, true)
+	visitedNodes = workflow.Visit(complexWorkflow, complexWorkflow.Start.Id, true)
 	u.AssertEquals(t, len(complexWorkflow.Tasks)-1, len(visitedNodes))
 
-	visitedNodes = workflow.Visit(complexWorkflow, choice, nodeList, false)
+	visitedNodes = workflow.Visit(complexWorkflow, choice, false)
 	u.AssertEquals(t, 8, len(visitedNodes))
 
-	visitedNodes = workflow.Visit(complexWorkflow, choice, nodeList, true)
+	visitedNodes = workflow.Visit(complexWorkflow, choice, true)
 	u.AssertEquals(t, 7, len(visitedNodes))
 
 }

internal/workflow/choice_task.go

@@ -75,7 +75,7 @@ func (c *ChoiceTask) VisitBranch(workflow *Workflow, branch int) []Task {
 		return branchTasks
 	}
 	taskId := c.NextTasks[branch]
-	return Visit(workflow, taskId, branchTasks, true)
+	return Visit(workflow, taskId, true)
 }
 
 // GetTasksToSkip skips all tasks that are in a branch that will not be executed.

internal/workflow/workflow.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/serverledge-faas/serverledge/internal/client"
+	"golang.org/x/exp/slices"
 	"io"
 	"net/http"
 	"sort"
@@ -114,89 +115,35 @@ func (w *Workflow) computePreviousTasks() {
 	}
 }
 
-// Visit visits the workflow starting from the given task and return a list of visited tasks.
-// If excludeEnd = true, the EndTask will not be in the output list
-func Visit(workflow *Workflow, taskId TaskId, tasks []Task, excludeEnd bool) []Task {
+func Visit(workflow *Workflow, taskId TaskId, excludeEnd bool) []Task {
+
 	task, ok := workflow.Find(taskId)
 	if !ok {
 		return []Task{}
 	}
-	if !isTaskPresent(task, tasks) {
+
+	tasks := make([]Task, 0)
+	visited := make(map[TaskId]bool)
+	toVisit := []Task{task}
+
+	for len(toVisit) > 0 {
+		task := toVisit[0]
 		tasks = append(tasks, task)
-	}
-	switch n := task.(type) {
-	case *StartTask:
-		toAdd := Visit(workflow, n.GetNext()[0], tasks, excludeEnd)
-		for _, add := range toAdd {
-			if !isTaskPresent(add, tasks) {
-				// only when isEndTask = true, excludeEnd = true -> we don't add the task
-				if !isEndTask(add) || !excludeEnd {
-					tasks = append(tasks, add)
-				}
-			}
-		}
-		return tasks
-	case *SimpleTask, *PassTask, *SuccessTask, *FailureTask:
-		toAdd := Visit(workflow, n.GetNext()[0], tasks, excludeEnd)
-		for _, add := range toAdd {
-			if !isTaskPresent(add, tasks) {
-				if !isEndTask(add) || !excludeEnd {
-					tasks = append(tasks, add)
-				}
-			}
-		}
-		return tasks
-	case *EndTask:
-		if !excludeEnd { // move end task to the end of the visit list
-			endTask := n
-			// get index of end task to remove\
-			indexToRemove := -1
-			for i, task := range tasks {
-				if isEndTask(task) {
-					indexToRemove = i
-					break
-				}
-			}
-			// remove end task
-			tasks = append(tasks[:indexToRemove], tasks[indexToRemove+1:]...)
-			// append at the end of the visited task list
-			tasks = append(tasks, endTask)
-		}
-		return tasks
-	case *ChoiceTask:
-		for _, alternative := range n.GetNext() {
-			toAdd := Visit(workflow, alternative, tasks, excludeEnd)
-			for _, add := range toAdd {
-				if !isTaskPresent(add, tasks) {
-					if !isEndTask(add) || !excludeEnd {
-						tasks = append(tasks, add)
-					}
-				}
-			}
-		}
-		return tasks
-	case *FanOutTask:
-		for _, parallelBranch := range n.GetNext() {
-			toAdd := Visit(workflow, parallelBranch, tasks, excludeEnd)
-			for _, add := range toAdd {
-				if !isTaskPresent(add, tasks) {
-					if !isEndTask(add) || !excludeEnd {
-						tasks = append(tasks, add)
+		toVisit = toVisit[1:]
+		visited[task.GetId()] = true
+
+		for _, nextTask := range task.GetNext() {
+			if _, ok := visited[nextTask]; !ok {
+				nt := workflow.Tasks[nextTask]
+				if !excludeEnd || nt.GetType() != End {
+					if !slices.Contains(toVisit, nt) {
+						toVisit = append(toVisit, nt)
 					}
 				}
 			}
 		}
-		return tasks
-	case *FanInTask:
-		toAdd := Visit(workflow, n.GetNext()[0], tasks, excludeEnd)
-		for _, add := range toAdd {
-			if !isTaskPresent(add, tasks) {
-				if !isEndTask(add) || !excludeEnd {
-					tasks = append(tasks, add)
-				}
-			}
-		}
 	}
+
 	return tasks
 }