diff --git a/changelog.md b/changelog.md index 623c64d..3acbc13 100644 --- a/changelog.md +++ b/changelog.md @@ -1,5 +1,18 @@ # ehttpc changes +## 0.7.6 + +- Zombie-connection detection no longer waits for the full request timeout. A connection + with requests in flight is reconnected when the latest send is older than + `request timeout + max_inactive`, capped at `max_inactive_cap` but never below + `max_inactive`. `max_inactive_cap` is a new pool option and defaults to 60 seconds. + Previously a 600-second request timeout delayed detection to 610 seconds, and requests + with an `infinity` timeout disabled detection entirely. +- Behaviour change: with a request timeout above about 50 seconds and the default pool + options, a connection whose response takes longer than 60 seconds is now killed at + 60 seconds. Set `max_inactive_cap` or `max_inactive` to the slowest expected response time + to raise the bound. + ## 0.7.5 - Added the possibility of choosing HTTP2 as the protocol. diff --git a/src/ehttpc.app.src b/src/ehttpc.app.src index 7fda509..6bbce1a 100644 --- a/src/ehttpc.app.src +++ b/src/ehttpc.app.src @@ -1,6 +1,6 @@ {application, ehttpc, [ {description, "HTTP Client for Erlang/OTP"}, - {vsn, "0.7.5"}, + {vsn, "0.7.6"}, {registered, []}, {applications, [ kernel, diff --git a/src/ehttpc.erl b/src/ehttpc.erl index 57b7fb4..6ea35c9 100644 --- a/src/ehttpc.erl +++ b/src/ehttpc.erl @@ -92,6 +92,7 @@ -define(undef, undefined). -define(IS_POOL(Pool), (not is_tuple(Pool) andalso not is_pid(Pool))). -define(DEFAULT_MAX_INACTIVE, 10_000). +-define(INACTIVE_CAP, 60_000). -define(IS_HEADERS_REQ(REQ), (tuple_size(REQ) =:= 2 andalso is_list(element(2, REQ))) @@ -119,6 +120,7 @@ %% In this case, host and port point to proxy server. origin :: undefined | map(), max_inactive :: pos_integer(), + max_inactive_cap :: pos_integer(), inactive_check_tref :: reference() | ?undef }). @@ -273,6 +275,7 @@ init([Pool, Id, Opts0]) -> PrioLatest = proplists:get_bool(prioritise_latest, Opts0), #{opts := Opts, origin := Origin} = parse_proxy_opts(Opts0), MaxInactive = proplists:get_value(max_inactive, Opts, ?DEFAULT_MAX_INACTIVE), + MaxInactiveCap = proplists:get_value(max_inactive_cap, Opts, ?INACTIVE_CAP), State = #state{ pool = Pool, id = Id, @@ -286,11 +289,13 @@ init([Pool, Id, Opts0]) -> pending => queue:new(), pending_count => 0, sent => #{}, - max_sent_expire => 0, + max_sent_at => 0, + max_sent_timeout => 0, prioritise_latest => PrioLatest }, origin = Origin, - max_inactive = MaxInactive + max_inactive = MaxInactive, + max_inactive_cap = MaxInactiveCap }, true = gproc_pool:connect_worker(ehttpc:name(Pool), {Pool, Id}), @@ -616,10 +621,21 @@ cancel_stream(nofin, Client, StreamRef) -> _ = gun:cancel(Client, StreamRef), ok. -timeout(infinity = _ExpireAt) -> +%% ExpireAt is {CalledAt, Timeout}: the caller's timestamp and its request timeout. +timeout({_CalledAt, infinity}) -> infinity; timeout(ExpireAt) -> - max(ExpireAt - now_(), 0). + max(deadline(ExpireAt) - now_(), 0). + +deadline({_CalledAt, infinity}) -> + infinity; +deadline({CalledAt, Timeout}) -> + CalledAt + Timeout. + +is_expired({_CalledAt, infinity}, _Now) -> + false; +is_expired(ExpireAt, Now) -> + Now > deadline(ExpireAt). now_() -> erlang:system_time(millisecond). @@ -628,25 +644,30 @@ now_() -> %% sent requests %% ================================================================================= +%% max_sent_at is the latest time a request was sent to gun. +%% max_sent_timeout is the largest request timeout among the sent requests. +%% Both reset to 0 when no request is in flight. put_sent_req( StreamRef, Req, #{ sent := Sent, - max_sent_expire := T + max_sent_at := SentAt, + max_sent_timeout := MaxTimeout } = Requests ) -> - ?SENT_REQ(_, Expire, _) = Req, + ?SENT_REQ(_, {_CalledAt, Timeout}, _) = Req, Requests#{ sent := maps:put(StreamRef, Req, Sent), - max_sent_expire := max_expire(T, Expire) + max_sent_at := max(SentAt, now_()), + max_sent_timeout := max_timeout(MaxTimeout, Timeout) }. -%% if a request has infinity timeout, ignore it -max_expire(T, infinity) -> T; -max_expire(T1, T2) when is_integer(T2) -> max(T1, T2). +max_timeout(_, infinity) -> infinity; +max_timeout(infinity, _) -> infinity; +max_timeout(T1, T2) -> max(T1, T2). -take_sent_req(StreamRef, #{sent := Sent, max_sent_expire := T} = Requests) -> +take_sent_req(StreamRef, #{sent := Sent} = Requests) -> case maps:take(StreamRef, Sent) of error -> error; @@ -655,27 +676,27 @@ take_sent_req(StreamRef, #{sent := Sent, max_sent_expire := T} = Requests) -> %% so there is no need to scan the map to find a new max %% or even if calls may use different timeout %% the impact of a wrong max is minimal: delayed detection of zombie connection - NewT = + NewRequests = case map_size(NewSent) of 0 -> - 0; + reset_sent(Requests); _ -> - T + Requests#{sent := NewSent} end, case is_sent_req_expired(Req, now_()) of true -> - {expired, Requests#{sent := NewSent, max_sent_expire := NewT}}; + {expired, NewRequests}; false -> - {Req, Requests#{sent := NewSent, max_sent_expire := NewT}} + {Req, NewRequests} end end. -is_sent_req_expired(?SENT_REQ(_From, infinity = _ExpireAt, _), _Now) -> +is_sent_req_expired(?SENT_REQ(_From, {_CalledAt, infinity}, _), _Now) -> false; is_sent_req_expired(?SENT_REQ({Pid, _Ref}, ExpireAt, _), Now) when is_pid(Pid) -> %% for gen_server:call, it is aborted after timeout, there is no need to send %% reply to the caller - Now > ExpireAt orelse (not erlang:is_process_alive(Pid)); + is_expired(ExpireAt, Now) orelse (not erlang:is_process_alive(Pid)); is_sent_req_expired(?SENT_REQ(_, _, _), _) -> %% for async requests, there is no way to tell if the caller %% the provided result-callback should be evaluated or not, @@ -698,7 +719,10 @@ reply_error_for_sent_reqs(#{sent := Sent} = R, Reason) -> end, maps:to_list(Sent) ), - R#{sent => #{}, max_sent_expire => 0}. + reset_sent(R). + +reset_sent(Requests) -> + Requests#{sent := #{}, max_sent_at := 0, max_sent_timeout := 0}. %% Continue droping expired requests, to avoid the state RAM usage %% explosion if http client can not keep up. @@ -712,7 +736,7 @@ drop_expired(#{pending_count := 0} = Requests, _Now) -> drop_expired(#{pending := Pending, pending_count := PC} = Requests, Now) -> {PeekFun, OutFun} = peek_oldest_fn(Requests), {value, ?PEND_REQ(ReplyTo, ?REQ(_, _, ExpireAt))} = PeekFun(Pending), - case is_integer(ExpireAt) andalso Now > ExpireAt of + case is_expired(ExpireAt, Now) of true -> {_, NewPendings} = OutFun(Pending), NewRequests = Requests#{pending => NewPendings, pending_count => PC - 1}, @@ -753,16 +777,19 @@ maybe_shoot( requests = #{ sent := Sent, - max_sent_expire := MaxExpire + max_sent_at := MaxSentAt, + max_sent_timeout := MaxTimeout } = Requests0, client = Client, max_inactive = MaxInactive, + max_inactive_cap = MaxInactiveCap, enable_pipelining = PipelineLimit } = State0 ) -> State = State0#state{requests = drop_expired(Requests0)}, SentCount = map_size(Sent), - case check_gun(Client, PipelineLimit, SentCount, MaxExpire, MaxInactive) of + Threshold = inactive_threshold(MaxTimeout, MaxInactive, MaxInactiveCap), + case check_gun(Client, PipelineLimit, SentCount, MaxSentAt, Threshold) of continue -> do_shoot(State); pause -> @@ -772,16 +799,17 @@ maybe_shoot( State; reconnect -> %% assert - true = (MaxExpire > 0), + true = (MaxSentAt > 0), %% the connection has been inactive for too long log( error, #{ msg => "force_reconnecting_zombie_http_connection", - last_request_expire => calendar:system_time_to_rfc3339(MaxExpire, [ + last_request_sent_at => calendar:system_time_to_rfc3339(MaxSentAt, [ {unit, millisecond} ]), - inactive_duration_threshold => MaxInactive, + inactive_duration_threshold => Threshold, + max_request_timeout => MaxTimeout, inflight_requests => SentCount, connection_pid => Client, pool => State#state.pool, @@ -794,10 +822,10 @@ maybe_shoot( State end. -check_gun(ClientPid, PipelineLimit, SentCount, MaxExpireTs, MaxInactiveDuration) -> +check_gun(ClientPid, PipelineLimit, SentCount, MaxSentAt, Threshold) -> maybe ok ?= check_gun_pid(ClientPid), - ok ?= check_gun_jammed(SentCount, MaxExpireTs, MaxInactiveDuration), + ok ?= check_gun_jammed(SentCount, MaxSentAt, Threshold), check_gun_limit(PipelineLimit, SentCount) end. @@ -816,19 +844,28 @@ check_gun_pid(Pid) -> pause end. -%% if there are sent requests, and the last reply is older than max_inactive, +%% if there are sent requests, and the latest send is older than the threshold, %% the connection is considered in zomebie state hence require a reconnect. -check_gun_jammed(_SentCount, 0, _MaxInactiveDuration) -> - %% there was no expire time recorded before +check_gun_jammed(_SentCount, 0, _Threshold) -> + %% no request in flight ok; -check_gun_jammed(_SentCount, MaxExpireTs, MaxInactiveDuration) -> - case (now_() - MaxExpireTs) > MaxInactiveDuration of +check_gun_jammed(_SentCount, MaxSentAt, Threshold) -> + case (now_() - MaxSentAt) > Threshold of true -> reconnect; false -> ok end. +%% How long after the latest send a connection with requests in flight +%% may stay silent before it is considered a zombie. +%% It is MaxTimeout + MaxInactive, capped at MaxInactiveCap, +%% but never less than MaxInactive. +inactive_threshold(infinity, MaxInactive, MaxInactiveCap) -> + max(MaxInactive, MaxInactiveCap); +inactive_threshold(MaxTimeout, MaxInactive, MaxInactiveCap) -> + max(MaxInactive, min(MaxInactiveCap, MaxTimeout + MaxInactive)). + %% allow 100 async requests maximum when enable_pipelining is 'true' %% allow only 1 async request when enable_pipelining is 'false' %% otherwise stop shooting at the number limited by enable_pipelining @@ -1129,10 +1166,8 @@ enqueue_latest_fn(#{prioritise_latest := true}) -> enqueue_latest_fn(_) -> fun queue:in/2. -fresh_expire_at(infinity = _Timeout) -> - infinity; -fresh_expire_at(Timeout) when is_integer(Timeout) -> - now_() + Timeout. +fresh_expire_at(Timeout) when Timeout =:= infinity; is_integer(Timeout) -> + {now_(), Timeout}. parse_proxy_opts(Opts) -> %% Target host and port @@ -1214,6 +1249,25 @@ log(Level, Data, #state{host = Host, port = Port}) -> -ifdef(TEST). +inactive_threshold_test_() -> + Cap = ?INACTIVE_CAP, + Cases = [ + %% {MaxTimeout, MaxInactive, Expected} + {5_000, 10_000, 15_000}, + {45_000, 10_000, 55_000}, + {50_000, 10_000, 60_000}, + {600_000, 10_000, 60_000}, + {infinity, 10_000, 60_000}, + {600_000, 180_000, 180_000}, + {infinity, 180_000, 180_000}, + {0, 10_000, 10_000}, + {0, 60_000, 60_000} + ], + [ + ?_assertEqual(Expected, inactive_threshold(MaxTimeout, MaxInactive, Cap)) + || {MaxTimeout, MaxInactive, Expected} <- Cases + ]. + prioritise_latest_test() -> Opts = #{prioritise_latest => true}, Seq = [1, 2, 3, 4], diff --git a/test/ehttpc_tests.erl b/test/ehttpc_tests.erl index c309d4b..4cf7755 100644 --- a/test/ehttpc_tests.erl +++ b/test/ehttpc_tests.erl @@ -651,7 +651,7 @@ zombie_detect_inflight_not_full_test() -> %% let the EXIT signal get to ehttpc process {ok, _} = ?block_until(#{?snk_kind := handle_client_down}, 1000, infinity), #{requests := Requests} = ehttpc:get_state(Pid, normal), - #{sent := Sent, pending_count := PendingCount, max_sent_expire := MaxTs} = Requests, + #{sent := Sent, pending_count := PendingCount, max_sent_at := MaxTs} = Requests, ?assertEqual(0, maps:size(Sent)), ?assertEqual(0, PendingCount), ?assertEqual(0, MaxTs), @@ -699,7 +699,7 @@ zombie_detect_inflight_full_test() -> error(timeout) end, #{requests := Requests} = ehttpc:get_state(Pid, normal), - #{sent := Sent, pending_count := PendingCount, max_sent_expire := MaxTs} = Requests, + #{sent := Sent, pending_count := PendingCount, max_sent_at := MaxTs} = Requests, ?assertEqual(0, maps:size(Sent)), ?assertEqual(0, PendingCount), ?assertEqual(0, MaxTs), @@ -707,6 +707,37 @@ zombie_detect_inflight_full_test() -> end ). +zombie_detect_long_timeout_test() -> + zombie_detect_capped(?PORT, ?FUNCTION_NAME, 30_000). + +zombie_detect_infinity_timeout_test() -> + zombie_detect_capped(?PORT, ?FUNCTION_NAME, infinity). + +%% With max_inactive = 1s and max_inactive_cap = 2s, a request timeout +%% longer than the cap must not delay zombie detection past the cap. +zombie_detect_capped(Port, Name, Timeout) -> + ServerOpts = #{ + port => Port, + name => Name, + %% no response during this test + delay => 30_000, + oneoff => false + }, + PoolOpts0 = pool_opts("127.0.0.1", Port, 5, _PrioritiseLatest = false), + PoolOpts = [{max_inactive, 1_000}, {max_inactive_cap, 2_000} | PoolOpts0], + ?WITH( + ServerOpts, + PoolOpts, + begin + T0 = erlang:monotonic_time(millisecond), + spawn_link(fun() -> ehttpc:request(?POOL, put, {<<"/">>, [], "foo"}, Timeout, 0) end), + {ok, _} = ?block_until(#{?snk_kind := reconnect}, 3_500, infinity), + Elapsed = erlang:monotonic_time(millisecond) - T0, + ?assert(Elapsed >= 2_000), + ok + end + ). + head_request_test() -> Port = ?PORT, Host = "127.0.0.1",