sdk/go/arvados/limiter.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: Apache-2.0
   4
   5 package arvados
   6
   7 import (
   8         "context"
   9         "errors"
  10         "net/http"
  11         "net/url"
  12         "sync"
  13         "time"
  14 )
  15
  16 var (
  17         requestLimiterQuietPeriod        = time.Second
  18         requestLimiterInitialLimit int64 = 8
  19 )
  20
  21 type requestLimiter struct {
  22         current    int64
  23         limit      int64
  24         lock       sync.Mutex
  25         cond       *sync.Cond
  26         quietUntil time.Time
  27 }
  28
  29 // Acquire reserves one request slot, waiting if necessary.
  30 //
  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) {
  35         rl.lock.Lock()
  36         if rl.cond == nil {
  37                 // First use of requestLimiter. Initialize.
  38                 rl.cond = sync.NewCond(&rl.lock)
  39                 rl.limit = requestLimiterInitialLimit
  40         }
  41         // Wait out the quiet period(s) immediately following a 503.
  42         for ctx.Err() == nil {
  43                 delay := rl.quietUntil.Sub(time.Now())
  44                 if delay < 0 {
  45                         break
  46                 }
  47                 // Wait for the end of the quiet period, which started
  48                 // when we last received a 503 response.
  49                 rl.lock.Unlock()
  50                 timer := time.NewTimer(delay)
  51                 select {
  52                 case <-timer.C:
  53                 case <-ctx.Done():
  54                         timer.Stop()
  55                 }
  56                 rl.lock.Lock()
  57         }
  58         ready := make(chan struct{})
  59         go func() {
  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 {
  64                         rl.cond.Wait()
  65                 }
  66                 close(ready)
  67         }()
  68         select {
  69         case <-ready:
  70                 // Wait() returned, so we have the lock.
  71                 rl.current++
  72                 rl.lock.Unlock()
  73         case <-ctx.Done():
  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).
  78                 go func() {
  79                         <-ready
  80                         rl.lock.Unlock()
  81                 }()
  82                 // Note we may have current > limit until the caller
  83                 // calls Release().
  84                 rl.lock.Lock()
  85                 rl.current++
  86                 rl.lock.Unlock()
  87         }
  88 }
  89
  90 // Release releases a slot that has been reserved with Acquire.
  91 func (rl *requestLimiter) Release() {
  92         rl.lock.Lock()
  93         rl.current--
  94         rl.lock.Unlock()
  95         rl.cond.Signal()
  96 }
  97
  98 // Report uses the return values from (*http.Client)Do() to adjust the
  99 // outgoing request limit (increase on success, decrease on 503).
 100 //
 101 // Return value is true if the response was a 503.
 102 func (rl *requestLimiter) Report(resp *http.Response, err error) bool {
 103         rl.lock.Lock()
 104         defer rl.lock.Unlock()
 105         is503 := false
 106         if err != nil {
 107                 uerr := &url.Error{}
 108                 if errors.As(err, &uerr) && uerr.Err.Error() == "Service Unavailable" {
 109                         // This is how http.Client reports 503 from proxy server
 110                         is503 = true
 111                 } else {
 112                         return false
 113                 }
 114         } else {
 115                 is503 = resp.StatusCode == http.StatusServiceUnavailable
 116         }
 117         if is503 {
 118                 if rl.limit == 0 {
 119                         // Concurrency was unlimited until now.
 120                         // Calculate new limit based on actual
 121                         // concurrency instead of previous limit.
 122                         rl.limit = rl.current
 123                 }
 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
 129                         // a second.
 130                         rl.quietUntil = time.Now().Add(requestLimiterQuietPeriod)
 131                 }
 132                 return true
 133         }
 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
 138                 // a failure.
 139                 increase := rl.limit / 10
 140                 if increase < 1 {
 141                         increase = 1
 142                 }
 143                 rl.limit += increase
 144                 if max := rl.current * 2; max < rl.limit {
 145                         rl.limit = max
 146                 }
 147                 rl.cond.Broadcast()
 148         }
 149         return false
 150 }