From 1df915697d8af18e7eb757a949758160bb14dcfb Mon Sep 17 00:00:00 2001 From: Brad Fitzpatrick Date: Fri, 11 Sep 2026 08:33:59 +0000 Subject: [PATCH 1/2] [tailscale-cherrypick] net/http/internal/http2: let the server's serve goroutine exit when idle Implement the TODO from 2014 at the top of server.go: a server connection with nothing to do used to pin a goroutine blocked in the serve loop (reached through net/http's conn.serve and ServeConn) in addition to the readFrames goroutine; now it pins only readFrames. When the serve loop has nothing runnable (nothing being written, unflushed, or poppable on the write side, and no shutdown in progress), the serve goroutine parks: it simply exits. Every sender into the serve loop's channels first calls beginServeSend, which counts in-flight sends and revives a parked loop on a new goroutine, so a frame from the peer, a handler write or body read, a timer firing, or a graceful shutdown request wakes the connection back up. Open streams and running handlers don't prevent parking, since the handler-side paths into the serve loop (response writes, body reads, stream deadline timers, pushes) all go through beginServeSend too. This helps SSE/long-poll-style servers: a connection whose handler is parked mid-long-poll for hours pins only the readFrames and handler goroutines, and the handler's next Write or Flush revives the loop. With open streams, the write scheduler can still hold DATA frames blocked on flow control; that's fine to park on, because only an incoming WINDOW_UPDATE or SETTINGS frame can unblock them, and incoming frames revive the loop. The serve loop's teardown became explicit rather than deferred, and it takes over the cleanup that ServeConn's callers used to do after it returned (closing the connection, unregistering it, and running net/http's StateClosed hook), signaled through the new ServeConnOpts.OnClose hook. Callers that don't set OnClose keep the old blocking ServeConn semantics and never park. Parking happens between the frames of an active request, so the park/resume cycle runs on hot paths. Spawning the resume goroutine with a pre-allocated method-value closure keeps it allocation-free: BenchmarkClientServer/h2 reports the same 68 allocs/op as before, with ns/op unchanged within noise. As a temporary safety measure, the GODEBUG=http2serveparking=0 setting disables parking, restoring the old behavior of keeping a goroutine parked for the lifetime of each connection. The setting is undocumented (#-prefixed), so it is not listed in doc/godebug.md or runtime/metrics. This is the server-side counterpart of CL 810780, which let the Transport's request-write goroutine exit early for the same reason: for servers with many mostly-idle connections, a parked goroutine and its stack per connection add up. Cherry-pick notes: the only conflict was in the serverConn struct field list, where this tree still has the srv field that upstream had removed and this change re-adds; the upstream field list was taken. The doc/next release note was dropped, since the 1.27 release branch has no doc/next directory. Updates golang/go#80735 Updates golang/go#81524 Updates tailscale/go#174 Change-Id: I4a9fd4e2ce7a7a4e0f84179e54a9cc97d1a62ad5 Reviewed-on: https://go-review.googlesource.com/c/go/+/831084 Reviewed-by: Nicholas Husin Reviewed-by: Nicholas Husin LUCI-TryBot-Result: golang-scoped@luci-project-accounts.iam.gserviceaccount.com Reviewed-by: Damien Neil (cherry picked from commit 5c51011e82ed754af072baaf4338ff35e76fe705) --- src/net/http/http2.go | 11 +- src/net/http/internal/http2/export_test.go | 10 + src/net/http/internal/http2/gotrack.go | 27 ++ src/net/http/internal/http2/server.go | 347 +++++++++++++++--- .../internal/http2/server_internal_test.go | 5 +- src/net/http/internal/http2/server_test.go | 145 ++++++++ src/net/http/omithttp2.go | 2 +- src/net/http/server.go | 25 +- 8 files changed, 505 insertions(+), 67 deletions(-) diff --git a/src/net/http/http2.go b/src/net/http/http2.go index 21468182ab0217..602459bf331fc1 100644 --- a/src/net/http/http2.go +++ b/src/net/http/http2.go @@ -87,11 +87,17 @@ func (s *Server) setHTTP2Config(conf http2ExternalServerConfig) { panic("http: HTTP/2 Server already registered") } s.h2Config = conf - s.h2Config.ServeConnFunc(s.serveHTTP2Conn) + s.h2Config.ServeConnFunc(func(ctx context.Context, nc net.Conn, h Handler, sawClientPreface bool, upgradeReq *Request, settings []byte) { + s.serveHTTP2Conn(ctx, nc, h, sawClientPreface, upgradeReq, settings, nil) + }) s.configureHTTP2() } -func (s *Server) serveHTTP2Conn(ctx context.Context, nc net.Conn, h Handler, sawClientPreface bool, upgradeReq *Request, settings []byte) { +// serveHTTP2Conn serves nc with the HTTP/2 server. It may return before the +// connection is done being served (an idle HTTP/2 connection doesn't hold +// onto a goroutine); onClose, if non-nil, runs once the connection is done +// and has been closed. +func (s *Server) serveHTTP2Conn(ctx context.Context, nc net.Conn, h Handler, sawClientPreface bool, upgradeReq *Request, settings []byte, onClose func()) { s.setupHTTP2_ServeTLS() var serverUpgradeReq *http2.ServerRequest if upgradeReq != nil { @@ -106,6 +112,7 @@ func (s *Server) serveHTTP2Conn(ctx context.Context, nc net.Conn, h Handler, saw SawClientPreface: sawClientPreface, UpgradeRequest: serverUpgradeReq, Settings: settings, + OnClose: onClose, }) } diff --git a/src/net/http/internal/http2/export_test.go b/src/net/http/internal/http2/export_test.go index e9d7c6bead5d9d..21e2f2ebdc1af7 100644 --- a/src/net/http/internal/http2/export_test.go +++ b/src/net/http/internal/http2/export_test.go @@ -82,6 +82,7 @@ func (sc *serverConn) TestFlowControlConsumed() (consumed int32) { func (sc *serverConn) TestStreamExists(id uint32) bool { ch := make(chan bool, 1) + sc.beginServeSend() sc.serveMsgCh <- func(int) { ch <- (sc.streams[id] != nil) } @@ -90,6 +91,7 @@ func (sc *serverConn) TestStreamExists(id uint32) bool { func (sc *serverConn) TestStreamState(id uint32) streamState { ch := make(chan streamState, 1) + sc.beginServeSend() sc.serveMsgCh <- func(int) { state, _ := sc.state(id) ch <- state @@ -97,6 +99,14 @@ func (sc *serverConn) TestStreamState(id uint32) streamState { return <-ch } +// TestServeParked reports whether the connection's serve goroutine has +// exited because the connection is idle. See serverConn.serveLoop. +func (sc *serverConn) TestServeParked() bool { + sc.parkMu.Lock() + defer sc.parkMu.Unlock() + return sc.parked +} + func (sc *serverConn) StartGracefulShutdown() { sc.startGracefulShutdown() } func (sc *serverConn) TestHPACKEncoder() *hpack.Encoder { diff --git a/src/net/http/internal/http2/gotrack.go b/src/net/http/internal/http2/gotrack.go index ba0db1690c77ef..14f651a6232f2c 100644 --- a/src/net/http/internal/http2/gotrack.go +++ b/src/net/http/internal/http2/gotrack.go @@ -57,6 +57,33 @@ func (g goroutineLock) checkNotOn() { } } +// A goroutineLocker is a goroutineLock whose owning goroutine can be +// atomically replaced, for values whose owning goroutine changes over +// their lifetime (such as the server's serve loop, which parks when +// the connection is idle and resumes on a new goroutine). +type goroutineLocker struct { + v atomic.Uint64 +} + +// setOwner records the current goroutine as the owner. +func (g *goroutineLocker) setOwner() { + g.v.Store(uint64(newGoroutineLock())) +} + +func (g *goroutineLocker) check() { + if !DebugGoroutines { + return + } + goroutineLock(g.v.Load()).check() +} + +func (g *goroutineLocker) checkNotOn() { + if !DebugGoroutines { + return + } + goroutineLock(g.v.Load()).checkNotOn() +} + var goroutineSpace = []byte("goroutine ") func curGoroutineID() uint64 { diff --git a/src/net/http/internal/http2/server.go b/src/net/http/internal/http2/server.go index bd2533a7c36e66..3bd6b1f1ad04bd 100644 --- a/src/net/http/internal/http2/server.go +++ b/src/net/http/internal/http2/server.go @@ -2,18 +2,6 @@ // Use of this source code is governed by a BSD-style // license that can be found in the LICENSE file. -// TODO: turn off the serve goroutine when idle, so -// an idle conn only has the readFrames goroutine active. (which could -// also be optimized probably to pin less memory in crypto/tls). This -// would involve tracking when the serve goroutine is active (atomic -// int32 read/CAS probably?) and starting it up when frames arrive, -// and shutting it down when all handlers exit. the occasional PING -// packets could use time.AfterFunc to call sc.wakeStartServeLoop() -// (which is a no-op if already running) and then queue the PING write -// as normal. The serve loop would then exit in most cases (if no -// Handlers running) and not be woken up again until the PING packet -// returns. - // TODO (maybe): add a mechanism for Handlers to going into // half-closed-local mode (rw.(io.Closer) test?) but not exit their // handler, and continue to be able to read from the @@ -33,6 +21,7 @@ import ( "crypto/tls" "errors" "fmt" + "internal/godebug" "internal/synctest" "io" "log" @@ -220,6 +209,16 @@ type ServeConnOpts struct { // SawClientPreface is set if the HTTP/2 connection preface // has already been read from the connection. SawClientPreface bool + + // OnClose is an optional callback that runs exactly once, after the + // connection is done being served and has been closed. It may run on + // any goroutine. + // + // Setting OnClose also grants ServeConn permission to return before + // the connection is done (see the ServeConn documentation), so + // callers providing it must release their per-connection resources + // here rather than after ServeConn returns. + OnClose func() } func (o *ServeConnOpts) context() context.Context { @@ -230,7 +229,11 @@ func (o *ServeConnOpts) context() context.Context { } // ServeConn serves HTTP/2 requests on the provided connection and -// blocks until the connection is no longer readable. +// blocks until the connection is no longer readable. As an exception, +// if opts.OnClose is set, ServeConn may instead return as soon as the +// connection goes idle, handing its servicing off to background +// goroutines; the connection is still being served after ServeConn +// returns, and opts.OnClose reports when it is done. // // ServeConn starts speaking HTTP/2 assuming that c has not had any // reads or writes. It writes its initial settings frame and expects @@ -266,7 +269,6 @@ var ( func (s *Server) serveConn(c net.Conn, opts *ServeConnOpts, newf func(*serverConn)) { baseCtx, cancel := serverConnBaseContext(c, opts) - defer cancel() conf := configFromServer(opts.BaseConfig) sc := &serverConn{ @@ -274,6 +276,8 @@ func (s *Server) serveConn(c net.Conn, opts *ServeConnOpts, newf func(*serverCon hs: opts.BaseConfig, conn: c, baseCtx: baseCtx, + baseCtxCancel: cancel, + onClose: opts.OnClose, remoteAddrStr: c.RemoteAddr().String(), bw: newBufferedWriter(c, conf.WriteByteTimeout), handler: opts.Handler, @@ -291,16 +295,33 @@ func (s *Server) serveConn(c net.Conn, opts *ServeConnOpts, newf func(*serverCon maxFrameSize: initialMaxFrameSize, pingTimeout: conf.PingTimeout, countErrorFunc: conf.CountError, - serveG: newGoroutineLock(), pushEnabled: true, sawClientPreface: opts.SawClientPreface, } + sc.serveG.setOwner() // serve is called below on this same goroutine + sc.resumeServeFunc = sc.resumeServe + if opts.OnClose != nil && http2serveparking.Value() == "0" { + sc.serveParkingDisabled = true + } if newf != nil { newf(sc) } s.registerConn(sc) - defer s.unregisterConn(sc) + served := false + defer func() { + if !served { + // We never made it to sc.serve, whose teardown owns + // releasing the connection's resources, so release them + // here instead. (We either rejected the connection or + // panicked during setup.) + s.unregisterConn(sc) + cancel() + if f := sc.onClose; f != nil { + f() + } + } + }() switch { case sc.hs.DisableClientPriority(): @@ -411,6 +432,7 @@ func (s *Server) serveConn(c net.Conn, opts *ServeConnOpts, newf func(*serverCon opts.UpgradeRequest = nil } + served = true sc.serve(conf) } @@ -428,28 +450,46 @@ func (sc *serverConn) rejectConn(err ErrCode, debug string) { type serverConn struct { // Immutable: - srv *Server - hs ServerConfig - conn net.Conn - bw *bufferedWriter // writing to conn - handler Handler - baseCtx context.Context - framer *Framer - doneServing chan struct{} // closed when serverConn.serve ends - readFrameCh chan readFrameResult // written by serverConn.readFrames - wantWriteFrameCh chan FrameWriteRequest // from handlers -> serve - wroteFrameCh chan frameWriteResult // from writeFrameAsync -> serve, tickles more frame writes - bodyReadCh chan bodyReadMsg // from handlers -> serve - serveMsgCh chan any // misc messages & code to send to / run on the serve loop - flow outflow // conn-wide (not stream-specific) outbound flow control - inflow inflow // conn-wide inbound flow control - tlsState *tls.ConnectionState // shared by all handlers, like net/http - remoteAddrStr string - writeSched WriteScheduler - countErrorFunc func(errType string) + srv *Server // for unregisterConn in teardown + hs ServerConfig + conn net.Conn + bw *bufferedWriter // writing to conn + handler Handler + baseCtx context.Context + baseCtxCancel context.CancelFunc + onClose func() // or nil; from ServeConnOpts.OnClose, called by teardown + + // serveParkingDisabled is whether GODEBUG=http2serveparking=0 + // disabled parking the serve goroutine for this connection. + serveParkingDisabled bool + framer *Framer + doneServing chan struct{} // closed by teardown when the connection is done + readFrameCh chan readFrameResult // written by serverConn.readFrames + wantWriteFrameCh chan FrameWriteRequest // from handlers -> serve + wroteFrameCh chan frameWriteResult // from writeFrameAsync -> serve, tickles more frame writes + bodyReadCh chan bodyReadMsg // from handlers -> serve + serveMsgCh chan any // misc messages & code to send to / run on the serve loop + flow outflow // conn-wide (not stream-specific) outbound flow control + inflow inflow // conn-wide inbound flow control + tlsState *tls.ConnectionState // shared by all handlers, like net/http + remoteAddrStr string + writeSched WriteScheduler + countErrorFunc func(errType string) + + // resumeServeFunc is sc.resumeServe, allocated once at setup so + // that beginServeSend's "go sc.resumeServeFunc()" doesn't allocate + // a method-value closure per park/resume cycle. + resumeServeFunc func() + + // parkMu guards the following fields, which coordinate parking the + // serve goroutine (exiting it while the connection is idle) with + // the goroutines that send it work. See serverConn.serveLoop. + parkMu sync.Mutex + parked bool // serve goroutine has exited; the next sender revives it + serveSendsPending int // sends to serve's channels begun but not yet received by serve // Everything following is owned by the serve loop; use serveG.check(): - serveG goroutineLock // used to verify funcs are on serve() + serveG goroutineLocker // used to verify funcs are on serve() pushEnabled bool sawClientPreface bool // preface has already been read, used in h2c upgrade sawFirstSettings bool // got the initial SETTINGS frame after the preface @@ -486,6 +526,9 @@ type serverConn struct { readIdleTimeout time.Duration pingTimeout time.Duration readIdleTimer *time.Timer // nil if unused + settingsTimer *time.Timer // fires if the client never sends SETTINGS; nil after the first frame arrives + lastFrameTime time.Time // when we last read a frame from the client + loopNum int // serve loop iteration counter, observable by tests // Owned by the writeFrameAsync goroutine: headerWriteBuf bytes.Buffer @@ -710,9 +753,11 @@ func (sc *serverConn) readFrames() { gateDone := func() { gate <- struct{}{} } for { f, err := sc.framer.ReadFrame() + sc.beginServeSend() select { case sc.readFrameCh <- readFrameResult{f, err, gateDone}: case <-sc.doneServing: + sc.endServeSend() return } select { @@ -744,6 +789,7 @@ func (sc *serverConn) writeFrameAsync(wr FrameWriteRequest, wd *writeData) { } else { err = sc.framer.endWrite() } + sc.beginServeSend() sc.wroteFrameCh <- frameWriteResult{wr: wr, err: err} } @@ -779,10 +825,12 @@ func (sc *serverConn) notePanic() { func (sc *serverConn) serve(conf Config) { sc.serveG.check() defer sc.notePanic() - defer sc.conn.Close() - defer sc.closeAllStreamsOnConnClose() - defer sc.stopShutdownTimer() - defer close(sc.doneServing) // unblocks handlers trying to send + parked := false + defer func() { + if !parked { + sc.teardown() + } + }() if VerboseLogs { sc.vlogf("http2: server connection from %v on %p", sc.conn.RemoteAddr(), sc.hs) @@ -825,71 +873,145 @@ func (sc *serverConn) serve(conf Config) { if idle := sc.hs.IdleTimeout(); idle > 0 { sc.idleTimer = time.AfterFunc(idle, sc.onIdleTimer) - defer sc.idleTimer.Stop() } if conf.SendPingTimeout > 0 { sc.readIdleTimeout = conf.SendPingTimeout sc.readIdleTimer = time.AfterFunc(conf.SendPingTimeout, sc.onReadIdleTimer) - defer sc.readIdleTimer.Stop() } - go sc.readFrames() // closed by defer sc.conn.Close above + go sc.readFrames() // closed by the conn.Close in sc.teardown + + sc.settingsTimer = time.AfterFunc(firstSettingsTimeout, sc.onSettingsTimer) + + sc.lastFrameTime = time.Now() + parked = sc.serveLoop() +} + +// resumeServe runs the serve loop on a new goroutine for a connection +// whose previous serve goroutine parked while the connection was idle. +// It is started by beginServeSend, which precedes every send to the +// serve loop's channels. +func (sc *serverConn) resumeServe() { + // The serve loop's first incarnation has net/http's conn.serve + // above it to recover from panics; this one has nothing, so + // recover here to keep a panic from crashing the process. + defer func() { + if err := recover(); err != nil && err != ErrAbortHandler { + const size = 64 << 10 + buf := make([]byte, size) + buf = buf[:runtime.Stack(buf, false)] + sc.logf("http2: panic serving %v: %v\n%s", sc.conn.RemoteAddr(), err, buf) + } + }() + sc.serveG.setOwner() + defer sc.notePanic() + parked := false + defer func() { + if !parked { + sc.teardown() + } + }() + parked = sc.serveLoop() +} - settingsTimer := time.AfterFunc(firstSettingsTimeout, sc.onSettingsTimer) - defer settingsTimer.Stop() +// teardown closes the connection and releases its resources. +// It runs (on the serve goroutine) when the serve loop exits. +// +// Because the serve goroutine may park while the connection is idle +// (making ServeConn return early), this is also responsible for the +// cleanup that ServeConn's callers used to do after it returned: +// unregistering the connection, canceling its base context, and +// notifying the ServeConnOpts.OnClose hook. +func (sc *serverConn) teardown() { + sc.serveG.check() + if t := sc.settingsTimer; t != nil { + t.Stop() + } + if t := sc.readIdleTimer; t != nil { + t.Stop() + } + if t := sc.idleTimer; t != nil { + t.Stop() + } + close(sc.doneServing) // unblocks handlers trying to send + sc.stopShutdownTimer() + sc.closeAllStreamsOnConnClose() + sc.conn.Close() + sc.srv.unregisterConn(sc) + sc.baseCtxCancel() + if f := sc.onClose; f != nil { + f() + } +} - lastFrameTime := time.Now() - loopNum := 0 +// serveLoop is the serve goroutine's main loop, processing frames and +// messages until the connection is done (returning false) or until it +// has nothing runnable and the goroutine parks (returning true). +// +// After it parks, the serve loop is resumed on a new goroutine by the +// next message sent to any of its channels: a frame from readFrames, a +// handler write or body read, a timer firing, or a graceful shutdown +// request. This way a connection with nothing to do only pins the +// readFrames goroutine (plus any handler goroutines), whether it's +// fully idle or its handlers are parked mid-long-poll. +func (sc *serverConn) serveLoop() (parked bool) { + sc.serveG.check() for { - loopNum++ + sc.loopNum++ select { case wr := <-sc.wantWriteFrameCh: + sc.endServeSend() if se, ok := wr.write.(StreamError); ok { sc.resetStream(se) break } sc.writeFrame(wr) case res := <-sc.wroteFrameCh: + sc.endServeSend() sc.wroteFrame(res) case res := <-sc.readFrameCh: - lastFrameTime = time.Now() + sc.endServeSend() + sc.lastFrameTime = time.Now() // Process any written frames before reading new frames from the client since a // written frame could have triggered a new stream to be started. if sc.writingFrameAsync { select { case wroteRes := <-sc.wroteFrameCh: + sc.endServeSend() sc.wroteFrame(wroteRes) default: } } if !sc.processFrameFromReader(res) { - return + return false } res.readMore() - if settingsTimer != nil { - settingsTimer.Stop() - settingsTimer = nil + if t := sc.settingsTimer; t != nil { + t.Stop() + sc.settingsTimer = nil } case m := <-sc.bodyReadCh: + sc.endServeSend() sc.noteBodyRead(m.st, m.n) case msg := <-sc.serveMsgCh: + sc.endServeSend() switch v := msg.(type) { case func(int): - v(loopNum) // for testing + v(sc.loopNum) // for testing case *serverMessage: switch v { case settingsTimerMsg: sc.logf("timeout waiting for SETTINGS frames from %v", sc.conn.RemoteAddr()) - return + return false case idleTimerMsg: sc.vlogf("connection is idle") sc.goAway(ErrCodeNo) case readIdleTimerMsg: - sc.handlePingTimer(lastFrameTime) + sc.handlePingTimer(sc.lastFrameTime) case shutdownTimerMsg: sc.vlogf("GOAWAY close timer fired; closing conn from %v", sc.conn.RemoteAddr()) - return + return false case gracefulShutdownMsg: sc.startGracefulShutdownInternal() case handlerDoneMsg: @@ -911,7 +1033,7 @@ func (sc *serverConn) serve(conf Config) { // run out of memory. if sc.queuedControlFrames > maxQueuedControlFrames { sc.vlogf("http2: too many control frames in send queue, closing connection") - return + return false } // Start the shutdown timer after sending a GOAWAY. When sending GOAWAY @@ -922,7 +1044,83 @@ func (sc *serverConn) serve(conf Config) { if sentGoAway && sc.shutdownTimer == nil && (sc.goAwayCode != ErrCodeNo || gracefulShutdownComplete) { sc.shutDownIn(goAwayTimeout) } + + // If the connection is idle, park: let this goroutine exit and + // have the next message revive the loop on a new goroutine. + if sc.canPark() && sc.tryPark() { + return true + } + } +} + +// canPark reports whether the serve loop has nothing runnable and can +// park: nothing being written or unflushed, nothing poppable in the +// write scheduler, and no shutdown in progress. +// +// Open streams and running handlers do not prevent parking: everything +// that could next give the serve loop work (a frame from the peer, a +// handler write or body read, a timer firing, a graceful shutdown +// request) reaches it through a channel send preceded by +// beginServeSend, which revives the loop. A connection whose handlers +// are parked mid-long-poll therefore doesn't pin a serve goroutine +// either. +// +// The write side is known to be drained because every state change +// ends in scheduleFrameWrite, which (except in a GOAWAY with an error +// code, and inGoAway is false here) pops the write scheduler until Pop +// fails and then flushes, so writingFrame, writingFrameAsync, and +// needsFrameFlush are all false only once nothing was left to write. +// With queuedControlFrames == 0, anything still in the scheduler after +// a failed Pop is a DATA frame blocked on flow control (non-DATA +// frames are always consumed whole; see FrameWriteRequest.Consume), +// and only an incoming WINDOW_UPDATE or SETTINGS frame, which arrives +// via readFrames and revives the loop, can unblock it. +// http2serveparking, if "0", disables parking the serve goroutine +// while a connection has nothing to do, restoring the old behavior of +// keeping a goroutine parked for the lifetime of each connection. +// +// It exists only as a temporary safety measure for the change that +// parks the serve goroutine. The # prefix marks it as undocumented, +// so it is not listed in doc/godebug.md or runtime/metrics. +// +// TODO: remove this setting; see #81524. +var http2serveparking = godebug.New("#http2serveparking") + +func (sc *serverConn) canPark() bool { + sc.serveG.check() + // Parking makes ServeConn return early, which only callers that set + // ServeConnOpts.OnClose are prepared for. + return !sc.serveParkingDisabled && + sc.onClose != nil && + !sc.writingFrame && + !sc.writingFrameAsync && + !sc.needsFrameFlush && + !sc.needToSendGoAway && + !sc.needToSendSettingsAck && + !sc.inGoAway && + sc.queuedControlFrames == 0 +} + +// tryPark attempts to mark the serve goroutine as parked, so that the +// next beginServeSend revives the serve loop on a new goroutine. It +// fails if a sender has already begun sending a message to one of the +// serve loop's channels, in which case the caller must keep serving to +// receive it. +func (sc *serverConn) tryPark() bool { + sc.serveG.check() + sc.parkMu.Lock() + defer sc.parkMu.Unlock() + if sc.serveSendsPending != 0 { + return false } + // With no sends pending, the channels must all be empty too, but + // parking with a queued message would stall the connection, so be + // defensive. + if len(sc.readFrameCh)+len(sc.wantWriteFrameCh)+len(sc.wroteFrameCh)+len(sc.bodyReadCh)+len(sc.serveMsgCh) != 0 { + return false + } + sc.parked = true + return true } func (sc *serverConn) handlePingTimer(lastFrameReadTime time.Time) { @@ -971,11 +1169,37 @@ func (sc *serverConn) onIdleTimer() { sc.sendServeMsg(idleTimerMsg) } func (sc *serverConn) onReadIdleTimer() { sc.sendServeMsg(readIdleTimerMsg) } func (sc *serverConn) onShutdownTimer() { sc.sendServeMsg(shutdownTimerMsg) } +// beginServeSend notes that the caller is about to send a message to +// one of the serve loop's channels, reviving the serve goroutine if it +// has parked. Every send to those channels must be preceded by a call +// to beginServeSend. The send is counted as pending until the serve +// loop receives the message, or until a sender that bailed out on +// another select case gives up; both call endServeSend. +func (sc *serverConn) beginServeSend() { + sc.parkMu.Lock() + defer sc.parkMu.Unlock() + sc.serveSendsPending++ + if sc.parked { + sc.parked = false + go sc.resumeServeFunc() + } +} + +// endServeSend balances an earlier beginServeSend, either because the +// serve loop received the message or because the sender never sent it. +func (sc *serverConn) endServeSend() { + sc.parkMu.Lock() + defer sc.parkMu.Unlock() + sc.serveSendsPending-- +} + func (sc *serverConn) sendServeMsg(msg any) { sc.serveG.checkNotOn() // NOT + sc.beginServeSend() select { case sc.serveMsgCh <- msg: case <-sc.doneServing: + sc.endServeSend() } } @@ -1070,12 +1294,14 @@ func (sc *serverConn) writeDataFromHandler(stream *stream, data []byte, endStrea // goroutine, call writeFrame instead. func (sc *serverConn) writeFrameFromHandler(wr FrameWriteRequest) error { sc.serveG.checkNotOn() // NOT + sc.beginServeSend() select { case sc.wantWriteFrameCh <- wr: return nil case <-sc.doneServing: // Serve loop is gone. // Client has closed their connection to the server. + sc.endServeSend() return errClientDisconnected } } @@ -2411,9 +2637,11 @@ type bodyReadMsg struct { func (sc *serverConn) noteBodyReadFromHandler(st *stream, n int, err error) { sc.serveG.checkNotOn() // NOT on if n > 0 { + sc.beginServeSend() select { case sc.bodyReadCh <- bodyReadMsg{st, n}: case <-sc.doneServing: + sc.endServeSend() } } } @@ -3088,10 +3316,13 @@ func (w *responseWriter) Push(target, method string, header Header) error { done: getErrChan(), } + sc.beginServeSend() select { case <-sc.doneServing: + sc.endServeSend() return errClientDisconnected case <-st.cw: + sc.endServeSend() return errStreamClosed case sc.serveMsgCh <- msg: } diff --git a/src/net/http/internal/http2/server_internal_test.go b/src/net/http/internal/http2/server_internal_test.go index 379b5de46dfc40..83487cfd51c180 100644 --- a/src/net/http/internal/http2/server_internal_test.go +++ b/src/net/http/internal/http2/server_internal_test.go @@ -58,9 +58,8 @@ func TestCheckValidHTTP2Request(t *testing.T) { func TestCanonicalHeaderCacheGrowth(t *testing.T) { for _, size := range []int{1, (1 << 20) - 10} { base := strings.Repeat("X", size) - sc := &serverConn{ - serveG: newGoroutineLock(), - } + sc := &serverConn{} + sc.serveG.setOwner() count := 0 added := 0 for added < 10*maxCachedCanonicalHeadersKeysSize { diff --git a/src/net/http/internal/http2/server_test.go b/src/net/http/internal/http2/server_test.go index f15ad9e666cb10..237648560986ef 100644 --- a/src/net/http/internal/http2/server_test.go +++ b/src/net/http/internal/http2/server_test.go @@ -3719,6 +3719,151 @@ func testServerIdleTimeout_AfterRequest(t *testing.T) { st.wantGoAway(1, ErrCodeNo) } +// wantParked asserts the state of the connection's serve goroutine. +func (st *serverTester) wantParked(want bool) { + st.t.Helper() + st.sync() + if got := st.sc.TestServeParked(); got != want { + st.t.Errorf("serve goroutine parked = %v, want %v", got, want) + } +} + +func TestServerParksWhenIdle(t *testing.T) { synctest.Test(t, testServerParksWhenIdle) } +func testServerParksWhenIdle(t *testing.T) { + unblock := make(chan struct{}) + st := newServerTester(t, func(w http.ResponseWriter, r *http.Request) { + <-unblock + }) + defer st.Close() + + // The serve goroutine parks once the connection is established + // and idle. + st.greet() + st.wantParked(true) + + // It parks even while a handler is running, as long as the + // handler isn't giving it any work. + st.bodylessReq1() + st.wantParked(true) + + // It parks again once the request is done. + close(unblock) + st.wantHeaders(wantHeader{ + streamID: 1, + endStream: true, + }) + st.wantParked(true) + + // A PING on the parked connection is acked as usual. + st.writePing(false, [8]byte{1, 2, 3, 4, 5, 6, 7, 8}) + st.wantFrameType(FramePing) + st.wantParked(true) +} + +func TestServerParksDuringLongPoll(t *testing.T) { synctest.Test(t, testServerParksDuringLongPoll) } +func testServerParksDuringLongPoll(t *testing.T) { + // An SSE-style handler: write an event, flush, park for a long + // time, write another event. + events := make(chan string) + st := newServerTester(t, func(w http.ResponseWriter, r *http.Request) { + for ev := range events { + io.WriteString(w, ev) + w.(http.Flusher).Flush() + } + }) + defer st.Close() + + st.greet() + st.bodylessReq1() + + events <- "hello" + st.wantHeaders(wantHeader{ + streamID: 1, + endStream: false, + }) + st.wantData(wantData{ + streamID: 1, + endStream: false, + data: []byte("hello"), + }) + + // The handler is parked between events, and so is the serve + // goroutine, even though the stream stays open for however long + // the long poll lasts. + st.wantParked(true) + st.advance(24 * time.Hour) + st.wantParked(true) + + // The next event revives the serve goroutine. + events <- "world" + st.wantData(wantData{ + streamID: 1, + endStream: false, + data: []byte("world"), + }) + st.wantParked(true) + + // So does the handler finishing. + close(events) + st.wantData(wantData{ + streamID: 1, + endStream: true, + data: []byte{}, + }) + st.wantParked(true) +} + +func TestServerParkedGracefulShutdown(t *testing.T) { + synctest.Test(t, testServerParkedGracefulShutdown) +} +func testServerParkedGracefulShutdown(t *testing.T) { + st := newServerTester(t, nil) + defer st.Close() + + st.greet() + st.wantParked(true) + + // Server.Shutdown revives the parked serve goroutine, which sends + // a GOAWAY and tears the connection down; it must not park again + // mid-shutdown. + st.sc.StartGracefulShutdown() + st.wantGoAway(0, ErrCodeNo) + st.wantParked(false) +} + +func TestServerParkedIdleTimeout(t *testing.T) { synctest.Test(t, testServerParkedIdleTimeout) } +func testServerParkedIdleTimeout(t *testing.T) { + const idleTimeout = 1 * time.Second + st := newServerTester(t, nil, func(s *http.Server) { + s.IdleTimeout = idleTimeout + }) + defer st.Close() + + // The idle timeout must still fire on a parked connection. + st.greet() + st.wantParked(true) + st.advance(idleTimeout) + st.wantGoAway(0, ErrCodeNo) +} + +func TestServerParkingDisabledByGODEBUG(t *testing.T) { + t.Setenv("GODEBUG", "http2serveparking=0") + synctest.Test(t, testServerParkingDisabledByGODEBUG) +} +func testServerParkingDisabledByGODEBUG(t *testing.T) { + st := newServerTester(t, func(w http.ResponseWriter, r *http.Request) {}) + defer st.Close() + + st.greet() + st.wantParked(false) + st.bodylessReq1() + st.wantHeaders(wantHeader{ + streamID: 1, + endStream: true, + }) + st.wantParked(false) +} + // grpc-go closes the Request.Body currently with a Read. // Verify that it doesn't race. // See https://github.com/grpc/grpc-go/pull/938 diff --git a/src/net/http/omithttp2.go b/src/net/http/omithttp2.go index 5745cd9797244e..775ee60093da95 100644 --- a/src/net/http/omithttp2.go +++ b/src/net/http/omithttp2.go @@ -20,7 +20,7 @@ const noHTTP2 = "no bundled HTTP/2" // should never see this func (s *Server) configureHTTP2() {} func (s *Server) setHTTP2Config(conf http2ExternalServerConfig) {} -func (s *Server) serveHTTP2Conn(ctx context.Context, nc net.Conn, h Handler, sawClientPreface bool, upgradeReq *Request, settings []byte) { +func (s *Server) serveHTTP2Conn(ctx context.Context, nc net.Conn, h Handler, sawClientPreface bool, upgradeReq *Request, settings []byte, onClose func()) { } func (t *Transport) configureHTTP2(protocols Protocols) {} diff --git a/src/net/http/server.go b/src/net/http/server.go index 5cd28fefe0782a..eef90e82dc4ab1 100644 --- a/src/net/http/server.go +++ b/src/net/http/server.go @@ -306,6 +306,12 @@ type conn struct { // by a Handler with the Hijacker interface. // It is guarded by mu. hijackedv bool + + // http2HandedOff is whether the connection has been handed off to + // the HTTP/2 server, which then owns closing the connection and + // running the final ConnState hook, possibly after (*conn).serve + // has returned. It is only accessed by the (*conn).serve goroutine. + http2HandedOff bool } func (c *conn) hijacked() bool { @@ -2014,7 +2020,7 @@ func (c *conn) serve(ctx context.Context) { inFlightResponse.cancelCtx() inFlightResponse.disableWriteContinue(true) } - if !c.hijacked() { + if !c.hijacked() && !c.http2HandedOff { if inFlightResponse != nil { inFlightResponse.conn.r.abortPendingRead() inFlightResponse.reqBody.Close() @@ -2072,7 +2078,7 @@ func (c *conn) serve(ctx context.Context) { // closing such connections. See issue https://golang.org/issue/39776. c.setState(c.rwc, StateActive, skipHooks) const sawClientPreface = false - c.server.serveHTTP2Conn(ctx, c.rwc, serverHandler{c.server}, sawClientPreface, nil, nil) + c.serveHTTP2(ctx, sawClientPreface) return } tlsConn, tlsConnOK := c.rwc.(*tls.Conn) @@ -2352,7 +2358,7 @@ func (c *conn) maybeServeUnencryptedHTTP2(ctx context.Context) bool { c.setState(c.rwc, StateActive, skipHooks) if c.server.h2 != nil { const sawClientPreface = true - c.server.serveHTTP2Conn(ctx, c.rwc, serverHandler{c.server}, sawClientPreface, nil, nil) + c.serveHTTP2(ctx, sawClientPreface) } else { c.rwc.SetReadDeadline(time.Time{}) c.rwc.SetWriteDeadline(time.Time{}) @@ -2362,6 +2368,19 @@ func (c *conn) maybeServeUnencryptedHTTP2(ctx context.Context) bool { return true } +// serveHTTP2 hands the connection off to the HTTP/2 server, which owns +// the connection from here on: it closes the connection and runs the +// final ConnState hook when it's done, possibly after this function has +// returned, since an idle HTTP/2 connection doesn't hold onto a +// goroutine. +func (c *conn) serveHTTP2(ctx context.Context, sawClientPreface bool) { + c.http2HandedOff = true + c.server.serveHTTP2Conn(ctx, c.rwc, serverHandler{c.server}, sawClientPreface, nil, nil, func() { + c.close() + c.setState(c.rwc, StateClosed, runHooks) + }) +} + func (w *response) sendExpectationFailed() { // TODO(bradfitz): let ServeHTTP handlers handle // requests with non-standard expectation[s]? Seems From fae738f339b4b14a9cdbfd022181d055b9756fe1 Mon Sep 17 00:00:00 2001 From: Brad Fitzpatrick Date: Mon, 14 Sep 2026 19:39:21 +0000 Subject: [PATCH 2/2] [tailscale-cherrypick] net/http/internal/http2: do some follow-ups to serve parking change Fix a rebase comment mishap: the doc comment describing canPark was attached to the http2serveparking var, leaving canPark undocumented. Skip the parkMu accounting in beginServeSend and endServeSend on connections that can never park (no OnClose hook, or parking disabled with GODEBUG=http2serveparking=0), removing a mutex round trip per frame on both the sender and receiver side. Add a test that a response blocked on the peer's flow control window parks the serve goroutine with its DATA queued in the write scheduler, and that a WINDOW_UPDATE revives it. Updates golang/go#80735 Updates golang/go#81524 Updates tailscale/go#174 Change-Id: Ic1a3f7b9d2e5a4c8f6b0d7a9e3c5f2b8d1a6e4c7 Reviewed-on: https://go-review.googlesource.com/c/go/+/832504 LUCI-TryBot-Result: golang-scoped@luci-project-accounts.iam.gserviceaccount.com Reviewed-by: Damien Neil Reviewed-by: Nicholas Husin Reviewed-by: Nicholas Husin (cherry picked from commit 0ac31b68d0787ed4a2b13a5dc35d3cba628e8bde) --- src/net/http/internal/http2/server.go | 77 ++++++++++++---------- src/net/http/internal/http2/server_test.go | 75 +++++++++++++++++++-- 2 files changed, 112 insertions(+), 40 deletions(-) diff --git a/src/net/http/internal/http2/server.go b/src/net/http/internal/http2/server.go index 3bd6b1f1ad04bd..b151d8fa51d216 100644 --- a/src/net/http/internal/http2/server.go +++ b/src/net/http/internal/http2/server.go @@ -300,8 +300,8 @@ func (s *Server) serveConn(c net.Conn, opts *ServeConnOpts, newf func(*serverCon } sc.serveG.setOwner() // serve is called below on this same goroutine sc.resumeServeFunc = sc.resumeServe - if opts.OnClose != nil && http2serveparking.Value() == "0" { - sc.serveParkingDisabled = true + if opts.OnClose != nil && http2serveparking.Value() != "0" { + sc.serveParkingEnabled = true } if newf != nil { newf(sc) @@ -459,22 +459,28 @@ type serverConn struct { baseCtxCancel context.CancelFunc onClose func() // or nil; from ServeConnOpts.OnClose, called by teardown - // serveParkingDisabled is whether GODEBUG=http2serveparking=0 - // disabled parking the serve goroutine for this connection. - serveParkingDisabled bool - framer *Framer - doneServing chan struct{} // closed by teardown when the connection is done - readFrameCh chan readFrameResult // written by serverConn.readFrames - wantWriteFrameCh chan FrameWriteRequest // from handlers -> serve - wroteFrameCh chan frameWriteResult // from writeFrameAsync -> serve, tickles more frame writes - bodyReadCh chan bodyReadMsg // from handlers -> serve - serveMsgCh chan any // misc messages & code to send to / run on the serve loop - flow outflow // conn-wide (not stream-specific) outbound flow control - inflow inflow // conn-wide inbound flow control - tlsState *tls.ConnectionState // shared by all handlers, like net/http - remoteAddrStr string - writeSched WriteScheduler - countErrorFunc func(errType string) + // serveParkingEnabled is whether this connection's serve goroutine + // may park when the connection is idle. It requires the OnClose + // hook (which is what lets ServeConn return early) and + // GODEBUG=http2serveparking to be enabled. It is set once in + // serveConn, before any goroutine that sends to the serve loop's + // channels exists, and never changes, so it can be read without a + // lock, letting beginServeSend and endServeSend skip the parkMu + // accounting on connections that can never park. + serveParkingEnabled bool + framer *Framer + doneServing chan struct{} // closed by teardown when the connection is done + readFrameCh chan readFrameResult // written by serverConn.readFrames + wantWriteFrameCh chan FrameWriteRequest // from handlers -> serve + wroteFrameCh chan frameWriteResult // from writeFrameAsync -> serve, tickles more frame writes + bodyReadCh chan bodyReadMsg // from handlers -> serve + serveMsgCh chan any // misc messages & code to send to / run on the serve loop + flow outflow // conn-wide (not stream-specific) outbound flow control + inflow inflow // conn-wide inbound flow control + tlsState *tls.ConnectionState // shared by all handlers, like net/http + remoteAddrStr string + writeSched WriteScheduler + countErrorFunc func(errType string) // resumeServeFunc is sc.resumeServe, allocated once at setup so // that beginServeSend's "go sc.resumeServeFunc()" doesn't allocate @@ -1075,23 +1081,9 @@ func (sc *serverConn) serveLoop() (parked bool) { // frames are always consumed whole; see FrameWriteRequest.Consume), // and only an incoming WINDOW_UPDATE or SETTINGS frame, which arrives // via readFrames and revives the loop, can unblock it. -// http2serveparking, if "0", disables parking the serve goroutine -// while a connection has nothing to do, restoring the old behavior of -// keeping a goroutine parked for the lifetime of each connection. -// -// It exists only as a temporary safety measure for the change that -// parks the serve goroutine. The # prefix marks it as undocumented, -// so it is not listed in doc/godebug.md or runtime/metrics. -// -// TODO: remove this setting; see #81524. -var http2serveparking = godebug.New("#http2serveparking") - func (sc *serverConn) canPark() bool { sc.serveG.check() - // Parking makes ServeConn return early, which only callers that set - // ServeConnOpts.OnClose are prepared for. - return !sc.serveParkingDisabled && - sc.onClose != nil && + return sc.serveParkingEnabled && !sc.writingFrame && !sc.writingFrameAsync && !sc.needsFrameFlush && @@ -1101,6 +1093,17 @@ func (sc *serverConn) canPark() bool { sc.queuedControlFrames == 0 } +// http2serveparking, if "0", disables parking the serve goroutine +// while a connection has nothing to do, restoring the old behavior of +// keeping a goroutine parked for the lifetime of each connection. +// +// It exists only as a temporary safety measure for the change that +// parks the serve goroutine. The # prefix marks it as undocumented, +// so it is not listed in doc/godebug.md or runtime/metrics. +// +// TODO: remove this setting; see #81524. +var http2serveparking = godebug.New("#http2serveparking") + // tryPark attempts to mark the serve goroutine as parked, so that the // next beginServeSend revives the serve loop on a new goroutine. It // fails if a sender has already begun sending a message to one of the @@ -1176,6 +1179,11 @@ func (sc *serverConn) onShutdownTimer() { sc.sendServeMsg(shutdownTimerMsg) } // loop receives the message, or until a sender that bailed out on // another select case gives up; both call endServeSend. func (sc *serverConn) beginServeSend() { + if !sc.serveParkingEnabled { + // The serve loop never parks, so there is nothing to count + // or revive, and no lock is needed. + return + } sc.parkMu.Lock() defer sc.parkMu.Unlock() sc.serveSendsPending++ @@ -1188,6 +1196,9 @@ func (sc *serverConn) beginServeSend() { // endServeSend balances an earlier beginServeSend, either because the // serve loop received the message or because the sender never sent it. func (sc *serverConn) endServeSend() { + if !sc.serveParkingEnabled { + return + } sc.parkMu.Lock() defer sc.parkMu.Unlock() sc.serveSendsPending-- diff --git a/src/net/http/internal/http2/server_test.go b/src/net/http/internal/http2/server_test.go index 237648560986ef..9358d3bb33601a 100644 --- a/src/net/http/internal/http2/server_test.go +++ b/src/net/http/internal/http2/server_test.go @@ -421,15 +421,21 @@ func (st *serverTester) Close() { // frames may be sent. func (st *serverTester) greet() { st.t.Helper() - st.greetAndCheckSettings(func(Setting) error { return nil }) + st.greetAndCheckSettings(nil, func(Setting) error { return nil }) } -func (st *serverTester) greetAndCheckSettings(checkSetting func(s Setting) error) { +// greetAndCheckSettings is like greet, but the client sends the +// provided settings in its initial SETTINGS frame, and the +// checkSetting callback, if non-nil, is run for each setting in the +// server's initial SETTINGS frame. +func (st *serverTester) greetAndCheckSettings(settings []Setting, checkSetting func(s Setting) error) { st.t.Helper() st.writePreface() - st.writeSettings() + st.writeSettings(settings...) st.sync() - readFrame[*SettingsFrame](st.t, st).ForeachSetting(checkSetting) + if f := readFrame[*SettingsFrame](st.t, st); checkSetting != nil { + f.ForeachSetting(checkSetting) + } st.writeSettingsAck() // The initial WINDOW_UPDATE and SETTINGS ACK can come in any order. @@ -2809,7 +2815,7 @@ func testServer_MaxDecoderHeaderTableSize(t *testing.T) { defer st.Close() var advHeaderTableSize *uint32 - st.greetAndCheckSettings(func(s Setting) error { + st.greetAndCheckSettings(nil, func(s Setting) error { switch s.ID { case SettingHeaderTableSize: advHeaderTableSize = &s.Val @@ -2932,7 +2938,7 @@ func testServerDoS_MaxHeaderListSize(t *testing.T) { // shake hands frameSize := DefaultMaxReadFrameSize var advHeaderListSize *uint32 - st.greetAndCheckSettings(func(s Setting) error { + st.greetAndCheckSettings(nil, func(s Setting) error { switch s.ID { case SettingMaxFrameSize: if s.Val < MinMaxFrameSize { @@ -3864,6 +3870,61 @@ func testServerParkingDisabledByGODEBUG(t *testing.T) { st.wantParked(false) } +func TestServerParksWithFlowControlledData(t *testing.T) { + synctest.Test(t, testServerParksWithFlowControlledData) +} +func testServerParksWithFlowControlledData(t *testing.T) { + // A handler whose response exceeds the client's stream flow + // control window. The server sends as much DATA as the window + // allows and then parks, with the rest of the response queued in + // the write scheduler, blocked on flow control. The client's + // WINDOW_UPDATE revives the parked serve goroutine to write it. + const window = 100 + const responseSize = 250 + rest := responseSize - window + st := newServerTester(t, func(w http.ResponseWriter, r *http.Request) { + w.(http.Flusher).Flush() + io.WriteString(w, strings.Repeat("a", responseSize)) + }) + defer st.Close() + + st.greetAndCheckSettings( + []Setting{{SettingInitialWindowSize, window}}, + nil, + ) + st.bodylessReq1() + + // The server sends the response headers and the first window bytes + // of DATA, then the write scheduler blocks the rest on the + // stream's flow control window. + st.wantHeaders(wantHeader{ + streamID: 1, + endStream: false, + }) + st.wantData(wantData{ + streamID: 1, + endStream: false, + size: window, + }) + + // The serve goroutine parks even though its write scheduler still + // holds the blocked DATA, and it stays parked, because only an + // incoming WINDOW_UPDATE can unblock it. + st.wantParked(true) + st.advance(24 * time.Hour) + st.wantParked(true) + + // The client's WINDOW_UPDATE revives the serve goroutine, which + // writes the rest of the response and ends the stream. + st.writeWindowUpdate(1, uint32(rest)) + st.wantData(wantData{ + streamID: 1, + endStream: true, + data: []byte(strings.Repeat("a", rest)), + }) + st.wantParked(true) +} + // grpc-go closes the Request.Body currently with a Read. // Verify that it doesn't race. // See https://github.com/grpc/grpc-go/pull/938 @@ -5008,7 +5069,7 @@ func testServerSettingNoRFC7540Priorities(t *testing.T) { defer st.Close() var gotNoRFC7540Setting bool - st.greetAndCheckSettings(func(s Setting) error { + st.greetAndCheckSettings(nil, func(s Setting) error { if s.ID != SettingNoRFC7540Priorities { return nil }