1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: Apache-2.0
17 requestLimiterQuietPeriod = time.Second
18 requestLimiterInitialLimit int64 = 8
21 type requestLimiter struct {
29 // Acquire reserves one request slot, waiting if necessary.
31 // Acquire returns early if ctx cancels before a slot is available. It
32 // is assumed in this case the caller will immediately notice
33 // ctx.Err() != nil and call Release().
34 func (rl *requestLimiter) Acquire(ctx context.Context) {
37 // First use of requestLimiter. Initialize.
38 rl.cond = sync.NewCond(&rl.lock)
39 rl.limit = requestLimiterInitialLimit
41 // Wait out the quiet period(s) immediately following a 503.
42 for ctx.Err() == nil {
43 delay := rl.quietUntil.Sub(time.Now())
47 // Wait for the end of the quiet period, which started
48 // when we last received a 503 response.
50 timer := time.NewTimer(delay)
58 ready := make(chan struct{})
60 // close ready when a slot is available _or_ we wake
61 // up and find ctx has been canceled (meaning Acquire
62 // has already returned, or is about to).
63 for rl.limit > 0 && rl.limit <= rl.current && ctx.Err() == nil {
70 // Wait() returned, so we have the lock.
74 // When Wait() returns the lock to our goroutine
75 // (which might have already happened) we need to
76 // release it (if we don't do this now, the following
77 // Lock() can deadlock).
82 // Note we may have current > limit until the caller
90 // Release releases a slot that has been reserved with Acquire.
91 func (rl *requestLimiter) Release() {
98 // Report uses the return values from (*http.Client)Do() to adjust the
99 // outgoing request limit (increase on success, decrease on 503).
101 // Return value is true if the response was a 503.
102 func (rl *requestLimiter) Report(resp *http.Response, err error) bool {
104 defer rl.lock.Unlock()
108 if errors.As(err, &uerr) && uerr.Err.Error() == "Service Unavailable" {
109 // This is how http.Client reports 503 from proxy server
115 is503 = resp.StatusCode == http.StatusServiceUnavailable
119 // Concurrency was unlimited until now.
120 // Calculate new limit based on actual
121 // concurrency instead of previous limit.
122 rl.limit = rl.current
124 if time.Now().After(rl.quietUntil) {
125 // Reduce concurrency limit by half.
126 rl.limit = (rl.limit + 1) / 2
127 // Don't start any new calls (or reduce the
128 // limit even further on additional 503s) for
130 rl.quietUntil = time.Now().Add(requestLimiterQuietPeriod)
134 if err == nil && resp.StatusCode >= 200 && resp.StatusCode < 400 && rl.limit > 0 {
135 // After each non-server-error response, increase
136 // concurrency limit by at least 10% -- but not beyond
137 // 2x the highest concurrency level we've seen without
139 increase := rl.limit / 10
144 if max := rl.current * 2; max > rl.limit {