WIP: introduce workflow offloading

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

Author: matnar

internal/workflow/request.go

@@ -3,7 +3,6 @@ package workflow
 import (
 	"time"
 
-	"github.com/cornelk/hashmap"
 	"github.com/serverledge-faas/serverledge/internal/function"
 )
 
@@ -19,6 +18,7 @@ type Request struct {
 	QoS             function.RequestQoS // every function should have its QoS
 	CanDoOffloading bool                // every function inherits this flag
 	Async           bool
+	Resuming        bool // indicating whether the function is resuming from a previous (partial) execution
 }
 
 func NewRequest(reqId string, workflow *Workflow, params map[string]interface{}) *Request {
@@ -28,10 +28,11 @@ func NewRequest(reqId string, workflow *Workflow, params map[string]interface{})
 		Params:  params,
 		Arrival: time.Now(),
 		ExecReport: ExecutionReport{
-			Reports: hashmap.New[ExecutionReportId, *function.ExecutionReport](), // make(map[ExecutionReportId]*function.ExecutionReport),
+			Reports: map[string]*function.ExecutionReport{},
 		},
 		CanDoOffloading: true,
 		Async:           false,
+		Resuming:        false,
 	}
 }
 

internal/workflow/workflow.go

@@ -428,11 +428,37 @@ func (workflow *Workflow) Invoke(r *Request) (ExecutionReport, error) {
 	var err error
 	requestId := ReqId(r.Id)
 
-	progress := InitProgress(requestId, workflow)
-	pd := NewPartialData(requestId, workflow.Start.Next, "", r.Params)
+	var progress *Progress
+	var pd *PartialData
+
+	// TODO: move into a function?
+	if !r.Resuming {
+		progress = InitProgress(requestId, workflow)
+		pd = NewPartialData(requestId, workflow.Start.Next, "", r.Params)
+	} else {
+		var found bool
+		progress, found = RetrieveProgress(requestId, true)
+		if !found {
+			return ExecutionReport{}, fmt.Errorf("failed to retrieve workflow progress: %v", requestId)
+		}
+		if len(progress.ReadyToExecute) == 0 {
+			return ExecutionReport{}, fmt.Errorf("workflow resumed but no task is ready for execution: %v", requestId)
+		} else if len(progress.ReadyToExecute) > 1 {
+			// TODO: manage case when len is > 1 (e.g., parallel branches)
+			return ExecutionReport{}, fmt.Errorf("workflow resumed with multiple tasks ready for execution not yet implemented!: %v", requestId)
+		}
+
+		pd, err = RetrieveSinglePartialData(requestId, progress.ReadyToExecute[0], true)
+		if err != nil {
+			return ExecutionReport{}, fmt.Errorf("workflow resumed but unable to retrieve partial data of next task: %v", requestId)
+		}
+	}
 
 	shouldContinue := true
 	for shouldContinue {
+		// TODO: introduce a workflow offloading policy
+
+		// TODO: if local execution:
 		// executing workflow
 		pd, progress, shouldContinue, err = workflow.Execute(r, pd, progress)
 		if err != nil {