make the number of incoming streams configurable, for gQUIC

Changelog.md

@@ -3,6 +3,7 @@
 ## v0.8.0 (unreleased)
 
 - Add support for unidirectional streams (for IETF QUIC).
+- Add a `quic.Config` option for the maximum number of incoming streams.
 
 ## v0.7.0 (2018-02-03)
 

interface.go

@@ -179,6 +179,7 @@ type Config struct {
 	// If set to a negative value, it doesn't allow any bidirectional streams.
 	MaxIncomingStreams int
 	// MaxIncomingUniStreams is the maximum number of concurrent unidirectional streams that a peer is allowed to open.
+	// This value doesn't have any effect in Google QUIC.
 	// If not set, it will default to 100.
 	// If set to a negative value, it doesn't allow any unidirectional streams.
 	MaxIncomingUniStreams int

internal/protocol/server_parameters.go

@@ -71,10 +71,6 @@ const MaxStreamsMultiplier = 1.1
 // MaxStreamsMinimumIncrement is the slack the client is allowed for the maximum number of streams per connection, needed e.g. when packets are out of order or dropped. The minimum of this absolute increment and the procentual increase specified by MaxStreamsMultiplier is used.
 const MaxStreamsMinimumIncrement = 10
 
-// MaxNewStreamIDDelta is the maximum difference between and a newly opened Stream and the highest StreamID that a client has ever opened
-// note that the number of streams is half this value, since the client can only open streams with open StreamID
-const MaxNewStreamIDDelta = 4 * DefaultMaxIncomingStreams
-
 // MaxSessionUnprocessedPackets is the max number of packets stored in each session that are not yet processed.
 const MaxSessionUnprocessedPackets = DefaultMaxCongestionWindow
 

session.go

@@ -157,7 +157,7 @@ func newSession(
 	transportParams := &handshake.TransportParameters{
 		StreamFlowControlWindow:     protocol.ReceiveStreamFlowControlWindow,
 		ConnectionFlowControlWindow: protocol.ReceiveConnectionFlowControlWindow,
-		MaxStreams:                  protocol.DefaultMaxIncomingStreams,
+		MaxStreams:                  uint32(s.config.MaxIncomingStreams),
 		IdleTimeout:                 s.config.IdleTimeout,
 	}
 	cs, err := newCryptoSetup(
@@ -205,7 +205,7 @@ var newClientSession = func(
 	transportParams := &handshake.TransportParameters{
 		StreamFlowControlWindow:     protocol.ReceiveStreamFlowControlWindow,
 		ConnectionFlowControlWindow: protocol.ReceiveConnectionFlowControlWindow,
-		MaxStreams:                  protocol.DefaultMaxIncomingStreams,
+		MaxStreams:                  uint32(s.config.MaxIncomingStreams),
 		IdleTimeout:                 s.config.IdleTimeout,
 		OmitConnectionID:            s.config.RequestConnectionIDOmission,
 	}
@@ -332,7 +332,7 @@ func (s *session) postSetup(initialPacketNumber protocol.PacketNumber) error {
 	if s.version.UsesTLS() {
 		s.streamsMap = newStreamsMap(s, s.newFlowController, s.config.MaxIncomingStreams, s.config.MaxIncomingUniStreams, s.perspective, s.version)
 	} else {
-		s.streamsMap = newStreamsMapLegacy(s.newStream, s.perspective)
+		s.streamsMap = newStreamsMapLegacy(s.newStream, s.config.MaxIncomingStreams, s.perspective)
 	}
 	s.streamFramer = newStreamFramer(s.cryptoStream, s.streamsMap, s.version)
 	s.packer = newPacketPacker(s.connectionID,

streams_map_legacy.go

@@ -39,11 +39,10 @@ var _ streamManager = &streamsMapLegacy{}
 
 var errMapAccess = errors.New("streamsMap: Error accessing the streams map")
 
-func newStreamsMapLegacy(newStream func(protocol.StreamID) streamI, pers protocol.Perspective) streamManager {
+func newStreamsMapLegacy(newStream func(protocol.StreamID) streamI, maxStreams int, pers protocol.Perspective) streamManager {
 	// add some tolerance to the maximum incoming streams value
-	maxStreams := uint32(protocol.DefaultMaxIncomingStreams)
 	maxIncomingStreams := utils.MaxUint32(
-		maxStreams+protocol.MaxStreamsMinimumIncrement,
+		uint32(maxStreams)+protocol.MaxStreamsMinimumIncrement,
 		uint32(float64(maxStreams)*float64(protocol.MaxStreamsMultiplier)),
 	)
 	sm := streamsMapLegacy{
@@ -131,7 +130,10 @@ func (m *streamsMapLegacy) openRemoteStream(id protocol.StreamID) (streamI, erro
 	if m.numIncomingStreams >= m.maxIncomingStreams {
 		return nil, qerr.TooManyOpenStreams
 	}
-	if id+protocol.MaxNewStreamIDDelta < m.highestStreamOpenedByPeer {
+	// maxNewStreamIDDelta is the maximum difference between and a newly opened Stream and the highest StreamID that a client has ever opened
+	// note that the number of streams is half this value, since the client can only open streams with open StreamID
+	maxStreamIDDelta := protocol.StreamID(4 * m.maxIncomingStreams)
+	if id+maxStreamIDDelta < m.highestStreamOpenedByPeer {
 		return nil, qerr.Error(qerr.InvalidStreamID, fmt.Sprintf("attempted to open stream %d, which is a lot smaller than the highest opened stream, %d", id, m.highestStreamOpenedByPeer))
 	}
 

streams_map_legacy_test.go

@@ -23,13 +23,23 @@ var _ = Describe("Streams Map (for gQUIC)", func() {
 	}
 
 	setNewStreamsMap := func(p protocol.Perspective) {
-		m = newStreamsMapLegacy(newStream, p).(*streamsMapLegacy)
+		m = newStreamsMapLegacy(newStream, protocol.DefaultMaxIncomingStreams, p).(*streamsMapLegacy)
 	}
 
 	deleteStream := func(id protocol.StreamID) {
 		ExpectWithOffset(1, m.DeleteStream(id)).To(Succeed())
 	}
 
+	It("applies the max stream limit for small number of streams", func() {
+		sm := newStreamsMapLegacy(newStream, 1, protocol.PerspectiveServer).(*streamsMapLegacy)
+		Expect(sm.maxIncomingStreams).To(BeEquivalentTo(1 + protocol.MaxStreamsMinimumIncrement))
+	})
+
+	It("applies the max stream limit for big number of streams", func() {
+		sm := newStreamsMapLegacy(newStream, 1000, protocol.PerspectiveServer).(*streamsMapLegacy)
+		Expect(sm.maxIncomingStreams).To(BeEquivalentTo(1000 * protocol.MaxStreamsMultiplier))
+	})
+
 	Context("getting and creating streams", func() {
 		Context("as a server", func() {
 			BeforeEach(func() {