forked from elastic/beats
-
Notifications
You must be signed in to change notification settings - Fork 7
/
window.go
83 lines (68 loc) · 1.92 KB
/
window.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
package logstash
import (
"math"
"sync/atomic"
)
type window struct {
windowSize int32
maxOkWindowSize int // max window size sending was successful for
maxWindowSize int
}
func newWindower(start, max int) *window {
w := &window{}
w.init(start, max)
return w
}
func (w *window) init(start, max int) {
*w = window{
windowSize: int32(start),
maxWindowSize: max,
}
}
func (w *window) get() int {
return int(atomic.LoadInt32(&w.windowSize))
}
// Increase window size by factor 1.5 until max window size
// (window size grows exponentially)
// TODO: use duration until ACK to estimate an ok max window size value
func (w *window) tryGrowWindow(batchSize int) {
windowSize := w.get()
if windowSize <= batchSize {
if w.maxOkWindowSize < windowSize {
debug("update max ok window size: %v < %v",
w.maxOkWindowSize, w.windowSize)
w.maxOkWindowSize = windowSize
newWindowSize := int(math.Ceil(1.5 * float64(windowSize)))
debug("increase window size to: %v", newWindowSize)
if windowSize <= batchSize && batchSize < newWindowSize {
debug("set to batchSize: %v", batchSize)
newWindowSize = batchSize
}
if newWindowSize > w.maxWindowSize {
debug("set to max window size: %v", w.maxWindowSize)
newWindowSize = int(w.maxWindowSize)
}
windowSize = newWindowSize
} else if windowSize < w.maxOkWindowSize {
debug("update current window size: %v", w.windowSize)
windowSize = int(math.Ceil(1.5 * float64(windowSize)))
if windowSize > w.maxOkWindowSize {
debug("set to max ok window size: %v", w.maxOkWindowSize)
windowSize = w.maxOkWindowSize
}
}
atomic.StoreInt32(&w.windowSize, int32(windowSize))
}
}
func (w *window) shrinkWindow() {
windowSize := w.get()
orig := windowSize
windowSize = windowSize / 2
if windowSize < minWindowSize {
windowSize = minWindowSize
if windowSize == orig {
return
}
}
atomic.StoreInt32(&w.windowSize, int32(windowSize))
}