diff --git a/NEWS.md b/NEWS.md index 575033f7..f226df12 100644 --- a/NEWS.md +++ b/NEWS.md @@ -5,6 +5,15 @@ unreleased ### Fixed +- A pooled HTTP/2 connection no longer closes when the caller that opened it + exits. It stayed owned by that caller, so its exit failed every other + caller's request on the connection with `{error, closed}`. A shared + connection now has no owner: each stream is tied to its own caller and is + reset if that caller dies, and the connection closes itself once it has had + no open stream for the pool `timeout` (#937, thanks @smartinio). +- Unregistering a pooled HTTP/2 connection no longer leaks its per-host slot. + The pool dropped its monitor on the connection, so the slot was never + released when the connection stopped. - A request that races a peer-initiated close now returns `{error, closed}` instead of `{error, invalid_state}`. A connection that sees the peer close stays alive briefly so late calls get an answer, and during that window every diff --git a/src/hackney.erl b/src/hackney.erl index 889e5ba1..b7cf6449 100644 --- a/src/hackney.erl +++ b/src/hackney.erl @@ -437,7 +437,9 @@ connect_pool_ssl(Transport, Host, Port, Options, FinalSslOpts, PoolHandler) -> maybe_register_h2(ConnPid, Host, Port, Transport, Options, PoolHandler) -> try hackney_conn:get_protocol(ConnPid) of http2 -> - %% HTTP/2 negotiated - register for connection sharing + %% HTTP/2 negotiated - register for connection sharing. Share it first + %% so it no longer dies with this caller while other callers use it. + _ = hackney_conn:share_h2(ConnPid), PoolHandler:register_h2(Host, Port, Transport, ConnPid, Options); http1 -> ok; diff --git a/src/hackney_conn.erl b/src/hackney_conn.erl index ca54fb99..ab1c1430 100644 --- a/src/hackney_conn.erl +++ b/src/hackney_conn.erl @@ -87,6 +87,7 @@ set_owner/2, set_owner/3, set_owner_async/2, + share_h2/1, %% Protocol info get_protocol/1 ]). @@ -139,7 +140,7 @@ %% State data record -record(conn_data, { %% Connection owner - owner :: pid(), + owner :: pid() | undefined, owner_mon :: reference() | undefined, %% Connection identity @@ -228,6 +229,11 @@ %% {stream, body_full, Status, Headers, Acc, From} %% {stream, done, Status, Headers, Buffer} h2_streams = #{} :: #{pos_integer() => {term(), tuple()}}, + %% Shared through the pool (share_h2/1): no owner, each stream monitors + %% its caller, and the connection closes itself once idle. + h2_shared = false :: boolean(), + %% Per-stream caller monitors of a shared connection: StreamId => ref + h2_stream_monitors = #{} :: #{pos_integer() => reference()}, %% Current HTTP/2 stream ID for streaming body mode (body = stream) h2_stream_id :: pos_integer() | undefined, %% Per-stream recv_timeout watchdog timers (sync one-shot reads): @@ -614,6 +620,15 @@ set_owner(Pid, NewOwner, Timeout) -> set_owner_async(Pid, NewOwner) -> gen_statem:cast(Pid, {set_owner, NewOwner}). +%% @doc Share a pooled HTTP/2 connection between callers. +%% The connection stops monitoring its owner, so the caller that dialed it +%% can exit without tearing down other callers' streams. Each stream is tied +%% to its own caller instead, and the connection closes itself after +%% `idle_timeout' with no stream open. Only pooled connections can be shared. +-spec share_h2(pid()) -> ok | {error, term()}. +share_h2(Pid) -> + safe_call(Pid, share_h2, 5000). + %% @doc Check if the connection's socket is still healthy. %% Returns ok if socket is open, {error, closed} otherwise. -spec verify_socket(pid()) -> ok | {error, closed | term()}. @@ -863,11 +878,29 @@ connected(enter, OldState, #conn_data{transport = Transport, socket = Socket, %% Transfer ownership back to pool and notify it auto_release_to_pool(Data) end, - case Timeout of - infinity -> {keep_state, Data2}; + case {Timeout, Data2#conn_data.protocol} of + {infinity, _} -> {keep_state, Data2}; + %% HTTP/2 multiplexes: the connection idles when its last stream ends, + %% not when it enters `connected' (#836). + {_, http2} -> {keep_state, Data2, h2_idle_actions(Data2)}; _ -> {keep_state, Data2, [{state_timeout, Timeout, idle_timeout}]} end; +connected({call, From}, share_h2, #conn_data{h2_shared = true}) -> + {keep_state_and_data, [{reply, From, ok}]}; + +connected({call, From}, share_h2, #conn_data{protocol = http2, pool_pid = PoolPid, + owner_mon = OwnerMon} = Data) + when is_pid(PoolPid) -> + %% The dialing caller no longer owns the connection: its exit must not + %% close streams other callers opened. Streams opened from now on monitor + %% their own caller, and the pool releases the per-host slot when the + %% connection stops. + demonitor(OwnerMon, [flush]), + Data2 = Data#conn_data{owner = undefined, owner_mon = undefined, + h2_shared = true}, + {keep_state, Data2, [{reply, From, ok} | h2_idle_actions(Data2)]}; + connected({call, From}, release_to_pool, #conn_data{pool_pid = PoolPid, owner_mon = OldMon, transport = Transport, socket = Socket} = Data) -> %% Reset owner to pool before notifying, to avoid deadlock @@ -890,6 +923,13 @@ connected({call, From}, release_to_pool, #conn_data{pool_pid = PoolPid, owner_mo notify_pool_available_sync(Data2), {keep_state, Data2, [{reply, From, ok}]}; +connected({call, From}, {set_owner, _NewOwner}, #conn_data{h2_shared = true}) -> + %% A shared connection has no owner to hand over. + {keep_state_and_data, [{reply, From, ok}]}; + +connected(cast, {set_owner, _NewOwner}, #conn_data{h2_shared = true}) -> + keep_state_and_data; + connected({call, From}, {set_owner, NewOwner}, #conn_data{owner_mon = OldMon} = Data) -> %% Update owner - demonitor old, monitor new demonitor(OldMon, [flush]), @@ -1895,6 +1935,28 @@ handle_common(cast, stop, _State, Data) -> %% Async stop - used by pool to avoid deadlock during sync checkin {stop, normal, Data}; +handle_common(info, {'DOWN', Ref, process, _Pid, _Reason}, State, Data) -> + handle_h2_stream_owner_down(Ref, State, Data); + +handle_common({timeout, h2_idle}, h2_idle, connected, + #conn_data{h2_shared = true, h2_streams = Streams, + h2_conn = H2Conn, h2_mon = H2Mon} = Data) + when map_size(Streams) =:= 0 -> + %% Idle with no stream open: close like a GOAWAY. The conn stops after + %% the closed grace window and the pool drops it on DOWN. + _ = case H2Mon of + undefined -> ok; + _ -> erlang:demonitor(H2Mon, [flush]) + end, + close_h2(H2Conn), + {next_state, closed, Data#conn_data{h2_conn = undefined, h2_mon = undefined, + socket = undefined}}; + +handle_common({timeout, h2_idle}, h2_idle, _State, _Data) -> + %% A stream opened since the timer was armed; the last one to finish + %% arms it again. + keep_state_and_data; + handle_common(cast, _Msg, _State, _Data) -> keep_state_and_data; @@ -2998,7 +3060,8 @@ start_h2_connection(Socket, Data, From, Origin) -> NewData = Data#conn_data{ h2_conn = H2Conn, h2_mon = Mon, - h2_streams = #{} + h2_streams = #{}, + h2_stream_monitors = #{} }, %% Cancel any pending idle_timeout armed by the %% TCP-first connected(enter): HTTP/2 connections @@ -3034,6 +3097,68 @@ h2_start_failure(after_upgrade, From, Reason) -> close_h2(H2Conn) -> try h2_connection:close(H2Conn) catch _:_ -> ok end. +%% An unshared connection lives and dies with its owner, so its streams need +%% no monitor of their own. +track_h2_stream(StreamId, Owner, StreamState, + #conn_data{h2_shared = false, h2_streams = Streams} = Data) -> + Data#conn_data{h2_streams = maps:put(StreamId, {Owner, StreamState}, Streams)}; +track_h2_stream(StreamId, Owner, StreamState, + #conn_data{h2_streams = Streams, + h2_stream_monitors = Monitors} = Data) -> + OwnerPid = h2_stream_owner_pid(Owner), + Monitor = erlang:monitor(process, OwnerPid), + Data#conn_data{ + h2_streams = maps:put(StreamId, {Owner, StreamState}, Streams), + h2_stream_monitors = maps:put(StreamId, Monitor, Monitors) + }. + +h2_stream_owner_pid({Pid, _Tag}) when is_pid(Pid) -> Pid; +h2_stream_owner_pid(Pid) when is_pid(Pid) -> Pid. + +drop_h2_stream(StreamId, + #conn_data{h2_streams = Streams, + h2_stream_monitors = Monitors} = Data) -> + Data1 = cancel_h2_timer(StreamId, Data), + Monitors2 = case maps:take(StreamId, Monitors) of + {Monitor, Rest} -> + _ = erlang:demonitor(Monitor, [flush]), + Rest; + error -> + Monitors + end, + Data1#conn_data{ + h2_streams = maps:remove(StreamId, Streams), + h2_stream_monitors = Monitors2 + }. + +clear_h2_stream_monitors(#conn_data{h2_stream_monitors = Monitors} = Data) -> + _ = maps:fold(fun(_StreamId, Monitor, ok) -> + _ = erlang:demonitor(Monitor, [flush]), + ok + end, ok, Monitors), + Data#conn_data{h2_stream_monitors = #{}}. + +handle_h2_stream_owner_down(Ref, State, + #conn_data{h2_stream_monitors = Monitors} = Data) -> + case [StreamId || {StreamId, Monitor} <- maps:to_list(Monitors), + Monitor =:= Ref] of + [StreamId] -> + _ = cancel_h2_stream(Data#conn_data.h2_conn, StreamId), + Data1 = drop_h2_stream(StreamId, Data), + h2_stream_owner_down_result(State, StreamId, Data1); + [] -> + keep_state_and_data + end. + +h2_stream_owner_down_result(streaming_body, StreamId, + #conn_data{h2_stream_id = StreamId} = Data) -> + %% The caller streaming a request body died: its stream is gone, so the + %% connection can take requests again. + {next_state, connected, + Data#conn_data{h2_stream_id = undefined, request_from = undefined}}; +h2_stream_owner_down_result(_State, _StreamId, Data) -> + h2_stream_result(Data, []). + %% @private Arm a per-stream recv_timeout watchdog for a sync HTTP/2 read so a %% lost frame fails fast with {error, timeout} instead of blocking until the %% connection dies. No-op when recv_timeout is infinity. @@ -3079,7 +3204,6 @@ handle_h2_recv_timeout(StreamId, TRef, h2_conn = H2Conn} = Data) -> case maps:get(StreamId, Timers, undefined) of TRef -> - Timers2 = maps:remove(StreamId, Timers), case maps:get(StreamId, Streams, undefined) of {From, Inner} when is_tuple(Inner), element(1, Inner) =:= sync -> %% RST_STREAM(CANCEL) the stalled stream so the peer stops @@ -3087,13 +3211,15 @@ handle_h2_recv_timeout(StreamId, TRef, %% pooled connection would be reused with an orphaned stream %% still open (h2_conn_usable only checks the conn state). _ = cancel_h2_stream(H2Conn, StreamId), - Streams2 = maps:remove(StreamId, Streams), - {keep_state, - Data#conn_data{h2_streams = Streams2, h2_timers = Timers2, - request_from = undefined}, - [{reply, From, {error, timeout}}]}; + Data2 = drop_h2_stream( + StreamId, + Data#conn_data{request_from = undefined}), + h2_stream_result( + Data2, + [{reply, From, {error, timeout}}]); _ -> - {keep_state, Data#conn_data{h2_timers = Timers2}} + {keep_state, + Data#conn_data{h2_timers = maps:remove(StreamId, Timers)}} end; _ -> {keep_state, Data} @@ -3167,10 +3293,8 @@ do_h2_send(From, Method, Path, Headers, Body, StreamState, Mode, SendTimeout, Da sync -> From; {async, _Ref0, StreamTo0, _AsyncMode0} -> StreamTo0 end, - Streams = maps:put(StreamId, {Owner, StreamState}, - Data#conn_data.h2_streams), - NewData0 = Data#conn_data{ - h2_streams = Streams, + NewData0 = track_h2_stream(StreamId, Owner, StreamState, Data), + NewData1 = NewData0#conn_data{ method = MethodBin, path = PathBin }, @@ -3178,9 +3302,9 @@ do_h2_send(From, Method, Path, Headers, Body, StreamState, Mode, SendTimeout, Da sync -> %% Watchdog the response so a lost frame fails fast rather %% than blocking on the infinity gen_statem:call. - arm_h2_timer(StreamId, NewData0#conn_data{request_from = From}); + arm_h2_timer(StreamId, NewData1#conn_data{request_from = From}); {async, Ref, StreamTo, AsyncMode} -> - NewData0#conn_data{ + NewData1#conn_data{ async = AsyncMode, async_ref = Ref, stream_to = StreamTo @@ -3198,7 +3322,7 @@ do_h2_send(From, Method, Path, Headers, Body, StreamState, Mode, SendTimeout, Da %% END_STREAM and transition to streaming_body so the caller can push body %% chunks via send_body_chunk/finish_send_body. Mirrors do_h3_send_headers/5. do_h2_send_headers(From, Method, Path, Headers, ReqOpts, Data) -> - #conn_data{h2_conn = H2Conn, h2_streams = Streams} = Data, + #conn_data{h2_conn = H2Conn} = Data, {MethodBin, PathBin, H2Headers} = build_h2_request_headers(Method, Path, Headers, Data), %% Effective send_timeout for this stream's body chunks. Stored in the @@ -3215,8 +3339,9 @@ do_h2_send_headers(From, Method, Path, Headers, ReqOpts, Data) -> end, case SendRes of {ok, StreamId} -> - NewData = Data#conn_data{ - h2_streams = maps:put(StreamId, {undefined, {stream, sending}}, Streams), + Owner = element(1, From), + NewData0 = track_h2_stream(StreamId, Owner, {stream, sending}, Data), + NewData = NewData0#conn_data{ h2_stream_id = StreamId, req_send_timeout = SendTimeout, method = MethodBin, @@ -3309,8 +3434,8 @@ handle_h2_stream_body(From, #conn_data{h2_stream_id = StreamId, h2_streams = Str [{reply, From, {ok, Buffer}}]} end; {_, {stream, done, _Status, _Hdrs, <<>>}} -> - Streams2 = maps:remove(StreamId, Streams), - {keep_state, Data#conn_data{h2_streams = Streams2}, [{reply, From, done}]}; + h2_stream_result(drop_h2_stream(StreamId, Data), + [{reply, From, done}]); {_, {stream, done, Status, Hdrs, Buffer}} -> %% Hand back the last buffered chunk; next call returns done. Streams2 = maps:put(StreamId, {undefined, {stream, done, Status, Hdrs, <<>>}}, Streams), @@ -3329,8 +3454,8 @@ handle_h2_read_body(From, #conn_data{h2_stream_id = StreamId, h2_streams = Strea Streams), {keep_state, Data#conn_data{h2_streams = Streams2}}; {_, {stream, done, _Status, _Hdrs, Buffer}} -> - Streams2 = maps:remove(StreamId, Streams), - {keep_state, Data#conn_data{h2_streams = Streams2}, [{reply, From, {ok, Buffer}}]}; + h2_stream_result(drop_h2_stream(StreamId, Data), + [{reply, From, {ok, Buffer}}]); _ -> {keep_state_and_data, [{reply, From, {error, no_stream}}]} end. @@ -3482,11 +3607,10 @@ deliver_once_item(StreamId, StreamTo, Ref, [{data, Body} | Rest], Data) -> {keep_state, Data#conn_data{h2_streams = Streams2}}; deliver_once_item(StreamId, StreamTo, Ref, [done], Data) -> StreamTo ! {hackney_response, Ref, done}, - Streams2 = maps:remove(StreamId, Data#conn_data.h2_streams), - {keep_state, Data#conn_data{h2_streams = Streams2, - async = false, - async_ref = undefined, - stream_to = undefined}}; + Data1 = drop_h2_stream(StreamId, Data), + h2_stream_result(Data1#conn_data{async = false, + async_ref = undefined, + stream_to = undefined}, []); deliver_once_item(StreamId, StreamTo, Ref, [], Data) -> Streams2 = maps:put(StreamId, {StreamTo, {async_once, StreamTo, Ref, [], 1}}, @@ -3516,12 +3640,11 @@ h2_on_data(StreamId, Body, EndStream, Data) -> NewAcc = <>, case EndStream of true -> - Streams2 = maps:remove(StreamId, Streams), - Data2 = cancel_h2_timer(StreamId, - Data#conn_data{h2_streams = Streams2, - request_from = undefined}), - {keep_state, Data2, - [{reply, From, {ok, Status, Headers, NewAcc}}]}; + Data2 = drop_h2_stream( + StreamId, + Data#conn_data{request_from = undefined}), + h2_stream_result(Data2, + [{reply, From, {ok, Status, Headers, NewAcc}}]); false -> Streams2 = maps:put(StreamId, {From, {sync, body, Status, Headers, NewAcc}}, @@ -3553,12 +3676,11 @@ h2_on_data(StreamId, Body, EndStream, Data) -> case EndStream of true -> StreamTo ! {hackney_response, Ref, done}, - Streams2 = maps:remove(StreamId, Streams), - {keep_state, - Data#conn_data{h2_streams = Streams2, - async = false, - async_ref = undefined, - stream_to = undefined}}; + Data2 = drop_h2_stream(StreamId, Data), + h2_stream_result( + Data2#conn_data{async = false, + async_ref = undefined, + stream_to = undefined}, []); false -> NewState = {async, AsyncMode, StreamTo, Ref, streaming, Status, Headers}, @@ -3586,9 +3708,8 @@ h2_on_data(StreamId, Body, EndStream, Data) -> [{reply, From, {ok, NewBuffer}}]}; From when EndStream -> %% Parked caller, no buffered bytes, stream ended -> done. - Streams2 = maps:remove(StreamId, Streams), - {keep_state, Data#conn_data{h2_streams = Streams2}, - [{reply, From, done}]}; + h2_stream_result(drop_h2_stream(StreamId, Data), + [{reply, From, done}]); _From -> %% Empty DATA frame without END_STREAM: keep the caller parked. {keep_state, Data} @@ -3598,9 +3719,8 @@ h2_on_data(StreamId, Body, EndStream, Data) -> NewAcc = <>, case EndStream of true -> - Streams2 = maps:remove(StreamId, Streams), - {keep_state, Data#conn_data{h2_streams = Streams2}, - [{reply, From, {ok, NewAcc}}]}; + h2_stream_result(drop_h2_stream(StreamId, Data), + [{reply, From, {ok, NewAcc}}]); false -> Streams2 = maps:put(StreamId, {From, {stream, body_full, Status, Headers, NewAcc, From}}, @@ -3615,34 +3735,32 @@ h2_on_stream_reset(StreamId, ErrorCode, Data) -> #conn_data{h2_streams = Streams} = Data, case maps:get(StreamId, Streams, undefined) of {From, Inner} when is_tuple(Inner), element(1, Inner) =:= sync -> - Streams2 = maps:remove(StreamId, Streams), - Data2 = cancel_h2_timer(StreamId, - Data#conn_data{h2_streams = Streams2, - request_from = undefined}), - {keep_state, Data2, - [{reply, From, {error, {stream_error, ErrorCode}}}]}; + Data2 = drop_h2_stream( + StreamId, + Data#conn_data{request_from = undefined}), + h2_stream_result(Data2, + [{reply, From, {error, {stream_error, ErrorCode}}}]); {StreamTo, {async, _, StreamTo, Ref, _, _, _}} -> StreamTo ! {hackney_response, Ref, {error, {stream_error, ErrorCode}}}, - Streams2 = maps:remove(StreamId, Streams), - {keep_state, Data#conn_data{h2_streams = Streams2}}; + h2_stream_result(drop_h2_stream(StreamId, Data), []); {StreamTo, {async, _, StreamTo, Ref, _}} -> StreamTo ! {hackney_response, Ref, {error, {stream_error, ErrorCode}}}, - Streams2 = maps:remove(StreamId, Streams), - {keep_state, Data#conn_data{h2_streams = Streams2}}; + h2_stream_result(drop_h2_stream(StreamId, Data), []); {StreamTo, {async_once, StreamTo, Ref, _, _}} -> StreamTo ! {hackney_response, Ref, {error, {stream_error, ErrorCode}}}, - Streams2 = maps:remove(StreamId, Streams), - {keep_state, Data#conn_data{h2_streams = Streams2}}; + h2_stream_result(drop_h2_stream(StreamId, Data), []); {_, Inner} when is_tuple(Inner), element(1, Inner) =:= stream -> %% Streaming-body stream: reply to any parked caller and drop it so a %% later stream_body/start_response sees {error, no_stream}. - Streams2 = maps:remove(StreamId, Streams), Replies = case h2_stream_parked_from(Inner) of undefined -> []; From -> [{reply, From, {error, {stream_error, ErrorCode}}}] end, - {keep_state, Data#conn_data{h2_streams = Streams2, request_from = undefined}, - Replies}; + h2_stream_result( + drop_h2_stream( + StreamId, + Data#conn_data{request_from = undefined}), + Replies); _ -> {keep_state, Data} end. @@ -3654,6 +3772,21 @@ h2_stream_parked_from({stream, headers, _, _, _, From}) -> From; h2_stream_parked_from({stream, body_full, _, _, _, From}) -> From; h2_stream_parked_from(_) -> undefined. +%% @private Result for a handler that just removed an HTTP/2 stream: a shared +%% connection left with no stream arms its idle timer. +h2_stream_result(Data, Replies) -> + {keep_state, Data, Replies ++ h2_idle_actions(Data)}. + +%% @private Arm the idle timer of a shared HTTP/2 connection with no stream +%% open. Re-arming replaces the previous timer, so it always counts from the +%% moment the last stream finished. +h2_idle_actions(#conn_data{h2_shared = true, h2_streams = Streams, + idle_timeout = Timeout}) + when map_size(Streams) =:= 0, Timeout =/= infinity -> + [{{timeout, h2_idle}, Timeout, h2_idle}]; +h2_idle_actions(_Data) -> + []. + h2_on_goaway(ErrorCode, #conn_data{h2_conn = H2Conn, h2_mon = H2Mon} = Data) -> %% A GOAWAY means the peer will not service new streams on this connection. %% AWS ALBs recycle connections this way, sending GOAWAY but keeping the @@ -3707,7 +3840,9 @@ collect_h2_aborts(Err, #conn_data{h2_streams = Streams} = Data) -> end; (_, _, Acc) -> Acc end, [], Streams), - {Replies, Data#conn_data{h2_streams = #{}, request_from = undefined}}. + Data1 = clear_h2_stream_monitors( + Data#conn_data{h2_streams = #{}, request_from = undefined}), + {Replies, Data1}. %%==================================================================== diff --git a/src/hackney_pool.erl b/src/hackney_pool.erl index b48ce0bc..d4522ecb 100644 --- a/src/hackney_pool.erl +++ b/src/hackney_pool.erl @@ -1253,7 +1253,8 @@ h2_conn_usable(Pid) -> %% @private Remove an HTTP/2 connection from the pool do_unregister_h2(Pid, State) -> - #state{h2_connections = H2Conns, pid_monitors = PidMonitors} = State, + #state{h2_connections = H2Conns, in_use = InUse, + pid_monitors = PidMonitors} = State, %% Find and remove the connection H2Conns2 = maps:fold( fun(Key, ConnPid, Acc) -> @@ -1265,12 +1266,18 @@ do_unregister_h2(Pid, State) -> H2Conns, H2Conns ), - %% Demonitor if no longer tracked - PidMonitors2 = case maps:take(Pid, PidMonitors) of - {MonRef, PM} -> - erlang:demonitor(MonRef, [flush]), - PM; - error -> PidMonitors + %% Demonitor if no longer tracked. A checked-out conn keeps its monitor: + %% it holds a per-host slot until it stops, and only the DOWN releases it. + PidMonitors2 = case maps:is_key(Pid, InUse) of + true -> + PidMonitors; + false -> + case maps:take(Pid, PidMonitors) of + {MonRef, PM} -> + erlang:demonitor(MonRef, [flush]), + PM; + error -> PidMonitors + end end, State#state{h2_connections = H2Conns2, pid_monitors = PidMonitors2}. diff --git a/test/hackney_http2_shared_conn_tests.erl b/test/hackney_http2_shared_conn_tests.erl new file mode 100644 index 00000000..9ac5671b --- /dev/null +++ b/test/hackney_http2_shared_conn_tests.erl @@ -0,0 +1,296 @@ +%%% Lifetime of pooled HTTP/2 connections shared between callers (#937). +%%% +%%% A pooled HTTP/2 connection used to stay owned by the caller that dialed +%%% it: when that caller exited, the connection stopped and every other +%%% caller's stream on it failed with {error, closed}. A shared connection +%%% now has no owner. Each stream is tied to its own caller, and the +%%% connection closes itself once it has been idle for the pool timeout. +-module(hackney_http2_shared_conn_tests). + +-include_lib("eunit/include/eunit.hrl"). + +shared_conn_test_() -> + [{timeout, 30, {Title, Fun}} || {Title, Fun} <- [ + {"dialer exiting does not close other callers' streams", + fun dialer_exit_keeps_other_streams/0}, + {"a killed caller resets only its own stream", + fun killed_caller_resets_its_stream/0}, + {"a dead async consumer resets only its own stream", + fun dead_async_consumer/0}, + {"a dead uploader frees the connection for new requests", + fun dead_uploader/0}, + {"stream monitors are removed when streams end", + fun stream_monitors_removed/0}, + {"an idle shared connection closes and releases its slot", + fun idle_conn_closes/0}, + {"the idle timer waits for open streams", + fun idle_waits_for_streams/0}, + {"an unregistered connection still releases its slot", + fun unregistered_conn_releases_slot/0}, + {"an unpooled connection is not shared", + fun unpooled_conn_not_shared/0} + ]]. + +%%==================================================================== +%% Tests +%%==================================================================== + +dialer_exit_keeps_other_streams() -> + with_server([], fun(URL, _Port, Opts) -> + {First, FirstRef} = spawn_monitor(fun() -> + Result = hackney:request(get, <>, [], <<>>, Opts), + exit({result, Result}) + end), + {FirstHandler, SConn} = started(<<"/first">>), + _ = sync_request(second, <>, Opts), + {SecondHandler, SConn2} = started(<<"/second">>), + ?assertEqual(SConn, SConn2), + FirstHandler ! respond, + receive + {'DOWN', FirstRef, process, First, {result, FirstResult}} -> + ?assertMatch({ok, 200, _, <<"ok">>}, FirstResult) + after 5000 -> error(no_first_result) + end, + SecondHandler ! respond, + ?assertMatch({ok, 200, _, <<"ok">>}, result(second)) + end). + +killed_caller_resets_its_stream() -> + with_server([], fun(URL, Port, Opts) -> + First = sync_request(first, <>, Opts), + {_FirstHandler, SConn} = started(<<"/first">>), + Conn = shared_conn(Opts, Port), + ?assert(lists:member(First, monitored(Conn))), + exit(First, kill), + ok = wait_until(fun() -> not lists:member(First, monitored(Conn)) end), + _ = sync_request(second, <>, Opts), + {SecondHandler, SConn2} = started(<<"/second">>), + ?assertEqual(SConn, SConn2), + SecondHandler ! respond, + ?assertMatch({ok, 200, _, <<"ok">>}, result(second)), + ?assertEqual(Conn, shared_conn(Opts, Port)) + end). + +dead_async_consumer() -> + with_server([], fun(URL, Port, Opts) -> + Parent = self(), + Consumer = spawn(fun() -> + {ok, _Ref} = hackney:request(get, <>, [], <<>>, + [async | Opts]), + Parent ! async_sent, + receive stop -> ok end + end), + receive async_sent -> ok after 5000 -> error(no_async_request) end, + %% The handler never answers: a response crossing the client's + %% RST_STREAM is a separate h2 HPACK issue, not what this covers. + {_AsyncHandler, _} = started(<<"/async">>), + Conn = shared_conn(Opts, Port), + ?assert(lists:member(Consumer, monitored(Conn))), + exit(Consumer, kill), + ok = wait_until(fun() -> not lists:member(Consumer, monitored(Conn)) end), + ?assertMatch({ok, 200, _, <<"ok">>}, + hackney:request(get, <>, [], <<>>, Opts)), + ?assertEqual(Conn, shared_conn(Opts, Port)) + end). + +dead_uploader() -> + with_server([], fun(URL, Port, Opts) -> + Parent = self(), + Uploader = spawn(fun() -> + {ok, UploadConn} = hackney:request(post, <>, [], + stream, Opts), + ok = hackney:send_body(UploadConn, <<"chunk">>), + Parent ! {uploading, UploadConn}, + receive stop -> ok end + end), + Conn = receive {uploading, C} -> C after 5000 -> error(no_upload) end, + {_UploadHandler, _} = started(<<"/upload">>), + ?assertEqual({ok, streaming_body}, hackney_conn:get_state(Conn)), + exit(Uploader, kill), + ok = wait_until(fun() -> + hackney_conn:get_state(Conn) =:= {ok, connected} + end), + ?assertMatch({ok, 200, _, <<"ok">>}, + hackney:request(get, <>, [], <<>>, Opts)), + ?assertEqual(Conn, shared_conn(Opts, Port)) + end). + +stream_monitors_removed() -> + with_server([], fun(URL, Port, Opts) -> + {ok, 200, _, <<"ok">>} = + hackney:request(get, <>, [], <<>>, Opts), + Conn = shared_conn(Opts, Port), + Baseline = monitored(Conn), + Callers = [begin + Pid = sync_request({fast, N}, <>, Opts), + ?assertMatch({ok, 200, _, _}, result({fast, N})), + Pid + end || N <- lists:seq(1, 20)], + %% The recv_timeout watchdog path drops the stream monitor too. + TimeoutOpts = lists:keystore(recv_timeout, 1, Opts, {recv_timeout, 100}), + ?assertEqual({error, timeout}, + hackney:request(get, <>, [], <<>>, + TimeoutOpts)), + ?assertEqual(Baseline, monitored(Conn)), + ?assertEqual([], [P || P <- Callers, lists:member(P, monitored(Conn))]), + ?assertEqual(Conn, shared_conn(Opts, Port)) + end). + +idle_conn_closes() -> + with_server([{timeout, 100}], fun(URL, Port, Opts) -> + {ok, 200, _, <<"ok">>} = + hackney:request(get, <>, [], <<>>, Opts), + Conn = shared_conn(Opts, Port), + Ref = erlang:monitor(process, Conn), + receive {'DOWN', Ref, process, Conn, _} -> ok + after 5000 -> error(conn_did_not_idle_out) + end, + ok = wait_until(fun() -> + hackney_load_regulation:current("localhost", Port) =:= 0 + end), + ?assertMatch({ok, 200, _, <<"ok">>}, + hackney:request(get, <>, [], <<>>, Opts)) + end). + +idle_waits_for_streams() -> + with_server([{timeout, 100}], fun(URL, Port, Opts) -> + _ = sync_request(slow, <>, Opts), + {Handler, _} = started(<<"/slow">>), + Conn = shared_conn(Opts, Port), + %% Hold the stream open well past the idle timeout. + receive after 400 -> ok end, + ?assertEqual({ok, connected}, hackney_conn:get_state(Conn)), + Handler ! respond, + ?assertMatch({ok, 200, _, <<"ok">>}, result(slow)) + end). + +unregistered_conn_releases_slot() -> + with_server([], fun(URL, Port, Opts) -> + {ok, 200, _, <<"ok">>} = + hackney:request(get, <>, [], <<>>, Opts), + Conn = shared_conn(Opts, Port), + ?assertEqual(1, hackney_load_regulation:current("localhost", Port)), + ok = hackney_pool:unregister_h2(Conn, Opts), + ok = wait_until(fun() -> no_shared_conn(Opts, Port) end), + hackney_conn:stop(Conn), + ok = wait_until(fun() -> + hackney_load_regulation:current("localhost", Port) =:= 0 + end) + end). + +unpooled_conn_not_shared() -> + with_server([], fun(URL, _Port, Opts) -> + Parent = self(), + UnpooledOpts = lists:keystore(pool, 1, Opts, {pool, false}), + Uploader = spawn(fun() -> + {ok, UploadConn} = hackney:request(post, <>, [], + stream, UnpooledOpts), + Parent ! {uploading, UploadConn}, + receive stop -> ok end + end), + Conn = receive {uploading, C} -> C after 5000 -> error(no_conn) end, + {_Handler, _} = started(<<"/upload">>), + ?assertEqual({error, invalid_state}, hackney_conn:share_h2(Conn)), + %% Its lifetime stays with its owner, so streams are not monitored. + ?assertNot(lists:member(Uploader, monitored(Conn))), + exit(Uploader, kill), + hackney_conn:stop(Conn) + end). + +%%==================================================================== +%% Helpers +%%==================================================================== + +%% Start an h2 server and a pool of one connection per host. `/fast' +%% answers at once and `/never' never answers; any other path reports +%% {request_started, Path, Handler, ServerConn} and answers on `respond'. +with_server(PoolOpts, Fun) -> + _ = application:ensure_all_started(hackney), + _ = application:ensure_all_started(h2), + Parent = self(), + Handler = fun(SConn, Sid, _Method, Path, _Headers) -> + case Path of + <<"/fast">> -> ok; + <<"/never">> -> receive after infinity -> ok end; + _ -> + Parent ! {request_started, Path, self(), SConn}, + receive respond -> ok end + end, + ok = h2:send_response(SConn, Sid, 200, + [{<<"content-type">>, <<"text/plain">>}]), + ok = h2:send_data(SConn, Sid, <<"ok">>, true) + end, + Certs = cert_dir(), + {ok, Server} = h2:start_server(0, #{ + cert => filename:join(Certs, "server.pem"), + key => filename:join(Certs, "server.key"), + handler => Handler}), + Port = h2:server_port(Server), + Pool = list_to_atom("hackney_h2_shared_" ++ integer_to_list(Port)), + ok = hackney_pool:start_pool(Pool, [{max_connections, 1} | PoolOpts]), + URL = iolist_to_binary([<<"https://localhost:">>, integer_to_list(Port)]), + Opts = [{pool, Pool}, {protocols, [http2]}, {recv_timeout, 5000}, + {ssl_options, [{insecure, true}, {verify, verify_none}]}], + try + Fun(URL, Port, Opts) + after + _ = hackney_pool:stop_pool(Pool), + _ = h2:stop_server(Server), + hackney_load_regulation:reset("localhost", Port) + end. + +cert_dir() -> + BeamDir = filename:dirname(code:which(?MODULE)), + Root = filename:join([BeamDir, "..", "..", "..", "..", ".."]), + filename:join([filename:absname(Root), "test", "certs"]). + +sync_request(Tag, URL, Opts) -> + Parent = self(), + spawn(fun() -> + Parent ! {result, Tag, hackney:request(get, URL, [], <<>>, Opts)} + end). + +result(Tag) -> + receive {result, Tag, Result} -> Result + after 5000 -> error({no_result, Tag}) + end. + +started(Path) -> + receive {request_started, Path, Handler, SConn} -> {Handler, SConn} + after 5000 -> error({not_started, Path}) + end. + +%% The pool keys shared connections by the TLS options hash, so read the one +%% registered for this port from the pool state instead of rebuilding the key. +shared_conn(Opts, Port) -> + [Conn] = shared_conns(Opts, Port), + Conn. + +no_shared_conn(Opts, Port) -> + shared_conns(Opts, Port) =:= []. + +shared_conns(Opts, Port) -> + PoolPid = hackney_pool:find_pool(proplists:get_value(pool, Opts)), + State = sys:get_state(PoolPid), + [Pid || Field <- tuple_to_list(State), is_map(Field), + {{_Host, P, hackney_ssl, _TlsKey}, Pid} <- maps:to_list(Field), + P =:= Port, is_pid(Pid)]. + +monitored(Pid) -> + {monitors, Monitors} = process_info(Pid, monitors), + lists:sort([P || {process, P} <- Monitors]). + +wait_until(Fun) -> + wait_until(Fun, erlang:monotonic_time(millisecond) + 5000). + +wait_until(Fun, Deadline) -> + case Fun() of + true -> ok; + false -> + case erlang:monotonic_time(millisecond) > Deadline of + true -> error(wait_until_timeout); + false -> + receive after 5 -> ok end, + wait_until(Fun, Deadline) + end + end.