Improve high-concurrency client throughput

src/http_client.jl

@@ -1,5 +1,13 @@
 # High-level HTTP client orchestration, HTTP/2 integration, cookies, response sinks, and convenience APIs.
 
+# Copy-loop buffer for response bodies that are known (or likely) to be large:
+# 64 KiB measured faster than 8 KiB and 256 KiB for high-throughput downloads.
+# Loops whose payload is typically small — unknown-size accumulation of
+# chunked responses and decompression output — keep 8 KiB buffers so a tiny
+# response doesn't pay a 64 KiB allocation per request.
+const _RESPONSE_COPY_BUFFER_BYTES = 64 * 1024
+const _RESPONSE_SMALL_COPY_BUFFER_BYTES = 8 * 1024
+
 """
     Client(; ...)
 
@@ -15,7 +23,11 @@ Keyword arguments:
   followed
 - `cookiejar`: cookie jar implementation, or `nothing` to disable cookies
 - `max_redirects`: maximum redirect hops before failing
-- `prefer_http2`: whether secure requests should try HTTP/2 when available
+- `prefer_http2`: whether secure requests should try HTTP/2 when available.
+  When automatic negotiation (`protocol = :auto`) learns an origin only
+  speaks HTTP/1.1, that result is cached and later automatic requests skip
+  the HTTP/2 attempt; [`close_idle_connections!`](@ref) clears the cache.
+  An explicit `protocol = :h2` always attempts HTTP/2.
 - `http2_settings`: an [`HTTP2Settings`](@ref) configuring HTTP/2 receive
   flow-control windows for connections this client opens
 - `default_headers`: headers applied to every request issued through this
@@ -48,6 +60,12 @@ mutable struct Client{CR}
     http2_settings::HTTP2Settings
     h2_lock::ReentrantLock
     h2_conns::Dict{String,Vector{H2Connection}}
+    # Origins that negotiated HTTP/1.1 during `protocol = :auto`. Guarded by
+    # its own lock (never `h2_lock`): `h2_lock` is held across full TCP+TLS
+    # dials in `_acquire_h2_conn!`, and the hot-path membership check in
+    # `_use_h2` must not serialize behind those dials.
+    h1_origins_lock::ReentrantLock
+    h1_origins::Set{String}
     default_headers::Headers
     default_query::Union{Nothing,Vector{Pair{String,String}}}
     default_basicauth::Any
@@ -342,6 +360,8 @@ function Client(;
         http2_settings,
         ReentrantLock(),
         Dict{String,Vector{H2Connection}}(),
+        ReentrantLock(),
+        Set{String}(),
         _normalize_headers_input(default_headers),
         _normalize_default_query(default_query),
         default_basicauth,
@@ -445,10 +465,16 @@ function _acquire_h2_conn!(
     address::String,
     secure::Bool,
     request::Union{Nothing,Request}=nothing,
-    server_name::Union{Nothing,String}=nothing,
+    server_name::Union{Nothing,String}=nothing;
     allow_h1_alpn::Bool=false,
+    auto_protocol::Bool=false,
 )::H2Connection
     key = _h2_key(plan)
+    # Fail fast (and without touching `h2_lock`) when another task cached this
+    # origin as HTTP/1.1 after our caller's `_use_h2` check.
+    if auto_protocol && _h1_origin_cached(client, key)
+        throw(H2NegotiationError("http2: origin previously negotiated HTTP/1.1"))
+    end
     base_host_resolver = client.transport.host_resolver
     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)
@@ -457,8 +483,15 @@ function _acquire_h2_conn!(
     # 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)
+    waited_for_h2_lock = !trylock(client.h2_lock)
+    waited_for_h2_lock && lock(client.h2_lock)
     try
+        # The origin can be cached while this task waits for `h2_lock`. Only
+        # contended acquisitions need this second check, so the uncontended h2
+        # fast path does not pay for another cache lookup.
+        if waited_for_h2_lock && auto_protocol && _h1_origin_cached(client, key)
+            throw(H2NegotiationError("http2: origin previously negotiated HTTP/1.1"))
+        end
         conns = get(() -> H2Connection[], client.h2_conns, key)
         idle_timeout_ns = client.transport.idle_timeout_ns
         now_ns = Int64(time_ns())
@@ -504,36 +537,43 @@ function _acquire_h2_conn!(
             nothing
         end
         conn = nothing
-        conn = if plan.mode == _ProxyPlanMode.DIRECT
-            connect_h2!(
-                address;
-                secure=secure,
-                host_resolver=connect_host_resolver,
-                tls_config=tls_cfg,
-                connect_deadline_ns=connect_deadline_ns,
-                http2_settings=client.http2_settings,
-            )
-        elseif plan.mode == _ProxyPlanMode.HTTP_TUNNEL || _proxy_plan_is_socks(plan)
-            proxy = plan.proxy
-            proxy === nothing && throw(ProtocolError("proxy connection is missing proxy config"))
-            tcp = _new_tcp_conn!(plan, address, connect_host_resolver, connect_deadline_ns)
-            try
+        conn = try
+            if plan.mode == _ProxyPlanMode.DIRECT
                 connect_h2!(
-                    tcp,
                     address;
                     secure=secure,
+                    host_resolver=connect_host_resolver,
                     tls_config=tls_cfg,
                     connect_deadline_ns=connect_deadline_ns,
                     http2_settings=client.http2_settings,
                 )
-            catch
-                @try_ignore begin
-                    TCP.close(tcp)
+            elseif plan.mode == _ProxyPlanMode.HTTP_TUNNEL || _proxy_plan_is_socks(plan)
+                proxy = plan.proxy
+                proxy === nothing && throw(ProtocolError("proxy connection is missing proxy config"))
+                tcp = _new_tcp_conn!(plan, address, connect_host_resolver, connect_deadline_ns)
+                try
+                    connect_h2!(
+                        tcp,
+                        address;
+                        secure=secure,
+                        tls_config=tls_cfg,
+                        connect_deadline_ns=connect_deadline_ns,
+                        http2_settings=client.http2_settings,
+                    )
+                catch
+                    @try_ignore begin
+                        TCP.close(tcp)
+                    end
+                    rethrow()
                 end
-                rethrow()
+            else
+                throw(ArgumentError("HTTP/2 is not supported for proxy plan mode $(plan.mode)"))
             end
-        else
-            throw(ArgumentError("HTTP/2 is not supported for proxy plan mode $(plan.mode)"))
+        catch err
+            if auto_protocol && _should_fallback_h2_to_h1(err)
+                _cache_h1_origin!(client, key)
+            end
+            rethrow()
         end
         # Claim a slot on the freshly opened connection up front so subsequent
         # acquirers that race in see this caller's pending request.
@@ -581,7 +621,7 @@ function _drop_h2_conn!(client::Client, plan::_ProxyPlan, target::Union{Nothing,
     return nothing
 end
 
-function _use_h2(client::Client, secure::Bool, protocol::Symbol)::Bool
+function _use_h2(client::Client, plan::_ProxyPlan, secure::Bool, protocol::Symbol)::Bool
     protocol == :h1 && return false
     protocol == :h2 && return true
     protocol == :auto || throw(ArgumentError("protocol must be :auto, :h1, or :h2"))
@@ -594,7 +634,26 @@ function _use_h2(client::Client, secure::Bool, protocol::Symbol)::Bool
     if cfg !== nothing && !isempty(cfg.alpn_protocols) && !in("h2", cfg.alpn_protocols)
         return false
     end
-    return true
+    return !_h1_origin_cached(client, _h2_key(plan))
+end
+
+@inline function _h1_origin_cached(client::Client, key::String)::Bool
+    lock(client.h1_origins_lock)
+    try
+        return key in client.h1_origins
+    finally
+        unlock(client.h1_origins_lock)
+    end
+end
+
+function _cache_h1_origin!(client::Client, key::String)::Nothing
+    lock(client.h1_origins_lock)
+    try
+        push!(client.h1_origins, key)
+    finally
+        unlock(client.h1_origins_lock)
+    end
+    return nothing
 end
 
 function _host_path_from_request(address::String, request::Request)::Tuple{String,String}
@@ -678,11 +737,10 @@ function _store_set_cookies!(
 end
 
 function _clone_bytes_body(body::BytesBody)::BytesBody
-    remaining = (length(body.data) - body.next_index) + 1
-    remaining <= 0 && return BytesBody(UInt8[])
-    copied = Vector{UInt8}(undef, remaining)
-    copyto!(copied, 1, body.data, body.next_index, remaining)
-    return BytesBody(copied)
+    # BytesBody retains its backing bytes by contract. A replay only needs an
+    # independent cursor and closed flag; copying the payload makes every
+    # buffered upload allocate and copy the complete request body again.
+    return BytesBody(body.data, body.next_index, false)
 end
 
 function _clone_body(body::AbstractBody)::AbstractBody
@@ -792,7 +850,8 @@ function _do_incoming!(
             send_request = _copy_request_for_send(current_request, retry_attempt == 1)
             request_url = _request_url(current_secure, current_address, current_request.target)
             proxy_plan = _proxy_plan(proxy_config, current_secure, current_address)
-            use_h2 = _use_h2(client, current_secure, protocol) && proxy_plan.mode != _ProxyPlanMode.HTTP_FORWARD
+            use_h2 = proxy_plan.mode != _ProxyPlanMode.HTTP_FORWARD &&
+                _use_h2(client, proxy_plan, current_secure, protocol)
             _emit_trace(trace, RequestEvent(send_request, request_url, retry_attempt, redirect_count, use_h2 ? :h2 : :h1))
             host, path = _host_path_from_request(current_address, current_request)
             manual_cookies = cookies === false ? Cookie[] : Cookies.readcookies(send_request.headers, "")
@@ -808,8 +867,9 @@ function _do_incoming!(
                             current_address,
                             current_secure,
                             send_request,
-                            current_server_name,
-                            protocol == :auto,
+                            current_server_name;
+                            allow_h1_alpn=(protocol == :auto),
+                            auto_protocol=(protocol == :auto),
                         )
                         _h2_roundtrip_incoming!(conn::H2Connection, send_request; pending_slot_claimed=true)
                     catch err
@@ -1186,6 +1246,14 @@ end
 
 function close_idle_connections!(client::Client)
     close_idle_connections!(client.transport)
+    # Drop cached HTTP/1.1 negotiation results so long-lived clients re-probe
+    # origins that may have enabled HTTP/2 since they were first contacted.
+    lock(client.h1_origins_lock)
+    try
+        empty!(client.h1_origins)
+    finally
+        unlock(client.h1_origins_lock)
+    end
     # 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
@@ -1238,7 +1306,7 @@ end
 
 function _read_all_response_bytes(io::IO, limit::Int=0)::Vector{UInt8}
     out = UInt8[]
-    buf = Vector{UInt8}(undef, 8192)
+    buf = Vector{UInt8}(undef, _RESPONSE_SMALL_COPY_BUFFER_BYTES)
     total = 0
     while true
         n = readbytes!(io, buf, length(buf))
@@ -1260,7 +1328,10 @@ function _read_all_response_bytes(body::AbstractBody, content_length_hint::Int64
     end
     out = UInt8[]
     content_length_hint > 0 && sizehint!(out, Int(min(content_length_hint, _MAX_EAGER_RESPONSE_PREALLOC)))
-    buf = Vector{UInt8}(undef, 8192)
+    # A hint above the prealloc cap means a known-large download; without a
+    # hint (chunked/EOF-framed) the body is usually small, so stay at 8 KiB.
+    buf_bytes = content_length_hint > _MAX_EAGER_RESPONSE_PREALLOC ? _RESPONSE_COPY_BUFFER_BYTES : _RESPONSE_SMALL_COPY_BUFFER_BYTES
+    buf = Vector{UInt8}(undef, buf_bytes)
     while true
         n = body_read!(body, buf)
         n == 0 && return out
@@ -1269,7 +1340,7 @@ function _read_all_response_bytes(body::AbstractBody, content_length_hint::Int64
 end
 
 function _copy_response_bytes!(dest::IO, io::IO, limit::Int=0)::Int64
-    buf = Vector{UInt8}(undef, 8192)
+    buf = Vector{UInt8}(undef, _RESPONSE_SMALL_COPY_BUFFER_BYTES)
     total = Int64(0)
     while true
         n = readbytes!(io, buf, length(buf))
@@ -1281,9 +1352,12 @@ function _copy_response_bytes!(dest::IO, io::IO, limit::Int=0)::Int64
 end
 
 function _copy_response_bytes!(dest::AbstractVector{UInt8}, io::IO, limit::Int=0)::Int64
-    buf = Vector{UInt8}(undef, 8192)
-    total = 0
     capacity = length(dest)
+    # `dest` is preallocated to the expected size; a small response should not
+    # pay a 64 KiB scratch allocation. The 1-byte floor keeps the capacity
+    # overflow check below reachable when `dest` is empty.
+    buf = Vector{UInt8}(undef, min(max(capacity, 1), _RESPONSE_COPY_BUFFER_BYTES))
+    total = 0
     while true
         n = readbytes!(io, buf, length(buf))
         n == 0 && break
@@ -1298,7 +1372,7 @@ function _copy_response_bytes!(dest::AbstractVector{UInt8}, io::IO, limit::Int=0
 end
 
 function _copy_response_bytes!(dest::IO, body::AbstractBody)::Int64
-    buf = Vector{UInt8}(undef, 8192)
+    buf = Vector{UInt8}(undef, _RESPONSE_COPY_BUFFER_BYTES)
     total = Int64(0)
     while true
         n = body_read!(body, buf)
@@ -1309,9 +1383,10 @@ function _copy_response_bytes!(dest::IO, body::AbstractBody)::Int64
 end
 
 function _copy_response_bytes!(dest::AbstractVector{UInt8}, body::AbstractBody)::Int64
-    buf = Vector{UInt8}(undef, 8192)
-    total = 0
     capacity = length(dest)
+    # See the sizing rationale on the `io::IO` method above.
+    buf = Vector{UInt8}(undef, min(max(capacity, 1), _RESPONSE_COPY_BUFFER_BYTES))
+    total = 0
     while true
         n = body_read!(body, buf)
         n == 0 && break
@@ -1356,7 +1431,7 @@ end
 end
 
 function _pump_response_body!(stream::Base.BufferStream, body::AbstractBody)::Nothing
-    buf = Vector{UInt8}(undef, 8192)
+    buf = Vector{UInt8}(undef, _RESPONSE_COPY_BUFFER_BYTES)
     try
         while true
             n = body_read!(body, buf)
@@ -1466,7 +1541,7 @@ end
 
 function Base.read(io::_BodyIO)::Vector{UInt8}
     out = UInt8[]
-    buf = Vector{UInt8}(undef, 8192)
+    buf = Vector{UInt8}(undef, _RESPONSE_SMALL_COPY_BUFFER_BYTES)
     while true
         n = readbytes!(io, buf)
         n == 0 && break

src/http_transport.jl

@@ -60,7 +60,14 @@ end
     return reader.stop - reader.next + 1
 end
 
+@inline function _request_write_deadline_needs_refresh(request::Request)::Bool
+    return _request_write_idle_timeout_ns(request) > 0
+end
+
 @inline function _apply_request_write_deadline!(io::_RequestDeadlineWriteIO)::Nothing
+    # The overall request deadline is applied once when the connection is
+    # acquired. Only an idle timeout moves after each write and needs refresh.
+    _request_write_deadline_needs_refresh(io.request) || return nothing
     _set_conn_write_deadline!(io.conn::Conn, _request_write_deadline_ns(io.request))
     return nothing
 end
@@ -1192,7 +1199,9 @@ 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.
+the HTTP/1 pool it owns. They additionally clear the client's cache of origins
+that negotiated HTTP/1.1 under `protocol = :auto`, so subsequent automatic
+requests re-attempt HTTP/2 against origins that may have enabled it since.
 """
 function close_idle_connections!(transport::Transport)
     to_close = Conn[]