Fixes ExecReport initialization

Author: grussorusso

internal/api/workflow.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"log"
+	"maps"
 	"net/http"
 	"time"
 
@@ -172,7 +173,7 @@ func ResumeWorkflow(e echo.Context) error {
 	req.Resuming = true
 	req.Id = clientReq.ReqId
 	req.ExecReport.ResponseTime = 0.0
-	req.ExecReport.Result = map[string]interface{}{}
+	req.ExecReport.Result = nil // NOTE: this should be nil until workflow completion
 	req.ExecReport.Reports = map[string]*function.ExecutionReport{}
 
 	if clientReq.Plan.ToExecute != nil {
@@ -217,7 +218,7 @@ func InvokeWorkflow(e echo.Context) error {
 	req.Resuming = false
 	req.Id = fmt.Sprintf("%v-%s%d", wflow.Name, node.LocalNode.String()[len(node.LocalNode.String())-5:], req.Arrival.Nanosecond())
 	req.ExecReport.ResponseTime = 0.0
-	req.ExecReport.Result = map[string]interface{}{}
+	req.ExecReport.Result = nil // NOTE: this should be nil until workflow completion
 	req.ExecReport.Reports = map[string]*function.ExecutionReport{}
 
 	return handleWorkflowInvocation(e, req)
@@ -263,6 +264,9 @@ func handleWorkflowInvocation(e echo.Context, req *workflow.Request) error {
 	} else {
 		req.ExecReport.ResponseTime = time.Now().Sub(req.Arrival).Seconds()
 
+		// TODO
+		log.Printf("Invocation succeeded. Returning %d reports for keys: %v", len(req.ExecReport.Reports), maps.Keys(req.ExecReport.Reports))
+
 		return e.JSON(http.StatusOK, workflow.InvocationResponse{
 			Success:      true,
 			Result:       req.ExecReport.Result,

internal/workflow/ilp_offloading_policy.go

@@ -85,7 +85,14 @@ func (policy *IlpOffloadingPolicy) Evaluate(r *Request, p *Progress) (Offloading
 	cacheSolution(r, &placement, defaultTTL)
 
 	for k, v := range placement {
-		fmt.Printf("Task: %s -> %s \n", k, v)
+		taskType := ""
+		task, found := r.W.Tasks[k]
+		if !found {
+			taskType = "?"
+		} else {
+			taskType = fmt.Sprintf("%v", task.GetType())
+		}
+		fmt.Printf("[Req %s] Task: %s (%v) -> %s \n", r.Id, k, taskType, v)
 	}
 
 	// parse results and make a decision

internal/workflow/remote_offloading_policy.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/serverledge-faas/serverledge/internal/config"
@@ -72,9 +73,13 @@ type cachedPlacement struct {
 	ttl       int
 }
 
+var placementCacheMutex sync.Mutex = sync.Mutex{}
 var placementCache map[string]*cachedPlacement
 
 func getCachedSolution(r *Request) (*taskPlacement, bool) {
+	placementCacheMutex.Lock()
+	defer placementCacheMutex.Unlock()
+
 	if placementCache == nil {
 		placementCache = make(map[string]*cachedPlacement)
 		return nil, false
@@ -96,6 +101,9 @@ func getCachedSolution(r *Request) (*taskPlacement, bool) {
 }
 
 func cacheSolution(r *Request, sol *taskPlacement, ttl int) {
+	placementCacheMutex.Lock()
+	defer placementCacheMutex.Unlock()
+
 	if placementCache == nil {
 		placementCache = make(map[string]*cachedPlacement)
 	}

internal/workflow/workflow.go

@@ -26,7 +26,7 @@ import (
 	"github.com/serverledge-faas/serverledge/internal/types"
 )
 
-var offloadingPolicy OffloadingPolicy = &IlpOffloadingPolicy{}
+var offloadingPolicy OffloadingPolicy = nil
 
 func CreateOffloadingPolicy() {
 	policyConf := config.GetString(config.WORKFLOW_OFFLOADING_POLICY, "disable")
@@ -475,6 +475,7 @@ func (wflow *Workflow) Invoke(r *Request) error {
 			if err != nil {
 				return fmt.Errorf("Could not retrieve progress after offloading: %v", err)
 			}
+
 			log.Printf("Ready to execute after offloading: %v", progress.ReadyToExecute)
 		} else {
 			// pick next executable task
@@ -506,7 +507,6 @@ func (wflow *Workflow) Invoke(r *Request) error {
 
 					input, found = dataMap[previousTask]
 					if !found {
-						log.Printf("Input not found in dataMap for previousTask %s", previousTask)
 						input, err = RetrievePartialData(requestId, previousTask)
 						if err != nil {
 							return fmt.Errorf("could not retrieve partial data: %v", err)