cmd/bpa-operator/vendor/golang.org/x/sync/semaphore/semaphore.go

   1 // Copyright 2017 The Go Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style
   3 // license that can be found in the LICENSE file.
   4
   5 // Package semaphore provides a weighted semaphore implementation.
   6 package semaphore // import "golang.org/x/sync/semaphore"
   7
   8 import (
   9         "container/list"
  10         "context"
  11         "sync"
  12 )
  13
  14 type waiter struct {
  15         n     int64
  16         ready chan<- struct{} // Closed when semaphore acquired.
  17 }
  18
  19 // NewWeighted creates a new weighted semaphore with the given
  20 // maximum combined weight for concurrent access.
  21 func NewWeighted(n int64) *Weighted {
  22         w := &Weighted{size: n}
  23         return w
  24 }
  25
  26 // Weighted provides a way to bound concurrent access to a resource.
  27 // The callers can request access with a given weight.
  28 type Weighted struct {
  29         size    int64
  30         cur     int64
  31         mu      sync.Mutex
  32         waiters list.List
  33 }
  34
  35 // Acquire acquires the semaphore with a weight of n, blocking until resources
  36 // are available or ctx is done. On success, returns nil. On failure, returns
  37 // ctx.Err() and leaves the semaphore unchanged.
  38 //
  39 // If ctx is already done, Acquire may still succeed without blocking.
  40 func (s *Weighted) Acquire(ctx context.Context, n int64) error {
  41         s.mu.Lock()
  42         if s.size-s.cur >= n && s.waiters.Len() == 0 {
  43                 s.cur += n
  44                 s.mu.Unlock()
  45                 return nil
  46         }
  47
  48         if n > s.size {
  49                 // Don't make other Acquire calls block on one that's doomed to fail.
  50                 s.mu.Unlock()
  51                 <-ctx.Done()
  52                 return ctx.Err()
  53         }
  54
  55         ready := make(chan struct{})
  56         w := waiter{n: n, ready: ready}
  57         elem := s.waiters.PushBack(w)
  58         s.mu.Unlock()
  59
  60         select {
  61         case <-ctx.Done():
  62                 err := ctx.Err()
  63                 s.mu.Lock()
  64                 select {
  65                 case <-ready:
  66                         // Acquired the semaphore after we were canceled.  Rather than trying to
  67                         // fix up the queue, just pretend we didn't notice the cancelation.
  68                         err = nil
  69                 default:
  70                         s.waiters.Remove(elem)
  71                 }
  72                 s.mu.Unlock()
  73                 return err
  74
  75         case <-ready:
  76                 return nil
  77         }
  78 }
  79
  80 // TryAcquire acquires the semaphore with a weight of n without blocking.
  81 // On success, returns true. On failure, returns false and leaves the semaphore unchanged.
  82 func (s *Weighted) TryAcquire(n int64) bool {
  83         s.mu.Lock()
  84         success := s.size-s.cur >= n && s.waiters.Len() == 0
  85         if success {
  86                 s.cur += n
  87         }
  88         s.mu.Unlock()
  89         return success
  90 }
  91
  92 // Release releases the semaphore with a weight of n.
  93 func (s *Weighted) Release(n int64) {
  94         s.mu.Lock()
  95         s.cur -= n
  96         if s.cur < 0 {
  97                 s.mu.Unlock()
  98                 panic("semaphore: bad release")
  99         }
 100         for {
 101                 next := s.waiters.Front()
 102                 if next == nil {
 103                         break // No more waiters blocked.
 104                 }
 105
 106                 w := next.Value.(waiter)
 107                 if s.size-s.cur < w.n {
 108                         // Not enough tokens for the next waiter.  We could keep going (to try to
 109                         // find a waiter with a smaller request), but under load that could cause
 110                         // starvation for large requests; instead, we leave all remaining waiters
 111                         // blocked.
 112                         //
 113                         // Consider a semaphore used as a read-write lock, with N tokens, N
 114                         // readers, and one writer.  Each reader can Acquire(1) to obtain a read
 115                         // lock.  The writer can Acquire(N) to obtain a write lock, excluding all
 116                         // of the readers.  If we allow the readers to jump ahead in the queue,
 117                         // the writer will starve — there is always one token available for every
 118                         // reader.
 119                         break
 120                 }
 121
 122                 s.cur += w.n
 123                 s.waiters.Remove(next)
 124                 close(w.ready)
 125         }
 126         s.mu.Unlock()
 127 }