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

Merge pull request #340 from lucas-clemente/fix-106

Browse Source 
implement receive flow control window auto tuning
This commit is contained in:
Lucas Clemente
2016-11-18 22:10:39 +01:00
committed by GitHub
parent 0d6356d9d8 b7a9fcf85f
commit 6d6e8f941a
9 changed files with 160 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
ackhandler/sent_packet_handler.go
View File

@@ -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,

3
ackhandler/sent_packet_handler_test.go
View File

@@ -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},

12
flowcontrol/flow_control_manager.go
View File

@@ -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
}

3
flowcontrol/flow_control_manager_test.go
View File

@@ -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() {

59
flowcontrol/flow_controller.go
View File

@@ -1,37 +1,48 @@
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 {
streamID protocol.StreamID
connectionParametersManager *handshake.ConnectionParametersManager
rttStats *congestion.RTTStats
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
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 {
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
@@ -99,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

79
flowcontrol/flow_controller_test.go
View File

@@ -2,8 +2,10 @@ package flowcontrol
import (
"reflect"
"time"
"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"
@@ -21,13 +23,16 @@ var _ = Describe("Flow controller", func() {
BeforeEach(func() {
controller = &flowController{}
controller.rttStats = &congestion.RTTStats{}
})
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 +40,26 @@ 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)))
Expect(fc.maxReceiveFlowControlWindowIncrement).To(Equal(protocol.MaxReceiveStreamFlowControlWindow))
})
It("reads the stream send and receive windows when acting as stream-level flow controller", func() {
fc := newFlowController(0, cpm)
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() {
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())
})
})
@@ -149,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() {
@@ -193,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

9
session.go
View File

@@ -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
@@ -92,13 +95,15 @@ 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
sentPacketHandler = ackhandler.NewSentPacketHandler()
rttStats := &congestion.RTTStats{}
sentPacketHandler = ackhandler.NewSentPacketHandler(rttStats)
receivedPacketHandler = ackhandler.NewReceivedPacketHandler()
flowControlManager := flowcontrol.NewFlowControlManager(connectionParametersManager, rttStats)
now := time.Now()
session := &Session{

3
stream_test.go
View File

@@ -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)
})