quic: don't block Writes on stream-level flow control

Data written to a stream can be sent to the peer in a STREAM
frame only when:

- congestion control window is available
- pacing does not block sending
- stream-level flow control is available
- connection-level flow control is available

There must be a pushback mechanism to limit the amount of
locally buffered stream data, but I no longer believe the
stream-level flow control needs to be part of that pushback.

Using connection-level flow control (not yet implemented)
to block stream Write calls is problematic, because it
makes it difficult to fairly divide available send capacity
between multiple streams. If writes to a stream consume
connection-level flow control before we commit that
data to the wire, it becomes very easy for one stream to
starve others.

It's confusing to use stream-level flow control to block
Write calls, but not connection-level flow control.
This will especially produce unexpected behavior
if the recipient chooses to provide unlimited stream-level
quota but limited connection-level quota.

Change Stream.Write to only block writes based on the
configured maximum send buffer size. We may now buffer data
which cannot be immediately sent, but that was the case already
when transmission is blocked by congestion control.

In the future, we may want to make the stream buffer sizes
adaptive in response to the amount of in-flight data.
Rename Config.Stream*BufferSize to MaxStream*BufferSize,
to allow for possibly adding a minimum size later.

For golang/go#58547

Change-Id: I528a611fefb16b323776965c5b2ab5644035ed7a
Reviewed-on: https://go-review.googlesource.com/c/net/+/524958
LUCI-TryBot-Result: Go LUCI <[email protected]>
Commit-Queue: Damien Neil <[email protected]>
Auto-Submit: Damien Neil <[email protected]>
Reviewed-by: Jonathan Amsterdam <[email protected]>

Author: neild

internal/quic/config.go

