diff --git a/src/http2_client.jl b/src/http2_client.jl index 155e34e46..f7e86a137 100644 --- a/src/http2_client.jl +++ b/src/http2_client.jl @@ -187,6 +187,14 @@ mutable struct H2Connection # `_h2_conn_available` so a second caller doesn't decide a peer-capped # connection is reusable while the first caller is still in flight. pending_stream_count::Int + # `time_ns()` at the moment the connection last became fully idle (no + # registered streams and no pending reservations); stamped at creation and + # on each transition into that state. Guarded by `state_lock`. The pool + # uses it to evict connections idle past the transport's `idle_timeout_ns`: + # NATs and load balancers silently drop idle flows, and a request written + # onto such a connection stalls until the kernel's retransmission limit + # (~15 min on Linux) because no RST/FIN ever arrives (#1331). + idle_since_ns::Int64 @atomic closed::Bool end @@ -539,6 +547,7 @@ function _release_h2_pending_slot!(conn::H2Connection) lock(conn.state_lock) try conn.pending_stream_count -= 1 + _h2_stamp_idle_locked!(conn) notify(conn.stream_condition; all=true) finally unlock(conn.state_lock) @@ -546,6 +555,51 @@ function _release_h2_pending_slot!(conn::H2Connection) return nothing end +# Record the moment the connection becomes fully idle (no registered streams, +# no pending reservations). Callers must hold `state_lock`. Stamping only on +# the transition into the idle state keeps `idle_since_ns` monotone with real +# inactivity: a connection that keeps serving traffic is never considered idle, +# and one that goes quiet ages from the moment its last stream finished. +@inline function _h2_stamp_idle_locked!(conn::H2Connection) + if isempty(conn.streams) && conn.pending_stream_count <= 0 + conn.idle_since_ns = Int64(time_ns()) + end + return nothing +end + +""" + _h2_conn_idle_expired(conn, idle_timeout_ns, now_ns) -> Bool + +`true` when `conn` has no registered streams or pending reservations and has +been idle longer than `idle_timeout_ns` (`0` disables the check). Pooled HTTP/2 +connections idle past the network's silent-drop window (NAT/load-balancer idle +timeouts are commonly a few minutes) must not be handed out again: the peer is +gone but no RST/FIN ever arrives, so the next request stalls until the kernel +retransmission limit (#1331). +""" +function _h2_conn_idle_expired(conn::H2Connection, idle_timeout_ns::Int64, now_ns::Int64)::Bool + idle_timeout_ns > 0 || return false + lock(conn.state_lock) + try + isempty(conn.streams) || return false + conn.pending_stream_count > 0 && return false + return (now_ns - conn.idle_since_ns) > idle_timeout_ns + finally + unlock(conn.state_lock) + end +end + +# `true` when the connection carries no registered streams and no pending +# reservations right now (regardless of how long it has been that way). +function _h2_conn_fully_idle(conn::H2Connection)::Bool + lock(conn.state_lock) + try + return isempty(conn.streams) && conn.pending_stream_count <= 0 + finally + unlock(conn.state_lock) + end +end + function _h2_conn_reusable(conn::H2Connection)::Bool lock(conn.state_lock) try @@ -695,6 +749,7 @@ function _unregister_stream!(conn::H2Connection, stream_id::UInt32) pop!(conn.streams, stream_id, nothing) pop!(conn.stream_send_window, stream_id, nothing) delete!(conn.send_closed_streams, stream_id) + _h2_stamp_idle_locked!(conn) notify(conn.stream_condition; all=true) finally unlock(conn.state_lock) @@ -1195,6 +1250,7 @@ function _connect_h2_from_tcp!( nothing, UInt8[], 0, + Int64(time_ns()), false, ) _verify_h2_alpn!(conn) diff --git a/src/http_client.jl b/src/http_client.jl index b722dac8b..f8c640773 100644 --- a/src/http_client.jl +++ b/src/http_client.jl @@ -453,18 +453,31 @@ function _acquire_h2_conn!( connect_host_resolver = request === nothing ? base_host_resolver : _request_connect_host_resolver(base_host_resolver, request::Request) connect_deadline_ns = request === nothing ? _phase_deadline_ns(base_host_resolver.timeout_ns, base_host_resolver.deadline_ns) : _request_connect_phase_deadline_ns(base_host_resolver, request::Request) tls_handshake_timeout_ns = request === nothing ? Int64(0) : _request_connect_phase_timeout_ns(base_host_resolver, request::Request) + # Connections culled from the pool are closed after `h2_lock` is released: + # closing an H2Connection waits for its read loop to exit, and that must + # not happen while holding the pool lock. + to_close = H2Connection[] lock(client.h2_lock) try conns = get(() -> H2Connection[], client.h2_conns, key) + idle_timeout_ns = client.transport.idle_timeout_ns + now_ns = Int64(time_ns()) i = 1 while i <= length(conns) existing = conns[i] + # Evict connections idle past the pool's idle timeout before + # considering them for reuse: a NAT or load balancer may have + # silently dropped the flow, and the first read on such a + # connection stalls until the kernel retransmission limit (#1331). + if _h2_conn_idle_expired(existing::H2Connection, idle_timeout_ns, now_ns) + deleteat!(conns, i) + push!(to_close, existing::H2Connection) + continue + end if !_h2_conn_available(existing::H2Connection) if !_h2_conn_reusable(existing::H2Connection) deleteat!(conns, i) - @try_ignore begin - close(existing::H2Connection) - end + push!(to_close, existing::H2Connection) continue end else @@ -530,6 +543,11 @@ function _acquire_h2_conn!( return conn::H2Connection finally unlock(client.h2_lock) + for culled in to_close + @try_ignore begin + close(culled) + end + end end end @@ -1167,7 +1185,42 @@ function _default_client!()::Client{Nothing} end end -close_idle_connections!(client::Client) = close_idle_connections!(client.transport) +function close_idle_connections!(client::Client) + close_idle_connections!(client.transport) + # Also close pooled HTTP/2 connections with no in-flight streams — these + # live on the Client (not the Transport pool) and are equally subject to + # silent idle drops by NATs/load balancers (#1331). Connections carrying + # active streams or pending reservations are left untouched. Closing + # happens after `h2_lock` is released (close waits for the read loop). + to_close = H2Connection[] + lock(client.h2_lock) + try + for key in collect(keys(client.h2_conns)) + conns = client.h2_conns[key] + kept = H2Connection[] + for conn in conns + if _h2_conn_fully_idle(conn) + push!(to_close, conn) + else + push!(kept, conn) + end + end + if isempty(kept) + pop!(client.h2_conns, key, nothing) + else + client.h2_conns[key] = kept + end + end + finally + unlock(client.h2_lock) + end + for conn in to_close + @try_ignore begin + close(conn) + end + end + return nothing +end function close_idle_connections!() lock(_DEFAULT_CLIENT_LOCK) @@ -1177,7 +1230,7 @@ function close_idle_connections!() unlock(_DEFAULT_CLIENT_LOCK) end client === nothing && return nothing - return close_idle_connections!(client.transport) + return close_idle_connections!(client) end function _status_throws(resp::Response)::Bool diff --git a/src/http_transport.jl b/src/http_transport.jl index 5de009636..36284f9ed 100644 --- a/src/http_transport.jl +++ b/src/http_transport.jl @@ -193,6 +193,12 @@ versus closed. bound the total live HTTP/1 connections (idle, in-flight, and dialing) for one pool key and cause additional acquires to wait for direct handoff or a freed dial slot. + +`idle_timeout_ns` (default 90s, `0` disables) bounds how long an idle pooled +connection may be handed out again; it also governs the owning `Client`'s +pooled HTTP/2 connections. Keep it under the network's silent idle-drop window +(NAT/load-balancer idle timeouts are commonly a few minutes) so a request is +never written onto a connection whose peer has silently vanished. """ mutable struct Transport host_resolver::_TransportHostResolver @@ -1183,6 +1189,10 @@ The no-argument form closes idle connections held by the default client used by `HTTP.get`, `HTTP.post`, `HTTP.request`, and friends (a no-op if no request has been made yet). Pass an `HTTP.Client` or `HTTP.Transport` to target a specific connection pool. + +The `Client` and no-argument forms also close the client's pooled HTTP/2 +connections that have no in-flight streams; the `Transport` form covers only +the HTTP/1 pool it owns. """ function close_idle_connections!(transport::Transport) to_close = Conn[] diff --git a/test/http2_client_tests.jl b/test/http2_client_tests.jl index 1ab3604d1..603780612 100644 --- a/test/http2_client_tests.jl +++ b/test/http2_client_tests.jl @@ -2102,6 +2102,7 @@ function _build_bare_h2_connection() nothing, UInt8[], 0, + Int64(time_ns()), false, ) cleanup = () -> begin @@ -2271,3 +2272,161 @@ end cleanup() end end + +# ── #1331: pooled h2 connections must not outlive the idle timeout ────────── + +function _h2_serve_one_get!(c::NC.Conn, reader, enc, dec) + sid = UInt32(0) + while true + f = HT.read_frame!(reader) + if f isa HT.HeadersFrame + hf = f::HT.HeadersFrame + _ = HT.decode_header_block(dec, hf.header_block_fragment) + sid = hf.stream_id + hf.end_stream && break + elseif f isa HT.DataFrame && (f::HT.DataFrame).end_stream + break + end + end + blk = HT.encode_header_block(enc, HT.HeaderField[HT.HeaderField(":status", "200", false)]) + _write_frame_to_conn!(c, HT.HeadersFrame(sid, true, true, blk)) + return nothing +end + +function _h2_handshake_server!(c::NC.Conn) + reader = HT._ConnReader(c) + _ = _read_exact_h2_tcp!(c, length(HT._H2_PREFACE)) + _ = HT.read_frame!(reader) + _write_frame_to_conn!(c, HT.SettingsFrame(false, Pair{UInt16, UInt32}[])) + return reader +end + +@testset "HTTP/2 pool evicts connections idle past the transport idle timeout (#1331)" begin + # NATs/load balancers silently drop idle flows; a pooled connection older + # than the idle timeout must be discarded at checkout, not handed out (the + # first read on a silently-dead connection stalls until the kernel + # retransmission limit, ~15 min on Linux). + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + address = ND.join_host_port("127.0.0.1", Int((NC.addr(listener)::NC.SocketAddrV4).port)) + accepts = Threads.Atomic{Int}(0) + server_task = errormonitor(Threads.@spawn begin + # conn 1: answer one request, then go mute (stand-in for a dead path) + c1 = NC.accept(listener) + Threads.atomic_add!(accepts, 1) + reader1 = _h2_handshake_server!(c1) + enc1 = HT.Encoder(); dec1 = HT.Decoder() + _h2_serve_one_get!(c1, reader1, enc1, dec1) + # conn 2: the client must arrive here on a fresh connection + c2 = NC.accept(listener) + Threads.atomic_add!(accepts, 1) + reader2 = _h2_handshake_server!(c2) + enc2 = HT.Encoder(); dec2 = HT.Decoder() + _h2_serve_one_get!(c2, reader2, enc2, dec2) + HTTP.@try_ignore NC.close(c1) + HTTP.@try_ignore NC.close(c2) + return nothing + end) + client = HT.Client(transport = HT.Transport(idle_timeout_ns = Int64(150_000_000))) # 150ms + try + r1 = HT.get("http://$(address)/one"; client = client, protocol = :h2) + @test r1.status == 200 + pooled = lock(client.h2_lock) do + only(only(values(client.h2_conns))) + end + sleep(0.4) # exceed the idle timeout + started = time() + r2 = HT.get("http://$(address)/two"; client = client, protocol = :h2, retry = false) + elapsed = time() - started + @test r2.status == 200 + @test elapsed < 5.0 # a reused dead connection would hang + @test accepts[] == 2 # request 2 ran on a fresh connection + # the evicted connection was closed, and the pool holds only the new one + @test timedwait(() -> (@atomic :acquire pooled.closed), 5.0; pollint = 0.01) == :ok + n = lock(client.h2_lock) do + sum(length(v) for v in values(client.h2_conns); init = 0) + end + @test n == 1 + _wait_task_h2!(server_task) + finally + close(client) + HTTP.@try_ignore NC.close(listener) + end +end + +@testset "HTTP/2 pool reuses fresh connections and honors idle_timeout_ns=0 (#1331)" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + address = ND.join_host_port("127.0.0.1", Int((NC.addr(listener)::NC.SocketAddrV4).port)) + accepts = Threads.Atomic{Int}(0) + server_task = errormonitor(Threads.@spawn begin + c = NC.accept(listener) + Threads.atomic_add!(accepts, 1) + reader = _h2_handshake_server!(c) + enc = HT.Encoder(); dec = HT.Decoder() + for _ in 1:3 + _h2_serve_one_get!(c, reader, enc, dec) + end + HTTP.@try_ignore NC.close(c) + return nothing + end) + # idle_timeout_ns = 0 disables eviction entirely (matching the h1 pool), + # and a fresh connection inside the window must be reused, not evicted. + client = HT.Client(transport = HT.Transport(idle_timeout_ns = Int64(0))) + try + for i in 1:3 + r = HT.get("http://$(address)/r$(i)"; client = client, protocol = :h2) + @test r.status == 200 + sleep(0.05) + end + @test accepts[] == 1 # every request reused the single pooled conn + _wait_task_h2!(server_task) + finally + close(client) + HTTP.@try_ignore NC.close(listener) + end +end + +@testset "close_idle_connections! closes idle pooled HTTP/2 connections (#1331)" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + address = ND.join_host_port("127.0.0.1", Int((NC.addr(listener)::NC.SocketAddrV4).port)) + server_task = errormonitor(Threads.@spawn begin + c = NC.accept(listener) + reader = _h2_handshake_server!(c) + enc = HT.Encoder(); dec = HT.Decoder() + _h2_serve_one_get!(c, reader, enc, dec) + # linger so closing is client-initiated + HTTP.@try_ignore begin + while true + _ = HT.read_frame!(reader) + end + end + HTTP.@try_ignore NC.close(c) + return nothing + end) + client = HT.Client() + try + r = HT.get("http://$(address)/warm"; client = client, protocol = :h2) + @test r.status == 200 + conn = lock(client.h2_lock) do + only(only(values(client.h2_conns))) + end + # a connection with a pending reservation is active: it must be kept + @test HT._try_claim_h2_pending_slot!(conn) + HT.close_idle_connections!(client) + kept = lock(client.h2_lock) do + sum(length(v) for v in values(client.h2_conns); init = 0) + end + @test kept == 1 + @test !(@atomic :acquire conn.closed) + # once fully idle again, it is closed and removed + HT._release_h2_pending_slot!(conn) + HT.close_idle_connections!(client) + remaining = lock(client.h2_lock) do + sum(length(v) for v in values(client.h2_conns); init = 0) + end + @test remaining == 0 + @test timedwait(() -> (@atomic :acquire conn.closed), 5.0; pollint = 0.01) == :ok + finally + close(client) + HTTP.@try_ignore NC.close(listener) + end +end