anchor.diff:              #concurrency_throttle_state{count = 0} = State) ->
   
       #concurrency_throttle_state{waiting_pool = WaitingPool,
  +                                active_pool = ActivePool,
                                   target_load = TargetLoad} = State,
   
       NewWaitingPool = dict:store(Pid, {Signal, TargetNode}, WaitingPool),
  
positive.code:                   active_pool = NewActivePool}};
  
  handle_call(done, {Pid, _Tag},
              #concurrency_throttle_state{count = Count, monitor_dict = MonDict,
                                          active_pool = ActivePool,
                                          target_load = TargetLoad} = State) ->
  
      true = erlang:demonitor(dict:fetch(Pid, MonDict), [flush]),
      %% update monitoring and active reps tables
      NewMonDict = dict:erase(Pid, MonDict),
      NewActivePool = dict:erase(Pid, ActivePool),
      %% update target_load
      TargetNode = dict:fetch(Pid, ActivePool),
      NewLoad = dict:fetch(TargetNode, TargetLoad) - 1,
      NewTargetLoad = case NewLoad > 0 of
                          true ->
                              dict:store(TargetNode, NewLoad, TargetLoad);
                          _ ->
                              dict:erase(TargetNode, TargetLoad)
                      end,
  
      NewCount = Count + 1,
  
      ?xdcr_debug("rep ~p to node ~p is done, available tokens: ~p",
                  [Pid, TargetNode, NewCount]),
  
      State2 = signal_waiting(State#concurrency_throttle_state{
                                count = NewCount,
                                monitor_dict = NewMonDict,
                                target_load = NewTargetLoad,
                                active_pool = NewActivePool}),
      {reply, ok, State2}.
  
  handle_cast(Msg, State) ->
      {stop, {error, {unexpected_cast, Msg}}, State}.
  
  handle_info({'DOWN', _MonRef, _Type, Pid, _Info},
              #concurrency_throttle_state{
                count = Count, monitor_dict = MonDict} = State) ->
      %% a process we already signalled died before we got a done call.
      %% Remove from dict and increment count.
      MonDict2 = dict:erase(Pid, MonDict),
      State2 = signal_waiting(State#concurrency_throttle_state{
                                count = Count + 1,
                                monitor_dict = MonDict2}),
      {noreply, State2}.
  
  
  
negative.code:                  waiting_pool = NewWaitingPool},
  
      ?xdcr_debug("no token available, put (pid:~p, signal: ~p, targetnode: ~p) "
                  "into waiting pool (active reps: ~p, waiting reps: ~p)",
                  [Pid, Signal, TargetNode, dict:size(ActivePool), dict:size(NewWaitingPool)]),
  
      {reply, ok, NewState};
  
  