From b7a9fcf85f38bf734045e11a39199f1f1100e3a8 Mon Sep 17 00:00:00 2001 From: Marten Seemann Date: Mon, 31 Oct 2016 12:51:26 +0700 Subject: [PATCH] implement auto-tuning of receive flow control windows fixes #106 --- flowcontrol/flow_controller.go | 54 +++++++++++++++++++++-- flowcontrol/flow_controller_test.go | 68 ++++++++++++++++++++++++++++- protocol/server_parameters.go | 14 ++++-- 3 files changed, 127 insertions(+), 9 deletions(-) diff --git a/flowcontrol/flow_controller.go b/flowcontrol/flow_controller.go index 562369aa..ea937a89 100644 --- a/flowcontrol/flow_controller.go +++ b/flowcontrol/flow_controller.go @@ -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 diff --git a/flowcontrol/flow_controller_test.go b/flowcontrol/flow_controller_test.go index 383fc578..9c1f8ac5 100644 --- a/flowcontrol/flow_controller_test.go +++ b/flowcontrol/flow_controller_test.go @@ -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 + }) + }) }) }) diff --git a/protocol/server_parameters.go b/protocol/server_parameters.go index 511bb469..bf5d5a51 100644 --- a/protocol/server_parameters.go +++ b/protocol/server_parameters.go @@ -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