anchor.diff:          end, gb_trees:values(S#state.nodes)),
       disco_proxy:update_nodes(gb_trees:keys(NewNodes)),
       update_nodes(NewNodes),
  -    {noreply, S#state{nodes = NewNodes}};
  -
  -handle_cast(schedule_next, #state{nodes = Nodes, workers = Workers} = S) ->
  -
  +    S#state{nodes = NewNodes}.
  +
  +-spec schedule_next() -> 'ok'.
  +schedule_next() ->
  +    gen_server:cast(?MODULE, schedule_next).
  +
  +-spec do_schedule_next(#state{}) -> #state{}.
  +do_schedule_next(#state{nodes = Nodes, workers = Workers} = S) ->
       {_, AvailableNodes} = lists:unzip(lists:keysort(1, [{Y, N} ||
           #dnode{slots = X, num_running = Y, name = N, blacklisted = false}
                   <- gb_trees:values(Nodes), X > Y])),
  
positive.code:                 M1 = M#dnode{num_running = M#dnode.num_running + 1},
                  UNodes = gb_trees:update(Node, M1, Nodes),
                  S1 = S#state{nodes = UNodes, workers = UWorkers},
                  handle_cast(schedule_next, S1);
              nojobs ->
                  {noreply, S}
          end;
      true -> {noreply, S}
      end;
  
  handle_cast({purge_job, JobName}, #state{purged = Purged} = S) ->
      handle_call({clean_job, JobName}, none, S),
      ddfs:delete(ddfs_master, disco:oob_name(JobName), internal),
      Key = list_to_binary(JobName),
  
negative.code:                 purged = gb_trees:empty()}}.
  
  handle_cast({update_config_table, Config}, S) ->
      {noreply, do_update_config_table(Config, S)};
  
  handle_cast(schedule_next, S) ->
      {noreply, do_schedule_next(S)};
  
  handle_cast({purge_job, JobName}, S) ->
      {noreply, do_purge_job(JobName, S)};
  
  handle_cast({exit_worker, Pid, Res}, S) ->
      {noreply, do_exit_worker(Pid, Res, S)}.
  
  handle_call(dbg_get_state, _, S) ->
      {reply, S, S};
  
  handle_call({new_job, JobName, JobCoord}, _, S) ->
      {reply, catch gen_server:call(scheduler, {new_job, JobName, JobCoord}), S};
  
  handle_call({new_task, Task}, _, S) ->
      NodeStats =
          [case gb_trees:lookup(Node, S#state.nodes) of
              none -> {false, Input};
              {value, N} -> {N#dnode.num_running, Input}
           end || {_Url, Node} = Input <- Task#task.input],
      case catch gen_server:call(scheduler, {new_task, Task, NodeStats}) of
          ok ->
              gen_server:cast(self(), schedule_next),
              {reply, ok, S};
          Error ->
              error_logger:warning_report({"Scheduling task failed",
                                           Task, Error}),
              {reply, failed, S}
      end;
  
  handle_call({get_active, all}, _From, #state{workers = Workers} = S) ->
      {reply, {ok, gb_trees:values(Workers)}, S};
  
  handle_call({get_active, JobName}, _From, #state{workers = Workers} = S) ->
      Active = [{Host, Task} || {Host, #task{jobname = N} = Task}
                                    <- gb_trees:values(Workers), N == JobName],
      {reply, {ok, Active}, S};
  
  handle_call({get_nodeinfo, all}, _From, #state{nodes = Nodes} = S) ->
      Info = [#nodeinfo{name = N#dnode.name,
                        slots = N#dnode.slots,
                        num_running = N#dnode.num_running,
                        stats_ok = N#dnode.stats_ok,
                        stats_failed = N#dnode.stats_failed,
                        stats_crashed = N#dnode.stats_crashed,
                        blacklisted = not (N#dnode.blacklisted == false)}
              || N <- gb_trees:values(Nodes)],
      {reply, {ok, Info}, S};
  
  handle_call(get_purged, _, #state{purged = Purged} = S) ->
      Now = now(),
      NPurged = gb_trees:from_orddict(
          [{Job, TStamp} || {Job, TStamp} <- gb_trees:to_list(Purged),
                            timer:now_diff(Now, TStamp) < ?PURGE_TIMEOUT * 1000]),
      {reply, {ok, gb_trees:keys(NPurged)}, S#state{purged = NPurged}};
  
  handle_call(get_num_cores, _, #state{nodes = Nodes} = S) ->
      NumCores = lists:sum([N#dnode.slots || N <- gb_trees:values(Nodes)]),
      {reply, {ok, NumCores}, S};
  
  handle_call({kill_job, JobName}, _From, S) ->
      event_server:event(JobName, "WARN: Job killed", [], []),
      % Make sure that scheduler don't accept new tasks from this job
      gen_server:cast(scheduler, {job_done, JobName}),
      {reply, ok, S};
  
  handle_call({clean_job, JobName}, From, State) ->
      handle_call({kill_job, JobName}, From, State),
      gen_server:cast(event_server, {clean_job, JobName}),
      {reply, ok, State};
  
  handle_call({blacklist, Node, Token}, _From, #state{nodes = Nodes} = S) ->
      {reply, ok, S#state{nodes = toggle_blacklist(Node, Nodes, true, Token)}};
  
  handle_call({whitelist, Node, Token}, _From, #state{nodes = Nodes} = S) ->
      {reply, ok, S#state{nodes = toggle_blacklist(Node, Nodes, false, Token)}}.
  
  handle_info({'EXIT', Pid, normal}, S) ->
      case gb_trees:lookup(Pid, S#state.workers) of
          none -> {noreply, S};
          _ -> error_logger:warning_report({"Task failed to call exit_worker",
                                            Pid}),
               process_exit(Pid, "Died unexpectedly without a reason",
              "unexpected", S)
      end;
  
  handle_info({'EXIT', Pid, {worker_dies, {Msg, Args}}}, S) ->
      process_exit(Pid, io_lib:fwrite(Msg, Args), "worker_dies", S);
  
  handle_info({'EXIT', Pid, noconnection}, S) ->
      process_exit(Pid, "Connection lost to the node (network busy?)",
                   "noconnection", S);
  
  handle_info({'EXIT', Pid, Reason}, S) when Pid == self() ->
      error_logger:warning_report(["Disco server dies on error!", Reason]),
      {stop, stop_requested, S};
  
  handle_info({'EXIT', Pid, Reason}, S) ->
      process_exit(Pid, io_lib:fwrite("Worked died unexpectedly: ~p", [Reason]),
                   "unexpected", S).
  
  %% ===================================================================
  %% gen_server callback stubs
  
  terminate(Reason, _State) ->
      error_logger:warning_report({"Disco server dies", Reason}).
  
  code_change(_OldVsn, State, _Extra) -> {ok, State}.
  
  %% ===================================================================
  %% internal functions
  
  -spec update_nodes(gb_tree()) -> 'ok'.
  update_nodes(Nodes) ->
      WhiteNodes = [{N#dnode.name, N#dnode.slots}
                    || #dnode{blacklisted = false} = N <- gb_trees:values(Nodes)],
      DDFSNodes = [{disco:node(N#dnode.name), not (N#dnode.blacklisted == false)}
                   || N <- gb_trees:values(Nodes)],
      gen_server:cast(ddfs_master, {update_nodes, DDFSNodes}),
      gen_server:cast(scheduler, {update_nodes, WhiteNodes}),
      gen_server:cast(self(), schedule_next).
  
  -spec update_stats(nonempty_string(), 'none'|{'value', dnode()}, _,
                     #state{}) -> #state{}.
  update_stats(_Node, none, _ReplyType, S) -> S;
  update_stats(Node, {value, N}, ReplyType, S) ->
      M = N#dnode{num_running = N#dnode.num_running - 1},
      M0 = case ReplyType of
               job_ok ->
                   M#dnode{stats_ok = M#dnode.stats_ok + 1};
               data_error ->
                   M#dnode{stats_failed = M#dnode.stats_failed + 1};
               job_error ->
                   M#dnode{stats_crashed = M#dnode.stats_crashed + 1};
               _ ->
                   M#dnode{stats_crashed = M#dnode.stats_crashed + 1}
           end,
      S#state{nodes = gb_trees:update(Node, M0, S#state.nodes)}.
  
  toggle_blacklist(Node, Nodes, IsBlacklisted, Token) ->
      UpdatedNodes =
          case gb_trees:lookup(Node, Nodes) of
              % blacklist
              {value, M} when IsBlacklisted == true,
                              M#dnode.blacklisted =/= manual ->
                  gb_trees:update(Node, M#dnode{blacklisted = Token}, Nodes);
              % whitelist if token is valid
              {value, M} when Token == any;
                              Token == M#dnode.blacklisted ->
                  error_logger:info_report({"Whitelisted", Node}),
                  gb_trees:update(Node, M#dnode{blacklisted = false}, Nodes);
              _ -> Nodes
          end,
      update_nodes(UpdatedNodes),
      UpdatedNodes.
  
  start_worker(Node, NodeMon, T) ->
      event_server:event(T#task.jobname, "~s:~B assigned to ~s",
                         [T#task.mode, T#task.taskid, Node], []),
      spawn_link(disco_worker, start_link_remote,
                 [self(), whereis(event_server), Node, NodeMon, T]).
  
  process_exit(Pid, Msg, Code, S) ->
      process_exit1(gb_trees:lookup(Pid, S#state.workers), Pid, Msg, Code, S).
  
  process_exit1(none, _, _, _, S) -> {noreply, S};
  process_exit1({_, {Node, T}}, Pid, Msg, Code, S) ->
      P = io_lib:fwrite("WARN: [~s:~B] ", [T#task.mode, T#task.taskid]),
      event_server:event(Node, T#task.jobname, lists:flatten(P, Msg), [],
                         {task_failed, T#task.mode}),
      gen_server:cast(self(), {exit_worker, Pid, {data_error, Code}}),
      {noreply, S}.
  
  -spec do_update_config_table([{nonempty_string(), integer()}],
                               #state{}) -> #state{}.
  do_update_config_table(Config, S) ->
      error_logger:info_report([{"Config table update"}]),
      NewNodes =
          lists:foldl(fun({Host, Slots}, NewNodes) ->
  