Merge pull request #521 from basho/pevm-client-specified-timeouts

Changes needed to expose FSM timeouts to clients

Author: evanmcc

src/riak_client.erl

@@ -250,7 +250,8 @@ delete(Bucket,Key,Options,Timeout) when is_list(Options) ->
     ReqId = mk_reqid(),
     riak_kv_delete_sup:start_delete(Node, [ReqId, Bucket, Key, Options, Timeout,
                                            Me, ClientId]),
-    wait_for_reqid(ReqId, Timeout);
+    RTimeout = recv_timeout(Options),
+    wait_for_reqid(ReqId, erlang:min(Timeout, RTimeout));
 delete(Bucket,Key,RW,Timeout) ->
     delete(Bucket,Key,[{rw, RW}], Timeout).
 
@@ -295,7 +296,8 @@ delete_vclock(Bucket,Key,VClock,Options,Timeout) when is_list(Options) ->
     ReqId = mk_reqid(),
     riak_kv_delete_sup:start_delete(Node, [ReqId, Bucket, Key, Options, Timeout,
                                            Me, ClientId, VClock]),
-    wait_for_reqid(ReqId, Timeout);
+    RTimeout = recv_timeout(Options),
+    wait_for_reqid(ReqId, erlang:min(Timeout, RTimeout));
 delete_vclock(Bucket,Key,VClock,RW,Timeout) ->
     delete_vclock(Bucket,Key,VClock,[{rw, RW}],Timeout).
 

src/riak_kv_pb_object.erl

@@ -65,8 +65,6 @@
                 req_ctx,   % context to go along with request (partial results, request ids etc)
                 client_id = <<0,0,0,0>> }). % emulate legacy API when vnode_vclocks is true
 
--define(DEFAULT_TIMEOUT, 60000).
-
 %% @doc init/0 callback. Returns the service internal start
 %% state.
 -spec init() -> any().
@@ -102,12 +100,14 @@ process(#rpbsetclientidreq{client_id = ClientId}, State) ->
 
 process(#rpbgetreq{bucket=B, key=K, r=R0, pr=PR0, notfound_ok=NFOk,
                    basic_quorum=BQ, if_modified=VClock,
-                   head=Head, deletedvclock=DeletedVClock}, #state{client=C} = State) ->
+                   head=Head, deletedvclock=DeletedVClock,
+                   timeout=Timeout}, #state{client=C} = State) ->
     R = decode_quorum(R0),
     PR = decode_quorum(PR0),
     case C:get(B, K, make_option(deletedvclock, DeletedVClock) ++
                    make_option(r, R) ++
                    make_option(pr, PR) ++
+                   make_option(timeout, Timeout) ++
                    make_option(notfound_ok, NFOk) ++
                    make_option(basic_quorum, BQ)) of
         {ok, O} ->
@@ -166,7 +166,7 @@ process(#rpbputreq{bucket=B, key=K, vclock=PbVC,
 
 process(#rpbputreq{bucket=B, key=K, vclock=PbVC, content=RpbContent,
                    w=W0, dw=DW0, pw=PW0, return_body=ReturnBody,
-                   return_head=ReturnHead},
+                   return_head=ReturnHead, timeout=Timeout},
         #state{client=C} = State) ->
 
     case K of
@@ -196,7 +196,8 @@ process(#rpbputreq{bucket=B, key=K, vclock=PbVC, content=RpbContent,
                       end
               end,
     case C:put(O, make_option(w, W) ++ make_option(dw, DW) ++
-                   make_option(pw, PW) ++ [{timeout, default_timeout()} | Options]) of
+                   make_option(pw, PW) ++ make_option(timeout, Timeout) ++ 
+                   Options) of
         ok when is_binary(ReturnKey) ->
             PutResp = #rpbputresp{key = ReturnKey},
             {reply, PutResp, State};
@@ -226,7 +227,8 @@ process(#rpbputreq{bucket=B, key=K, vclock=PbVC, content=RpbContent,
     end;
 
 process(#rpbdelreq{bucket=B, key=K, vclock=PbVc,
-                   r=R0, w=W0, pr=PR0, pw=PW0, dw=DW0, rw=RW0},
+                   r=R0, w=W0, pr=PR0, pw=PW0, dw=DW0, rw=RW0,
+                   timeout=Timeout},
         #state{client=C} = State) ->
     W = decode_quorum(W0),
     PW = decode_quorum(PW0),
