anchor.diff:      {reply, {ok, Nodes}, S};
   
   handle_call(get_read_nodes, _F, #state{nodes = Nodes, read_blacklist = RB} = S) ->
  -    {ReadNodes, NumUnreadableNodes} = get_readable_nodes(Nodes, RB),
  -    {reply, {ok, ReadNodes, NumUnreadableNodes}, S};
  +    {reply, do_get_readable_nodes(Nodes, RB), S};
   
   handle_call({choose_write_nodes, K, Exclude}, _, #state{write_blacklist = BL} = S) ->
  +    {reply, do_choose_write_nodes(S#state.nodes, K, Exclude, BL), S};
  +
  +handle_call({new_blob, Obj, K, Exclude}, _, #state{nodes = Nodes,
  +                                                   write_blacklist = BL} = S) ->
  +    {reply, do_new_blob(Obj, K, Exclude, BL, Nodes), S};
  +
  +handle_call({tag, _M, _Tag}, _From, #state{nodes = []} = S) ->
  +    {reply, {error, no_nodes}, S};
  +
  +handle_call({tag, M, Tag}, From, S) ->
  +    {noreply, do_tag_request(M, Tag, From, S)};
  +
  +handle_call({get_tags, Mode}, From, #state{nodes = Nodes} = S) ->
  +    spawn(fun() ->
  +        gen_server:reply(From, get_tags(Mode, [N || {N, _} <- Nodes]))
  +    end),
  +    {noreply, S}.
  +
  +handle_cast({update_tag_cache, TagCache}, S) ->
  +    {noreply, S#state{tag_cache = TagCache}};
  +
  +handle_cast({update_nodes, NewNodes}, S) ->
  +    {noreply, do_update_nodes(NewNodes, S)};
  +
  +handle_cast({update_nodestats, NewNodes}, S) ->
  +    {noreply, do_update_nodestats(NewNodes, S)}.
  +
  +handle_info({'DOWN', _, _, Pid, _}, S) ->
  +    {noreply, do_tag_exit(Pid, S)}.
  +
  +%% ===================================================================
  +%% gen_server callback stubs
  +
  +terminate(Reason, _State) ->
  +    error_logger:warning_report({"DDFS master dies", Reason}).
  +
  +code_change(_OldVsn, State, _Extra) -> {ok, State}.
  +
  +%% ===================================================================
  +%% internal functions
  +
  +do_get_readable_nodes(Nodes, ReadBlacklist) ->
  +    NodeSet = gb_sets:from_ordset(lists:sort([Node || {Node, _} <- Nodes])),
  +    BlackSet = gb_sets:from_ordset(ReadBlacklist),
  +    ReadableNodeSet = gb_sets:subtract(NodeSet, BlackSet),
  +    {ok, gb_sets:to_list(ReadableNodeSet), gb_sets:size(BlackSet)}.
  +
  +do_choose_write_nodes(Nodes, K, Exclude, BlackList) ->
       % Node selection algorithm:
       % 1. try to choose K nodes randomly from all the nodes which have
       %    more than ?MIN_FREE_SPACE bytes free space available and which
       %    are not excluded or blacklisted.
       % 2. if K nodes cannot be found this way, choose the K emptiest
       %    nodes which are not excluded or blacklisted.
  -    Primary = [Node || {Node, {Free, _Total}} <- S#state.nodes,
  -                Free > ?MIN_FREE_SPACE / 1024] -- (Exclude ++ BL),
  +    Primary = ([N || {N, {Free, _Total}} <- Nodes, Free > ?MIN_FREE_SPACE / 1024]
  +               -- (Exclude ++ BlackList)),
       if length(Primary) >= K ->
  -        {reply, {ok, ddfs_util:choose_random(Primary, K)}, S};
  -    true ->
  -        Secondary = lists:sublist([N || {N, _} <- lists:reverse(
  -            lists:keysort(2, S#state.nodes))] -- (Exclude ++ BL), K),
  -        {reply, {ok, Secondary}, S}
  -    end;
  -
  -handle_call({new_blob, _, K, _}, _, #state{nodes = N} = S) when K > length(N) ->
  -    {reply, too_many_replicas, S};
  -
  -handle_call({new_blob, Obj, K, Exclude}, _, S) ->
  -    {_, {ok, Nodes}, _} = handle_call({choose_write_nodes, K, Exclude}, none, S),
  -    Urls = [["http://", disco:host(Node), ":", get(put_port), "/ddfs/", Obj]
  -            || Node <- Nodes],
  -    {reply, {ok, Urls}, S};
  +            {ok, ddfs_util:choose_random(Primary, K)};
  +       true ->
  +            Preferred = [N || {N, _} <- lists:reverse(lists:keysort(2, Nodes))],
  +            Secondary = lists:sublist(Preferred -- (Exclude ++ BlackList), K),
  +            {ok, Secondary}
  +    end.
  +
  +do_new_blob(_Obj, K, _Exclude, _BlackList, Nodes) when K > length(Nodes) ->
  +    too_many_replicas;
  +do_new_blob(Obj, K, Exclude, BlackList, Nodes) ->
  +    {ok, WriteNodes} = do_choose_write_nodes(Nodes, K, Exclude, BlackList),
  +    Urls = [["http://", disco:host(N), ":", get(put_port), "/ddfs/", Obj]
  +            || N <- WriteNodes],
  +    {ok, Urls}.
   
   % Tag request: Start a new tag server if one doesn't exist already. Forward
   % the request to the tag server.
  -handle_call({tag, _M, _Tag}, _From, #state{nodes = []} = S) ->
  -    {reply, {error, no_nodes}, S};
  -
  -handle_call({tag, M, Tag}, From, #state{tags = Tags, tag_cache = Cache} = S) ->
  +do_tag_request(M, Tag, From, #state{tags = Tags, tag_cache = Cache} = S) ->
       {Pid, TagsN} =
           case gb_trees:lookup(Tag, Tags) of
               none ->
  -                NotFound = Cache =/= false andalso
  -                           not gb_sets:is_element(Tag, Cache),
  +                NotFound = (Cache =/= false
  +                            andalso not gb_sets:is_element(Tag, Cache)),
                   {ok, Server} = ddfs_tag:start(Tag, NotFound),
                   erlang:monitor(process, Server),
                   {Server, gb_trees:insert(Tag, Server, Tags)};
  
positive.code:                 {P, Tags}
          end,
      gen_server:cast(Pid, {M, From}),
      {noreply, S#state{tags = TagsN,
          tag_cache = Cache =/= false andalso gb_sets:add(Tag, Cache)}};
  
  handle_call({get_tags, Mode}, From, #state{nodes = Nodes} = S) ->
      spawn(fun() ->
          gen_server:reply(From, get_tags(Mode, [N || {N, _} <- Nodes]))
      end),
      {noreply, S}.
  
  handle_cast({update_tag_cache, TagCache}, S) ->
      {noreply, S#state{tag_cache = TagCache}};
  
  handle_cast({update_nodes, NewNodes}, #state{nodes = Nodes, tags = Tags} = S) ->
      error_logger:info_report({"DDFS UPDATE NODES", NewNodes}),
      WriteBlacklist = lists:sort([Node || {Node, false, _} <- NewNodes]),
      ReadBlacklist = lists:sort([Node || {Node, _, false} <- NewNodes]),
  
negative.code:         UpdatedNodes =/= Nodes ->
              [gen_server:cast(Pid, {die, none}) || Pid <- gb_trees:values(Tags)],
              spawn(fun() ->
                            {ReadableNodes, RBSize} =
                                do_get_readable_nodes(UpdatedNodes, ReadBlacklist),
                            refresh_tag_cache(ReadableNodes, RBSize)
                    end),
              S#state{nodes = UpdatedNodes,
                      write_blacklist = WriteBlacklist,
                      read_blacklist = ReadBlacklist,
                      tag_cache = false,
                      tags = gb_trees:empty()};
          true ->
              S#state{write_blacklist = WriteBlacklist,
                      read_blacklist = ReadBlacklist}
      end.
  
  do_update_nodestats(NewNodes, #state{nodes = Nodes} = S) ->
      UpdatedNodes = [case gb_trees:lookup(Node, NewNodes) of
                          none ->
                              {Node, Stats};
                          {value, NewStats} ->
                              {Node, NewStats}
                      end || {Node, Stats} <- Nodes],
      S#state{nodes = UpdatedNodes}.
  
  do_tag_exit(Pid, S) ->
      NewTags = [X || {_, V} = X <- gb_trees:to_list(S#state.tags), V =/= Pid],
      S#state{tags = gb_trees:from_orddict(NewTags)}.
  
  -spec get_tags('all' | 'filter', [node()]) -> {[node()], [node()], [binary()]};
                ('safe', [node()]) -> {'ok', [binary()]} | 'too_many_failed_nodes'.
  