From bb64088c02083e531c3e352cd416792cee6463ad Mon Sep 17 00:00:00 2001 From: Brad Fitzpatrick Date: Wed, 29 Jul 2026 03:45:18 +0000 Subject: [PATCH] [tailscale] net/http/internal/http2: let request-write goroutines exit early The goroutine spawned per RoundTrip to write the request previously parked until the stream ended, even after the request was fully sent, just to wait for the stream-end events and run cleanupWriteRequest. For clients with many concurrent long-lived response streams (long polls, event streams), that's a parked goroutine and its stack per stream doing nothing, which adds up to a large fraction of such a client's memory use. Once the request is fully sent, detach: the goroutine exits, and cleanupWriteRequest instead runs (on a short-lived goroutine) from whichever stream-end event fires first. Both stream-end events (peer half-close and abort) are raised under cc.mu, so a mutex-guarded flag gives exactly-once cleanup dispatch. Request context cancellation is watched via context.AfterFunc, and ResponseHeaderTimeout is enforced by a time.AfterFunc timer armed at detach and disarmed when response headers arrive. Only the deprecated Request.Cancel channel still requires a parked goroutine to watch it, so requests using it keep the previous behavior. Intended for upstreaming; carried in the Tailscale fork until then. Updates tailscale/corp#29053 Signed-off-by: Brad Fitzpatrick Change-Id: I5500c691194457195b0869c1612bf93718ca96c1 --- src/net/http/internal/http2/transport.go | 134 +++++++++++++++++- src/net/http/internal/http2/transport_test.go | 97 +++++++++++++ 2 files changed, 230 insertions(+), 1 deletion(-) 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")) +}