[tailscale] net/http/internal/http2: let request-write goroutines exit early

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)
+		}
 	}
 }
 

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"))
+}