anchor.diff:      NewPLS = lists:append(PLS, [W_PL || {_W_Idx,_W_Node,W_PL} <- WPL]),
       reduce_pls(StateData#state{pls=NewPLS,wait_pls=[]}).
   
  +finish(StateData=#state{req_id=ReqId,client=Client,client_type=ClientType, listers=[]}) ->
  +    case ClientType of
  +        mapred -> 
  +            %% No nodes are available for key listing so all
  +            %% we can do now is die so that the rest of the
  +            %% MapReduce processes will also die and be cleaned up.
  +            exit(all_nodes_unavailable);
  +        plain -> 
  +            %%Notify the requesting client that the key 
  +            %% listing is complete or that no nodes are 
  +            %% available to fulfil the request.
  +            Client ! {ReqId, all_nodes_unavailable}
  +    end,
  +    {stop,normal,StateData};
   finish(StateData=#state{req_id=ReqId,client=Client,client_type=ClientType}) ->
       case ClientType of
  -        mapred -> luke_flow:finish_inputs(Client);
  -        plain -> Client ! {ReqId, done}
  +        mapred ->
  +            luke_flow:finish_inputs(Client);
  +        plain -> 
  +            Client ! {ReqId, done}
       end,
       {stop,normal,StateData}.
   
  
positive.code:     {ok, StateName, State}.
  
  %% @private
  start_listers(ReqId, Bucket) ->
      Nodes = riak_core_node_watcher:nodes(riak_kv),
      start_listers(Nodes, ReqId, Bucket, []).
  
  start_listers([], _ReqId, _Bucket, Accum) ->
      Accum;
  start_listers([H|T], ReqId, Bucket, Accum) ->
      {ok, Pid} = riak_kv_keylister_master:start_keylist(H, ReqId, Bucket),
      start_listers(T, ReqId, Bucket, [{H, Pid}|Accum]).
  
negative.code: %% @private
  handle_info({'EXIT', Pid, normal}, _StateName, #state{client=Pid}=StateData) ->
      {stop,normal,StateData};
  handle_info({_ReqId, {ok, _Pid}}, StateName, StateData=#state{timeout=Timeout}) ->
      %% Received a message from a key lister node that 
      %% did not start up within the timeout. Just ignore
      %% the message and move on.
      {next_state, StateName, StateData, Timeout};
  handle_info(_Info, _StateName, StateData) ->
      {stop,badmsg,StateData}.
  
  