diff --git a/src/net/http/internal/http2/transport.go b/src/net/http/internal/http2/transport.go index 0b32ea72da0246..53ffb5c0c38782 100644 --- a/src/net/http/internal/http2/transport.go +++ b/src/net/http/internal/http2/transport.go @@ -247,6 +247,27 @@ type clientStream struct { donec chan struct{} // closed after the stream is in the closed state on100 chan struct{} // buffered; written to if a 100 is received + // detached, guarded by cc.mu, indicates that the writeRequest + // goroutine has exited without waiting for the stream to end, and + // that cleanupWriteRequest should instead be run (on a new goroutine) + // by whichever of abortStreamLocked or clientConnReadLoop.endStream + // ends the stream. It is cleared when that cleanup is scheduled. + // See clientStream.detach. + detached bool + + // stopCtxWatch, if non-nil, cancels the context.AfterFunc watching + // for request context cancellation on behalf of a detached stream. + // It is set (under cc.mu) at most once, by detach, before detached + // is set, and is called by cleanupWriteRequest. + stopCtxWatch func() bool + + // respHeaderTimeoutTimer, guarded by cc.mu, is a timer enforcing + // Transport.ResponseHeaderTimeout on behalf of a detached stream. + // It is armed by detach if response headers haven't yet arrived, and + // stopped when they do (clientConnReadLoop.processHeaders) or when + // the stream ends (cleanupWriteRequest). + respHeaderTimeoutTimer *time.Timer + respHeaderRecv chan struct{} // closed when headers are received res *ClientResponse // set if respHeaderRecv is closed @@ -299,6 +320,10 @@ func (cs *clientStream) abortStreamLocked(err error) { cs.abortErr = err close(cs.abort) }) + if cs.detached { + cs.detached = false + go cs.cleanupWriteRequest(cs.abortErr) + } if cs.reqBody != nil { cs.closeReqBodyLocked() } @@ -1211,12 +1236,81 @@ func (cc *ClientConn) roundTrip(req *ClientRequest, streamf func(*clientStream)) // doRequest runs for the duration of the request lifetime. // -// It sends the request and performs post-request cleanup (closing Request.Body, etc.). +// It sends the request and performs post-request cleanup (closing Request.Body, etc.), +// except when writeRequest detaches from the stream, in which case cleanup is +// performed at stream end by whoever ends it. See clientStream.detach. func (cs *clientStream) doRequest(req *ClientRequest, streamf func(*clientStream)) { err := cs.writeRequest(req, streamf) + if err == errStreamDetached { + return + } cs.cleanupWriteRequest(err) } +// errStreamDetached is a sentinel returned by writeRequest to tell doRequest +// that the stream detached and cleanupWriteRequest will be called at stream +// end by whoever ends it. It is never returned to users. +var errStreamDetached = errors.New("http2: internal sentinel; stream detached from writeRequest goroutine") + +// detach arranges for cleanupWriteRequest to run when the stream ends (the +// peer half-closes it, it's aborted, or the request context is canceled), +// letting the writeRequest goroutine exit instead of parking until then. +// +// This matters for servers and proxies with many concurrent long-lived +// response streams (long polls): without it, each in-flight request pins a +// goroutine and its stack for the stream's lifetime doing nothing but +// waiting. +// +// respHeaderTimeout, if non-zero, gives the Transport.ResponseHeaderTimeout +// to enforce on the detached stream if response headers haven't arrived yet. +// +// It reports whether the stream was detached. It returns false if the stream +// has already ended, in which case the caller should wait for the stream end +// events itself (they're already pending). +func (cs *clientStream) detach(respHeaderTimeout time.Duration) bool { + cc := cs.cc + cc.mu.Lock() + defer cc.mu.Unlock() + select { + case <-cs.peerClosed: + return false + case <-cs.abort: + return false + default: + } + if respHeaderTimeout != 0 { + select { + case <-cs.respHeaderRecv: + // Headers already arrived; nothing to enforce. + default: + cs.respHeaderTimeoutTimer = time.AfterFunc(respHeaderTimeout, func() { + select { + case <-cs.respHeaderRecv: + // Headers arrived after all; we lost a race + // with the Stop in processHeaders. Not a + // timeout. + return + default: + } + cs.abortStream(errTimeout) + }) + } + } + // Watch for request context cancellation without parking a goroutine + // on ctx.Done(). If the context was canceled already, AfterFunc runs + // the func in a new goroutine, which blocks acquiring cc.mu until we + // return. + // + // stopCtxWatch must be assigned before detached is set: once detached + // is set, an abort or peer close can schedule cleanupWriteRequest + // (which calls stopCtxWatch) as soon as we release cc.mu. + cs.stopCtxWatch = context.AfterFunc(cs.ctx, func() { + cs.abortStream(cs.ctx.Err()) + }) + cs.detached = true + return true +} + var errExtendedConnectNotSupported = errors.New("net/http: extended connect not supported by peer") // writeRequest sends a request. @@ -1340,6 +1434,24 @@ func (cs *clientStream) writeRequest(req *ClientRequest, streamf func(*clientStr traceWroteRequest(cs.trace, err) + // If the request is fully sent and there's nothing left for this + // goroutine to do but wait for the stream to end, detach from the + // stream and exit rather than pinning this goroutine (and its stack) + // for the lifetime of what may be a very long-lived response stream. + // The remaining cases below then run cleanupWriteRequest from the + // stream-end event sites instead: + // - peerClosed and abort schedule it directly + // (abortStreamLocked, clientConnReadLoop.endStream) + // - ctx.Done is handled via context.AfterFunc in detach + // - ResponseHeaderTimeout is enforced by a time.AfterFunc timer, + // armed in detach and stopped when headers arrive + // The deprecated Request.Cancel channel can only be watched by a + // goroutine, so that (rare) case keeps the historical behavior of + // waiting here. + if cs.sentEndStream && cs.reqCancel == nil && cs.detach(cc.responseHeaderTimeout()) { + return errStreamDetached + } + var respHeaderTimer <-chan time.Time var respHeaderRecv chan struct{} if d := cc.responseHeaderTimeout(); d != 0 { @@ -1348,6 +1460,7 @@ func (cs *clientStream) writeRequest(req *ClientRequest, streamf func(*clientStr respHeaderTimer = timer.C respHeaderRecv = cs.respHeaderRecv } + // Wait until the peer half-closes its end of the stream, // or until the request is aborted (via context, error, or otherwise), // whichever comes first. @@ -1433,6 +1546,10 @@ func encodeRequestHeaders(req *ClientRequest, addGzipHeader bool, peerMaxHeaderL func (cs *clientStream) cleanupWriteRequest(err error) { cc := cs.cc + if cs.stopCtxWatch != nil { + cs.stopCtxWatch() + } + if cs.ID == 0 { // We were canceled before creating the stream, so return our reservation. cc.decrStreamReservations() @@ -1443,6 +1560,10 @@ func (cs *clientStream) cleanupWriteRequest(err error) { // and in multiple cases: server replies <=299 and >299 // while still writing request body cc.mu.Lock() + if t := cs.respHeaderTimeoutTimer; t != nil { + t.Stop() + cs.respHeaderTimeoutTimer = nil + } mustCloseBody := false if cs.reqBody != nil && cs.reqBodyClosed == nil { mustCloseBody = true @@ -2154,6 +2275,13 @@ func (rl *clientConnReadLoop) processHeaders(f *MetaHeadersFrame) error { } cs.res = res close(cs.respHeaderRecv) + // Stop a detached stream's response header timeout, if armed. + rl.cc.mu.Lock() + if t := cs.respHeaderTimeoutTimer; t != nil { + t.Stop() + cs.respHeaderTimeoutTimer = nil + } + rl.cc.mu.Unlock() if f.StreamEnded() { rl.endStream(cs) } @@ -2566,6 +2694,10 @@ func (rl *clientConnReadLoop) endStream(cs *clientStream) { defer rl.cc.mu.Unlock() cs.bufPipe.closeWithErrorAndCode(io.EOF, cs.copyTrailers) close(cs.peerClosed) + if cs.detached { + cs.detached = false + go cs.cleanupWriteRequest(nil) + } } } diff --git a/src/net/http/internal/http2/transport_test.go b/src/net/http/internal/http2/transport_test.go index e3093f64e39380..a92b9237e264f5 100644 --- a/src/net/http/internal/http2/transport_test.go +++ b/src/net/http/internal/http2/transport_test.go @@ -26,6 +26,7 @@ import ( "net/url" "os" "reflect" + "runtime" "sort" "strconv" "strings" @@ -5651,3 +5652,99 @@ func testExtendedConnectReadFrameError(t *testing.T) { t.Fatalf("after connection closed: RoundTrip succeeded; want error") } } + +// TestTransportRequestGoroutineExits verifies that the goroutine spawned to +// write a request exits once the request has been fully sent, rather than +// parking for the lifetime of the response stream. For clients with many +// concurrent long-lived streams (long polls), a parked goroutine and its +// stack per stream is a significant memory cost. +func TestTransportRequestGoroutineExits(t *testing.T) { + synctest.Test(t, testTransportRequestGoroutineExits) +} +func testTransportRequestGoroutineExits(t *testing.T) { + tc := newTestClientConn(t) + tc.greet() + + req, _ := http.NewRequest("GET", "https://dummy.tld/", nil) + rt := tc.roundTrip(req) + + tc.wantFrameType(FrameHeaders) + tc.writeHeaders(HeadersFrameParam{ + StreamID: rt.streamID(), + EndHeaders: true, + EndStream: false, + BlockFragment: tc.makeHeaderBlockFragment(":status", "200"), + }) + rt.wantStatus(200) + + // The request is fully sent and the response is streaming with no + // end in sight. The request-writing goroutine should be gone. + synctest.Wait() + if n := requestWriteGoroutines(); n != 0 { + t.Errorf("got %d request-writing goroutines parked during long-lived response stream; want 0", n) + } + + // The stream still works and still cleans up at END_STREAM. + tc.writeData(rt.streamID(), false, []byte("hello, ")) + tc.writeData(rt.streamID(), true, []byte("world")) + rt.wantBody([]byte("hello, world")) +} + +// requestWriteGoroutines returns the number of goroutines in +// clientStream.doRequest or clientStream.writeRequest. +func requestWriteGoroutines() int { + buf := make([]byte, 1<<20) + buf = buf[:runtime.Stack(buf, true)] + n := 0 + for g := range strings.SplitSeq(string(buf), "\n\n") { + if strings.Contains(g, ").writeRequest(") || strings.Contains(g, ").doRequest(") { + n++ + } + } + return n +} + +// TestTransportRequestGoroutineExitsRespHeaderTimeout is like +// TestTransportRequestGoroutineExits, but with a ResponseHeaderTimeout +// configured: the timeout is enforced by a timer rather than a parked +// goroutine, and once response headers arrive the timer is disarmed and +// must not fire even long after the timeout elapses. +func TestTransportRequestGoroutineExitsRespHeaderTimeout(t *testing.T) { + synctest.Test(t, testTransportRequestGoroutineExitsRespHeaderTimeout) +} +func testTransportRequestGoroutineExitsRespHeaderTimeout(t *testing.T) { + const timeout = 1 * time.Second + tc := newTestClientConn(t, func(t1 *http.Transport) { + t1.ResponseHeaderTimeout = timeout + }) + tc.greet() + + req, _ := http.NewRequest("GET", "https://dummy.tld/", nil) + rt := tc.roundTrip(req) + + tc.wantFrameType(FrameHeaders) + + // The request-writing goroutine should be gone even before response + // headers arrive; the response header timeout is enforced by a timer. + synctest.Wait() + if n := requestWriteGoroutines(); n != 0 { + t.Errorf("got %d request-writing goroutines parked awaiting response headers; want 0", n) + } + + // Response headers arrive within the timeout. + time.Sleep(timeout / 2) + tc.writeHeaders(HeadersFrameParam{ + StreamID: rt.streamID(), + EndHeaders: true, + EndStream: false, + BlockFragment: tc.makeHeaderBlockFragment(":status", "200"), + }) + rt.wantStatus(200) + + // Long after the response header timeout has elapsed, the + // still-streaming response must be unaffected. + time.Sleep(10 * timeout) + synctest.Wait() + tc.writeData(rt.streamID(), true, []byte("hello")) + rt.wantBody([]byte("hello")) +}