anchor.diff:  %% Description: Handling call messages
   %% @hidden
   %%--------------------------------------------------------------------
  -handle_call({register_worker, {WorkerType, Url, Reader}}, {From, _Ref}, State = #state{ maps=Maps }) ->
  -    {WorkerType, Ring} = lists:keyfind(WorkerType, 1, Maps),
  -    Ring1 = case lists:keyfind(Url, 1, Ring) of
  -               {Url, {async, From}} ->
  -                   lists:keyreplace(Url, 1, Ring, {Url, Reader});
  -               _ -> Ring
  -           end,
  -    Ring2 = lists:map(fun ({Frag, {async, _}}) ->
  -                              {Frag, undefined};
  -                          ({_Frag, Pool}=Mapping) when is_pid(Pool) ->
  -                              Mapping
  -                      end, Ring1),
  -    io:format("RING2 ~p~n", [Ring2]),
  -    {ok, Map, Interval} = redis_shard:generate_map_and_interval(Ring2),
  -    io:format("~p ~p ~p~n", [WorkerType, Map, Interval]),
  +notify_complete(State0=#state{ reply_to=Pid }) 
  +  when Pid =:= self; Pid =:= undefined ->
  +    State0#state{ reply_to=undefined };
  +
  +notify_complete(State0=#state{ maps=TempTable, reply_to=Pid }) ->
  +    case dict:size(TempTable) of
  +        0 ->
  +            gen_server:reply(Pid, ok),
  +            State0#state{ reply_to=undefined };
  +        _ -> State0
  +    end.
  +
  +update_worker_pid(Url, From, Pid, Ring) ->
  +    case lists:keyfind(Url, 1, Ring) of
  +        {Url, {async, From}} ->
  +            lists:keyreplace(Url, 1, Ring, {Url, Pid});
  +        _ -> Ring
  +    end.
  +
  +save_shard_info(WorkerType, Ring0) ->
  +    MapFun = fun ({Frag, {async, _}}, AccIn) ->
  +                     {{Frag, undefined}, AccIn};
  +                 ({_Frag, Pool}=Mapping, AccIn) when is_pid(Pool) ->
  +                     {Mapping, AccIn+1}
  +             end,
  +    {Ring1, Completed} = lists:mapfoldl(MapFun, 0, Ring0),
  +    {ok, Map, Interval} = redis_shard:generate_map_and_interval(Ring1),
       logplex_shard_info:save(WorkerType, Map, Interval),
  -    Maps1 = lists:keyreplace(WorkerType, 1, Maps, {WorkerType, Ring1}),
  -    {reply, ok, State#state{ maps=Maps1 }};
  +    case Completed =:= length(Ring1) of
  +        true -> done;
  +        false -> incomplete
  +    end.
  +
  +handle_call({register_worker, {WorkerType, Url, Reader}}, {From, _Ref}, State = #state{ maps=TempTable0 }) ->
  +    Ring = dict:fetch(WorkerType, TempTable0),
  +    Ring1 = update_worker_pid(Url, From, Reader, Ring),
  +    TempTable = case save_shard_info(WorkerType, Ring1) of
  +                    done ->
  +                        dict:erase(WorkerType, TempTable0);
  +                    incomplete ->
  +                        dict:store(WorkerType, Ring1, TempTable0)
  +                end,
  +    State1 = notify_complete(State#state{ maps=TempTable }),
  +    {reply, ok, State1};
   
   handle_call({commit, new_shard_info}, _From, State) ->
       backup_shard_info(),
  
positive.code: 
      ets:new(logplex_shard_info, [protected, set, named_table]),
  
      Maps = populate_info_table(Urls),
  
      {ok, #state{urls=Urls, maps=Maps}}.
  
  %%--------------------------------------------------------------------
  %% Function: %% handle_call(Request, From, State) -> {reply, Reply, State} |
  
negative.code:     WritePools = [ {Url, async_add_buffer(WriteMap, Url)}
                     || Url <- redis_sort(Urls)],
  
      dict:from_list([{ReadMap, ReadPools}, {WriteMap, WritePools}]).
  
  register_worker(WorkerType, Url, Pid) ->
      gen_server:call(?MODULE, {register_worker, {WorkerType, Url, Pid}}).
  