anchor.diff:                         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}).
   
  -%% @private
  -initialize(timeout, StateData0=#state{timeout=Timeout, req_id=ReqId,
  -                                      bucket=Bucket, ring=Ring}) ->
  -    [FirstVN|VNodes] = riak_core_ring:all_owners(Ring),
  -    ask_vn(FirstVN,ReqId,{self(), Bucket, ReqId}),
  -    StateData = StateData0#state{vnodes_left=VNodes},
  -    {next_state, waiting_kl, StateData, Timeout}.
  +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.
   
  -waiting_kl({kl, Keys, _Idx, ReqId},
  -           StateData=#state{vnodes_left=[],bucket=Bucket,
  -            bloom=Bloom,req_id=ReqId,client=Client,client_type=ClientType}) ->
  -    process_keys(Keys,Bucket,ClientType,Bloom,ReqId,Client),
  -    case ClientType of
  -        mapred -> luke_flow:finish_inputs(Client);
  -        plain -> Client ! {ReqId, done}
  -    end,
  -    {stop,normal,StateData};
  -waiting_kl({kl, Keys, _Idx, ReqId},
  -           StateData=#state{vnodes_left=[FirstVN|VNodes],bloom=Bloom,
  -                   req_id=ReqId,client=Client,timeout=Timeout,
  -                            bucket=Bucket,client_type=ClientType}) ->
  -    ask_vn(FirstVN,ReqId,{self(), Bucket, ReqId}),
  -    {next_state, waiting_kl,
  -     StateData#state{
  -       bloom=process_keys(Keys,Bucket,ClientType,Bloom,ReqId,Client),
  -       vnodes_left=VNodes},
  -     Timeout}.
  +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) ->
  
positive.code: -record(state, {client :: {pid(), reference()},
                  client_type :: atom(),
                  bloom :: term(),
                  vnodes_left :: [node()],
                  bucket :: riak_object:bucket(),
                  timeout :: pos_integer(),
                  req_id :: pos_integer(),
  
negative.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
  