anchor.diff:  -record(state, {client :: {pid(), reference()},
                   client_type :: atom(),
                   bloom :: term(),
  -                vnodes_left :: [node()],
  +                pls :: [list()],
  +                wait_pls :: [term()],
  +                simul_pls :: integer(),
  +                vns :: term(),
                   bucket :: riak_object:bucket(),
                   timeout :: pos_integer(),
                   req_id :: pos_integer(),
  
positive.code:     lists:foldl(fun(E,A) -> bloom:add(E,A) end, Bloom, Acc);
  process_keys([K|Rest],Bucket,ClientType,Bloom,ReqId,Client,Acc) ->
      case bloom:member(K,Bloom) of
          true -> process_keys(Rest,Bucket,ClientType,Bloom,ReqId,Client,Acc);
          false -> process_keys(Rest,Bucket,ClientType,bloom:add(K,Bloom),ReqId,Client,[K|Acc])
      end.
  
  %% @private
  
negative.code:                        bloom=Bloom, req_id=ReqId, bucket=Bucket, ring=Ring},
      {ok,initialize,StateData,0}.
  
  %% @private
  initialize(timeout, StateData0=#state{bucket=Bucket, ring=Ring}) ->
      BucketProps = riak_core_bucket:get_bucket(Bucket, Ring),
      N = proplists:get_value(n_val,BucketProps),
      PLS0 = riak_core_ring:all_preflists(Ring,N),
      {LA1, LA2} = lists:partition(fun({A,_B}) -> A rem N == 0 end,
                                lists:zip(lists:seq(0,(length(PLS0)-1)), PLS0)),
      {_, PLS} = lists:unzip(lists:append(LA1,LA2)),
      Simul_PLS = trunc(length(PLS) / N),
      StateData = StateData0#state{pls=PLS,simul_pls=Simul_PLS,
                                   wait_pls=[],vns=sets:from_list([])},
      reduce_pls(StateData).
  
  waiting_kl({kl, Keys, Idx, ReqId},
             StateData0=#state{pls=PLS,vns=VNS0,wait_pls=WPL0,bloom=Bloom0,
                              req_id=ReqId,client=Client,timeout=Timeout,
                              bucket=Bucket,client_type=ClientType}) ->
      Bloom = process_keys(Keys,Bucket,ClientType,Bloom0,ReqId,Client),
      WPL = [{W_Idx,W_Node,W_PL} || {W_Idx,W_Node,W_PL} <- WPL0, W_Idx /= Idx],
      [Node] = [W_Node || {W_Idx,W_Node,_W_PL} <- WPL0, W_Idx =:= Idx],
      VNS = sets:add_element({Idx,Node},VNS0),
      StateData = StateData0#state{bloom=Bloom,wait_pls=WPL,vns=VNS},
      case PLS of
          [] ->
              case WPL of
                  [] -> finish(StateData);
                  _ -> {next_state, waiting_kl, StateData, Timeout}
              end;
          _ -> reduce_pls(StateData)
      end;
  
  waiting_kl(timeout, StateData=#state{pls=PLS,wait_pls=WPL,timeout=Timeout}) ->
      NewPLS = lists:append(PLS, [W_PL || {_W_Idx,_W_Node,W_PL} <- WPL]),
      {next_state, waiting_kl, StateData#state{pls=NewPLS,wait_pls=[]}, Timeout}.
  
  finish(StateData=#state{req_id=ReqId,client=Client,client_type=ClientType}) ->
      case ClientType of
          mapred -> luke_flow:finish_inputs(Client);
          plain -> Client ! {ReqId, done}
      end,
      {stop,normal,StateData}.
                                               
  ask_vn({Index,Node},ReqId,Msg) ->
      gen_server:cast({riak_kv_vnode_master, Node},
                      {vnode_list_bucket,{Index,ReqId},Msg}).
  
  reduce_pls(StateData0=#state{timeout=Timeout, req_id=ReqId,wait_pls=WPL,
                               simul_pls=Simul_PLS, bucket=Bucket}) ->
      case find_free_pl(StateData0) of
          {none_free,NewPLS} ->
              StateData = StateData0#state{pls=NewPLS},
              case NewPLS =:= [] andalso WPL =:= [] of
                  true -> finish(StateData);
                  false -> {next_state, waiting_kl, StateData, Timeout}
              end;
          {[{Idx,Node}|RestPL],PLS} ->
              ask_vn({Idx,Node},ReqId,{self(), Bucket, ReqId}),
              WaitPLS = [{Idx,Node,RestPL}|WPL],
              StateData = StateData0#state{pls=PLS, wait_pls=WaitPLS},
              case length(WaitPLS) > Simul_PLS of
                  true ->
                      {next_state, waiting_kl, StateData, Timeout};
                  false ->
                      reduce_pls(StateData)
              end
      end.
  
  find_free_pl(StateData) -> find_free_pl1(StateData, []).
  find_free_pl1(_StateData=#state{pls=[]}, NotFree) -> {none_free,NotFree};
  find_free_pl1(StateData=#state{wait_pls=WPL,pls=[PL|PLS],vns=VNS}, NotFree) ->
      case PL of
          [] -> find_free_pl1(StateData#state{pls=PLS}, NotFree);
          _ ->
              case check_pl(PL,VNS,WPL) of
                  redundant -> find_free_pl1(StateData#state{pls=PLS},NotFree);
                  notfree -> find_free_pl1(StateData#state{pls=PLS},[PL|NotFree]);
                  free -> {PL,lists:append(PLS,NotFree)}
              end
      end.
  
  check_pl(PL,VNS,WPL) ->
      case sets:is_disjoint(sets:from_list(PL),VNS) of
          false -> redundant;
          true ->
              PL_Nodes = sets:from_list([Node || {_Idx,Node} <- PL]),
              WaitNodes = sets:from_list([Node || {_Idx,Node,_RestPL} <- WPL]),
              case sets:is_disjoint(PL_Nodes,WaitNodes) of
                  false -> notfree;
                  true -> free
              end
      end.
  
  %% @private
  process_keys(Keys,Bucket,ClientType,Bloom,ReqId,Client) ->
  