-
-
Notifications
You must be signed in to change notification settings - Fork 1.3k
/
base_flow_controller_test.go
236 lines (209 loc) · 8.94 KB
/
base_flow_controller_test.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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
package flowcontrol
import (
"os"
"strconv"
"time"
"github.com/lucas-clemente/quic-go/internal/utils"
"github.com/lucas-clemente/quic-go/internal/protocol"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
)
// on the CIs, the timing is a lot less precise, so scale every duration by this factor
//
//nolint:unparam
func scaleDuration(t time.Duration) time.Duration {
scaleFactor := 1
if f, err := strconv.Atoi(os.Getenv("TIMESCALE_FACTOR")); err == nil { // parsing "" errors, so this works fine if the env is not set
scaleFactor = f
}
Expect(scaleFactor).ToNot(BeZero())
return time.Duration(scaleFactor) * t
}
var _ = Describe("Base Flow controller", func() {
var controller *baseFlowController
BeforeEach(func() {
controller = &baseFlowController{}
controller.rttStats = &utils.RTTStats{}
})
Context("send flow control", func() {
It("adds bytes sent", func() {
controller.bytesSent = 5
controller.AddBytesSent(6)
Expect(controller.bytesSent).To(Equal(protocol.ByteCount(5 + 6)))
})
It("gets the size of the remaining flow control window", func() {
controller.bytesSent = 5
controller.sendWindow = 12
Expect(controller.sendWindowSize()).To(Equal(protocol.ByteCount(12 - 5)))
})
It("updates the size of the flow control window", func() {
controller.AddBytesSent(5)
controller.UpdateSendWindow(15)
Expect(controller.sendWindow).To(Equal(protocol.ByteCount(15)))
Expect(controller.sendWindowSize()).To(Equal(protocol.ByteCount(15 - 5)))
})
It("says that the window size is 0 if we sent more than we were allowed to", func() {
controller.AddBytesSent(15)
controller.UpdateSendWindow(10)
Expect(controller.sendWindowSize()).To(BeZero())
})
It("does not decrease the flow control window", func() {
controller.UpdateSendWindow(20)
Expect(controller.sendWindowSize()).To(Equal(protocol.ByteCount(20)))
controller.UpdateSendWindow(10)
Expect(controller.sendWindowSize()).To(Equal(protocol.ByteCount(20)))
})
It("says when it's blocked", func() {
controller.UpdateSendWindow(100)
Expect(controller.IsNewlyBlocked()).To(BeFalse())
controller.AddBytesSent(100)
blocked, offset := controller.IsNewlyBlocked()
Expect(blocked).To(BeTrue())
Expect(offset).To(Equal(protocol.ByteCount(100)))
})
It("doesn't say that it's newly blocked multiple times for the same offset", func() {
controller.UpdateSendWindow(100)
controller.AddBytesSent(100)
newlyBlocked, offset := controller.IsNewlyBlocked()
Expect(newlyBlocked).To(BeTrue())
Expect(offset).To(Equal(protocol.ByteCount(100)))
newlyBlocked, _ = controller.IsNewlyBlocked()
Expect(newlyBlocked).To(BeFalse())
controller.UpdateSendWindow(150)
controller.AddBytesSent(150)
newlyBlocked, _ = controller.IsNewlyBlocked()
Expect(newlyBlocked).To(BeTrue())
})
})
Context("receive flow control", func() {
var (
receiveWindow protocol.ByteCount = 10000
receiveWindowSize protocol.ByteCount = 1000
)
BeforeEach(func() {
controller.bytesRead = receiveWindow - receiveWindowSize
controller.receiveWindow = receiveWindow
controller.receiveWindowSize = receiveWindowSize
})
It("adds bytes read", func() {
controller.bytesRead = 5
controller.addBytesRead(6)
Expect(controller.bytesRead).To(Equal(protocol.ByteCount(5 + 6)))
})
It("triggers a window update when necessary", func() {
bytesConsumed := float64(receiveWindowSize)*protocol.WindowUpdateThreshold + 1 // consumed 1 byte more than the threshold
bytesRemaining := receiveWindowSize - protocol.ByteCount(bytesConsumed)
readPosition := receiveWindow - bytesRemaining
controller.bytesRead = readPosition
offset := controller.getWindowUpdate()
Expect(offset).To(Equal(readPosition + receiveWindowSize))
Expect(controller.receiveWindow).To(Equal(readPosition + receiveWindowSize))
})
It("doesn't trigger a window update when not necessary", func() {
bytesConsumed := float64(receiveWindowSize)*protocol.WindowUpdateThreshold - 1 // consumed 1 byte less than the threshold
bytesRemaining := receiveWindowSize - protocol.ByteCount(bytesConsumed)
readPosition := receiveWindow - bytesRemaining
controller.bytesRead = readPosition
offset := controller.getWindowUpdate()
Expect(offset).To(BeZero())
})
Context("receive window size auto-tuning", func() {
var oldWindowSize protocol.ByteCount
BeforeEach(func() {
oldWindowSize = controller.receiveWindowSize
controller.maxReceiveWindowSize = 5000
})
// update the congestion such that it returns a given value for the smoothed RTT
setRtt := func(t time.Duration) {
controller.rttStats.UpdateRTT(t, 0, time.Now())
Expect(controller.rttStats.SmoothedRTT()).To(Equal(t)) // make sure it worked
}
It("doesn't increase the window size for a new stream", func() {
controller.maybeAdjustWindowSize()
Expect(controller.receiveWindowSize).To(Equal(oldWindowSize))
})
It("doesn't increase the window size when no RTT estimate is available", func() {
setRtt(0)
controller.startNewAutoTuningEpoch(time.Now())
controller.addBytesRead(400)
offset := controller.getWindowUpdate()
Expect(offset).ToNot(BeZero()) // make sure a window update is sent
Expect(controller.receiveWindowSize).To(Equal(oldWindowSize))
})
It("increases the window size if read so fast that the window would be consumed in less than 4 RTTs", func() {
bytesRead := controller.bytesRead
rtt := scaleDuration(50 * time.Millisecond)
setRtt(rtt)
// consume more than 2/3 of the window...
dataRead := receiveWindowSize*2/3 + 1
// ... in 4*2/3 of the RTT
controller.epochStartOffset = controller.bytesRead
controller.epochStartTime = time.Now().Add(-rtt * 4 * 2 / 3)
controller.addBytesRead(dataRead)
offset := controller.getWindowUpdate()
Expect(offset).ToNot(BeZero())
// check that the window size was increased
newWindowSize := controller.receiveWindowSize
Expect(newWindowSize).To(Equal(2 * oldWindowSize))
// check that the new window size was used to increase the offset
Expect(offset).To(Equal(bytesRead + dataRead + newWindowSize))
})
It("doesn't increase the window size if data is read so fast that the window would be consumed in less than 4 RTTs, but less than half the window has been read", func() {
bytesRead := controller.bytesRead
rtt := scaleDuration(20 * time.Millisecond)
setRtt(rtt)
// consume more than 2/3 of the window...
dataRead := receiveWindowSize*1/3 + 1
// ... in 4*2/3 of the RTT
controller.epochStartOffset = controller.bytesRead
controller.epochStartTime = time.Now().Add(-rtt * 4 * 1 / 3)
controller.addBytesRead(dataRead)
offset := controller.getWindowUpdate()
Expect(offset).ToNot(BeZero())
// check that the window size was not increased
newWindowSize := controller.receiveWindowSize
Expect(newWindowSize).To(Equal(oldWindowSize))
// check that the new window size was used to increase the offset
Expect(offset).To(Equal(bytesRead + dataRead + newWindowSize))
})
It("doesn't increase the window size if read too slowly", func() {
bytesRead := controller.bytesRead
rtt := scaleDuration(20 * time.Millisecond)
setRtt(rtt)
// consume less than 2/3 of the window...
dataRead := receiveWindowSize*2/3 - 1
// ... in 4*2/3 of the RTT
controller.epochStartOffset = controller.bytesRead
controller.epochStartTime = time.Now().Add(-rtt * 4 * 2 / 3)
controller.addBytesRead(dataRead)
offset := controller.getWindowUpdate()
Expect(offset).ToNot(BeZero())
// check that the window size was not increased
Expect(controller.receiveWindowSize).To(Equal(oldWindowSize))
// check that the new window size was used to increase the offset
Expect(offset).To(Equal(bytesRead + dataRead + oldWindowSize))
})
It("doesn't increase the window size to a value higher than the maxReceiveWindowSize", func() {
resetEpoch := func() {
// make sure the next call to maybeAdjustWindowSize will increase the window
controller.epochStartTime = time.Now().Add(-time.Millisecond)
controller.epochStartOffset = controller.bytesRead
controller.addBytesRead(controller.receiveWindowSize/2 + 1)
}
setRtt(scaleDuration(20 * time.Millisecond))
resetEpoch()
controller.maybeAdjustWindowSize()
Expect(controller.receiveWindowSize).To(Equal(2 * oldWindowSize)) // 2000
// because the lastWindowUpdateTime is updated by MaybeTriggerWindowUpdate(), we can just call maybeAdjustWindowSize() multiple times and get an increase of the window size every time
resetEpoch()
controller.maybeAdjustWindowSize()
Expect(controller.receiveWindowSize).To(Equal(2 * 2 * oldWindowSize)) // 4000
resetEpoch()
controller.maybeAdjustWindowSize()
Expect(controller.receiveWindowSize).To(Equal(controller.maxReceiveWindowSize)) // 5000
controller.maybeAdjustWindowSize()
Expect(controller.receiveWindowSize).To(Equal(controller.maxReceiveWindowSize)) // 5000
})
})
})
})