quic: avoid redundant MAX_DATA updates

When Stream.Read determines that we should send a MAX_DATA update,
it sends a message to the Conn to mark us as needing one.

If a second Read happens before the message from the first read
is processed, we may send a redundant MAX_DATA update.
This is harmless, but inefficient.

Double check that we still need to send an update before
marking one as necessary.

Change-Id: I0eb5a591eae6929b91da68b1ab6834a7795323ee
Reviewed-on: https://go-review.googlesource.com/c/net/+/530035
LUCI-TryBot-Result: Go LUCI <[email protected]>
Reviewed-by: Jonathan Amsterdam <[email protected]>

internal/quic/conn_flow.go

@@ -54,7 +54,10 @@ func (c *Conn) handleStreamBytesReadOffLoop(n int64) {
 		// We should send a MAX_DATA update to the peer.
 		// Record this on the Conn's main loop.
 		c.sendMsg(func(now time.Time, c *Conn) {
-			c.sendMaxDataUpdate()
+			// A MAX_DATA update may have already happened, so check again.
+			if c.shouldUpdateFlowControl(c.streams.inflow.credit.Load()) {
+				c.sendMaxDataUpdate()
+			}
 		})
 	}
 }

internal/quic/conn_flow_test.go

@@ -7,6 +7,7 @@
 package quic
 
 import (
+	"context"
 	"testing"
 )
 
@@ -36,6 +37,56 @@ func TestConnInflowReturnOnRead(t *testing.T) {
 		})
 }
 
+func TestConnInflowReturnOnRacingReads(t *testing.T) {
+	// Perform two reads at the same time,
+	// one for half of MaxConnReadBufferSize
+	// and one for one byte.
+	//
+	// We should observe a single MAX_DATA update.
+	// Depending on the ordering of events,
+	// this may include the credit from just the larger read
+	// or the credit from both.
+	ctx := canceledContext()
+	tc := newTestConn(t, serverSide, func(c *Config) {
+		c.MaxConnReadBufferSize = 64
+	})
+	tc.handshake()
+	tc.ignoreFrame(frameTypeAck)
+	tc.writeFrames(packetType1RTT, debugFrameStream{
+		id:   newStreamID(clientSide, uniStream, 0),
+		data: make([]byte, 32),
+	})
+	tc.writeFrames(packetType1RTT, debugFrameStream{
+		id:   newStreamID(clientSide, uniStream, 1),
+		data: make([]byte, 32),
+	})
+	s1, err := tc.conn.AcceptStream(ctx)
+	if err != nil {
+		t.Fatalf("conn.AcceptStream() = %v", err)
+	}
+	s2, err := tc.conn.AcceptStream(ctx)
+	if err != nil {
+		t.Fatalf("conn.AcceptStream() = %v", err)
+	}
+	read1 := runAsync(tc, func(ctx context.Context) (int, error) {
+		return s1.ReadContext(ctx, make([]byte, 16))
+	})
+	read2 := runAsync(tc, func(ctx context.Context) (int, error) {
+		return s2.ReadContext(ctx, make([]byte, 1))
+	})
+	// This MAX_DATA might extend the window by 16 or 17, depending on
+	// whether the second write occurs before the update happens.
+	tc.wantFrameType("MAX_DATA update is sent",
+		packetType1RTT, debugFrameMaxData{})
+	tc.wantIdle("redundant MAX_DATA is not sent")
+	if _, err := read1.result(); err != nil {
+		t.Errorf("ReadContext #1 = %v", err)
+	}
+	if _, err := read2.result(); err != nil {
+		t.Errorf("ReadContext #2 = %v", err)
+	}
+}
+
 func TestConnInflowReturnOnClose(t *testing.T) {
 	tc, s := newTestConnAndRemoteStream(t, serverSide, uniStream, func(c *Config) {
 		c.MaxConnReadBufferSize = 64