gr1ffon/quic-go
1
0
Fork 0
forked from quic-go/quic-go
Code Pull Requests Activity

implement auto-tuning of receive flow control windows

Browse Source 
fixes #106
This commit is contained in:
Marten Seemann
2016-10-31 12:51:26 +07:00
parent 32d89eee02
commit b7a9fcf85f
3 changed files with 127 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
flowcontrol/flow_controller.go
View File

@@ -1,9 +1,12 @@
package flowcontrol
import (
"time"
"github.com/lucas-clemente/quic-go/congestion"
"github.com/lucas-clemente/quic-go/handshake"
"github.com/lucas-clemente/quic-go/protocol"
"github.com/lucas-clemente/quic-go/utils"
)
type flowController struct {
@@ -15,10 +18,13 @@ type flowController struct {
bytesSent protocol.ByteCount
sendFlowControlWindow protocol.ByteCount
bytesRead protocol.ByteCount
highestReceived protocol.ByteCount
receiveFlowControlWindow protocol.ByteCount
receiveFlowControlWindowIncrement protocol.ByteCount
lastWindowUpdateTime time.Time
bytesRead protocol.ByteCount
highestReceived protocol.ByteCount
receiveFlowControlWindow protocol.ByteCount
receiveFlowControlWindowIncrement protocol.ByteCount
maxReceiveFlowControlWindowIncrement protocol.ByteCount
}
// newFlowController gets a new flow controller
@@ -32,9 +38,11 @@ func newFlowController(streamID protocol.StreamID, connectionParametersManager *
if streamID == 0 {
fc.receiveFlowControlWindow = connectionParametersManager.GetReceiveConnectionFlowControlWindow()
fc.receiveFlowControlWindowIncrement = fc.receiveFlowControlWindow
fc.maxReceiveFlowControlWindowIncrement = protocol.MaxReceiveConnectionFlowControlWindow
} else {
fc.receiveFlowControlWindow = connectionParametersManager.GetReceiveStreamFlowControlWindow()
fc.receiveFlowControlWindowIncrement = fc.receiveFlowControlWindow
fc.maxReceiveFlowControlWindowIncrement = protocol.MaxReceiveStreamFlowControlWindow
}
return &fc
@@ -102,14 +110,52 @@ func (c *flowController) AddBytesRead(n protocol.ByteCount) {
// if so, it returns true and the offset of the window
func (c *flowController) MaybeTriggerWindowUpdate() (bool, protocol.ByteCount) {
diff := c.receiveFlowControlWindow - c.bytesRead
// Chromium implements the same threshold
if diff < (c.receiveFlowControlWindowIncrement / 2) {
c.maybeAdjustWindowIncrement()
c.lastWindowUpdateTime = time.Now()
c.receiveFlowControlWindow = c.bytesRead + c.receiveFlowControlWindowIncrement
return true, c.receiveFlowControlWindow
}
return false, 0
}
// maybeAdjustWindowIncrement increases the receiveFlowControlWindowIncrement if we're sending WindowUpdates too often
func (c *flowController) maybeAdjustWindowIncrement() {
if c.lastWindowUpdateTime.IsZero() {
return
}
rtt := c.rttStats.SmoothedRTT()
if rtt == 0 {
return
}
timeSinceLastWindowUpdate := time.Now().Sub(c.lastWindowUpdateTime)
// interval between the window updates is sufficiently large, no need to increase the increment
if timeSinceLastWindowUpdate >= 2*rtt {
return
}
oldWindowSize := c.receiveFlowControlWindowIncrement
c.receiveFlowControlWindowIncrement = utils.MinByteCount(2*c.receiveFlowControlWindowIncrement, c.maxReceiveFlowControlWindowIncrement)
// debug log, if the window size was actually increased
if oldWindowSize < c.receiveFlowControlWindowIncrement {
newWindowSize := c.receiveFlowControlWindowIncrement / (1 << 10)
if c.streamID == 0 {
utils.Debugf("Increasing receive flow control window for the connection to %d kB", newWindowSize)
} else {
utils.Debugf("Increasing receive flow control window increment for stream %d to %d kB", c.streamID, newWindowSize)
}
}
}
func (c *flowController) CheckFlowControlViolation() bool {
if c.highestReceived > c.receiveFlowControlWindow {
return true

68
flowcontrol/flow_controller_test.go
View File

@@ -2,6 +2,7 @@ package flowcontrol
import (
"reflect"
"time"
"unsafe"
"github.com/lucas-clemente/quic-go/congestion"
@@ -22,6 +23,7 @@ var _ = Describe("Flow controller", func() {
BeforeEach(func() {
controller = &flowController{}
controller.rttStats = &congestion.RTTStats{}
})
Context("Constructor", func() {
@@ -41,12 +43,14 @@ var _ = Describe("Flow controller", func() {
fc := newFlowController(5, cpm, rttStats)
Expect(fc.streamID).To(Equal(protocol.StreamID(5)))
Expect(fc.receiveFlowControlWindow).To(Equal(protocol.ByteCount(2000)))
Expect(fc.maxReceiveFlowControlWindowIncrement).To(Equal(protocol.MaxReceiveStreamFlowControlWindow))
})
It("reads the stream send and receive windows when acting as stream-level flow controller", func() {
It("reads the stream send and receive windows when acting as connection-level flow controller", func() {
fc := newFlowController(0, cpm, rttStats)
Expect(fc.streamID).To(Equal(protocol.StreamID(0)))
Expect(fc.receiveFlowControlWindow).To(Equal(protocol.ByteCount(4000)))
Expect(fc.maxReceiveFlowControlWindowIncrement).To(Equal(protocol.MaxReceiveConnectionFlowControlWindow))
})
It("does not set the stream flow control windows for sending", func() {
@@ -152,19 +156,24 @@ var _ = Describe("Flow controller", func() {
})
It("triggers a window update when necessary", func() {
controller.lastWindowUpdateTime = time.Now().Add(-time.Hour)
readPosition := receiveFlowControlWindow - receiveFlowControlWindowIncrement/2 + 1
controller.bytesRead = readPosition
updateNecessary, offset := controller.MaybeTriggerWindowUpdate()
Expect(updateNecessary).To(BeTrue())
Expect(offset).To(Equal(readPosition + receiveFlowControlWindowIncrement))
Expect(controller.receiveFlowControlWindow).To(Equal(readPosition + receiveFlowControlWindowIncrement))
Expect(controller.lastWindowUpdateTime).To(BeTemporally("~", time.Now(), 5*time.Millisecond))
})
It("triggers a window update when not necessary", func() {
It("doesn't trigger a window update when not necessary", func() {
lastWindowUpdateTime := time.Now().Add(-time.Hour)
controller.lastWindowUpdateTime = lastWindowUpdateTime
readPosition := receiveFlowControlWindow - receiveFlowControlWindow/2 - 1
controller.bytesRead = readPosition
updateNecessary, _ := controller.MaybeTriggerWindowUpdate()
Expect(updateNecessary).To(BeFalse())
Expect(controller.lastWindowUpdateTime).To(Equal(lastWindowUpdateTime))
})
It("updates the highestReceived", func() {
@@ -196,5 +205,60 @@ var _ = Describe("Flow controller", func() {
controller.UpdateHighestReceived(receiveFlowControlWindow)
Expect(controller.CheckFlowControlViolation()).To(BeFalse())
})
Context("receive window increment auto-tuning", func() {
var oldIncrement protocol.ByteCount
BeforeEach(func() {
oldIncrement = controller.receiveFlowControlWindowIncrement
controller.maxReceiveFlowControlWindowIncrement = 3000
})
// 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 increment for a new stream", func() {
controller.maybeAdjustWindowIncrement()
Expect(controller.receiveFlowControlWindowIncrement).To(Equal(oldIncrement))
})
It("doesn't increase the increment when no RTT estimate is available", func() {
setRtt(0)
controller.lastWindowUpdateTime = time.Now()
controller.maybeAdjustWindowIncrement()
Expect(controller.receiveFlowControlWindowIncrement).To(Equal(oldIncrement))
})
It("increases the increment when the last WindowUpdate was sent less than two RTTs ago", func() {
setRtt(10 * time.Millisecond)
controller.lastWindowUpdateTime = time.Now().Add(-19 * time.Millisecond)
controller.maybeAdjustWindowIncrement()
Expect(controller.receiveFlowControlWindowIncrement).To(Equal(2 * oldIncrement))
})
It("doesn't increase the increase increment when the last WindowUpdate was sent more than two RTTs ago", func() {
setRtt(10 * time.Millisecond)
controller.lastWindowUpdateTime = time.Now().Add(-21 * time.Millisecond)
controller.maybeAdjustWindowIncrement()
Expect(controller.receiveFlowControlWindowIncrement).To(Equal(oldIncrement))
})
It("doesn't increase the increment to a value higher than the maxReceiveFlowControlWindowIncrement", func() {
setRtt(10 * time.Millisecond)
controller.lastWindowUpdateTime = time.Now().Add(-19 * time.Millisecond)
controller.maybeAdjustWindowIncrement()
Expect(controller.receiveFlowControlWindowIncrement).To(Equal(2 * oldIncrement)) // 1200
// because the lastWindowUpdateTime is updated by MaybeTriggerWindowUpdate(), we can just call maybeAdjustWindowIncrement() multiple times and get an increase of the increment every time
controller.maybeAdjustWindowIncrement()
Expect(controller.receiveFlowControlWindowIncrement).To(Equal(2 * 2 * oldIncrement)) // 2400
controller.maybeAdjustWindowIncrement()
Expect(controller.receiveFlowControlWindowIncrement).To(Equal(controller.maxReceiveFlowControlWindowIncrement)) // 3000
controller.maybeAdjustWindowIncrement()
Expect(controller.receiveFlowControlWindowIncrement).To(Equal(controller.maxReceiveFlowControlWindowIncrement)) // 3000
})
})
})
})

14
protocol/server_parameters.go
View File

@@ -17,11 +17,19 @@ const AckSendDelay = 5 * time.Millisecond
// ReceiveStreamFlowControlWindow is the stream-level flow control window for receiving data
// This is the value that Google servers are using
const ReceiveStreamFlowControlWindow ByteCount = (1 << 20) // 1 MB
const ReceiveStreamFlowControlWindow ByteCount = (1 << 10) * 32 // 32 kB
// ReceiveConnectionFlowControlWindow is the stream-level flow control window for receiving data
// ReceiveConnectionFlowControlWindow is the connection-level flow control window for receiving data
// This is the value that Google servers are using
const ReceiveConnectionFlowControlWindow ByteCount = (1 << 20) * 1.5 // 1.5 MB
const ReceiveConnectionFlowControlWindow ByteCount = (1 << 10) * 48 // 48 kB
// MaxReceiveStreamFlowControlWindow is the maximum stream-level flow control window for receiving data
// This is the value that Google servers are using
const MaxReceiveStreamFlowControlWindow ByteCount = 1 * (1 << 20) // 1 MB
// MaxReceiveConnectionFlowControlWindow is the connection-level flow control window for receiving data
// This is the value that Google servers are using
const MaxReceiveConnectionFlowControlWindow ByteCount = 1.5 * (1 << 20) // 1.5 MB
// MaxStreamsPerConnection is the maximum value accepted for the number of streams per connection
const MaxStreamsPerConnection = 100