create FlowController interface

2016-05-19 19:40:21 +07:00
parent d148943bc8
commit c748a8dfc0
8 changed files with 107 additions and 43 deletions
--- a/flowcontrol/flow_controller.go
+++ b/flowcontrol/flow_controller.go
@@ -0,0 +1,170 @@
+package flowcontrol
+
+import (
+	"sync"
+
+	"github.com/lucas-clemente/quic-go/handshake"
+	"github.com/lucas-clemente/quic-go/protocol"
+)
+
+type flowController struct {
+	streamID protocol.StreamID
+
+	connectionParametersManager *handshake.ConnectionParametersManager
+
+	bytesSent                protocol.ByteCount
+	sendFlowControlWindow    protocol.ByteCount
+	lastBlockedSentForOffset protocol.ByteCount
+
+	bytesRead                         protocol.ByteCount
+	highestReceived                   protocol.ByteCount
+	receiveFlowControlWindow          protocol.ByteCount
+	receiveFlowControlWindowIncrement protocol.ByteCount
+
+	mutex sync.RWMutex
+}
+
+// NewFlowController gets a new flow controller
+func NewFlowController(streamID protocol.StreamID, connectionParametersManager *handshake.ConnectionParametersManager) FlowController {
+	fc := flowController{
+		streamID:                    streamID,
+		connectionParametersManager: connectionParametersManager,
+	}
+
+	if streamID == 0 {
+		fc.receiveFlowControlWindow = connectionParametersManager.GetReceiveConnectionFlowControlWindow()
+		fc.receiveFlowControlWindowIncrement = fc.receiveFlowControlWindow
+	} else {
+		fc.receiveFlowControlWindow = connectionParametersManager.GetReceiveStreamFlowControlWindow()
+		fc.receiveFlowControlWindowIncrement = fc.receiveFlowControlWindow
+	}
+
+	return &fc
+}
+
+func (c *flowController) getSendFlowControlWindow() protocol.ByteCount {
+	c.mutex.RLock()
+	defer c.mutex.RUnlock()
+
+	if c.sendFlowControlWindow == 0 {
+		if c.streamID == 0 {
+			return c.connectionParametersManager.GetSendConnectionFlowControlWindow()
+		}
+		return c.connectionParametersManager.GetSendStreamFlowControlWindow()
+	}
+	return c.sendFlowControlWindow
+}
+
+func (c *flowController) AddBytesSent(n protocol.ByteCount) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	c.bytesSent += n
+}
+
+// UpdateSendWindow should be called after receiving a WindowUpdateFrame
+// it returns true if the window was actually updated
+func (c *flowController) UpdateSendWindow(newOffset protocol.ByteCount) bool {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	if newOffset > c.sendFlowControlWindow {
+		c.sendFlowControlWindow = newOffset
+		return true
+	}
+	return false
+}
+
+func (c *flowController) SendWindowSize() protocol.ByteCount {
+	c.mutex.RLock()
+	defer c.mutex.RUnlock()
+
+	sendFlowControlWindow := c.getSendFlowControlWindow()
+
+	if c.bytesSent > sendFlowControlWindow { // should never happen, but make sure we don't do an underflow here
+		return 0
+	}
+	return sendFlowControlWindow - c.bytesSent
+}
+
+// UpdateHighestReceived updates the highestReceived value, if the byteOffset is higher
+// Should **only** be used for the stream-level FlowController
+func (c *flowController) UpdateHighestReceived(byteOffset protocol.ByteCount) protocol.ByteCount {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	if byteOffset > c.highestReceived {
+		increment := byteOffset - c.highestReceived
+		c.highestReceived = byteOffset
+		return increment
+	}
+	return 0
+}
+
+// IncrementHighestReceived adds an increment to the highestReceived value
+// Should **only** be used for the connection-level FlowController
+func (c *flowController) IncrementHighestReceived(increment protocol.ByteCount) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	c.highestReceived += increment
+}
+
+func (c *flowController) AddBytesRead(n protocol.ByteCount) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	c.bytesRead += n
+}
+
+// MaybeTriggerBlocked determines if it is necessary to send a Blocked for this stream
+// it makes sure that only one Blocked is sent for each offset
+func (c *flowController) MaybeTriggerBlocked() bool {
+	if c.SendWindowSize() != 0 {
+		return false
+	}
+
+	sendFlowControlWindow := c.getSendFlowControlWindow()
+
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	if c.lastBlockedSentForOffset == sendFlowControlWindow {
+		return false
+	}
+
+	c.lastBlockedSentForOffset = sendFlowControlWindow
+	return true
+}
+
+// MaybeTriggerWindowUpdate determines if it is necessary to send a WindowUpdate
+// if so, it returns true and the offset of the window
+func (c *flowController) MaybeTriggerWindowUpdate() (bool, protocol.ByteCount) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	diff := c.receiveFlowControlWindow - c.bytesRead
+	// Chromium implements the same threshold
+	if diff < (c.receiveFlowControlWindowIncrement / 2) {
+		c.receiveFlowControlWindow += c.receiveFlowControlWindowIncrement
+		return true, c.bytesRead + c.receiveFlowControlWindowIncrement
+	}
+	return false, 0
+}
+
+func (c *flowController) CheckFlowControlViolation() bool {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	if c.highestReceived > c.receiveFlowControlWindow {
+		return true
+	}
+	return false
+}
+
+func (c *flowController) GetHighestReceived() protocol.ByteCount {
+	c.mutex.RLock()
+	defer c.mutex.RUnlock()
+
+	return c.highestReceived
+}
--- a/flowcontrol/flow_controller_test.go
+++ b/flowcontrol/flow_controller_test.go
@@ -0,0 +1,217 @@
+package flowcontrol
+
+import (
+	"reflect"
+	"unsafe"
+
+	"github.com/lucas-clemente/quic-go/handshake"
+	"github.com/lucas-clemente/quic-go/protocol"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+// set private variables of the ConnectionParametersManager
+// those are normally read from the server parameter constants in the constructor of the ConnectionParametersManager
+func setConnectionParametersManagerWindow(cpm *handshake.ConnectionParametersManager, name string, value protocol.ByteCount) {
+	*(*protocol.ByteCount)(unsafe.Pointer(reflect.ValueOf(cpm).Elem().FieldByName(name).UnsafeAddr())) = value
+}
+
+var _ = Describe("Flow controller", func() {
+	var controller *flowController
+
+	BeforeEach(func() {
+		controller = &flowController{}
+	})
+
+	Context("Constructor", func() {
+		var cpm *handshake.ConnectionParametersManager
+
+		BeforeEach(func() {
+			cpm = &handshake.ConnectionParametersManager{}
+			setConnectionParametersManagerWindow(cpm, "sendStreamFlowControlWindow", 1000)
+			setConnectionParametersManagerWindow(cpm, "receiveStreamFlowControlWindow", 2000)
+			setConnectionParametersManagerWindow(cpm, "sendConnectionFlowControlWindow", 3000)
+			setConnectionParametersManagerWindow(cpm, "receiveConnectionFlowControlWindow", 4000)
+		})
+
+		It("reads the stream send and receive windows when acting as stream-level flow controller", func() {
+			fc := NewFlowController(5, cpm).(*flowController)
+			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).(*flowController)
+			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).(*flowController)
+			Expect(fc.sendFlowControlWindow).To(BeZero())
+		})
+
+		It("does not set the connection flow control windows for sending", func() {
+			fc := NewFlowController(0, cpm).(*flowController)
+			Expect(fc.sendFlowControlWindow).To(BeZero())
+		})
+	})
+
+	Context("send flow control", func() {
+		var cpm *handshake.ConnectionParametersManager
+
+		BeforeEach(func() {
+			cpm = &handshake.ConnectionParametersManager{}
+			setConnectionParametersManagerWindow(cpm, "sendStreamFlowControlWindow", 1000)
+			setConnectionParametersManagerWindow(cpm, "sendConnectionFlowControlWindow", 3000)
+			controller.connectionParametersManager = cpm
+		})
+
+		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.sendFlowControlWindow = 12
+			Expect(controller.SendWindowSize()).To(Equal(protocol.ByteCount(12 - 5)))
+		})
+
+		It("updates the size of the flow control window", func() {
+			controller.bytesSent = 5
+			updateSuccessful := controller.UpdateSendWindow(15)
+			Expect(updateSuccessful).To(BeTrue())
+			Expect(controller.SendWindowSize()).To(Equal(protocol.ByteCount(15 - 5)))
+		})
+
+		It("does not decrease the flow control window", func() {
+			updateSuccessful := controller.UpdateSendWindow(20)
+			Expect(updateSuccessful).To(BeTrue())
+			Expect(controller.SendWindowSize()).To(Equal(protocol.ByteCount(20)))
+			updateSuccessful = controller.UpdateSendWindow(10)
+			Expect(updateSuccessful).To(BeFalse())
+			Expect(controller.SendWindowSize()).To(Equal(protocol.ByteCount(20)))
+		})
+
+		It("asks the ConnectionParametersManager for the stream flow control window size", func() {
+			controller.streamID = 5
+			Expect(controller.getSendFlowControlWindow()).To(Equal(protocol.ByteCount(1000)))
+			// make sure the value is not cached
+			setConnectionParametersManagerWindow(cpm, "sendStreamFlowControlWindow", 2000)
+			Expect(controller.getSendFlowControlWindow()).To(Equal(protocol.ByteCount(2000)))
+		})
+
+		It("stops asking the ConnectionParametersManager for the flow control stream window size once a window update has arrived", func() {
+			controller.streamID = 5
+			Expect(controller.UpdateSendWindow(8000))
+			setConnectionParametersManagerWindow(cpm, "sendStreamFlowControlWindow", 9000)
+			Expect(controller.getSendFlowControlWindow()).To(Equal(protocol.ByteCount(8000)))
+		})
+
+		It("asks the ConnectionParametersManager for the connection flow control window size", func() {
+			controller.streamID = 0
+			Expect(controller.getSendFlowControlWindow()).To(Equal(protocol.ByteCount(3000)))
+			// make sure the value is not cached
+			setConnectionParametersManagerWindow(cpm, "sendConnectionFlowControlWindow", 5000)
+			Expect(controller.getSendFlowControlWindow()).To(Equal(protocol.ByteCount(5000)))
+		})
+
+		It("stops asking the ConnectionParametersManager for the connection flow control window size once a window update has arrived", func() {
+			controller.streamID = 0
+			Expect(controller.UpdateSendWindow(7000))
+			setConnectionParametersManagerWindow(cpm, "sendConnectionFlowControlWindow", 9000)
+			Expect(controller.getSendFlowControlWindow()).To(Equal(protocol.ByteCount(7000)))
+		})
+
+		Context("Blocked", func() {
+			var sendFlowControlWindow protocol.ByteCount = 20
+
+			BeforeEach(func() {
+				controller.sendFlowControlWindow = sendFlowControlWindow
+			})
+
+			It("sends a Blocked when there's no space left in the window", func() {
+				controller.bytesSent = sendFlowControlWindow
+				Expect(controller.MaybeTriggerBlocked()).To(BeTrue())
+			})
+
+			It("does not send a Blocked when there's still space in the window", func() {
+				controller.bytesSent = sendFlowControlWindow - 1
+				Expect(controller.MaybeTriggerBlocked()).To(BeFalse())
+			})
+
+			It("only sends one Blocked for one offset", func() {
+				controller.bytesSent = sendFlowControlWindow
+				Expect(controller.MaybeTriggerBlocked()).To(BeTrue())
+				Expect(controller.MaybeTriggerBlocked()).To(BeFalse())
+				updateSuccessfull := controller.UpdateSendWindow(sendFlowControlWindow + 1)
+				Expect(updateSuccessfull).To(BeTrue())
+				controller.bytesSent = sendFlowControlWindow + 1
+				Expect(controller.MaybeTriggerBlocked()).To(BeTrue())
+			})
+		})
+	})
+
+	Context("receive flow control", func() {
+		var receiveFlowControlWindow protocol.ByteCount = 10000
+		var receiveFlowControlWindowIncrement protocol.ByteCount = 600
+
+		BeforeEach(func() {
+			controller.receiveFlowControlWindow = receiveFlowControlWindow
+			controller.receiveFlowControlWindowIncrement = receiveFlowControlWindowIncrement
+		})
+
+		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() {
+			readPosition := receiveFlowControlWindow - receiveFlowControlWindowIncrement/2 + 1
+			controller.bytesRead = readPosition
+			updateNecessary, offset := controller.MaybeTriggerWindowUpdate()
+			Expect(updateNecessary).To(BeTrue())
+			Expect(offset).To(Equal(readPosition + receiveFlowControlWindowIncrement))
+		})
+
+		It("triggers a window update when not necessary", func() {
+			readPosition := receiveFlowControlWindow - receiveFlowControlWindow/2 - 1
+			controller.bytesRead = readPosition
+			updateNecessary, _ := controller.MaybeTriggerWindowUpdate()
+			Expect(updateNecessary).To(BeFalse())
+		})
+
+		It("updates the highestReceived", func() {
+			controller.highestReceived = 1337
+			increment := controller.UpdateHighestReceived(1338)
+			Expect(increment).To(Equal(protocol.ByteCount(1338 - 1337)))
+			Expect(controller.highestReceived).To(Equal(protocol.ByteCount(1338)))
+		})
+
+		It("does not decrease the highestReceived", func() {
+			controller.highestReceived = 1337
+			increment := controller.UpdateHighestReceived(1000)
+			Expect(increment).To(Equal(protocol.ByteCount(0)))
+			Expect(controller.highestReceived).To(Equal(protocol.ByteCount(1337)))
+		})
+
+		It("increases the highestReceived by a given increment", func() {
+			controller.highestReceived = 1337
+			controller.IncrementHighestReceived(123)
+			Expect(controller.highestReceived).To(Equal(protocol.ByteCount(1337 + 123)))
+		})
+
+		It("detects a flow control violation", func() {
+			controller.UpdateHighestReceived(receiveFlowControlWindow + 1)
+			Expect(controller.CheckFlowControlViolation()).To(BeTrue())
+		})
+
+		It("does not give a flow control violation when using the window completely", func() {
+			controller.UpdateHighestReceived(receiveFlowControlWindow)
+			Expect(controller.CheckFlowControlViolation()).To(BeFalse())
+		})
+	})
+})
--- a/flowcontrol/flowcontrol_suite_test.go
+++ b/flowcontrol/flowcontrol_suite_test.go
@@ -0,0 +1,13 @@
+package flowcontrol
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"testing"
+)
+
+func TestCrypto(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "FlowControl Suite")
+}
--- a/flowcontrol/interface.go
+++ b/flowcontrol/interface.go
@@ -0,0 +1,17 @@
+package flowcontrol
+
+import "github.com/lucas-clemente/quic-go/protocol"
+
+// A FlowController handles the flow control
+type FlowController interface {
+	AddBytesSent(n protocol.ByteCount)
+	UpdateSendWindow(newOffset protocol.ByteCount) bool
+	SendWindowSize() protocol.ByteCount
+	UpdateHighestReceived(byteOffset protocol.ByteCount) protocol.ByteCount
+	IncrementHighestReceived(increment protocol.ByteCount)
+	AddBytesRead(n protocol.ByteCount)
+	MaybeTriggerBlocked() bool
+	MaybeTriggerWindowUpdate() (bool, protocol.ByteCount)
+	CheckFlowControlViolation() bool
+	GetHighestReceived() protocol.ByteCount
+}