anchor.diff:          _ -> undefined
       end,
       VNS = sets:add_element({Idx,Node},VNS0),
  -    StateData = StateData0#state{bloom=Bloom,wait_pls=WPL,vns=VNS},
  +    StateData = StateData0#state{wait_pls=WPL,vns=VNS},
       case PLS of
           [] ->
               case WPL of
  
positive.code: process_keys(Keys,Bucket,ClientType,Bloom,ReqId,Client) ->
      process_keys(Keys,Bucket,ClientType,Bloom,ReqId,Client,[]).
  %% @private
  process_keys([],Bucket,ClientType,Bloom,ReqId,Client,Acc) ->
      case ClientType of
          mapred -> luke_flow:add_inputs(Client, [{Bucket,K} || K <- Acc]);
          plain -> Client ! {ReqId, {keys, Acc}}
      end,
      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:     reduce_pls(StateData).
  
  waiting_kl({kl, Keys, Idx, ReqId},
             StateData0=#state{pls=PLS,vns=VNS0,wait_pls=WPL0,bloom=Bloom,
                              req_id=ReqId,client=Client,timeout=Timeout,
                              bucket=Bucket,client_type=ClientType}) ->
      process_keys(Keys,Bucket,ClientType,Bloom,ReqId,Client),
      WPL = [{W_Idx,W_Node,W_PL} || {W_Idx,W_Node,W_PL} <- WPL0, W_Idx /= Idx],
      WNs = [W_Node || {W_Idx,W_Node,_W_PL} <- WPL0, W_Idx =:= Idx],
      Node = case WNs of
  