diff --git a/http2/client_conn_pool.go b/http2/client_conn_pool.go index e754d344..9303fb0b 100644 --- a/http2/client_conn_pool.go +++ b/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. diff --git a/http2/transport.go b/http2/transport.go index e47f3868..b2ad5ac9 100644 --- a/http2/transport.go +++ b/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)) diff --git a/http2/transport_flow_test.go b/http2/transport_flow_test.go index 6c29ffe2..15a107a9 100644 --- a/http2/transport_flow_test.go +++ b/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") + } +} diff --git a/http2/transport_test.go b/http2/transport_test.go index 0d563f03..a58bf681 100644 --- a/http2/transport_test.go +++ b/http2/transport_test.go @@ -2755,6 +2755,9 @@ func testTransportUsesGoAwayDebugError(t *testing.T, failMidBody bool) { func testTransportReturnsUnusedFlowControl(t *testing.T, oneDataFrame bool) { ct := newClientTester(t) + // Use a small connection window so the 5000-byte refund crosses the + // announce threshold (half the window) and shows up on the wire. + ct.tr.ConnectionFlow = 8000 clientClosed := make(chan struct{}) serverWroteFirstByte := make(chan struct{}) @@ -2811,11 +2814,13 @@ func testTransportReturnsUnusedFlowControl(t *testing.T, oneDataFrame bool) { // - Send one DATA frame with 5000 bytes. // - Send two DATA frames with 1 and 4999 bytes each. // - // In both cases, the client should consume one byte of data, - // refund that byte, then refund the following 4999 bytes. + // In both cases, the client should return all 5000 bytes of + // connection-level flow control. The first case returns the credit + // when the data is buffered; the second also returns the data received + // after the stream has been reset. // // In the second case, the server waits for the client connection to - // close before seconding the second DATA frame. This tests the case + // close before sending the second DATA frame. This tests the case // where the client receives a DATA frame after it has reset the stream. if oneDataFrame { ct.fr.WriteData(hf.StreamID, false /* don't end stream */, make([]byte, 5000)) @@ -2828,28 +2833,44 @@ func testTransportReturnsUnusedFlowControl(t *testing.T, oneDataFrame bool) { ct.fr.WriteData(hf.StreamID, false /* don't end stream */, make([]byte, 4999)) } - waitingFor := "RSTStreamFrame" - for { + var gotReset bool + var gotWindowUpdate uint32 + for !gotReset || gotWindowUpdate < 5000 { f, err := ct.fr.ReadFrame() if err != nil { - return fmt.Errorf("ReadFrame while waiting for %s: %v", waitingFor, err) + return fmt.Errorf("ReadFrame while waiting for flow-control cleanup: %v", err) } if _, ok := f.(*SettingsFrame); ok { continue } - switch waitingFor { - case "RSTStreamFrame": - if rf, ok := f.(*RSTStreamFrame); !ok || rf.ErrCode != ErrCodeCancel { - return fmt.Errorf("Expected a RSTStreamFrame with code cancel; got %v", summarizeFrame(f)) + switch f := f.(type) { + case *RSTStreamFrame: + if f.ErrCode != ErrCodeCancel { + return fmt.Errorf("expected a RSTStreamFrame with code cancel; got %v", summarizeFrame(f)) } - waitingFor = "WindowUpdateFrame" - case "WindowUpdateFrame": - if wuf, ok := f.(*WindowUpdateFrame); !ok || wuf.Increment != 4999 { - return fmt.Errorf("Expected WindowUpdateFrame for 4999 bytes; got %v", summarizeFrame(f)) + gotReset = true + case *WindowUpdateFrame: + if f.StreamID == 0 { + gotWindowUpdate += f.Increment } - return nil } } + if gotWindowUpdate != 5000 { + return fmt.Errorf("connection-level WINDOW_UPDATE credit = %d, want exactly 5000", gotWindowUpdate) + } + // The client must not refund the same bytes twice: drain until the + // read deadline and fail on any further connection-level credit. + ct.sc.SetReadDeadline(time.Now().Add(250 * time.Millisecond)) + for { + f, err := ct.fr.ReadFrame() + if err != nil { + break + } + if wu, ok := f.(*WindowUpdateFrame); ok && wu.StreamID == 0 { + return fmt.Errorf("unexpected extra connection-level WINDOW_UPDATE: %v", summarizeFrame(f)) + } + } + return nil } ct.run() } @@ -2941,6 +2962,9 @@ func TestTransportAdjustsFlowControl(t *testing.T) { // See golang.org/issue/16556 func TestTransportReturnsDataPaddingFlowControl(t *testing.T) { ct := newClientTester(t) + // Use a connection window small enough that the 6-byte padding refund + // crosses the announce threshold (half the window) immediately. + ct.tr.ConnectionFlow = 12 unblockClient := make(chan bool, 1)