anchor.diff:      end.
   
   -spec setup_stage_tasks([{task_id(), [task_output()]}], stage_name(), label_grouping(),
  -                        boolean(), string(), state()) -> {[task_id()], state()}.
  -setup_stage_tasks(PrevStageOutputs, Stage, Grouping, SaveOutputs, SaveInfo, S) ->
  +                        state()) -> {[task_id()], state()}.
  +setup_stage_tasks(PrevStageOutputs, Stage, Grouping, S) ->
       % The outputs of the previous stage are grouped in the specified
       % way, and these grouped outputs form the inputs for the current
       % stage.
       GOutputs = pipeline_utils:group_outputs(Grouping, PrevStageOutputs),
  -    make_stage_tasks(Stage, Grouping, SaveOutputs, SaveInfo, GOutputs, S, {0, []}).
  -
  -make_stage_tasks(Stage, _Grouping, _SaveOutputs, _SaveInfo, [],
  +    make_stage_tasks(Stage, Grouping, GOutputs, S, {0, []}).
  +
  +make_stage_tasks(Stage, _Grouping, [],
                    #state{stage_info = SI} = S, {TaskNum, Tasks}) ->
       StageInfo = #stage_info{start = now(), all = TaskNum},
       SI1 = jc_utils:update_stage(Stage, StageInfo, SI),
       {Tasks, S#state{stage_info = SI1}};
  -make_stage_tasks(Stage, Grouping, SaveOutputs, SaveInfo, [{G, Inputs}|Rest],
  +make_stage_tasks(Stage, Grouping, [{G, Inputs}|Rest],
                    #state{jobinfo = #jobinfo{jobname = JN,
                                              jobenvs = JE,
  +                                           save_info = SaveInfo,
  +                                           save_results = Save,
                                              worker  = W},
  +                        pipeline   = P,
                           tasks = Tasks,
                           schedule    = Schedule,
                           next_taskid = NextTaskId,
  
positive.code:                         jc_utils:add_input(InputId, DInfo, DM)
                  end, OldDataMap, Inputs),
      {InputIds, _DataInputs} = lists:unzip(Inputs),
      TaskSpec = #task_spec{jobname = JN,
                            stage   = Stage,
                            taskid  = NextTaskId,
  
negative.code:             % we need to start the tasks in the next stage.
              case jc_utils:stage_info_opt(Next, SI) of
                  none ->
                      PrevStageOutputs = stage_outputs(Stage, S),
                      start_next_stage(PrevStageOutputs, Next, Grouping, S);
                  _ ->
                      S
              end
      end.
  start_next_stage(PrevStageOutputs, Stage, Grouping,
                   #state{jobinfo = #jobinfo{jobname = JobName}} = S) ->
      {Tasks, S1} = setup_stage_tasks(PrevStageOutputs, Stage, Grouping, S),
      case Tasks of
          [] ->
              do_stage_done(Stage, S1);
  