anchor.diff:              {reply,Error,State}
       end;
   handle_call(handoff_status,_From,State=#state{handoffs=HS}) ->
  -    Handoffs=[{H,D,active,S} || #handoff_status{ handoff=H,direction=D,status=S } <- HS],
  +    Handoffs=[{{M,I,N},D,active,S} ||
  +                 #handoff_status{modindex={M,I},node=N,direction=D,status=S} <- HS],
       {reply, {ok, Handoffs}, State};
   handle_call({set_concurrency,Limit},_From,State=#state{handoffs=HS}) ->
       application:set_env(riak_core,handoff_concurrency,Limit),
  
positive.code:     Reply =  [I || {M, I} <- ordsets:to_list(Excl), M =:= Module],
      {reply, {ok, Reply}, State};
  handle_call({add_outbound,Mod,Idx,Node,Pid},_From,State=#state{handoffs=HS}) ->
      ShouldStart=enqueue_handoff(Mod,Idx,Node,Pid,HS),
      case ShouldStart andalso send_handoff(Mod,Idx,Node,Pid) of
          {ok,Handoff=#handoff_status{transport_pid=Sender}} ->
              {reply,{ok,Sender},State#state{handoffs=HS ++ [Handoff]}};
          Error ->
  
negative.code:     ActiveSenders=proplists:get_value(active,Senders),
      get_concurrency_limit() =< (ActiveReceivers + ActiveSenders).
  
  %% spawn a sender process
  send_handoff (Mod,Idx,Node,Vnode,HS) ->
      case handoff_concurrency_limit_reached() of
          true ->
              {error, max_concurrency};
          false ->
              ShouldHandoff=
                  case lists:keyfind({Mod,Idx},#handoff_status.modindex,HS) of
                      false ->
                          true;
                      Handoff=#handoff_status{node=Node,vnode_pid=Vnode} ->
                          {false,Handoff};
                      #handoff_status{transport_pid=Sender} ->
                          %% found a running handoff with a different vnode
                          %% source or a different arget ndoe, kill the current
                          %% one and the new one will start up
                          erlang:exit(Sender,resubmit_handoff_change),
                          true
                  end,
  
              case ShouldHandoff of
                  true ->
                      %% start the sender process
                      {ok,Pid}=riak_core_handoff_sender_sup:start_sender(Node,
                                                                         Mod,
                                                                         Idx,
                                                                         Vnode),
                      erlang:monitor(process,Pid),
  
                      %% successfully started up a new sender handoff
                      {ok, #handoff_status{ transport_pid=Pid,
                                            direction=outbound,
                                            timestamp=now(),
                                            node=Node,
                                            modindex={Mod,Idx},
                                            vnode_pid=Vnode
                                          }
                      };
                  {false,CurrentHandoff} ->
                      %% handoff already going, just return it
                      {ok, CurrentHandoff}
              end
      end.
  
  %% spawn a receiver process
  