Refund HTTP/2 connection credit for buffered DATA

http2/client_conn_pool.go

@@ -270,6 +270,25 @@ func filterOutClientConn(in []*ClientConn, exclude *ClientConn) []*ClientConn {
 	return out
 }
 
+// bufferedUnreadBytes returns the sum of BufferedUnreadBytes over all pooled
+// connections. Connection references are collected under the pool lock and
+// summed outside it, since BufferedUnreadBytes takes each conn's own mutex.
+func (p *clientConnPool) bufferedUnreadBytes() int64 {
+	p.mu.Lock()
+	var conns []*ClientConn
+	for _, vv := range p.conns {
+		conns = append(conns, vv...)
+	}
+	p.mu.Unlock()
+
+	var n int64
+	for _, cc := range conns {
+		n += cc.BufferedUnreadBytes()
+	}
+
+	return n
+}
+
 // noDialClientConnPool is an implementation of http2.ClientConnPool
 // which never dials. We let the HTTP/1.1 client dial and use its TLS
 // connection instead.

http2/transport.go

@@ -293,6 +293,21 @@ func (t *Transport) connPool() ClientConnPool {
 	return t.connPoolOrDef
 }
 
+// BufferedUnreadBytes returns the number of response DATA bytes buffered by
+// this transport's pooled connections but not yet consumed by Response.Body
+// reads. It reports 0 when a custom ConnPool is configured, since the
+// transport cannot enumerate connections it does not own.
+func (t *Transport) BufferedUnreadBytes() int64 {
+	switch p := t.connPool().(type) {
+	case *clientConnPool:
+		return p.bufferedUnreadBytes()
+	case noDialClientConnPool:
+		return p.bufferedUnreadBytes()
+	}
+
+	return 0
+}
+
 func (t *Transport) initConnPool() {
 	if t.ConnPool != nil {
 		t.connPoolOrDef = t.ConnPool
@@ -324,6 +339,7 @@ type ClientConn struct {
 	idleTimer   *time.Timer
 
 	inflow            inflow // peer's conn-level flow control
+	unsentConnRefund  int32  // conn-level credit not yet announced with a WINDOW_UPDATE; guarded by mu
 	initialWindowSize uint32
 
 	lastActive           time.Time
@@ -934,6 +950,24 @@ func (cc *ClientConn) CanTakeNewRequest() bool {
 	return cc.canTakeNewRequestLocked()
 }
 
+// BufferedUnreadBytes returns the number of response DATA bytes buffered by
+// this connection's streams but not yet consumed by Response.Body reads.
+// Connection-level flow control is refunded once DATA is buffered, so this
+// memory is bounded per stream by its window but no longer bounded per
+// connection; callers can export the sum as a gauge to watch for unread
+// bodies accumulating.
+func (cc *ClientConn) BufferedUnreadBytes() int64 {
+	cc.mu.Lock()
+	defer cc.mu.Unlock()
+
+	var n int64
+	for _, cs := range cc.streams {
+		n += int64(cs.bufPipe.Len())
+	}
+
+	return n
+}
+
 // clientConnIdleState describes the suitability of a client
 // connection to initiate a new RoundTrip request.
 type clientConnIdleState struct {
@@ -2436,28 +2470,18 @@ func (b transportResponseBody) Read(p []byte) (n int, err error) {
 	cc.mu.Lock()
 	defer cc.mu.Unlock()
 
-	// Credit every byte the application consumes back to the flow control
-	// windows. The inflow accounting batches wire updates (at least
-	// inflowMinRefresh bytes, or enough to at least double the peer's
-	// remaining window), and unlike a threshold refresh it conserves
-	// credit exactly: bytes read from a body that is closed shortly after
-	// are still refunded by the next stream's reads instead of stranding
-	// until the connection window strangles to zero.
-	connAdd := cc.inflow.add(n)
+	// Credit bytes consumed by the application back to the stream-level
+	// flow-control window. Connection-level credit is refunded when DATA is
+	// buffered, so it must not be refunded again here.
 	var streamAdd int32
 	if err == nil { // No need to refresh if the stream is over or failed.
 		streamAdd = cs.inflow.add(n)
 	}
 
-	if connAdd != 0 || streamAdd != 0 {
+	if streamAdd != 0 {
 		cc.wmu.Lock()
 		defer cc.wmu.Unlock()
-		if connAdd != 0 {
-			cc.fr.WriteWindowUpdate(0, mustUint31(connAdd))
-		}
-		if streamAdd != 0 {
-			cc.fr.WriteWindowUpdate(cs.ID, mustUint31(streamAdd))
-		}
+		cc.fr.WriteWindowUpdate(cs.ID, mustUint31(streamAdd))
 		cc.bw.Flush()
 	}
 
@@ -2466,35 +2490,40 @@ func (b transportResponseBody) Read(p []byte) (n int, err error) {
 
 var errClosedResponseBody = errors.New("http2: response body closed")
 
+// refundConnFlow returns n bytes of connection-level flow control credit.
+// Every byte taken from the connection window is refunded exactly once:
+// padding and data for reset or unknown streams are refunded as the frame is
+// processed, and response data is refunded once buffered in the stream's
+// pipe. Refunds accumulate in cc.unsentConnRefund and are only announced to
+// the peer once they reach half the connection window, matching the
+// refresh-below-half cadence browsers use; cc.inflow tracks the window as
+// announced. It returns the increment to send in a stream-0 WINDOW_UPDATE,
+// or 0 if the refund was buffered.
+//
+// cc.mu must be held.
+func (cc *ClientConn) refundConnFlow(n int32) int32 {
+	cc.unsentConnRefund += n
+	if cc.unsentConnRefund < int32(cc.connFlow/2) {
+		return 0
+	}
+	send := cc.unsentConnRefund
+	cc.unsentConnRefund = 0
+	cc.inflow.add(int(send))
+
+	return send
+}
+
 func (b transportResponseBody) Close() error {
 	cs := b.cs
 	cc := cs.cc
 
 	serverSentStreamEnd := cs.bufPipe.Err() == io.EOF
 
-	// Break the pipe before returning flow control credit for unread data.
-	// Pipe writes fail from here on, so no data can land in the pipe (and
-	// silently lose its connection-level credit) between the refund below and
-	// the break. See golang.org/x/net commit 9f24bb44.
-	cs.bufPipe.BreakWithError(errClosedResponseBody)
-
-	unread := cs.bufPipe.Len()
-
-	if unread > 0 || !serverSentStreamEnd {
+	if !serverSentStreamEnd {
 		cc.mu.Lock()
-		var connAdd int32
-		if unread > 0 {
-			// Return connection-level flow control.
-			connAdd = cc.inflow.add(unread)
-		}
 		cc.wmu.Lock()
-		if !serverSentStreamEnd {
-			cc.fr.WriteRSTStream(cs.ID, ErrCodeCancel)
-			cs.didReset = true
-		}
-		if connAdd > 0 {
-			cc.fr.WriteWindowUpdate(0, uint32(connAdd))
-		}
+		cc.fr.WriteRSTStream(cs.ID, ErrCodeCancel)
+		cs.didReset = true
 		cc.bw.Flush()
 		cc.wmu.Unlock()
 		cc.mu.Unlock()
@@ -2530,7 +2559,7 @@ func (rl *clientConnReadLoop) processData(f *DataFrame) error {
 		if f.Length > 0 {
 			cc.mu.Lock()
 			ok := cc.inflow.take(f.Length)
-			connAdd := cc.inflow.add(int(f.Length))
+			connAdd := cc.refundConnFlow(int32(f.Length))
 			cc.mu.Unlock()
 			if !ok {
 				return ConnectionError(ErrCodeFlowControl)
@@ -2596,18 +2625,21 @@ func (rl *clientConnReadLoop) processData(f *DataFrame) error {
 		if didReset {
 			refund += len(data)
 		}
-
-		sendConn := cc.inflow.add(refund)
+		connRefund := refund
+		if len(data) > 0 && !didReset {
+			connRefund += len(data)
+		}
+		connAdd := cc.refundConnFlow(int32(connRefund))
 		var sendStream int32
 		if !didReset {
 			sendStream = cs.inflow.add(refund)
 		}
 		cc.mu.Unlock()
 
-		if sendConn > 0 || sendStream > 0 {
+		if connAdd > 0 || sendStream > 0 {
 			cc.wmu.Lock()
-			if sendConn > 0 {
-				cc.fr.WriteWindowUpdate(0, uint32(sendConn))
+			if connAdd > 0 {
+				cc.fr.WriteWindowUpdate(0, uint32(connAdd))
 			}
 			if sendStream > 0 {
 				cc.fr.WriteWindowUpdate(cs.ID, uint32(sendStream))

http2/transport_flow_test.go

@@ -275,3 +275,158 @@ func TestTransportBodyCloseRaceRefundsConnFlow(t *testing.T) {
 
 	ct.run()
 }
+
+// TestTransportPausedBodiesDoNotExhaustConnectionWindow verifies that response
+// bodies which stop being read do not prevent other streams from making
+// progress once their DATA has been buffered.
+//
+// The three paused bodies need more connection credit than the window holds,
+// so before connection-level credit was refunded at buffer time this test
+// failed with "paused bodies did not fill their stream windows": buffering
+// itself stalled once the connection window ran dry.
+func TestTransportPausedBodiesDoNotExhaustConnectionWindow(t *testing.T) {
+	const (
+		streamWindow = 6 << 20
+		connWindow   = 15663105
+		bodySize     = streamWindow + 1<<20
+	)
+
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ln.Close()
+
+	serverDone := make(chan struct{})
+	go func() {
+		defer close(serverDone)
+		conn, err := ln.Accept()
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+
+		(&Server{}).ServeConn(conn, &ServeConnOpts{
+			Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.Header().Set("Content-Length", strconv.Itoa(bodySize))
+				w.WriteHeader(http.StatusOK)
+				chunk := make([]byte, 16<<10)
+				for written := 0; written < bodySize; written += len(chunk) {
+					if _, err := w.Write(chunk); err != nil {
+						return
+					}
+				}
+			}),
+		})
+	}()
+
+	conn, err := net.Dial("tcp", ln.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	tr := &Transport{
+		Settings: map[SettingID]uint32{
+			SettingInitialWindowSize: streamWindow,
+		},
+		SettingsOrder:     []SettingID{SettingInitialWindowSize},
+		ConnectionFlow:    connWindow,
+		PseudoHeaderOrder: []string{":method", ":authority", ":scheme", ":path"},
+	}
+	cc, err := tr.NewClientConn(conn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cc.Close()
+
+	responses := make([]*http.Response, 0, 4)
+	defer func() {
+		for _, resp := range responses {
+			resp.Body.Close()
+		}
+	}()
+
+	for i := 0; i < 3; i++ {
+		req, err := http.NewRequest(http.MethodGet, "https://example.test/paused/"+strconv.Itoa(i), nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		resp, err := cc.RoundTrip(req)
+		if err != nil {
+			t.Fatal(err)
+		}
+		responses = append(responses, resp)
+	}
+
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		cc.mu.Lock()
+		available := cc.inflow.avail
+		pending := cc.unsentConnRefund
+		streams := make([]*clientStream, 0, len(cc.streams))
+		for _, cs := range cc.streams {
+			streams = append(streams, cs)
+		}
+		cc.mu.Unlock()
+
+		allBuffered := len(streams) == 3
+		for _, cs := range streams {
+			if cs.bufPipe.Len() > streamWindow {
+				t.Fatalf("stream buffered %d bytes, more than its %d byte window", cs.bufPipe.Len(), streamWindow)
+			}
+			if cs.bufPipe.Len() < streamWindow {
+				allBuffered = false
+				break
+			}
+		}
+		if allBuffered {
+			// Refunds may still be batched in unsentConnRefund rather than
+			// added to the announced window; both count as returned credit.
+			if available+pending < connWindow {
+				t.Fatalf("paused bodies reduced connection window: available=%d pending=%d, want at least %d", available, pending, connWindow)
+			}
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("paused bodies did not fill their stream windows")
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	if got, want := cc.BufferedUnreadBytes(), int64(3*streamWindow); got != want {
+		t.Fatalf("BufferedUnreadBytes = %d, want %d", got, want)
+	}
+
+	req, err := http.NewRequest(http.MethodGet, "https://example.test/fourth", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	fourth, err := cc.RoundTrip(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	responses = append(responses, fourth)
+
+	readDone := make(chan error, 1)
+	go func() {
+		buf := make([]byte, 1)
+		_, err := fourth.Body.Read(buf)
+		readDone <- err
+	}()
+
+	select {
+	case err := <-readDone:
+		if err != nil {
+			t.Fatalf("fourth response body read failed: %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("fourth response body remained blocked behind paused bodies")
+	}
+
+	cc.Close()
+	select {
+	case <-serverDone:
+	case <-time.After(2 * time.Second):
+		t.Fatal("server did not stop after client connection closed")
+	}
+}