@@ -240,7 +242,8 @@ process(#rpbdelreq{bucket=B, key=K, vclock=PbVc,
         make_option(rw, RW) ++
         make_option(pr, PR) ++
         make_option(pw, PW) ++
-        make_option(dw, DW),
+        make_option(dw, DW) ++
+        make_option(timeout, Timeout),
     Result = case PbVc of
                  undefined ->
                      C:delete(B, K, Options);
@@ -298,6 +301,3 @@ erlify_rpbvc(PbVc) ->
 %% Convert a vector clock to protocol buffers
 pbify_rpbvc(Vc) ->
     zlib:zip(term_to_binary(Vc)).
-
-default_timeout() ->
-    ?DEFAULT_TIMEOUT.

src/riak_kv_wm_object.erl

@@ -168,20 +168,6 @@ init(Props) ->
 %%      bindings from the dispatch, as well as any vtag
 %%      query parameter.
 service_available(RD, Ctx=#ctx{riak=RiakProps}) ->
-    Timeout = case wrq:get_req_header(?HEAD_TIMEOUT, RD) of 
-                  undefined -> ?DEFAULT_TIMEOUT;
-                  TimeoutStr -> 
-                      try
-                          list_to_integer(TimeoutStr)
-                      catch
-                          _:_ ->
-                              lager:error("Bad timeout value ~p, "
-                                          "using ~d~n", 
-                                          [TimeoutStr, ?DEFAULT_TIMEOUT]),
-                              ?DEFAULT_TIMEOUT
-                      end
-              end,
-    
     case riak_kv_wm_utils:get_riak_client(RiakProps, riak_kv_wm_utils:get_client_id(RD)) of
         {ok, C} ->
             {true,
@@ -197,8 +183,7 @@ service_available(RD, Ctx=#ctx{riak=RiakProps}) ->
                        undefined -> undefined;
                        K -> list_to_binary(riak_kv_wm_utils:maybe_decode_uri(RD, K))
                    end,
-               vtag=wrq:get_qs_value(?Q_VTAG, RD),
-               timeout=Timeout
+               vtag=wrq:get_qs_value(?Q_VTAG, RD)
               }};
         Error ->
             {false,
@@ -245,29 +230,64 @@ malformed_request(RD, Ctx) when Ctx#ctx.method =:= 'POST'
         undefined ->
             {true, missing_content_type(RD), Ctx};
         _ ->
-            case malformed_rw_params(RD, Ctx) of
-                Result={true, _, _} -> 
+            case malformed_timeout_param(RD, Ctx) of
+                Result={true, _, _} ->
                     Result;
-                {false, RWRD, RWCtx} ->
-                    case malformed_link_headers(RWRD, RWCtx) of
-                        Result = {true, _, _} ->
+                {false, ToRD, ToCtx} ->
+                    case malformed_rw_params(ToRD, ToCtx) of
+                        Result={true, _, _} -> 
                             Result;
-                        {false, RWLH, LHCtx} ->
-                            malformed_index_headers(RWLH, LHCtx)
+                        {false, RWRD, RWCtx} ->
+                            case malformed_link_headers(RWRD, RWCtx) of
+                                Result = {true, _, _} ->
+                                    Result;
+                                {false, RWLH, LHCtx} ->
+                                    malformed_index_headers(RWLH, LHCtx)
+                            end
                     end
             end
     end;
-malformed_request(RD, Ctx) ->
-    case malformed_rw_params(RD, Ctx) of
-        Result = {true, _, _} ->
+malformed_request(RD, Ctx) -> 
+    case malformed_timeout_param(RD, Ctx) of
+        Result={true, _, _} ->
             Result;
-        {false, ResRD, ResCtx} ->
-            DocCtx = ensure_doc(ResCtx),
-            case DocCtx#ctx.doc of
-                {error, Reason} ->
-                    handle_common_error(Reason, ResRD, DocCtx);
-                _ ->
-                    {false, ResRD, DocCtx}
+        {false, ToRD, ToCtx} ->
+            case malformed_rw_params(ToRD, ToCtx) of
+                Result = {true, _, _} ->
+                    Result;
+                {false, ResRD, ResCtx} ->
+                    DocCtx = ensure_doc(ResCtx),
+                    case DocCtx#ctx.doc of
+                        {error, Reason} ->
+                            handle_common_error(Reason, ResRD, DocCtx);
+                        _ ->
+                            {false, ResRD, DocCtx}
+                    end
+            end
+    end.
+
+%% @spec malformed_timeout_param(reqdata(), context()) ->
+%%          {boolean(), reqdata(), context()}
+%% @doc Check that the timeout parameter is are a
+%%      string-encoded integer.  Store the integer value
+%%      in context() if so.
+malformed_timeout_param(RD, Ctx) ->
+    case wrq:get_qs_value("timeout", none, RD) of
+        none ->
+            {false, RD, Ctx};
+        TimeoutStr -> 
+            try
+                Timeout = list_to_integer(TimeoutStr),
+                {false, RD, Ctx#ctx{timeout=Timeout}}
+            catch
+                _:_ ->
+                    {true,
+                     wrq:append_to_resp_body(io_lib:format("Bad timeout "
+                                                           "value ~p~n",
+                                                           [TimeoutStr]),
+                                             wrq:set_resp_header(?HEAD_CTYPE, 
+                                                                 "text/plain", RD)),
+                     Ctx}
             end
     end.
 
@@ -611,10 +631,9 @@ accept_doc_body(RD, Ctx=#ctx{bucket=B, key=K, client=C, links=L, index_fields=IF
     IndexMD = dict:store(?MD_INDEX, IF, UserMetaMD),
     MDDoc = riak_object:update_metadata(VclockDoc, IndexMD),
     Doc = riak_object:update_value(MDDoc, riak_kv_wm_utils:accept_value(CType, wrq:req_body(RD))),
-    Options = case wrq:get_qs_value(?Q_RETURNBODY, RD) of ?Q_TRUE -> [returnbody]; _ -> [] end,
-    case C:put(Doc, [{w, Ctx#ctx.w}, {dw, Ctx#ctx.dw}, {pw, Ctx#ctx.pw}, 
-                     {timeout, Ctx#ctx.timeout} |
-                Options]) of
+    Options0 = case wrq:get_qs_value(?Q_RETURNBODY, RD) of ?Q_TRUE -> [returnbody]; _ -> [] end,
+    Options = make_options(Options0, Ctx),
+    case C:put(Doc, Options) of
         {error, Reason} ->
             handle_common_error(Reason, RD, Ctx);
         ok ->
@@ -846,21 +865,18 @@ decode_vclock_header(RD) ->
 %%      worry about the order of executing of those places.
 ensure_doc(Ctx=#ctx{doc=undefined, key=undefined}) ->
     Ctx#ctx{doc={error, notfound}};
-ensure_doc(Ctx=#ctx{doc=undefined, bucket=B, key=K, client=C, r=R,
-        pr=PR, basic_quorum=Quorum, notfound_ok=NotFoundOK}) ->
-    Ctx#ctx{doc=C:get(B, K, [deletedvclock, {r, R}, {pr, PR},
-                             {basic_quorum, Quorum}, 
-                             {notfound_ok, NotFoundOK}, 
-                             {timeout, Ctx#ctx.timeout}])};
+ensure_doc(Ctx=#ctx{doc=undefined, bucket=B, key=K, client=C,
+                    basic_quorum=Quorum, notfound_ok=NotFoundOK}) ->
+    Options0 = [deletedvclock, {basic_quorum, Quorum},
+                {notfound_ok, NotFoundOK}],
+    Options = make_options(Options0, Ctx),
+    Ctx#ctx{doc=C:get(B, K, Options)};
 ensure_doc(Ctx) -> Ctx.
 
 %% @spec delete_resource(reqdata(), context()) -> {true, reqdata(), context()}
 %% @doc Delete the document specified.
-delete_resource(RD, Ctx=#ctx{bucket=B, key=K, client=C, rw=RW, r=R, w=W,
-        pr=PR, pw=PW, dw=DW, timeout=Timeout}) ->
-    Options = lists:filter(fun({_, default}) -> false; (_) -> true end,
-        [{rw, RW}, {r, R}, {w, W}, {pr, PR}, {pw, PW}, {dw, DW}, 
-         {timeout, Timeout}]),
+delete_resource(RD, Ctx=#ctx{bucket=B, key=K, client=C}) ->
+    Options = make_options([], Ctx),
     Result = case wrq:get_req_header(?HEAD_VCLOCK, RD) of
         undefined -> 
             C:delete(B,K,Options);
@@ -1083,4 +1099,10 @@ handle_common_error(Reason, RD, Ctx) ->
                 Ctx}
     end.
 
-
+make_options(Prev, Ctx) ->
+    NewOpts0 = [{rw, Ctx#ctx.rw}, {r, Ctx#ctx.r}, {w, Ctx#ctx.w}, 
+                {pr, Ctx#ctx.pr}, {pw, Ctx#ctx.pw}, {dw, Ctx#ctx.dw}, 
+                {timeout, Ctx#ctx.timeout}],
+    NewOpts = [ {Opt, Val} || {Opt, Val} <- NewOpts0, 
+                              Val /= undefined, Val /= default ],
+    Prev ++ NewOpts.