Code review: report and workflow invocation

Co-authored-by: Gabriele Russo Russo <gabri.russo17@gmail.com>

Author: matnar

internal/api/workflow.go

@@ -9,7 +9,6 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/cornelk/hashmap"
 	"github.com/labstack/echo/v4"
 	"github.com/labstack/gommon/log"
 	"github.com/serverledge-faas/serverledge/internal/client"
@@ -160,7 +159,6 @@ func InvokeWorkflow(e echo.Context) error {
 	}
 
 	req := workflowInvocationRequestPool.Get().(*workflow.Request)
-	defer workflowInvocationRequestPool.Put(req) // at the end of the function, the function.NewRequest is added to the pool.
 	req.W = wflow
 	req.Params = clientReq.Params
 	req.Arrival = time.Now()
@@ -169,38 +167,49 @@ 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())
-	req.ExecReport.Reports = hashmap.New[workflow.ExecutionReportId, *function.ExecutionReport]() // make(map[workflow.ExecutionReportId]*function.ExecutionReport)
+	req.ExecReport.Reports = map[string]*function.ExecutionReport{}
 
 	if req.Async {
-		go workflow.SubmitAsyncWorkflowInvocationRequest(req)
+		go func() {
+			_, errInvoke := req.W.Invoke(req)
+
+			defer workflowInvocationRequestPool.Put(req)
+
+			if errInvoke != nil {
+				log.Printf("Invocation failed: %v", errInvoke)
+				workflow.PublishAsyncInvocationResponse(req.Id, workflow.InvocationResponse{Success: false})
+				return
+			}
+
+			req.ExecReport.ResponseTime = time.Now().Sub(req.Arrival).Seconds()
+			workflow.PublishAsyncInvocationResponse(req.Id, workflow.InvocationResponse{
+				Success:      true,
+				Result:       req.ExecReport.Result,
+				Reports:      req.ExecReport.Reports,
+				ResponseTime: req.ExecReport.ResponseTime,
+			})
+		}()
+
 		return e.JSON(http.StatusOK, function.AsyncResponse{ReqId: req.Id})
 	}
 
-	err = workflow.SubmitWorkflowInvocationRequest(req)
+	// Synchronous execution of the workflow
+	_, err = req.W.Invoke(req)
+
+	defer workflowInvocationRequestPool.Put(req)
 
 	if errors.Is(err, node.OutOfResourcesErr) {
 		return e.String(http.StatusTooManyRequests, "")
 	} else if err != nil {
 		log.Printf("Invocation failed: %v", err)
-		v := struct {
-			Error    string
-			Progress string
-		}{
-			Error:    err.Error(),
-			Progress: req.ExecReport.Progress.PrettyString(),
-		}
-		return e.JSON(http.StatusInternalServerError, v)
+		return e.JSON(http.StatusInternalServerError, err.Error())
 	} else {
-		reports := make(map[string]*function.ExecutionReport)
-		req.ExecReport.Reports.Range(func(id workflow.ExecutionReportId, report *function.ExecutionReport) bool {
-			reports[string(id)] = report
-			return true
-		})
+		req.ExecReport.ResponseTime = time.Now().Sub(req.Arrival).Seconds()
 
 		return e.JSON(http.StatusOK, workflow.InvocationResponse{
 			Success:      true,
 			Result:       req.ExecReport.Result,
-			Reports:      reports,
+			Reports:      req.ExecReport.Reports,
 			ResponseTime: req.ExecReport.ResponseTime,
 		})
 	}

internal/workflow/report.go

@@ -2,49 +2,38 @@ package workflow
 
 import (
 	"fmt"
-	"github.com/cornelk/hashmap"
 	"github.com/serverledge-faas/serverledge/internal/function"
 )
 
