Method AddOneToOneStep

flow/context.go:40–59

the tasks should run on the source dataset shard

(input *Dataset, output *Dataset)

// the tasks should run on the source dataset shard
func (f *FlowContext) AddOneToOneStep(input *Dataset, output *Dataset) (step *Step) {
	step = f.NewStep()
	FromStepToDataset(step, output)
	FromDatasetToStep(input, step)

	// setup the network
	if output != nil && len(output.ExternalInputChans) > 0 {
		task := step.NewTask()
		FromTaskToDatasetShard(task, output.GetShards()[0])
	} else {
		for i, shard := range input.GetShards() {
			task := step.NewTask()
			if output != nil && output.Shards != nil {
				FromTaskToDatasetShard(task, output.GetShards()[i])
			}
			FromDatasetShardToTask(shard, task)
		}
	}
	return
}

// the task should run on the destination dataset shard
func (f *FlowContext) AddAllToOneStep(input *Dataset, output *Dataset) (step *Step) {