anchor.diff:  kill_job(Reason) ->
       gen_server:cast(self(), {kill_job, Reason}).
   
  +-spec use_inputs(pid(), [task_output()]) -> ok.
  +use_inputs(Coord, Inputs) ->
  +    gen_server:cast(Coord, {inputs, Inputs}).
  +
   -spec stage_done(stage_name()) -> ok.
   stage_done(Stage) ->
       gen_server:cast(self(), {stage_done, Stage}).
  
positive.code:     {JobInfo#jobinfo{jobname = JobName, jobfile = JobFile}, Hosts}.
  
  -spec init_state(jobinfo(), [host()]) -> state().
  init_state(#jobinfo{schedule = Schedule,
                      inputs   = Inputs,
                      pipeline = Pipeline} = JobInfo, Hosts) ->
      % Create a dummy completed 'input' task.
      Tasks = gb_trees:from_orddict([{input,
                                      #task_info{spec = input,
                                                 outputs = Inputs}}]),
      % Mark the 'input' stage as done, and send notification.
      InputStage = #stage_info{all = 1, done = [input]},
      SI = gb_trees:from_orddict([{?INPUT, InputStage}]),
      stage_done(?INPUT),
      #state{jobinfo    = JobInfo,
             pipeline   = Pipeline,
             schedule   = Schedule,
             hosts      = gb_sets:from_list(Hosts),
             tasks      = Tasks,
             stage_info = SI}.
  
  %% ===================================================================
  %% state access and update utils
  
  -spec stage_outputs(stage_name(), state()) -> [{task_id(), [task_output()]}].
  stage_outputs(Stage, #state{stage_info = SI, tasks = Tasks}) ->
      Done = (jc_utils:stage_info(Stage, SI))#stage_info.done,
      [{Id, jc_utils:task_outputs(Id, Tasks)} || Id <- Done].
  
  %% ===================================================================
  %% Callback implementations.
  
  -spec do_update_nodes([host()], state()) -> state().
  do_update_nodes(Hosts, S) ->
  
negative.code:                  ({kill_job, term()}, state()) -> gs_noreply().
  handle_cast({update_nodes, Hosts}, S) ->
      {noreply, do_update_nodes(Hosts, S)};
  handle_cast({inputs, Inputs}, S) ->
      {noreply, do_use_inputs(Inputs, S)};
  handle_cast({stage_done, Stage}, S) ->
      {noreply, do_stage_done(Stage, S)};
  handle_cast({submit_tasks, Mode, Tasks}, S) ->
      {noreply, do_submit_tasks(Mode, Tasks, S)};
  handle_cast({task_done, TaskId, Host, Results}, S) ->
      {noreply, do_task_done(TaskId, Host, Results, S)};
  handle_cast(pipeline_done, S) ->
  