/
throttle.go
130 lines (117 loc) · 2.89 KB
/
throttle.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
package main
import (
"math"
"sync"
"time"
)
type SamplingThrottle struct {
mut sync.Mutex
limit time.Duration
hwm float64
windowSize int
backoffFactor float64
samples []float64
}
type ThrottleOpt func(s *SamplingThrottle)
func NewSamplingThrottle(opts ...ThrottleOpt) *SamplingThrottle {
s := new(SamplingThrottle)
// Set defaults
s.limit = 30 * time.Second
s.hwm = 0.5
s.windowSize = 10
s.backoffFactor = 1.0
// Then optionally override
for _, opt := range opts {
opt(s)
}
return s
}
func SetHWM(hwm float64) ThrottleOpt {
return func(s *SamplingThrottle) {
s.hwm = hwm
}
}
func SetLimit(limit time.Duration) ThrottleOpt {
return func(s *SamplingThrottle) {
s.limit = limit
}
}
func SetWindowSize(size int) ThrottleOpt {
return func(s *SamplingThrottle) {
s.windowSize = size
}
}
func SetBackoffFactor(factor float64) ThrottleOpt {
return func(s *SamplingThrottle) {
s.backoffFactor = factor
}
}
func (s *SamplingThrottle) Collect(t time.Duration) {
defer s.mut.Unlock()
s.mut.Lock()
// Shift sliding window forward if it's already full
if len(s.samples) == s.windowSize {
s.samples = s.samples[1:]
}
s.samples = append(s.samples, float64(t.Seconds()))
}
func (s *SamplingThrottle) computeOLS() (bool, float64, float64) {
if len(s.samples) >= s.windowSize {
num := 0.0
den := 0.0
xs := make([]float64, len(s.samples))
for i := 0; i < len(s.samples); i++ {
xs[i] = float64(i)
}
xavg := computeAvg(xs)
yavg := computeAvg(s.samples)
for x, y := range s.samples {
xdelta := float64(x) - xavg
num += xdelta * (y - yavg)
den += math.Pow(xdelta, 2)
}
m := num / den
b := yavg - m*xavg
return true, m, b
}
return false, 0, 0
}
func (s *SamplingThrottle) computeBackoff(m, b float64) (time.Duration, time.Duration) {
var backoffTime time.Duration
// Predict the duration of the next call
x := float64(len(s.samples))
pred := m*x + b
// Then compute the backoff as the "distance" the predicted value squared is from the high water
// mark (in seconds)
strength := math.Pow(pred, 2.0) / (s.limit.Seconds() * s.hwm)
backoff := strength * s.backoffFactor
if backoff > 0 {
backoffTime = time.Duration(backoff * float64(time.Second))
}
return time.Duration(pred * float64(time.Second)), backoffTime
}
func (s *SamplingThrottle) Wait() <-chan time.Time {
defer s.mut.Unlock()
s.mut.Lock()
log := Log.WithField("proc", "throttle")
var backoff time.Duration
ready, m, b := s.computeOLS()
if ready {
log.Debugf("y = %.4fx + %.4f", m, b)
pred, backoff := s.computeBackoff(m, b)
log.Debugf("pred = %v, backoff = %v", pred, backoff)
if backoff > 0 {
log.Warnf("Throttling worker for %.2fs", backoff.Seconds())
}
}
return time.After(backoff)
}
func computeAvg(samples []float64) float64 {
size := float64(len(samples))
sum := 0.0
for _, s := range samples {
sum += s
}
avg := sum / size
return avg
}