From c8b724615908b9a31bbb37c57dec556a117f54d7 Mon Sep 17 00:00:00 2001 From: Marten Seemann Date: Sun, 30 Oct 2016 17:27:21 +0700 Subject: [PATCH 1/3] create congestion RTTstats in Session ref #106 --- ackhandler/sent_packet_handler.go | 4 +--- ackhandler/sent_packet_handler_test.go | 3 ++- session.go | 7 ++++++- 3 files changed, 9 insertions(+), 5 deletions(-) diff --git a/ackhandler/sent_packet_handler.go b/ackhandler/sent_packet_handler.go index 531a99a0..686fd9d7 100644 --- a/ackhandler/sent_packet_handler.go +++ b/ackhandler/sent_packet_handler.go @@ -47,9 +47,7 @@ type sentPacketHandler struct { } // NewSentPacketHandler creates a new sentPacketHandler -func NewSentPacketHandler() SentPacketHandler { - rttStats := &congestion.RTTStats{} - +func NewSentPacketHandler(rttStats *congestion.RTTStats) SentPacketHandler { congestion := congestion.NewCubicSender( congestion.DefaultClock{}, rttStats, diff --git a/ackhandler/sent_packet_handler_test.go b/ackhandler/sent_packet_handler_test.go index 59cc5514..00642b60 100644 --- a/ackhandler/sent_packet_handler_test.go +++ b/ackhandler/sent_packet_handler_test.go @@ -57,7 +57,8 @@ var _ = Describe("SentPacketHandler", func() { ) BeforeEach(func() { - handler = NewSentPacketHandler().(*sentPacketHandler) + rttStats := &congestion.RTTStats{} + handler = NewSentPacketHandler(rttStats).(*sentPacketHandler) streamFrame = frames.StreamFrame{ StreamID: 5, Data: []byte{0x13, 0x37}, diff --git a/session.go b/session.go index 2f752999..376538f5 100644 --- a/session.go +++ b/session.go @@ -9,6 +9,7 @@ import ( "time" "github.com/lucas-clemente/quic-go/ackhandler" + "github.com/lucas-clemente/quic-go/congestion" "github.com/lucas-clemente/quic-go/flowcontrol" "github.com/lucas-clemente/quic-go/frames" "github.com/lucas-clemente/quic-go/handshake" @@ -51,6 +52,8 @@ type Session struct { streamsMap *streamsMap + rttStats *congestion.RTTStats + sentPacketHandler ackhandler.SentPacketHandler receivedPacketHandler ackhandler.ReceivedPacketHandler streamFramer *streamFramer @@ -97,7 +100,9 @@ func newSession(conn connection, v protocol.VersionNumber, connectionID protocol var sentPacketHandler ackhandler.SentPacketHandler var receivedPacketHandler ackhandler.ReceivedPacketHandler - sentPacketHandler = ackhandler.NewSentPacketHandler() + rttStats := &congestion.RTTStats{} + + sentPacketHandler = ackhandler.NewSentPacketHandler(rttStats) receivedPacketHandler = ackhandler.NewReceivedPacketHandler() now := time.Now() From 32d89eee0270d754d0fce16abbf8c64a42635816 Mon Sep 17 00:00:00 2001 From: Marten Seemann Date: Sun, 30 Oct 2016 17:36:36 +0700 Subject: [PATCH 2/3] pass RTTStats to the FlowControllers ref #106 --- flowcontrol/flow_control_manager.go | 12 ++++++++---- flowcontrol/flow_control_manager_test.go | 3 ++- flowcontrol/flow_controller.go | 5 ++++- flowcontrol/flow_controller_test.go | 11 +++++++---- session.go | 2 +- stream_test.go | 3 ++- 6 files changed, 24 insertions(+), 12 deletions(-) diff --git a/flowcontrol/flow_control_manager.go b/flowcontrol/flow_control_manager.go index 0ab63634..0af52e9c 100644 --- a/flowcontrol/flow_control_manager.go +++ b/flowcontrol/flow_control_manager.go @@ -4,13 +4,16 @@ import ( "errors" "sync" + "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 flowControlManager struct { - connectionParametersManager *handshake.ConnectionParametersManager + connectionParametersManager *handshake.ConnectionParametersManager + rttStats *congestion.RTTStats + streamFlowController map[protocol.StreamID]*flowController contributesToConnectionFlowControl map[protocol.StreamID]bool mutex sync.RWMutex @@ -26,14 +29,15 @@ var ( var errMapAccess = errors.New("Error accessing the flowController map.") // NewFlowControlManager creates a new flow control manager -func NewFlowControlManager(connectionParametersManager *handshake.ConnectionParametersManager) FlowControlManager { +func NewFlowControlManager(connectionParametersManager *handshake.ConnectionParametersManager, rttStats *congestion.RTTStats) FlowControlManager { fcm := flowControlManager{ connectionParametersManager: connectionParametersManager, + rttStats: rttStats, streamFlowController: make(map[protocol.StreamID]*flowController), contributesToConnectionFlowControl: make(map[protocol.StreamID]bool), } // initialize connection level flow controller - fcm.streamFlowController[0] = newFlowController(0, connectionParametersManager) + fcm.streamFlowController[0] = newFlowController(0, connectionParametersManager, rttStats) fcm.contributesToConnectionFlowControl[0] = false return &fcm } @@ -47,7 +51,7 @@ func (f *flowControlManager) NewStream(streamID protocol.StreamID, contributesTo return } - f.streamFlowController[streamID] = newFlowController(streamID, f.connectionParametersManager) + f.streamFlowController[streamID] = newFlowController(streamID, f.connectionParametersManager, f.rttStats) f.contributesToConnectionFlowControl[streamID] = contributesToConnectionFlow } diff --git a/flowcontrol/flow_control_manager_test.go b/flowcontrol/flow_control_manager_test.go index aaebfbcf..362e5849 100644 --- a/flowcontrol/flow_control_manager_test.go +++ b/flowcontrol/flow_control_manager_test.go @@ -1,6 +1,7 @@ package flowcontrol import ( + "github.com/lucas-clemente/quic-go/congestion" "github.com/lucas-clemente/quic-go/handshake" "github.com/lucas-clemente/quic-go/protocol" . "github.com/onsi/ginkgo" @@ -15,7 +16,7 @@ var _ = Describe("Flow Control Manager", func() { cpm = &handshake.ConnectionParametersManager{} setConnectionParametersManagerWindow(cpm, "receiveStreamFlowControlWindow", 0x100) setConnectionParametersManagerWindow(cpm, "receiveConnectionFlowControlWindow", 0x200) - fcm = NewFlowControlManager(cpm).(*flowControlManager) + fcm = NewFlowControlManager(cpm, &congestion.RTTStats{}).(*flowControlManager) }) It("creates a connection level flow controller", func() { diff --git a/flowcontrol/flow_controller.go b/flowcontrol/flow_controller.go index 21020ad9..562369aa 100644 --- a/flowcontrol/flow_controller.go +++ b/flowcontrol/flow_controller.go @@ -1,6 +1,7 @@ package flowcontrol import ( + "github.com/lucas-clemente/quic-go/congestion" "github.com/lucas-clemente/quic-go/handshake" "github.com/lucas-clemente/quic-go/protocol" ) @@ -9,6 +10,7 @@ type flowController struct { streamID protocol.StreamID connectionParametersManager *handshake.ConnectionParametersManager + rttStats *congestion.RTTStats bytesSent protocol.ByteCount sendFlowControlWindow protocol.ByteCount @@ -20,10 +22,11 @@ type flowController struct { } // newFlowController gets a new flow controller -func newFlowController(streamID protocol.StreamID, connectionParametersManager *handshake.ConnectionParametersManager) *flowController { +func newFlowController(streamID protocol.StreamID, connectionParametersManager *handshake.ConnectionParametersManager, rttStats *congestion.RTTStats) *flowController { fc := flowController{ streamID: streamID, connectionParametersManager: connectionParametersManager, + rttStats: rttStats, } if streamID == 0 { diff --git a/flowcontrol/flow_controller_test.go b/flowcontrol/flow_controller_test.go index e2489788..383fc578 100644 --- a/flowcontrol/flow_controller_test.go +++ b/flowcontrol/flow_controller_test.go @@ -4,6 +4,7 @@ import ( "reflect" "unsafe" + "github.com/lucas-clemente/quic-go/congestion" "github.com/lucas-clemente/quic-go/handshake" "github.com/lucas-clemente/quic-go/protocol" . "github.com/onsi/ginkgo" @@ -25,9 +26,11 @@ var _ = Describe("Flow controller", func() { Context("Constructor", func() { var cpm *handshake.ConnectionParametersManager + var rttStats *congestion.RTTStats BeforeEach(func() { cpm = &handshake.ConnectionParametersManager{} + rttStats = &congestion.RTTStats{} setConnectionParametersManagerWindow(cpm, "sendStreamFlowControlWindow", 1000) setConnectionParametersManagerWindow(cpm, "receiveStreamFlowControlWindow", 2000) setConnectionParametersManagerWindow(cpm, "sendConnectionFlowControlWindow", 3000) @@ -35,24 +38,24 @@ var _ = Describe("Flow controller", func() { }) It("reads the stream send and receive windows when acting as stream-level flow controller", func() { - fc := newFlowController(5, cpm) + fc := newFlowController(5, cpm, rttStats) Expect(fc.streamID).To(Equal(protocol.StreamID(5))) Expect(fc.receiveFlowControlWindow).To(Equal(protocol.ByteCount(2000))) }) It("reads the stream send and receive windows when acting as stream-level flow controller", func() { - fc := newFlowController(0, cpm) + fc := newFlowController(0, cpm, rttStats) Expect(fc.streamID).To(Equal(protocol.StreamID(0))) Expect(fc.receiveFlowControlWindow).To(Equal(protocol.ByteCount(4000))) }) It("does not set the stream flow control windows for sending", func() { - fc := newFlowController(5, cpm) + fc := newFlowController(5, cpm, rttStats) Expect(fc.sendFlowControlWindow).To(BeZero()) }) It("does not set the connection flow control windows for sending", func() { - fc := newFlowController(0, cpm) + fc := newFlowController(0, cpm, rttStats) Expect(fc.sendFlowControlWindow).To(BeZero()) }) }) diff --git a/session.go b/session.go index 376538f5..3d13025d 100644 --- a/session.go +++ b/session.go @@ -95,7 +95,6 @@ type Session struct { // newSession makes a new session func newSession(conn connection, v protocol.VersionNumber, connectionID protocol.ConnectionID, sCfg *handshake.ServerConfig, streamCallback StreamCallback, closeCallback closeCallback) (packetHandler, error) { connectionParametersManager := handshake.NewConnectionParamatersManager() - flowControlManager := flowcontrol.NewFlowControlManager(connectionParametersManager) var sentPacketHandler ackhandler.SentPacketHandler var receivedPacketHandler ackhandler.ReceivedPacketHandler @@ -104,6 +103,7 @@ func newSession(conn connection, v protocol.VersionNumber, connectionID protocol sentPacketHandler = ackhandler.NewSentPacketHandler(rttStats) receivedPacketHandler = ackhandler.NewReceivedPacketHandler() + flowControlManager := flowcontrol.NewFlowControlManager(connectionParametersManager, rttStats) now := time.Now() session := &Session{ diff --git a/stream_test.go b/stream_test.go index ce73b231..1eaa5479 100644 --- a/stream_test.go +++ b/stream_test.go @@ -5,6 +5,7 @@ import ( "io" "time" + "github.com/lucas-clemente/quic-go/congestion" "github.com/lucas-clemente/quic-go/flowcontrol" "github.com/lucas-clemente/quic-go/frames" "github.com/lucas-clemente/quic-go/handshake" @@ -110,7 +111,7 @@ var _ = Describe("Stream", func() { onDataCalled = false var streamID protocol.StreamID = 1337 cpm := handshake.NewConnectionParamatersManager() - flowControlManager := flowcontrol.NewFlowControlManager(cpm) + flowControlManager := flowcontrol.NewFlowControlManager(cpm, &congestion.RTTStats{}) flowControlManager.NewStream(streamID, true) str, _ = newStream(streamID, onData, flowControlManager) }) From b7a9fcf85f38bf734045e11a39199f1f1100e3a8 Mon Sep 17 00:00:00 2001 From: Marten Seemann Date: Mon, 31 Oct 2016 12:51:26 +0700 Subject: [PATCH 3/3] 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