@@ -30,17 +30,17 @@ type Config struct {
 	// If negative, the limit is zero.
 	MaxUniRemoteStreams int64
 
-	// StreamReadBufferSize is the maximum amount of data sent by the peer that a
+	// MaxStreamReadBufferSize is the maximum amount of data sent by the peer that a
 	// stream will buffer for reading.
 	// If zero, the default value of 1MiB is used.
 	// If negative, the limit is zero.
-	StreamReadBufferSize int64
+	MaxStreamReadBufferSize int64
 
-	// StreamWriteBufferSize is the maximum amount of data a stream will buffer for
+	// MaxStreamWriteBufferSize is the maximum amount of data a stream will buffer for
 	// sending to the peer.
 	// If zero, the default value of 1MiB is used.
 	// If negative, the limit is zero.
-	StreamWriteBufferSize int64
+	MaxStreamWriteBufferSize int64
 }
 
 func configDefault(v, def, limit int64) int64 {
@@ -62,10 +62,10 @@ func (c *Config) maxUniRemoteStreams() int64 {
 	return configDefault(c.MaxUniRemoteStreams, 100, maxStreamsLimit)
 }
 
-func (c *Config) streamReadBufferSize() int64 {
-	return configDefault(c.StreamReadBufferSize, 1<<20, maxVarint)
+func (c *Config) maxStreamReadBufferSize() int64 {
+	return configDefault(c.MaxStreamReadBufferSize, 1<<20, maxVarint)
 }
 
-func (c *Config) streamWriteBufferSize() int64 {
-	return configDefault(c.StreamWriteBufferSize, 1<<20, maxVarint)
+func (c *Config) maxStreamWriteBufferSize() int64 {
+	return configDefault(c.MaxStreamWriteBufferSize, 1<<20, maxVarint)
 }

internal/quic/config_test.go

@@ -17,7 +17,7 @@ func TestConfigTransportParameters(t *testing.T) {
 	tc := newTestConn(t, clientSide, func(c *Config) {
 		c.MaxBidiRemoteStreams = wantInitialMaxStreamsBidi
 		c.MaxUniRemoteStreams = wantInitialMaxStreamsUni
-		c.StreamReadBufferSize = wantInitialMaxStreamData
+		c.MaxStreamReadBufferSize = wantInitialMaxStreamData
 	})
 	tc.handshake()
 	if tc.sentTransportParameters == nil {

internal/quic/conn.go

@@ -117,9 +117,9 @@ func newConn(now time.Time, side connSide, initialConnID []byte, peerAddr netip.
 		maxUDPPayloadSize:              maxUDPPayloadSize,
 		maxAckDelay:                    maxAckDelay,
 		disableActiveMigration:         true,
-		initialMaxStreamDataBidiLocal:  config.streamReadBufferSize(),
-		initialMaxStreamDataBidiRemote: config.streamReadBufferSize(),
-		initialMaxStreamDataUni:        config.streamReadBufferSize(),
+		initialMaxStreamDataBidiLocal:  config.maxStreamReadBufferSize(),
+		initialMaxStreamDataBidiRemote: config.maxStreamReadBufferSize(),
+		initialMaxStreamDataUni:        config.maxStreamReadBufferSize(),
 		initialMaxStreamsBidi:          c.streams.remoteLimit[bidiStream].max,
 		initialMaxStreamsUni:           c.streams.remoteLimit[uniStream].max,
 		activeConnIDLimit:              activeConnIDLimit,

internal/quic/conn_loss_test.go

@@ -297,7 +297,7 @@ func TestLostMaxStreamDataFrame(t *testing.T) {
 		const maxWindowSize = 10
 		buf := make([]byte, maxWindowSize)
 		tc, s := newTestConnAndRemoteStream(t, serverSide, uniStream, func(c *Config) {
-			c.StreamReadBufferSize = maxWindowSize
+			c.MaxStreamReadBufferSize = maxWindowSize
 		})
 
 		// We send MAX_STREAM_DATA = 19.
@@ -339,7 +339,7 @@ func TestLostMaxStreamDataFrameAfterStreamFinReceived(t *testing.T) {
 		const maxWindowSize = 10
 		buf := make([]byte, maxWindowSize)
 		tc, s := newTestConnAndRemoteStream(t, serverSide, uniStream, func(c *Config) {
-			c.StreamReadBufferSize = maxWindowSize
+			c.MaxStreamReadBufferSize = maxWindowSize
 		})
 
 		tc.writeFrames(packetType1RTT, debugFrameStream{

internal/quic/conn_streams.go

@@ -76,11 +76,11 @@ func (c *Conn) newLocalStream(ctx context.Context, styp streamType) (*Stream, er
 	}
 
 	s := newStream(c, newStreamID(c.side, styp, num))
-	s.outmaxbuf = c.config.streamWriteBufferSize()
+	s.outmaxbuf = c.config.maxStreamWriteBufferSize()
 	s.outwin = c.streams.peerInitialMaxStreamDataRemote[styp]
 	if styp == bidiStream {
-		s.inmaxbuf = c.config.streamReadBufferSize()
-		s.inwin = c.config.streamReadBufferSize()
+		s.inmaxbuf = c.config.maxStreamReadBufferSize()
+		s.inwin = c.config.maxStreamReadBufferSize()
 	}
 	s.inUnlock()
 	s.outUnlock()
@@ -170,10 +170,10 @@ func (c *Conn) streamForFrame(now time.Time, id streamID, ftype streamFrameType)
 	}
 
 	s = newStream(c, id)
-	s.inmaxbuf = c.config.streamReadBufferSize()
-	s.inwin = c.config.streamReadBufferSize()
+	s.inmaxbuf = c.config.maxStreamReadBufferSize()
+	s.inwin = c.config.maxStreamReadBufferSize()
 	if id.streamType() == bidiStream {
-		s.outmaxbuf = c.config.streamWriteBufferSize()
+		s.outmaxbuf = c.config.maxStreamWriteBufferSize()
 		s.outwin = c.streams.peerInitialMaxStreamDataBidiLocal
 	}
 	s.inUnlock()

internal/quic/conn_streams_test.go

@@ -207,7 +207,7 @@ func TestStreamsWriteQueueFairness(t *testing.T) {
 		p.initialMaxData = 1<<62 - 1
 		p.initialMaxStreamDataBidiRemote = dataLen
 	}, func(c *Config) {
-		c.StreamWriteBufferSize = dataLen
+		c.MaxStreamWriteBufferSize = dataLen
 	})
 	tc.handshake()
 	tc.ignoreFrame(frameTypeAck)

internal/quic/crypto_stream.go

@@ -118,7 +118,7 @@ func (s *cryptoStream) ackOrLoss(start, end int64, fate packetFate) {
 // copy the data it wants into position.
 func (s *cryptoStream) dataToSend(pto bool, f func(off, size int64) (sent int64)) {
 	for {
-		off, size := dataToSend(s.out, s.outunsent, s.outacked, pto)
+		off, size := dataToSend(s.out.start, s.out.end, s.outunsent, s.outacked, pto)
 		if size == 0 {
 			return
 		}

internal/quic/stream.go

@@ -202,14 +202,7 @@ func (s *Stream) WriteContext(ctx context.Context, b []byte) (n int, err error)
 		// We exit the loop after writing all data, so on subsequent passes through
 		// the loop we are always write blocked.
 		if len(b) > 0 && !canWrite {
-			// We're blocked, either by flow control or by our own buffer limit.
-			// We either need the peer to extend our flow control window,
-			// or ack some of our outstanding packets.
-			if s.out.end == s.outwin {
-				// We're blocked by flow control.
-				// Send a STREAM_DATA_BLOCKED frame to let the peer know.
-				s.outblocked.setUnsent()
-			}
+			// Our send buffer is full. Wait for the peer to ack some data.
 			s.outUnlock()
 			if err := s.outgate.waitAndLock(ctx, s.conn.testHooks); err != nil {
 				return n, err
@@ -233,18 +226,24 @@ func (s *Stream) WriteContext(ctx context.Context, b []byte) (n int, err error)
 		if len(b) == 0 {
 			break
 		}
-		s.outblocked.clear()
-		// Write limit is min(our own buffer limit, the peer-provided flow control window).
+		// Write limit is our send buffer limit.
 		// This is a stream offset.
-		lim := min(s.out.start+s.outmaxbuf, s.outwin)
+		lim := s.out.start + s.outmaxbuf
 		// Amount to write is min(the full buffer, data up to the write limit).
 		// This is a number of bytes.
 		nn := min(int64(len(b)), lim-s.out.end)
 		// Copy the data into the output buffer and mark it as unsent.
-		s.outunsent.add(s.out.end, s.out.end+nn)
+		if s.out.end <= s.outwin {
+			s.outunsent.add(s.out.end, min(s.out.end+nn, s.outwin))
+		}
 		s.out.writeAt(b[:nn], s.out.end)
 		b = b[nn:]
 		n += int(nn)
+		if s.out.end > s.outwin {
+			// We're blocked by flow control.
+			// Send a STREAM_DATA_BLOCKED frame to let the peer know.
+			s.outblocked.set()
+		}
 		// If we have bytes left to send, we're blocked.
 		canWrite = false
 	}
@@ -425,8 +424,8 @@ func (s *Stream) outUnlockNoQueue() streamState {
 			}
 		}
 	}
-	lim := min(s.out.start+s.outmaxbuf, s.outwin)
-	canWrite := lim > s.out.end || // available flow control
+	lim := s.out.start + s.outmaxbuf
+	canWrite := lim > s.out.end || // available send buffer
 		s.outclosed.isSet() || // closed locally
 		s.outreset.isSet() // reset locally
 	defer s.outgate.unlock(canWrite)
@@ -533,7 +532,19 @@ func (s *Stream) handleStopSending(code uint64) error {
 func (s *Stream) handleMaxStreamData(maxStreamData int64) error {
 	s.outgate.lock()
 	defer s.outUnlock()
-	s.outwin = max(maxStreamData, s.outwin)
+	if maxStreamData <= s.outwin {
+		return nil
+	}
+	if s.out.end > s.outwin {
+		s.outunsent.add(s.outwin, min(maxStreamData, s.out.end))
+	}
+	s.outwin = maxStreamData
+	if s.out.end > s.outwin {
+		// We've still got more data than flow control window.
+		s.outblocked.setUnsent()
+	} else {
+		s.outblocked.clear()
+	}
 	return nil
 }
 
@@ -635,7 +646,7 @@ func (s *Stream) appendOutFramesLocked(w *packetWriter, pnum packetNumber, pto b
 	if s.outreset.isSet() {
 		// RESET_STREAM
 		if s.outreset.shouldSendPTO(pto) {
-			if !w.appendResetStreamFrame(s.id, s.outresetcode, s.out.end) {
+			if !w.appendResetStreamFrame(s.id, s.outresetcode, min(s.outwin, s.out.end)) {
 				return false
 			}
 			s.outreset.setSent(pnum)
@@ -645,15 +656,15 @@ func (s *Stream) appendOutFramesLocked(w *packetWriter, pnum packetNumber, pto b
 	}
 	if s.outblocked.shouldSendPTO(pto) {
 		// STREAM_DATA_BLOCKED
-		if !w.appendStreamDataBlockedFrame(s.id, s.out.end) {
+		if !w.appendStreamDataBlockedFrame(s.id, s.outwin) {
 			return false
 		}
 		s.outblocked.setSent(pnum)
 		s.frameOpensStream(pnum)
 	}
-	// STREAM
 	for {
-		off, size := dataToSend(s.out, s.outunsent, s.outacked, pto)
+		// STREAM
+		off, size := dataToSend(min(s.out.start, s.outwin), min(s.out.end, s.outwin), s.outunsent, s.outacked, pto)
 		fin := s.outclosed.isSet() && off+size == s.out.end
 		shouldSend := size > 0 || // have data to send
 			s.outopened.shouldSendPTO(pto) || // should open the stream
@@ -691,7 +702,7 @@ func (s *Stream) frameOpensStream(pnum packetNumber) {
 }
 
 // dataToSend returns the next range of data to send in a STREAM or CRYPTO_STREAM.
-func dataToSend(out pipe, outunsent, outacked rangeset[int64], pto bool) (start, size int64) {
+func dataToSend(start, end int64, outunsent, outacked rangeset[int64], pto bool) (sendStart, size int64) {
 	switch {
 	case pto:
 		// On PTO, resend unacked data that fits in the probe packet.
@@ -702,14 +713,14 @@ func dataToSend(out pipe, outunsent, outacked rangeset[int64], pto bool) (start,
 		// This may miss unacked data starting after that acked byte,
 		// but avoids resending data the peer has acked.
 		for _, r := range outacked {
-			if r.start > out.start {
-				return out.start, r.start - out.start
+			if r.start > start {
+				return start, r.start - start
 			}
 		}
-		return out.start, out.end - out.start
+		return start, end - start
 	case outunsent.numRanges() > 0:
 		return outunsent.min(), outunsent[0].size()
 	default:
-		return out.end, 0
+		return end, 0
 	}
 }