-type ExecutionReportId string
-
-func CreateExecutionReportId(task Task) ExecutionReportId {
-	return ExecutionReportId(printType(task.GetType()) + "_" + string(task.GetId()))
+func CreateExecutionReportId(task Task) string {
+	return printType(task.GetType()) + "_" + string(task.GetId())
 }
 
 type ExecutionReport struct {
 	Result       map[string]interface{}
-	Reports      *hashmap.Map[ExecutionReportId, *function.ExecutionReport]
-	ResponseTime float64   // time waited by the user to get the output of the entire workflow
-	Progress     *Progress `json:"-"` // skipped in Json marshaling
+	Reports      map[string]*function.ExecutionReport
+	ResponseTime float64 // time waited by the user to get the output of the entire workflow
 }
 
 func (cer *ExecutionReport) String() string {
 	str := "["
 	str += fmt.Sprintf("\n\tResponseTime: %f,", cer.ResponseTime)
-	str += "\n\tReports: ["
-	if cer.Reports.Len() > 0 {
-		j := 0
-		cer.Reports.Range(func(id ExecutionReportId, report *function.ExecutionReport) bool {
-			schedAction := "''"
-			if report.SchedAction != "" {
-				schedAction = report.SchedAction
-			}
-			output := "''"
-			if report.Output != "" {
-				output = report.Output
-			}
+	str += "\n\tReports: "
+
+	for id, report := range cer.Reports {
+		schedAction := "''"
+		if report.SchedAction != "" {
+			schedAction = report.SchedAction
+		}
+		output := "''"
+		if report.Output != "" {
+			output = report.Output
+		}
 
-			str += fmt.Sprintf("\n\t\t%s: {ResponseTime: %f, IsWarmStart: %v, InitTime: %f, OffloadLatency: %f, Duration: %f, SchedAction: %v, Output: %s, Result: %s}", id, report.ResponseTime, report.IsWarmStart, report.InitTime, report.OffloadLatency, report.Duration, schedAction, output, report.Result)
-			if j < cer.Reports.Len()-1 {
-				str += ","
-			}
-			if j == cer.Reports.Len()-1 {
-				str += "\n\t]"
-			}
-			j++
-			return true
-		})
+		str += fmt.Sprintf("\n\t\t%s: {ResponseTime: %f, IsWarmStart: %v, InitTime: %f, OffloadLatency: %f,"+
+			" Duration: %f, SchedAction: %v, Output: %s, Result: %s}", id, report.ResponseTime, report.IsWarmStart,
+			report.InitTime, report.OffloadLatency, report.Duration, schedAction, output, report.Result)
+		str += ",\n"
 	}
 
 	str += "\n\tResult: {"

internal/workflow/scheduler.go

@@ -103,11 +103,10 @@ func (s *SimpleTask) exec(compRequest *Request, params ...map[string]interface{}
 				return nil, fmt.Errorf("output type checking failed: %v", err)
 			}
 		}
-
 	}
 
 	// saving execution report for this function
-	compRequest.ExecReport.Reports.Set(CreateExecutionReportId(s), &report)
+	compRequest.ExecReport.Reports[CreateExecutionReportId(s)] = &report
 
 	return outputData, nil
 }

internal/workflow/simple_task.go

@@ -297,7 +297,6 @@ func (workflow *Workflow) Execute(r *Request, input *PartialData, progress *Prog
 		}
 		if err != nil {
 			progress.Fail(n.GetId())
-			r.ExecReport.Progress = progress
 			return output, progress, false, err
 		}
 	} else {
@@ -437,12 +436,13 @@ func (workflow *Workflow) Invoke(r *Request) (ExecutionReport, error) {
 		// executing workflow
 		pd, progress, shouldContinue, err = workflow.Execute(r, pd, progress)
 		if err != nil {
-			return ExecutionReport{Result: nil, Progress: progress}, fmt.Errorf("failed workflow execution: %v", err)
+			return ExecutionReport{}, fmt.Errorf("failed workflow execution: %v", err)
 		}
 	}
 
 	r.ExecReport.Result = pd.Data
 
+	// TODO: remove r.ExecReport
 	return r.ExecReport, nil
 }