anchor.diff:  
   %% Common code for sending a single bucket or multiple inputs map/request
   %% @private
  -send_mapred_req(Pid, MapRed, ClientPid) ->
  +send_mapred_req(Pid, MapRed, ClientPid, CallTimeout) ->
       ReqMsg = #rpbmapredreq{request = encode_mapred_req(MapRed),
                              content_type = <<"application/x-erlang-binary">>},
       ReqId = mk_reqid(),
       %% Add an extra 100ms to the mapred timeout and use that for the 
       %% socket timeout.  This should give the map/reduce a chance to fail and let us know.
       Timeout = proplists:get_value(timeout, MapRed, default_timeout(mapred_timeout)) + 100,
  -    gen_server:call(Pid, {req, ReqMsg, Timeout, {ReqId, ClientPid}}).
  +    gen_server:call(Pid, {req, ReqMsg, Timeout, {ReqId, ClientPid}}, CallTimeout).
   
   %% @private
   %% Make a new request that can be sent or queued
  
positive.code:     %% Run a Map/Reduce operation using reduce_identity to get a list
      %% of BKeys.
      IdentityQuery = [{reduce, {modfun, riak_kv_mapreduce, reduce_identity}, none, true}],
      case search(Pid, Bucket, SearchQuery, IdentityQuery, ?DEFAULT_TIMEOUT) of
          {ok, [{_, Results}]} -> 
              %% Unwrap the results.
              {ok, Results};
  
negative.code: %%      across the cluster.
  %%      See the map/reduce documentation for explanation of behavior.
  mapred_bucket(Pid, Bucket, Query, Timeout) ->
      mapred_bucket(Pid, Bucket, Query, Timeout, default_timeout(mapred_bucket_call_timeout)).
  
  %% @spec mapred_bucket(Pid :: pid(),
  %%                     Bucket :: bucket(),
  %%                     Query :: [riak_kv_mapred_query:mapred_queryterm()],
  %%                     TimeoutMillisecs :: integer() | 'infinity') ->
  %%       {ok, {ReqId :: term(), MR_FSM_PID :: pid()}} |
  %%       {error, Err :: term()}
  %% @doc Perform a map/reduce job against a bucket with a timeout
  %%      across the cluster and local call timeout.
  %%      See the map/reduce documentation for explanation of behavior.
  mapred_bucket(Pid, Bucket, Query, Timeout, CallTimeout) ->
      case mapred_bucket_stream(Pid, Bucket, Query, self(), Timeout, CallTimeout) of
          {ok, ReqId} ->
              wait_for_mapred(ReqId, Timeout);
          Error ->
  