diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c6895a2a6..bc24ee265 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -26,6 +26,7 @@ jobs: fail-fast: false matrix: version: + - '1.10' - '1' - 'pre' os: diff --git a/Project.toml b/Project.toml index 3dbc01c9c..8e0e8a2e6 100644 --- a/Project.toml +++ b/Project.toml @@ -1,6 +1,6 @@ name = "HTTP" uuid = "cd3eb016-35fb-5094-929b-558a96fad6f3" -version = "2.6.6" +version = "2.6.7" authors = ["Jacob Quinn", "contributors: https://github.com/JuliaWeb/HTTP.jl/graphs/contributors"] [deps] diff --git a/src/http_retry.jl b/src/http_retry.jl index 70c4f21a3..7ecb05953 100644 --- a/src/http_retry.jl +++ b/src/http_retry.jl @@ -61,10 +61,11 @@ mutable struct RetryBucket capacity::Int partitions::Dict{String,_RetryPartition} lock::ReentrantLock - # Copy-on-write snapshot of partition keys below full capacity. Published - # snapshots are treated as immutable. Writers publish under `lock`; readers - # use the snapshot to avoid locking for healthy traffic to unrelated keys. - @atomic depleted_partitions::Set{String} + # Number of partition keys below full capacity. Writers update it under + # `lock`; readers use it to avoid locking while every partition is full. + # Keep this field pointer-free: atomic references to GC-managed containers + # can corrupt the referenced object on Julia 1.10 (#1355). + @atomic depleted_partitions::Int end """Handle returned by `acquire` and consumed by `release` to refund retry budget.""" @@ -105,7 +106,7 @@ function RetryBucket(; Int(capacity), Dict{String,_RetryPartition}(), ReentrantLock(), - Set{String}(), + 0, ) end @@ -118,9 +119,7 @@ function RetryBucket( partitions::Dict{String,_RetryPartition}, lock::ReentrantLock, ) - depleted_partitions = Set( - key for (key, state) in partitions if state.capacity < capacity - ) + depleted_partitions = count(state -> state.capacity < capacity, values(partitions)) return RetryBucket( backoff_scale_factor_ms, max_backoff_secs, @@ -131,6 +130,27 @@ function RetryBucket( ) end +# Preserve the six-field constructor exposed in HTTP 2.6.6. The set argument +# is accepted only for compatibility; the pointer-free depleted-partition +# count is derived from `partitions` so the count invariant holds even when +# the caller's set disagrees with the partition states. +function RetryBucket( + backoff_scale_factor_ms::Int, + max_backoff_secs::Int, + capacity::Int, + partitions::Dict{String,_RetryPartition}, + lock::ReentrantLock, + ::Set{String}, +) + return RetryBucket( + backoff_scale_factor_ms, + max_backoff_secs, + capacity, + partitions, + lock, + ) +end + function RetryBucket( backoff_scale_factor_ms, max_backoff_secs, @@ -147,11 +167,10 @@ function RetryBucket( ) end -# Set a partition's capacity while keeping the published depleted-key snapshot -# in sync. Must be called with `bucket.lock` held. +# Set a partition's capacity while keeping the depleted-partition count in sync. +# Must be called with `bucket.lock` held. @inline function _retry_partition_set_capacity!( bucket::RetryBucket, - partition_key::String, state::_RetryPartition, new_capacity::Int, )::Nothing @@ -159,13 +178,12 @@ end now_full = new_capacity >= bucket.capacity state.capacity = new_capacity if was_full && !now_full - depleted = copy(@atomic :acquire bucket.depleted_partitions) - push!(depleted, partition_key) - @atomic :release bucket.depleted_partitions = depleted + depleted = @atomic :monotonic bucket.depleted_partitions + @atomic :release bucket.depleted_partitions = depleted + 1 elseif !was_full && now_full - depleted = copy(@atomic :acquire bucket.depleted_partitions) - delete!(depleted, partition_key) - @atomic :release bucket.depleted_partitions = depleted + depleted = @atomic :monotonic bucket.depleted_partitions + depleted > 0 || error("retry bucket depleted-partition count underflow") + @atomic :release bucket.depleted_partitions = depleted - 1 end return nothing end @@ -190,7 +208,7 @@ function acquire(bucket::RetryBucket, partition) if state.capacity < _RETRY_BUCKET_ACQUIRE_COST throw(RetryDeniedError(partition_key)) end - _retry_partition_set_capacity!(bucket, partition_key, state, state.capacity - _RETRY_BUCKET_ACQUIRE_COST) + _retry_partition_set_capacity!(bucket, state, state.capacity - _RETRY_BUCKET_ACQUIRE_COST) return RetryToken(bucket, partition_key, _RETRY_BUCKET_ACQUIRE_COST, false) end end @@ -222,7 +240,7 @@ end reserved = _retry_bucket_reserved_cost(token) consumed = min(reserved, max(0, failure_cost)) refund = reserved - consumed - _retry_partition_set_capacity!(bucket, token.partition, state, min(bucket.capacity, state.capacity + refund)) + _retry_partition_set_capacity!(bucket, state, min(bucket.capacity, state.capacity + refund)) token.released = true return nothing finally @@ -238,21 +256,20 @@ non-retried request, capped at the bucket's full capacity. This is the slow recovery path that lets a partition legitimately drained by a burst of real failures regain retry budget from healthy traffic instead of staying empty for the transport's lifetime. Partitions that have never spent capacity are left -untouched. The published depleted-key snapshot also keeps this lock-free for -healthy traffic to other partitions. +untouched. The depleted-partition count keeps this lock-free while all +partitions are healthy. """ function _retry_bucket_replenish!(bucket::RetryBucket, partition)::Nothing depleted = @atomic :acquire bucket.depleted_partitions - isempty(depleted) && return nothing + depleted == 0 && return nothing partition_key = _retry_bucket_partition_key(partition) - partition_key in depleted || return nothing lock(bucket.lock) try state = get(() -> nothing, bucket.partitions, partition_key) state === nothing && return nothing partition_state = state::_RetryPartition partition_state.capacity >= bucket.capacity && return nothing - _retry_partition_set_capacity!(bucket, partition_key, partition_state, partition_state.capacity + 1) + _retry_partition_set_capacity!(bucket, partition_state, partition_state.capacity + 1) return nothing finally unlock(bucket.lock) diff --git a/test/http_retry_tests.jl b/test/http_retry_tests.jl index 6362493ec..9286c61d7 100644 --- a/test/http_retry_tests.jl +++ b/test/http_retry_tests.jl @@ -116,7 +116,15 @@ end partitions = Dict{String,HT._RetryPartition}("depleted.example" => HT._RetryPartition(5)) positional = HT.RetryBucket(25, 20, 10, partitions, ReentrantLock()) @test positional.partitions === partitions - @test (@atomic :acquire positional.depleted_partitions) == Set(["depleted.example"]) + @test (@atomic :acquire positional.depleted_partitions) == 1 + + six_field = HT.RetryBucket(25, 20, 10, partitions, ReentrantLock(), Set(["depleted.example"])) + @test (@atomic :acquire six_field.depleted_partitions) == 1 + + # The six-field constructor derives the count from the partition states; + # a stale legacy set must not break the count invariant. + stale_set = HT.RetryBucket(25, 20, 10, partitions, ReentrantLock(), Set{String}()) + @test (@atomic :acquire stale_set.depleted_partitions) == 1 converted = HT.RetryBucket(Int32(25), Int16(20), Int8(10), copy(partitions), ReentrantLock()) @test converted.backoff_scale_factor_ms === 25 @@ -210,7 +218,7 @@ end @testset "HTTP retry bucket replenishes consumed capacity (#1353)" begin bucket = HT.RetryBucket(capacity = 20) - @test isempty(@atomic :acquire bucket.depleted_partitions) + @test (@atomic :acquire bucket.depleted_partitions) == 0 # Replenish before any capacity was ever spent is a lock-free no-op and # creates no partitions. @@ -218,7 +226,7 @@ end @test isempty(bucket.partitions) token = Base.acquire(bucket, "svc.example") - @test (@atomic :acquire bucket.depleted_partitions) == Set(["svc.example"]) + @test (@atomic :acquire bucket.depleted_partitions) == 1 Base.release(bucket, token, HT._RETRY_BUCKET_ACQUIRE_COST) @test bucket.partitions["svc.example"].capacity == 10 @@ -226,22 +234,48 @@ end HT._retry_bucket_replenish!(bucket, "svc.example") end @test bucket.partitions["svc.example"].capacity == 15 - @test (@atomic :acquire bucket.depleted_partitions) == Set(["svc.example"]) + @test (@atomic :acquire bucket.depleted_partitions) == 1 # Case-insensitive, and capped at full capacity. for _ in 1:10 HT._retry_bucket_replenish!(bucket, "SVC.example") end @test bucket.partitions["svc.example"].capacity == 20 - @test isempty(@atomic :acquire bucket.depleted_partitions) + @test (@atomic :acquire bucket.depleted_partitions) == 0 # Untouched partitions are not affected by another partition's depletion. other = Base.acquire(bucket, "other.example") HT._retry_bucket_replenish!(bucket, "svc.example") @test bucket.partitions["svc.example"].capacity == 20 - @test (@atomic :acquire bucket.depleted_partitions) == Set(["other.example"]) + @test (@atomic :acquire bucket.depleted_partitions) == 1 Base.release(bucket, other, 0) - @test isempty(@atomic :acquire bucket.depleted_partitions) + @test (@atomic :acquire bucket.depleted_partitions) == 0 +end + +@testset "HTTP retry bucket uses a pointer-free concurrent fast path (#1355)" begin + bucket = HT.RetryBucket(capacity = 20) + @test isbitstype(fieldtype(HT.RetryBucket, :depleted_partitions)) + workers = max(4, 2 * Threads.nthreads()) + @sync begin + for worker in 1:workers + Threads.@spawn begin + key = "svc-$worker.example" + for _ in 1:2_000 + token = Base.acquire(bucket, key) + Base.release(bucket, token, HT._RETRY_BUCKET_ACQUIRE_COST) + for _ in 1:HT._RETRY_BUCKET_ACQUIRE_COST + HT._retry_bucket_replenish!(bucket, key) + end + end + end + end + Threads.@spawn for _ in 1:100 + GC.gc(false) + yield() + end + end + @test all(state.capacity == bucket.capacity for state in values(bucket.partitions)) + @test (@atomic :acquire bucket.depleted_partitions) == 0 end @testset "HTTP retry bucket heals only after successful responses" begin @@ -734,7 +768,7 @@ end end @test err === trace_err @test bucket.partitions["127.0.0.1"].capacity == 10 - @test isempty(@atomic :acquire bucket.depleted_partitions) + @test (@atomic :acquire bucket.depleted_partitions) == 0 @test lock(transport.lock) do isempty(transport.conns_per_host) end @@ -779,7 +813,7 @@ end end @test err === policy_err @test bucket.partitions["127.0.0.1"].capacity == 10 - @test isempty(@atomic :acquire bucket.depleted_partitions) + @test (@atomic :acquire bucket.depleted_partitions) == 0 @test lock(transport.lock) do isempty(transport.conns_per_host) end @@ -925,7 +959,7 @@ end # The armed retry reserved 10 and recovered with a 200, so the # reservation was refunded in full instead of consumed (#1353). @test bucket.partitions["127.0.0.1"].capacity == 20 - @test isempty(@atomic :acquire bucket.depleted_partitions) + @test (@atomic :acquire bucket.depleted_partitions) == 0 finally HTTP.@try_ignore NC.close(listener) end diff --git a/test/public_api_tests.jl b/test/public_api_tests.jl index 0a05190c8..9b9661cc1 100644 --- a/test/public_api_tests.jl +++ b/test/public_api_tests.jl @@ -48,8 +48,11 @@ using HTTP @test true # `public` unsupported before 1.11; nothing to assert end - # Internals must stay private regardless of Julia version. - @test !Base.ispublic(HTTP, :_retryable_request_error) - @test !Base.ispublic(HTTP, :_normalize_local_addr) - @test !Base.ispublic(HTTP.WebSockets, :_ws_mask_into!) + if isdefined(Base, :ispublic) + @test !Base.ispublic(HTTP, :_retryable_request_error) + @test !Base.ispublic(HTTP, :_normalize_local_addr) + @test !Base.ispublic(HTTP.WebSockets, :_ws_mask_into!) + else + @test true + end end