Make sure that FanIn returns output in consistent order

Author: grussorusso

internal/api/workflow.go

@@ -10,12 +10,12 @@ import (
 	"time"
 
 	"github.com/cornelk/hashmap"
+	"github.com/labstack/echo/v4"
+	"github.com/labstack/gommon/log"
 	"github.com/serverledge-faas/serverledge/internal/client"
 	"github.com/serverledge-faas/serverledge/internal/function"
 	"github.com/serverledge-faas/serverledge/internal/node"
 	"github.com/serverledge-faas/serverledge/internal/workflow"
-	"github.com/labstack/echo/v4"
-	"github.com/labstack/gommon/log"
 )
 
 func CreateWorkflowFromASL(e echo.Context) error {
@@ -169,24 +169,13 @@ func InvokeWorkflow(e echo.Context) error {
 	req.Async = clientReq.Async
 
 	req.Id = fmt.Sprintf("%v-%s%d", wflow.Name, node.NodeIdentifier[len(node.NodeIdentifier)-5:], req.Arrival.Nanosecond())
-
-	// TODO: do we really need to initialize all the reports at this point?
 	req.ExecReport.Reports = hashmap.New[workflow.ExecutionReportId, *function.ExecutionReport]() // make(map[workflow.ExecutionReportId]*function.ExecutionReport)
-	for nodeId := range wflow.Nodes {
-		task := wflow.Nodes[nodeId]
-		execReportId := workflow.CreateExecutionReportId(task)
-		req.ExecReport.Reports.Set(execReportId, &function.ExecutionReport{
-			OffloadLatency: 0,
-			SchedAction:    "",
-		})
-	}
 
 	if req.Async {
 		go workflow.SubmitAsyncWorkflowInvocationRequest(req)
 		return e.JSON(http.StatusOK, function.AsyncResponse{ReqId: req.Id})
 	}
 
-	// sync execution
 	err = workflow.SubmitWorkflowInvocationRequest(req)
 
 	if errors.Is(err, node.OutOfResourcesErr) {

internal/workflow/fanin_task.go

@@ -35,17 +35,6 @@ func (f *FanInNode) Equals(cmp types.Comparable) bool {
 	}
 }
 
-// Exec already have all inputs when executing, so it simply merges them with the chosen policy
-func (f *FanInNode) Exec(_ *Request, params ...map[string]interface{}) (map[string]interface{}, error) {
-	output := make(map[string]interface{})
-
-	for i, inputMap := range params {
-		output[fmt.Sprintf("%d", i)] = inputMap
-	}
-
-	return output, nil
-}
-
 func (f *FanInNode) AddOutput(workflow *Workflow, taskId TaskId) error {
 	f.OutputTo = taskId
 	return nil

internal/workflow/workflow.go

@@ -10,9 +10,9 @@ import (
 	"strings"
 	"time"
 
+	"github.com/labstack/gommon/log"
 	"github.com/serverledge-faas/serverledge/internal/cache"
 	"github.com/serverledge-faas/serverledge/utils"
-	"github.com/labstack/gommon/log"
 
 	"github.com/serverledge-faas/serverledge/internal/asl"
 	"github.com/serverledge-faas/serverledge/internal/function"
@@ -406,8 +406,7 @@ func (workflow *Workflow) executeParallel(progress *Progress, input *PartialData
 
 	for i, outChan := range outputChannels {
 		out := <-outChan
-		task := parallelTasks[i]
-		outputMap[fmt.Sprintf("%s", task.GetId())] = out
+		outputMap[fmt.Sprintf("%d", i)] = out
 		err := progress.CompleteNode(parallelTasks[i].GetId())
 		if err != nil {
 			return nil, progress, err
@@ -424,20 +423,8 @@ func (workflow *Workflow) executeParallel(progress *Progress, input *PartialData
 }
 
 func (workflow *Workflow) executeFanIn(progress *Progress, input *PartialData, task *FanInNode, r *Request) (*PartialData, *Progress, bool, error) {
-	inputs := make([]map[string]interface{}, task.FanInDegree)
-	i := 0
-	for _, inputMap := range input.Data {
-		inputs[i] = inputMap.(map[string]interface{})
-		i++
-	}
-
-	output, err := task.Exec(r, inputs...)
-	if err != nil {
-		return nil, progress, false, err
-	}
-
-	outputData := NewPartialData(ReqId(r.Id), task.GetNext()[0], task.GetId(), output)
-	err = progress.CompleteNode(task.GetId())
+	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
 	}