anchor.diff:                  hosts      = gb_sets:empty()  :: disco_gbset(host()),
                   tasks      = gb_trees:empty() :: disco_gbtree(task_id(), task_info()),
                   data_map   = gb_trees:empty() :: disco_gbtree(input_id(), data_info()),
  -                pending    = []               :: [{task_id(), submit_mode()}],
  +                pending    = gb_sets:empty()  :: disco_gbset({task_id(), submit_mode()}),
                   stage_info = gb_trees:empty() :: disco_gbtree(stage_name(), stage_info())}).
   -type state() :: #state{}.
   
  
positive.code:             TInfo1 = TInfo#task_info{failed_count = FC,
                                       failed_hosts = Blacklist},
              S#state{tasks = jc_utils:update_task_info(TaskId, TInfo1, Tasks),
                      pending = [{TaskId, re_run}|Pending]}
      end.
  
  -spec do_kill_job(term(), state()) -> ok.
  
negative.code:     #task_spec{stage = Stage} = TaskSpec,
      case jc_utils:new_task_permitted_for_stage(P, Stage, SI) of
          false ->
              do_submit_tasks(Mode, Rest,
                              S#state{pending = gb_sets:add_element({TaskId, Mode}, Pending)},
                              NFailuresAllowed);
          true ->
              NewPending = case gb_sets:is_element({TaskId, Mode}, Pending) of
                  true ->
                      gb_sets:del_element({TaskId, Mode}, Pending);
                  false -> Pending
              end,
              S1 = S#state{pending = NewPending},
              do_submit_tasks_in(Mode, [TaskId|Rest], S1, NFailuresAllowed)
      end.
  
  