1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: Apache-2.0
16 var requestLimiterQuietPeriod = time.Second
18 type requestLimiter struct {
26 // Acquire reserves one request slot, waiting if necessary.
28 // Acquire returns early if ctx cancels before a slot is available. It
29 // is assumed in this case the caller will immediately notice
30 // ctx.Err() != nil and call Release().
31 func (rl *requestLimiter) Acquire(ctx context.Context) {
34 // First use of requestLimiter. Initialize.
35 rl.cond = sync.NewCond(&rl.lock)
37 // Wait out the quiet period(s) immediately following a 503.
38 for ctx.Err() == nil {
39 delay := rl.quietUntil.Sub(time.Now())
43 // Wait for the end of the quiet period, which started
44 // when we last received a 503 response.
46 timer := time.NewTimer(delay)
54 ready := make(chan struct{})
56 // close ready when a slot is available _or_ we wake
57 // up and find ctx has been canceled (meaning Acquire
58 // has already returned, or is about to).
59 for rl.limit > 0 && rl.limit <= rl.current && ctx.Err() == nil {
66 // Wait() returned, so we have the lock.
70 // When Wait() returns the lock to our goroutine
71 // (which might have already happened) we need to
72 // release it (if we don't do this now, the following
73 // Lock() can deadlock).
78 // Note we may have current > limit until the caller
86 // Release releases a slot that has been reserved with Acquire.
87 func (rl *requestLimiter) Release() {
94 // Report uses the return values from (*http.Client)Do() to adjust the
95 // outgoing request limit (increase on success, decrease on 503).
97 // Return value is true if the response was a 503.
98 func (rl *requestLimiter) Report(resp *http.Response, err error) bool {
100 defer rl.lock.Unlock()
104 if errors.As(err, &uerr) && uerr.Err.Error() == "Service Unavailable" {
105 // This is how http.Client reports 503 from proxy server
111 is503 = resp.StatusCode == http.StatusServiceUnavailable
115 // Concurrency was unlimited until now.
116 // Calculate new limit based on actual
117 // concurrency instead of previous limit.
118 rl.limit = rl.current
120 if time.Now().After(rl.quietUntil) {
121 // Reduce concurrency limit by half.
122 rl.limit = (rl.limit + 1) / 2
123 // Don't start any new calls (or reduce the
124 // limit even further on additional 503s) for
126 rl.quietUntil = time.Now().Add(requestLimiterQuietPeriod)
130 if err == nil && resp.StatusCode >= 200 && resp.StatusCode < 400 && rl.limit > 0 {
131 // After each non-server-error response, increase
132 // concurrency limit by at least 10% -- but not beyond
133 // 2x the highest concurrency level we've seen without
135 increase := rl.limit / 10
140 if max := rl.current * 2; max > rl.limit {