Removed fromTask field in PartialData

Author: grussorusso

internal/test/partial_data_test.go

@@ -16,10 +16,9 @@ func TestPartialDataMarshaling(t *testing.T) {
 	data["num"] = 2
 	data["list"] = []string{"uno", "due", "tre"}
 	partialData := workflow.PartialData{
-		ReqId:    workflow.ReqId("abc"),
-		ForTask:  "fai13p102",
-		FromTask: "120e8d12d",
-		Data:     data,
+		ReqId:   workflow.ReqId("abc"),
+		ForTask: "fai13p102",
+		Data:    data,
 	}
 	marshal, errMarshal := json.Marshal(partialData)
 	u.AssertNilMsg(t, errMarshal, "error during marshaling")
@@ -104,9 +103,8 @@ func TestPartialDataCache(t *testing.T) {
 
 func initPartialData(reqId workflow.ReqId, to, from workflow.TaskId, data map[string]interface{}) *workflow.PartialData {
 	return &workflow.PartialData{
-		ReqId:    reqId,
-		ForTask:  to,
-		FromTask: from,
-		Data:     data,
+		ReqId:   reqId,
+		ForTask: to,
+		Data:    data,
 	}
 }

internal/test/workflow_integration_test.go

@@ -601,7 +601,7 @@ func TestResumeWorkflow(t *testing.T) {
 	request := workflow.NewRequest(shortuuid.New(), wflow, params)
 
 	progress := workflow.InitProgress(workflow.ReqId(request.Id), wflow)
-	pd := workflow.NewPartialData(workflow.ReqId(request.Id), wflow.Start.Id, "", request.Params)
+	pd := workflow.NewPartialData(workflow.ReqId(request.Id), wflow.Start.Id, request.Params)
 
 	err = workflow.SaveProgress(progress, true)
 	u.AssertNil(t, err)

internal/workflow/choice_task.go

@@ -62,7 +62,7 @@ func (c *ChoiceTask) AddOutput(workflow *Workflow, taskId TaskId) error {
 
 func (c *ChoiceTask) execute(progress *Progress, input *PartialData, r *Request) (*PartialData, *Progress, bool, error) {
 
-	outputData := NewPartialData(ReqId(r.Id), "", c.GetId(), nil) // partial initialization of outputData
+	outputData := NewPartialData(ReqId(r.Id), c.GetId(), nil) // partial initialization of outputData
 
 	// NOTE: we do not call task.CheckInput() as this task has no signature to match against
 

internal/workflow/fail_task.go

@@ -44,7 +44,7 @@ func (f *FailureTask) execute(progress *Progress, r *Request) (*PartialData, *Pr
 
 	output := make(map[string]interface{})
 	output[f.Error] = f.Cause
-	outputData := NewPartialData(ReqId(r.Id), f.GetNext()[0], f.GetId(), output)
+	outputData := NewPartialData(ReqId(r.Id), f.GetNext()[0], output)
 
 	progress.Complete(f.GetId())
 

internal/workflow/fanin_task.go

@@ -26,7 +26,7 @@ func NewFanInTask(fanInDegree int) *FanInTask {
 }
 
 func (f *FanInTask) execute(progress *Progress, input *PartialData, r *Request) (*PartialData, *Progress, bool, error) {
-	outputData := NewPartialData(ReqId(r.Id), f.GetNext()[0], f.GetId(), input.Data)
+	outputData := NewPartialData(ReqId(r.Id), f.GetNext()[0], input.Data)
 	progress.Complete(f.GetId())
 	err := progress.AddReadyTask(f.GetNext()[0])
 	if err != nil {

internal/workflow/fanout_task.go

@@ -93,7 +93,8 @@ func (f *FanOutTask) execute(progress *Progress, input *PartialData, r *Request)
 	 * case with Data field which contains a map[string]interface{} with the key set
 	 * to taskId and the value which is also a map[string]interface{} containing the
 	 * effective input for the nth-parallel task */
-	outputData := NewPartialData(ReqId(r.Id), "", f.GetId(), output)
+	// TODO: fix this
+	outputData := NewPartialData(ReqId(r.Id), "", output)
 	//newOutputDataMap := make(map[string]interface{})        // TODO: consider using a map of PartialData rather than a single PartialData object
 
 	progress.Complete(f.GetId())

internal/workflow/partial_data.go

@@ -24,10 +24,9 @@ func newPartialDataId(reqId ReqId) PartialDataId {
 
 // PartialData is saved separately from progressData to avoid cluttering the Progress struct and each Serverledge node's cache
 type PartialData struct {
-	ReqId    ReqId  // request referring to this partial data
-	ForTask  TaskId // task that should receive this partial data
-	FromTask TaskId // useful for fanIn
-	Data     map[string]interface{}
+	ReqId   ReqId  // request referring to this partial data
+	ForTask TaskId // task that should receive this partial data
+	Data    map[string]interface{}
 }
 
 var pdCache = sync.Map{}
@@ -47,24 +46,22 @@ func (pd PartialData) Equals(pd2 *PartialData) bool {
 		}
 	}
 
-	return pd.ReqId == pd2.ReqId && pd.FromTask == pd2.FromTask && pd.ForTask == pd2.ForTask
+	return pd.ReqId == pd2.ReqId && pd.ForTask == pd2.ForTask
 }
 
 func (pd PartialData) String() string {
 	return fmt.Sprintf(`PartialData{
 		Id:    %s,
 		ForTask:  %s,
-		FromTask: %s,
 		Data:     %v,
-	}`, pd.ReqId, pd.ForTask, pd.FromTask, pd.Data)
+	}`, pd.ReqId, pd.ForTask, pd.Data)
 }
 
-func NewPartialData(reqId ReqId, forTask TaskId, fromTask TaskId, data map[string]interface{}) *PartialData {
+func NewPartialData(reqId ReqId, forTask TaskId, data map[string]interface{}) *PartialData {
 	return &PartialData{
-		ReqId:    reqId,
-		ForTask:  forTask,
-		FromTask: fromTask,
-		Data:     data,
+		ReqId:   reqId,
+		ForTask: forTask,
+		Data:    data,
 	}
 }
 

internal/workflow/simple_task.go

@@ -41,7 +41,7 @@ func (s *SimpleTask) execute(progress *Progress, input *PartialData, r *Request)
 	}
 
 	nextTask := s.GetNext()[0]
-	outputData := NewPartialData(ReqId(r.Id), nextTask, s.Id, output)
+	outputData := NewPartialData(ReqId(r.Id), nextTask, output)
 
 	progress.Complete(s.Id)
 	err = progress.AddReadyTask(nextTask)

internal/workflow/workflow.go

@@ -232,16 +232,16 @@ func (workflow *Workflow) executeParallel(progress *Progress, input *PartialData
 		progress.Complete(parallelTasks[i].GetId())
 	}
 
-	outputData := NewPartialData(ReqId(r.Id), "", "", outputMap) // partial initialization of outputData
-	outputData.ForTask = parallelTasks[0].GetNext()[0]           // TODO: we are assuming that the next task is unique for all the parallel tasks (i.e. a FanIn)
+	outputData := NewPartialData(ReqId(r.Id), "", outputMap) // partial initialization of outputData
+	outputData.ForTask = parallelTasks[0].GetNext()[0]       // TODO: we are assuming that the next task is unique for all the parallel tasks (i.e. a FanIn)
 	progress.AddReadyTask(parallelTasks[0].GetNext()[0])
 	return outputData, progress, nil
 }
 
 func (workflow *Workflow) doNothingExec(progress *Progress, input *PartialData, task Task, r *Request) (*PartialData, *Progress, bool, error) {
 
 	output := input.Data
-	outputData := NewPartialData(ReqId(r.Id), task.GetNext()[0], task.GetId(), output)
+	outputData := NewPartialData(ReqId(r.Id), task.GetNext()[0], output)
 
 	progress.Complete(task.GetId())
 
@@ -436,7 +436,7 @@ func (workflow *Workflow) Invoke(r *Request) (ExecutionReport, error) {
 	// TODO: move into a function?
 	if !r.Resuming {
 		progress = InitProgress(requestId, workflow)
-		pd = NewPartialData(requestId, workflow.Start.Id, "", r.Params)
+		pd = NewPartialData(requestId, workflow.Start.Id, r.Params)
 	} else {
 		var found bool
 		progress, found = RetrieveProgress(requestId, true)