From 1ee38488df73c22fc01bfd7c76c098facd874fac Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Mon, 28 Sep 2026 08:36:59 +0200 Subject: [PATCH 1/7] fix(epoll, iouring, std): graceful shutdown waits for the HTTP/2 streams on the shared worker pool, and std for its h2c streams (celeris#759) A stream on an async route (Route.Async, or a route AsyncHandlers has made async) runs its handler on the shared HTTP/2 worker pool, off the event loop, and its response comes back through the connection's write queue, which only the loop drains. At shutdown epoll and io_uring cancelled such streams (CloseH2) and closed their connections under the handlers: the client got unexpected EOF, and the OnShutdown hooks ran before the handlers had finished. On std, net/http hands an h2c connection over (hijack) and stops tracking it, so http.Server.Shutdown waited for no h2c stream: the hooks ran, and a direct Shutdown returned, while the handler was still running. epoll, io_uring (and so adaptive): the Processor counts its pool handlers (poolRunning: incremented before Submit, decremented after executeHandler has returned). Once its context is cancelled, a loop or worker sends every HTTP/2 connection GOAWAY(NO_ERROR, last client stream), so its client opens no new stream, and keeps turning, reading and writing as usual, until no HTTP/2 connection has a pool handler running or a response in its write queue, then shuts down as before (on io_uring the 250 ms send drain starts after). The wait ends at the deadline of the budget the last Engine.Shutdown handed over, never before 250 ms: io_uring's Engine.Shutdown now records its ctx, as epoll's does since celeris#760. std: Bridge counts the HTTP/2 requests in their handler, and the drain, after http.Server.Shutdown, waits for that count to reach zero, bounded by the drain's context (celeris#753). Tests: TestShutdownHooksRunAfterTheDrain gains the h2c cases on std and the h2c-async-route cases on every engine; TestShutdownSendsH2GoAwayThenFinishesStreams reads the frames of a raw h2c connection: GOAWAY first, then the held stream's response, then the close; TestShutdownH2PoolWaitIsBounded: a pool handler that does not return holds the shutdown for its budget, no longer. BenchmarkPoolDispatch and BenchmarkBridgeServeHTTP measure the two new counters. Stacked on #803 (celeris#753: std's drainCtx) and #807 (celeris#760: epoll's drain budget); merge after them. Fixes #759 --- engine/epoll/conn.go | 6 + engine/epoll/loop.go | 67 +++- engine/iouring/conn.go | 2 + engine/iouring/engine.go | 16 +- engine/iouring/worker.go | 105 ++++++- engine/std/bridge.go | 5 + engine/std/bridge_bench_test.go | 50 +++ engine/std/engine.go | 39 ++- internal/conn/h2.go | 24 ++ .../h2/stream/pool_dispatch_bench_test.go | 33 ++ protocol/h2/stream/processor.go | 21 ++ server.go | 24 +- shutdown_drain_order_linux_test.go | 39 ++- shutdown_h2_pool_linux_test.go | 296 ++++++++++++++++++ 14 files changed, 693 insertions(+), 34 deletions(-) create mode 100644 engine/std/bridge_bench_test.go create mode 100644 protocol/h2/stream/pool_dispatch_bench_test.go create mode 100644 shutdown_h2_pool_linux_test.go diff --git a/engine/epoll/conn.go b/engine/epoll/conn.go index cb11d0ac..574ee4df 100644 --- a/engine/epoll/conn.go +++ b/engine/epoll/conn.go @@ -273,6 +273,11 @@ type connState struct { // this set and does nothing (celeris#668). hijackSettled bool + // h2GoAwaySent records that a graceful shutdown has sent this HTTP/2 + // conn its GOAWAY (celeris#759; Loop.h2PoolSettled). Loop thread; reset + // on release. + h2GoAwaySent bool + // relinkOwed (guarded by asyncInMu) is set by the dirty pass or the // EPOLLOUT resume when they give the conn up because its dispatch // goroutine holds detachMu across a handler (celeris#669). The goroutine @@ -377,6 +382,7 @@ func releaseConnState(cs *connState) { cs.liveIdx = -1 cs.hijacked.Store(false) cs.hijackSettled = false + cs.h2GoAwaySent = false cs.closeOwed = false cs.closeErr = nil cs.relinkOwed = false diff --git a/engine/epoll/loop.go b/engine/epoll/loop.go index 0b89068e..f403b472 100644 --- a/engine/epoll/loop.go +++ b/engine/epoll/loop.go @@ -80,6 +80,10 @@ const shutdownSendDrainFloor = 250 * time.Millisecond // the drain notices a budget that ends early (a cancelled Shutdown ctx). const shutdownSendDrainPoll = 20 * time.Millisecond +// h2PoolDrainPollMs caps one epoll_wait while the loop waits out the HTTP/2 +// pool handlers at shutdown (celeris#759). +const h2PoolDrainPollMs = 10 + // Loop is an epoll-based event loop worker. type Loop struct { id int @@ -197,6 +201,11 @@ type Loop struct { // floor, shutdownSendDrainFloor. drainBudget *atomic.Pointer[context.Context] + // h2DrainStart is when the loop began waiting, its context cancelled, + // for the HTTP/2 stream handlers running on the shared worker pool + // (celeris#759; h2PoolSettled). Zero until then. Loop thread. + h2DrainStart time.Time + // transplantInFlight counts connections this loop has detached for a // transplant whose hand-off is not finished yet — the deferred async // path, where tryTransplant detaches and drainDetachQueue completes. @@ -473,7 +482,7 @@ func (l *Loop) run(ctx context.Context) { l.cachedNow = time.Now().UnixNano() for { - if ctx.Err() != nil { + if ctx.Err() != nil && l.h2PoolSettled() { l.shutdown() return } @@ -548,6 +557,12 @@ func (l *Loop) run(ctx context.Context) { if l.listenHot { timeoutMs = 0 } + // Waiting out HTTP/2 pool handlers at shutdown (celeris#759): a + // handler that ends without a write the loop hears of must not + // leave the loop blocked. + if !l.h2DrainStart.IsZero() && (timeoutMs < 0 || timeoutMs > h2PoolDrainPollMs) { + timeoutMs = h2PoolDrainPollMs + } n, err := unix.EpollWait(l.epollFD, l.events, timeoutMs) if err != nil { @@ -3670,6 +3685,56 @@ func (l *Loop) shutdown() { l.closeEpollFD() } +// h2PoolSettled reports whether the loop, its context cancelled, may shut +// down as far as HTTP/2 is concerned (celeris#759). A stream on an async +// route runs its handler on the shared worker pool, off this loop, and its +// response comes back through the conn's write queue, which only this loop +// drains; shutdown cancelled such streams and closed their conns under their +// handlers, so the client got unexpected EOF, and the hooks ran before the +// handlers had finished. So the loop keeps turning, reading and writing as +// usual, until no HTTP/2 conn has a pool handler running or a response still +// in its write queue. Every HTTP/2 conn is sent GOAWAY first (again for a +// conn that arrives meanwhile), so its client opens no new stream on it, as +// net/http's graceful shutdown does. The wait is bounded like the send drain +// (sendDrainWait): the budget of the last Engine.Shutdown, and never less +// than shutdownSendDrainFloor. Loop thread. +func (l *Loop) h2PoolSettled() bool { + if len(l.h2Conns) == 0 { + return true + } + if l.h2DrainStart.IsZero() { + l.h2DrainStart = time.Now() + } + busy := false + for _, fd := range l.h2Conns { + cs := l.conns[fd] + if cs == nil || cs.h2State == nil { + continue + } + if !cs.h2GoAwaySent { + mu := cs.detachMu + if mu != nil { + mu.Lock() + } + cs.h2GoAwaySent = cs.h2State.GoAway(cs.writeFn) + if mu != nil { + mu.Unlock() + } + if cs.h2GoAwaySent { + l.markDirty(cs) // the dirty pass flushes the GOAWAY + } + } + if cs.h2State.PoolHandlersRunning() || cs.h2State.WriteQueuePending() { + busy = true + } + } + if !busy { + return true + } + _, more := l.sendDrainWait(l.h2DrainStart) + return !more +} + // drainSends is shutdown's send drain (celeris#760). It flushes every live // conn with response bytes still queued, and waits for their sockets to take // more, until nothing is queued or the drain's time is up (sendDrainWait). A diff --git a/engine/iouring/conn.go b/engine/iouring/conn.go index 6f554bff..54d3d040 100644 --- a/engine/iouring/conn.go +++ b/engine/iouring/conn.go @@ -122,6 +122,7 @@ type connState struct { needsRecv bool // 1: recv arm was dropped (SQ ring full); retry on next opportunity recvIntoBody bool // 1: next recv CQE fills h1State.bodyBuf directly (skips ProcessH1 + cs.buf memcpy) zcNotifPending bool // 1: waiting for SEND_ZC notification CQE + h2GoAwaySent bool // 1: a graceful shutdown sent this H2 conn its GOAWAY (celeris#759) // sendIsZC records how the send SQE currently in flight for this // connection was ARMED: true for IORING_OP_SEND_ZC, false for a plain // SEND / WRITEV / linked SEND. It is the provenance flag the error @@ -560,6 +561,7 @@ func releaseConnState(cs *connState) { cs.needsRecv = false cs.recvIntoBody = false cs.zcNotifPending = false + cs.h2GoAwaySent = false cs.sendIsZC = false cs.zcSentBytes = 0 cs.lastActivity = 0 diff --git a/engine/iouring/engine.go b/engine/iouring/engine.go index 78b5e385..0918d325 100644 --- a/engine/iouring/engine.go +++ b/engine/iouring/engine.go @@ -109,6 +109,11 @@ type Engine struct { // and when the probe got no answer. Every worker gets a copy; the // hand-off's REAP needs them (celeris#657). asyncCancelFlags bool + + // drainBudget is the ctx of the last Shutdown call: the budget of the + // wait for HTTP/2 pool handlers the workers run when Listen's context is + // cancelled (celeris#759; Worker.h2PoolSettled). + drainBudget atomic.Pointer[context.Context] } // New creates a new io_uring engine. @@ -465,6 +470,7 @@ func (e *Engine) createWorkers(tier TierStrategy, cpus []int, w.sweepCnt = &e.metrics.sweep // celeris#657 P9 sweep witnesses w.asyncCancelFlags = e.asyncCancelFlags w.pause = &e.pause // celeris#662 pause linger + w.drainBudget = &e.drainBudget workers[i] = w } return workers, nil @@ -491,7 +497,7 @@ func fallbackTier(current TierStrategy) TierStrategy { } } -// Shutdown is a no-op for the io_uring engine — graceful shutdown is +// Shutdown does not stop the io_uring engine itself — graceful shutdown is // driven by context cancellation on Listen's parent context. Workers // exit their run loops on ctx.Done, drain the responses still queued for // the ring (Worker.hasPendingSends, celeris#595) and call Worker.shutdown, @@ -503,7 +509,13 @@ func fallbackTier(current TierStrategy) TierStrategy { // point owns one and Server.Shutdown cancels it after the graceful phase. // Handing Listen a context.Background() is what made Start hang here // (celeris#595), since this method cannot wake it. -func (e *Engine) Shutdown(_ context.Context) error { +// +// What Shutdown does is hand ctx to the workers as the budget of their wait +// for the HTTP/2 stream handlers still running on the shared worker pool +// when Listen's context is cancelled (celeris#759): Server.Shutdown calls it +// before it cancels that context. +func (e *Engine) Shutdown(ctx context.Context) error { + e.drainBudget.Store(&ctx) e.mu.Lock() defer e.mu.Unlock() return nil diff --git a/engine/iouring/worker.go b/engine/iouring/worker.go index a446788b..5f99052a 100644 --- a/engine/iouring/worker.go +++ b/engine/iouring/worker.go @@ -515,6 +515,14 @@ type Worker struct { // thread only, never read on the hot path. shutdownDrainDeadline int64 + // drainBudget points at the engine's record of the budget of the last + // Engine.Shutdown call, and h2DrainStart is when the worker, its + // context cancelled, began waiting for the HTTP/2 stream handlers on + // the shared worker pool (celeris#759; h2PoolSettled). Worker thread + // only; h2DrainStart is zero until then. + drainBudget *atomic.Pointer[context.Context] + h2DrainStart int64 + dirtyHead *connState // head of intrusive doubly-linked dirty list hasBufReturns bool // set when provided buffers need publishing sendsPending bool // true when SEND SQEs are in the SQ ring (guarantees CQE production) @@ -1168,12 +1176,22 @@ func (w *Worker) run(ctx context.Context) { // that window — it is the ordinary iteration — which is the // graceful side of the trade: a connection that arrives // inside it is answered rather than reset. - if w.shutdownDrainDeadline == 0 { - w.shutdownDrainDeadline = time.Now().UnixNano() + shutdownSendDrainNanos - } - if !w.hasPendingSends() || time.Now().UnixNano() > w.shutdownDrainDeadline { - w.shutdown() - return + // + // First, though, the HTTP/2 stream handlers on the shared + // worker pool (celeris#759): they run off this loop, and their + // responses come back through the conns' write queues, which + // only this loop drains. The send drain's clock starts once they + // are done. + if !w.h2PoolSettled() { + w.shutdownDrainDeadline = 0 + } else { + if w.shutdownDrainDeadline == 0 { + w.shutdownDrainDeadline = time.Now().UnixNano() + shutdownSendDrainNanos + } + if !w.hasPendingSends() || time.Now().UnixNano() > w.shutdownDrainDeadline { + w.shutdown() + return + } } } @@ -2048,6 +2066,12 @@ func (w *Worker) adaptiveTimeout() time.Duration { // smallest of the three: the celeris#657 sweep's next pass, and the // celeris#662 pause linger's deadline. d := w.sweptTimeout() + // Waiting out HTTP/2 pool handlers at shutdown (celeris#759): a + // handler that ends without a completion the ring hears of must not + // leave the worker waiting. + if w.h2DrainStart != 0 && d > h2PoolDrainPoll { + d = h2PoolDrainPoll + } if w.lingerUntil != 0 { return capToDeadline(d, w.lingerUntil) } @@ -6037,6 +6061,75 @@ func (w *Worker) checkTimeouts() { } } +// h2PoolDrainFloor is the least time the worker, its context cancelled, +// waits for the HTTP/2 stream handlers on the shared worker pool before it +// shuts down (celeris#759); the budget of the last Engine.Shutdown extends it +// to that budget's deadline. h2PoolDrainPoll caps one wait meanwhile. +const ( + h2PoolDrainFloor = 250 * time.Millisecond + h2PoolDrainPoll = 10 * time.Millisecond +) + +// h2PoolSettled reports whether the worker, its context cancelled, may go on +// to its send drain and shutdown as far as HTTP/2 is concerned (celeris#759). +// A stream on an async route runs its handler on the shared worker pool, off +// this loop, and its response comes back through the conn's write queue, +// which only this loop drains; shutdown cancelled such streams and closed +// their conns under their handlers, so the client got unexpected EOF, and +// the hooks ran before the handlers had finished. So the loop keeps turning, +// reading and writing as usual, until no HTTP/2 conn has a pool handler +// running or a response still in its write queue. Every HTTP/2 conn is sent +// GOAWAY first (again for a conn that arrives meanwhile), so its client opens +// no new stream on it, as net/http's graceful shutdown does. The wait ends at +// the deadline of the budget the last Engine.Shutdown handed over, and never +// before h2PoolDrainFloor. Worker thread. +func (w *Worker) h2PoolSettled() bool { + if len(w.h2Conns) == 0 { + return true + } + now := time.Now().UnixNano() + if w.h2DrainStart == 0 { + w.h2DrainStart = now + } + busy := false + for _, fd := range w.h2Conns { + cs := w.conns[fd] + if cs == nil || cs.h2State == nil { + continue + } + if !cs.h2GoAwaySent { + mu := cs.detachMu + if mu != nil { + mu.Lock() + } + cs.h2GoAwaySent = cs.h2State.GoAway(cs.writeFn) + if cs.h2GoAwaySent && w.flushSend(cs) { + w.markDirty(cs) + } + if mu != nil { + mu.Unlock() + } + } + if cs.h2State.PoolHandlersRunning() || cs.h2State.WriteQueuePending() { + busy = true + } + } + if !busy { + return true + } + end := w.h2DrainStart + int64(h2PoolDrainFloor) + if w.drainBudget != nil { + // A budget that is done, at its deadline or by a cancel before it, + // leaves the floor. + if p := w.drainBudget.Load(); p != nil && (*p).Err() == nil { + if d, has := (*p).Deadline(); has && d.UnixNano() > end { + end = d.UnixNano() + } + } + } + return now > end +} + // hasPendingSends reports whether any live connection still has response bytes // queued for the ring or a SEND in flight in the kernel. Called only from the // shutdown drain in run() (celeris#595), never on the hot path, so the diff --git a/engine/std/bridge.go b/engine/std/bridge.go index 4fc0c0b9..17390636 100644 --- a/engine/std/bridge.go +++ b/engine/std/bridge.go @@ -22,6 +22,11 @@ type Bridge struct { // ServeHTTP converts an http.Request to a stream.Stream, calls the handler, and writes the response. func (b *Bridge) ServeHTTP(w http.ResponseWriter, r *http.Request) { b.engine.metrics.reqCount.Add(1) + if r.ProtoMajor == 2 { + // An h2c stream: the drain waits for it (celeris#759). + b.engine.h2Streams.Add(1) + defer b.engine.h2Streams.Add(-1) + } s := stream.NewH1Stream(1) defer s.Release() diff --git a/engine/std/bridge_bench_test.go b/engine/std/bridge_bench_test.go new file mode 100644 index 00000000..f53b2bd0 --- /dev/null +++ b/engine/std/bridge_bench_test.go @@ -0,0 +1,50 @@ +package std + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + "github.com/goceleris/celeris/engine" + "github.com/goceleris/celeris/protocol/h2/stream" + "github.com/goceleris/celeris/resource" +) + +type okStreamHandler struct{} + +func (okStreamHandler) HandleStream(_ context.Context, s *stream.Stream) error { + return s.ResponseWriter.WriteResponse(s, 200, [][2]string{{"content-type", "text/plain"}}, []byte("ok")) +} + +type discardResponseWriter struct{ h http.Header } + +func (w *discardResponseWriter) Header() http.Header { return w.h } +func (w *discardResponseWriter) Write(p []byte) (int, error) { return len(p), nil } +func (w *discardResponseWriter) WriteHeader(int) {} + +// BenchmarkBridgeServeHTTP measures one request through the std bridge, over +// HTTP/1.1 and over HTTP/2 (an h2c stream), whose in-handler count +// celeris#759 adds (an atomic add and a deferred one, HTTP/2 only). +func BenchmarkBridgeServeHTTP(b *testing.B) { + for _, tc := range []struct { + name string + major int + }{{"HTTP1", 1}, {"HTTP2", 2}} { + b.Run(tc.name, func(b *testing.B) { + e, err := New(resource.Config{Addr: "127.0.0.1:0", Engine: engine.Std, Protocol: engine.HTTP1}, okStreamHandler{}) + if err != nil { + b.Fatal(err) + } + br := &Bridge{engine: e, handler: okStreamHandler{}} + req := httptest.NewRequest(http.MethodGet, "/", nil) + req.ProtoMajor = tc.major + w := &discardResponseWriter{h: http.Header{}} + b.ReportAllocs() + for b.Loop() { + clear(w.h) + br.ServeHTTP(w, req) + } + }) + } +} diff --git a/engine/std/engine.go b/engine/std/engine.go index 351622f8..35e35114 100644 --- a/engine/std/engine.go +++ b/engine/std/engine.go @@ -51,7 +51,12 @@ type Engine struct { drainCtx context.Context drainCancel context.CancelFunc drainErr error - metrics struct { + // h2Streams counts the HTTP/2 (h2c) requests in their handler + // (Bridge.ServeHTTP). net/http hands an h2c connection over (hijack) + // and stops tracking it, so http.Server.Shutdown does not wait for + // its streams; the drain waits for this count instead (celeris#759). + h2Streams atomic.Int64 + metrics struct { reqCount atomic.Uint64 activeConns atomic.Int64 // errs is the per-cause ErrorCount breakdown (celeris#645). @@ -211,11 +216,41 @@ func (e *Engine) Listen(ctx context.Context) error { // once.Do. func (e *Engine) drain() error { e.once.Do(func() { - e.drainErr = e.server.Shutdown(e.drainCtx) + err := e.server.Shutdown(e.drainCtx) + if err == nil { + err = e.waitH2Streams(e.drainCtx) + } + e.drainErr = err }) return e.drainErr } +// h2StreamsPoll is how often the drain looks at h2Streams while it waits. +const h2StreamsPoll = 5 * time.Millisecond + +// waitH2Streams waits, bounded by ctx, until no HTTP/2 (h2c) request is in +// its handler (celeris#759). http.Server.Shutdown does not wait for them: +// net/http serves h2c on a connection it has handed over and no longer +// tracks, so the OnShutdown hooks ran, and a direct Shutdown returned, while +// an h2c handler was still running. The connection itself is left as +// http.Server.Shutdown leaves a hijacked one: its handlers are what the drain +// waits for. +func (e *Engine) waitH2Streams(ctx context.Context) error { + if e.h2Streams.Load() <= 0 { + return nil + } + t := time.NewTicker(h2StreamsPoll) + defer t.Stop() + for e.h2Streams.Load() > 0 { + select { + case <-ctx.Done(): + return ctx.Err() + case <-t.C: + } + } + return nil +} + // Shutdown gracefully shuts down the server: in-flight requests drain // until ctx expires, and whatever is still running once that budget is // spent is woken through its request context. diff --git a/internal/conn/h2.go b/internal/conn/h2.go index 517ec143..0158d8de 100644 --- a/internal/conn/h2.go +++ b/internal/conn/h2.go @@ -774,6 +774,30 @@ func ProcessH2(ctx context.Context, data []byte, state *H2State, _ stream.Handle return nil } +// PoolHandlersRunning reports whether a stream of this connection has a +// handler running on the shared HTTP/2 worker pool (an async route), which +// runs off the event loop (celeris#759). Safe from any goroutine. +func (s *H2State) PoolHandlersRunning() bool { + return s.processor.PoolHandlersRunning() +} + +// GoAway starts a graceful close of the connection (celeris#759): it sends +// GOAWAY(NO_ERROR) naming the last stream the client has opened, so the +// client opens no new stream on the connection, and leaves the streams in +// flight to finish. It reports false, sending nothing, while the server's +// preface has not gone out yet. Called on the engine's event loop thread, +// like ProcessH2. +func (s *H2State) GoAway(write func([]byte)) bool { + s.mu.Lock() + defer s.mu.Unlock() + if !s.serverPrefaceSent { + return false + } + _ = s.processor.SendGoAway(s.processor.GetManager().GetLastClientStreamID(), http2.ErrCodeNo, nil) + flushOutBuf(&s.outBuf, write) + return true +} + // CloseH2 cleans up H2 state. Releases all streams still held by the // manager to prevent memory leaks on connection close. func CloseH2(state *H2State) { diff --git a/protocol/h2/stream/pool_dispatch_bench_test.go b/protocol/h2/stream/pool_dispatch_bench_test.go new file mode 100644 index 00000000..a5f1e50a --- /dev/null +++ b/protocol/h2/stream/pool_dispatch_bench_test.go @@ -0,0 +1,33 @@ +package stream + +import ( + "context" + "testing" +) + +// doneHandler signals every stream it has handled; the benchmark waits for +// it so each iteration is one full dispatch round trip. +type doneHandler struct{ done chan struct{} } + +func (h *doneHandler) HandleStream(_ context.Context, _ *Stream) error { + h.done <- struct{}{} + return nil +} + +// BenchmarkPoolDispatch measures one stream handed to the shared worker pool +// (runHandler → Submit → executeHandler), the path whose per-connection +// handler count celeris#759 adds (an atomic add before Submit and one after +// executeHandler returns). The stream is not END_STREAM, so it is not +// eligible to run inline. +func BenchmarkPoolDispatch(b *testing.B) { + h := &doneHandler{done: make(chan struct{}, 1)} + proc := NewProcessor(h, newTestFrameWriter(), newTestResponseWriter()) + id := uint32(1) + b.ReportAllocs() + for b.Loop() { + s := proc.manager.CreateStream(id) + id += 2 + proc.runHandler(s) + <-h.done + } +} diff --git a/protocol/h2/stream/processor.go b/protocol/h2/stream/processor.go index 28d4ee9a..749ff875 100644 --- a/protocol/h2/stream/processor.go +++ b/protocol/h2/stream/processor.go @@ -169,6 +169,22 @@ type Processor struct { // processing is single-threaded per connection (event loop under // H2State.mu), so a per-processor scratch slice is safe. connFlushScratch []*Stream + + // poolRunning counts this connection's streams whose handler has been + // handed to the shared worker pool and has not returned yet + // (celeris#759). The native engines' graceful shutdown waits for it to + // reach zero before it closes the connection: those handlers run off the + // event loop, and the drain that waits for the loop's own work did not + // see them. Incremented in runHandler before Submit, decremented after + // executeHandler has returned. + poolRunning atomic.Int32 +} + +// PoolHandlersRunning reports whether a stream of this connection has a +// handler running on the shared worker pool (celeris#759). Safe from any +// goroutine. +func (p *Processor) PoolHandlersRunning() bool { + return p.poolRunning.Load() > 0 } // rstRateLimit and rstBurstLimit bound RST_STREAM arrivals. An honest @@ -609,6 +625,7 @@ func (p *Processor) runHandler(stream *Stream) { } stream.flags.Or(flagAsyncRunning) + p.poolRunning.Add(1) globalH2Pool.Submit(p, stream) } @@ -732,6 +749,10 @@ func (p *Processor) executeHandlerInline(stream *Stream) { // releases it back to the pool. If outbound data is buffered (flow control), // the stream stays in the map for the event loop to flush via WINDOW_UPDATE. func (p *Processor) executeHandler(stream *Stream) { + // Deferred first, so it runs last: the response is queued and the + // stream settled before a shutdown that waits for this count can close + // the connection (celeris#759). + defer p.poolRunning.Add(-1) defer func() { if r := recover(); r != nil { _ = r // last-resort panic recovery diff --git a/server.go b/server.go index e5fa22d6..4c6c8c78 100644 --- a/server.go +++ b/server.go @@ -523,18 +523,18 @@ func (s *Server) cancelListen() { // call it from another goroutine (celeris#703). The Start* call that the // server was started with returns only after Shutdown has returned. // -// The drain waits for every HTTP/1.1 request, and on epoll, io_uring and -// adaptive for every HTTP/2 stream whose handler runs on the connection's -// worker. It does not wait for an HTTP/2 stream on an async route (marked -// Async, or promoted to async under [Config.AsyncHandlers]), which runs on -// the shared HTTP/2 worker pool, nor on std for any h2c stream: the hooks can -// run while such a handler is still running, and on the native engines its -// response is lost (celeris#759). Once the handlers have returned, the native -// engines send what the sockets have not taken yet before they close the -// connections: epoll (and adaptive while it runs epoll) while ctx is live, -// until its deadline or, for a ctx without one such as context.Background(), -// until it is done, but no longer than [Config.WriteTimeout], and never for -// less than 250 ms (celeris#760); io_uring for 250 ms (celeris#806). +// The drain waits for every HTTP/1.1 request and every HTTP/2 stream. An +// HTTP/2 stream on an async route (marked Async, or promoted to async under +// [Config.AsyncHandlers]) runs on the shared HTTP/2 worker pool: epoll, +// io_uring and adaptive send each HTTP/2 connection GOAWAY and serve it until +// those handlers have returned and their responses have gone out, while ctx +// is live (until its deadline or, for a ctx without one, until it is done, +// but no longer than [Config.WriteTimeout]) and never for less than 250 ms; +// std waits for its h2c streams' handlers, bounded by ctx (celeris#759). Once +// the handlers have returned, the native engines send what the sockets have +// not taken yet before they close the connections: epoll (and adaptive while +// it runs epoll) for as long, and never for less than 250 ms (celeris#760); +// io_uring for 250 ms (celeris#806). // // The listen context published by the Start* entry points is cancelled AFTER // the engine's graceful phase, never before: on std, Engine.Shutdown IS the diff --git a/shutdown_drain_order_linux_test.go b/shutdown_drain_order_linux_test.go index 5c304434..7444d503 100644 --- a/shutdown_drain_order_linux_test.go +++ b/shutdown_drain_order_linux_test.go @@ -33,10 +33,14 @@ import ( // server, did the Start call return only after that Shutdown's hook had // returned (a main that exits when Start returns must not lose its hooks). // -// The h2c cases send the request over HTTP/2 (prior knowledge) to the native -// engines, on a route that is not async, so its handler runs on the -// connection's worker. std's h2c streams and async-route HTTP/2 streams run -// outside the drain and are not asserted here (celeris#759). +// The h2c cases send the request over HTTP/2 (prior knowledge). On a route +// that is not async the native engines run its handler on the connection's +// worker; on an async route (the h2c-async-route cases) they run it on the +// shared HTTP/2 worker pool, which the drain did not wait for: the shutdown +// closed the connection under the handler and the client got unexpected EOF. +// std serves h2c on a connection net/http has handed over (hijacked) and no +// longer tracks, so its drain did not wait for any h2c stream: the hooks ran +// with the handler still running (celeris#759). // // The order is forced, not raced. The handler returns only when the test // releases it, and the test releases it as soon as a hook starts or the @@ -75,17 +79,24 @@ func TestShutdownHooksRunAfterTheDrain(t *testing.T) { for _, ec := range engines { for _, mode := range modes { t.Run(ec.name+"/"+mode.String(), func(t *testing.T) { - runDrainOrderCase(t, ec.eng, ec.async, false, mode, releaseAfter) + runDrainOrderCase(t, ec.eng, ec.async, false, false, mode, releaseAfter) }) } } for _, ec := range engines { - if ec.eng == celeris.Std { - continue // celeris#759: std's h2c streams are not drained - } for _, mode := range modes { t.Run("h2c-"+ec.name+"/"+mode.String(), func(t *testing.T) { - runDrainOrderCase(t, ec.eng, ec.async, true, mode, releaseAfter) + runDrainOrderCase(t, ec.eng, ec.async, true, false, mode, releaseAfter) + }) + } + } + for _, ec := range engines { + if ec.async { + continue // an async route is async whatever AsyncHandlers says + } + for _, mode := range modes { + t.Run("h2c-async-route-"+ec.name+"/"+mode.String(), func(t *testing.T) { + runDrainOrderCase(t, ec.eng, ec.async, true, true, mode, releaseAfter) }) } } @@ -160,7 +171,7 @@ func (m drainOrderMode) String() string { return "unknown" } -func runDrainOrderCase(t *testing.T, engType celeris.EngineType, async, h2c bool, mode drainOrderMode, releaseAfter time.Duration) { +func runDrainOrderCase(t *testing.T, engType celeris.EngineType, async, h2c, asyncRoute bool, mode drainOrderMode, releaseAfter time.Duration) { t.Helper() // One sequence for every event, so the order is read from numbers, not // from clocks. Zero means "has not happened". @@ -194,13 +205,16 @@ func runDrainOrderCase(t *testing.T, engType celeris.EngineType, async, h2c bool Addr: addr, }) s.GET("/ping", func(c *celeris.Context) error { return c.String(http.StatusOK, "ok") }) - s.GET("/slow", func(c *celeris.Context) error { + slow := s.GET("/slow", func(c *celeris.Context) error { close(handlerEntered) <-release handlerDoneAt.Store(int64(since())) handlerDone.Store(seq.Add(1)) return c.String(http.StatusOK, "done") }) + if asyncRoute { + slow.Async() + } s.OnShutdown(func(context.Context) { hookSawHandlerDone.Store(handlerDone.Load() != 0) hookStartAt.Store(int64(since())) @@ -372,6 +386,9 @@ func runDrainOrderCase(t *testing.T, engType celeris.EngineType, async, h2c bool if h2c { proto = "h2c" } + if asyncRoute { + proto += "-async-route" + } t.Logf("RESULT engine=%s async=%v proto=%s mode=%s handler_done_ms=%.1f hook_start_ms=%.1f call_return_ms=%.1f order(handler,hook,call)=(%d,%d,%d) hook_end=%d start_return=%d response=%d/%q err=%v", engType, async, proto, mode, ms(handlerDoneAt.Load()), ms(hookStartAt.Load()), ms(callReturnAt.Load()), handlerDone.Load(), hookStart.Load(), callReturn.Load(), hookEnd.Load(), startReturn.Load(), r.status, r.body, r.err) diff --git a/shutdown_h2_pool_linux_test.go b/shutdown_h2_pool_linux_test.go new file mode 100644 index 00000000..4822b1a0 --- /dev/null +++ b/shutdown_h2_pool_linux_test.go @@ -0,0 +1,296 @@ +//go:build linux + +package celeris_test + +import ( + "bytes" + "context" + "errors" + "io" + "net" + "net/http" + "strings" + "testing" + "time" + + "golang.org/x/net/http2" + "golang.org/x/net/http2/hpack" + + "github.com/goceleris/celeris" +) + +// TestShutdownSendsH2GoAwayThenFinishesStreams pins the HTTP/2 half of the +// native engines' graceful shutdown (celeris#759), on the wire: with a stream +// on an async route held in its handler, which runs on the shared HTTP/2 +// worker pool, the shutdown must first send GOAWAY (NO_ERROR, naming the +// held stream or a later one), so the client opens no new stream, and then +// let the held stream finish: its response arrives after the GOAWAY, and only +// then does the connection close. Before, the shutdown cancelled the stream +// and closed the connection under the handler, with no GOAWAY. +// +// The order is forced: the handler is released only once the client has read +// the GOAWAY, or after releaseAfter if none comes, so a shutdown that closes +// at once, or sends no GOAWAY, fails every time. +func TestShutdownSendsH2GoAwayThenFinishesStreams(t *testing.T) { + const releaseAfter = 2 * time.Second + for _, e := range []struct { + name string + eng celeris.EngineType + }{{"epoll", celeris.Epoll}, {"io_uring", celeris.IOUring}, {"adaptive", celeris.Adaptive}} { + for _, mode := range []string{"Shutdown", "cancel"} { + t.Run(e.name+"/"+mode, func(t *testing.T) { + entered := make(chan struct{}) + release := make(chan struct{}) + srv := startH2PoolServer759(t, e.eng, 30*time.Second, func(s *celeris.Server) { + s.GET("/held", func(c *celeris.Context) error { + close(entered) + <-release + return c.String(http.StatusOK, "done") + }).Async() + }) + c, fr := dialH2759(t, srv.addr) + writeH2Get759(t, fr, 1, "/held") + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("the held handler did not start within 5s") + } + srv.beginShutdown(mode, 30*time.Second) + + // Read until GOAWAY, releasing the handler only then. + released := false + releaseTimer := time.AfterFunc(releaseAfter, func() { close(release) }) + var sawGoAway bool + var goAwayCode http2.ErrCode + var goAwayLast uint32 + var status string + var body bytes.Buffer + var events []string + dec := hpack.NewDecoder(4096, nil) + for { + _ = c.SetReadDeadline(time.Now().Add(10 * time.Second)) + f, err := fr.ReadFrame() + if err != nil { + events = append(events, "read: "+describeH2ReadEnd759(err)) + break + } + switch f := f.(type) { + case *http2.GoAwayFrame: + sawGoAway, goAwayCode, goAwayLast = true, f.ErrCode, f.LastStreamID + events = append(events, "GOAWAY") + if !released && releaseTimer.Stop() { + released = true + close(release) + } + case *http2.HeadersFrame: + if f.StreamID == 1 { + hf, _ := dec.DecodeFull(f.HeaderBlockFragment()) + for _, h := range hf { + if h.Name == ":status" { + status = h.Value + } + } + events = append(events, "HEADERS") + } + case *http2.DataFrame: + if f.StreamID == 1 { + body.Write(f.Data()) + if f.StreamEnded() { + events = append(events, "DATA(end)") + } + } + case *http2.RSTStreamFrame: + events = append(events, "RST_STREAM "+f.ErrCode.String()) + case *http2.SettingsFrame: + if !f.IsAck() { + _ = fr.WriteSettingsAck() + } + case *http2.PingFrame: + if !f.IsAck() { + _ = fr.WritePing(true, f.Data) + } + } + } + if !released && releaseTimer.Stop() { + close(release) // no GOAWAY came: let the handler go + } + srv.waitStart(t, 20*time.Second) + trace := strings.Join(events, ", ") + if !sawGoAway || goAwayCode != http2.ErrCodeNo || goAwayLast < 1 { + t.Fatalf("%s/%s: GOAWAY seen=%v code=%v last=%d, want NO_ERROR naming stream 1 or later; frames: %s", e.name, mode, sawGoAway, goAwayCode, goAwayLast, trace) + } + if status != "200" || body.String() != "done" { + t.Fatalf("%s/%s: stream 1 got status %q body %q; frames: %s", e.name, mode, status, body.String(), trace) + } + if i, j := strings.Index(trace, "GOAWAY"), strings.Index(trace, "HEADERS"); i < 0 || j < i { + t.Fatalf("%s/%s: the response came before the GOAWAY; frames: %s", e.name, mode, trace) + } + }) + } + } +} + +// TestShutdownH2PoolWaitIsBounded: a handler on the HTTP/2 worker pool that +// does not return must not hold the native engines' shutdown past its budget +// (celeris#759). The wait for pool handlers ends at the budget's deadline, +// and the Start call returns within bound of it, with the handler still +// running. +func TestShutdownH2PoolWaitIsBounded(t *testing.T) { + const budget = 500 * time.Millisecond + const bound = 3 * time.Second + for _, e := range []struct { + name string + eng celeris.EngineType + }{{"epoll", celeris.Epoll}, {"io_uring", celeris.IOUring}, {"adaptive", celeris.Adaptive}} { + for _, mode := range []string{"Shutdown", "cancel"} { + t.Run(e.name+"/"+mode, func(t *testing.T) { + entered := make(chan struct{}) + release := make(chan struct{}) + defer close(release) + srv := startH2PoolServer759(t, e.eng, budget, func(s *celeris.Server) { + s.GET("/stuck", func(c *celeris.Context) error { + close(entered) + <-release + return c.String(http.StatusOK, "late") + }).Async() + }) + _, fr := dialH2759(t, srv.addr) + writeH2Get759(t, fr, 1, "/stuck") + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("the stuck handler did not start within 5s") + } + start := time.Now() + srv.beginShutdown(mode, budget) + srv.waitStart(t, budget+bound) + t.Logf("%s/%s: Start returned %v after the shutdown began, with the handler still running", e.name, mode, time.Since(start).Round(time.Millisecond)) + }) + } + } +} + +type h2PoolServer759 struct { + s *celeris.Server + addr string + cancel context.CancelFunc + startDone chan error + shutErr chan error +} + +func startH2PoolServer759(t *testing.T, eng celeris.EngineType, budget time.Duration, routes func(*celeris.Server)) *h2PoolServer759 { + t.Helper() + retryUntil := time.Now().Add(30 * time.Second) + for tries := 1; ; tries++ { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + addr := ln.Addr().String() + _ = ln.Close() + s := celeris.New(celeris.Config{Engine: eng, Addr: addr, ShutdownTimeout: budget}) + s.GET("/ping", func(c *celeris.Context) error { return c.String(http.StatusOK, "ok") }) + routes(s) + ctx, cancel := context.WithCancel(context.Background()) + srv := &h2PoolServer759{s: s, addr: addr, cancel: cancel, startDone: make(chan error, 1)} + go func() { srv.startDone <- s.StartWithContext(ctx) }() + err = waitDrainOrderReady(addr, srv.startDone) + if err == nil { + if tries > 1 { + t.Logf("server start retried on ring ENOMEM: %d tries", tries) + } + t.Cleanup(func() { + cancel() + select { + case err := <-srv.startDone: + srv.startDone <- err + case <-time.After(30 * time.Second): + t.Errorf("StartWithContext did not return within 30s of the cleanup cancel") + } + }) + return srv + } + cancel() + if strings.Contains(err.Error(), "cannot allocate memory") && time.Now().Before(retryUntil) { + time.Sleep(20 * time.Millisecond) + continue + } + t.Fatalf("server did not start: %v", err) + } +} + +func (srv *h2PoolServer759) beginShutdown(mode string, budget time.Duration) { + if mode == "cancel" { + srv.cancel() + return + } + srv.shutErr = make(chan error, 1) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), budget) + defer cancel() + srv.shutErr <- srv.s.Shutdown(ctx) + }() +} + +func (srv *h2PoolServer759) waitStart(t *testing.T, limit time.Duration) { + t.Helper() + deadline := time.After(limit) + if srv.shutErr != nil { + select { + case <-srv.shutErr: + case <-deadline: + t.Fatalf("Shutdown had not returned %v after it began", limit) + } + } + select { + case err := <-srv.startDone: + srv.startDone <- err // for the cleanup + case <-deadline: + t.Fatalf("StartWithContext had not returned %v after the shutdown began", limit) + } +} + +// dialH2759 opens a raw h2c (prior knowledge) connection: the client preface +// and an empty SETTINGS frame. +func dialH2759(t *testing.T, addr string) (net.Conn, *http2.Framer) { + t.Helper() + c, err := net.Dial("tcp", addr) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = c.Close() }) + if _, err := io.WriteString(c, http2.ClientPreface); err != nil { + t.Fatal(err) + } + fr := http2.NewFramer(c, c) + if err := fr.WriteSettings(); err != nil { + t.Fatal(err) + } + return c, fr +} + +func writeH2Get759(t *testing.T, fr *http2.Framer, id uint32, path string) { + t.Helper() + var hb bytes.Buffer + enc := hpack.NewEncoder(&hb) + for _, f := range []hpack.HeaderField{ + {Name: ":method", Value: "GET"}, {Name: ":scheme", Value: "http"}, + {Name: ":authority", Value: "x"}, {Name: ":path", Value: path}, + } { + _ = enc.WriteField(f) + } + if err := fr.WriteHeaders(http2.HeadersFrameParam{StreamID: id, BlockFragment: hb.Bytes(), EndStream: true, EndHeaders: true}); err != nil { + t.Fatal(err) + } +} + +func describeH2ReadEnd759(err error) string { + var ne net.Error + switch { + case errors.As(err, &ne) && ne.Timeout(): + return "no frame for 10s, connection still open" + case errors.Is(err, io.EOF), errors.Is(err, io.ErrUnexpectedEOF): + return "EOF" + } + return err.Error() +} From 071c08d157675359fa6d6f4ba460aa9445150b21 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Mon, 28 Sep 2026 11:44:36 +0200 Subject: [PATCH 2/7] fix(epoll, iouring): the HTTP/2 shutdown wait accepts no new connection, refuses streams above the GOAWAY, and waits for DATA held by flow control (celeris#759) Round 2 of #808, from its review. - The loops and workers kept their listeners while they waited for HTTP/2 pool handlers, and accepted and served new connections for up to the whole budget, then cut them at its end. They close the listener (io_uring cancels its accept first) when their context is cancelled, as net/http's Shutdown does, and never re-create it after. - The GOAWAY did not enforce its last-stream-id: a stream the client opened above it, which the client counts as not processed and may retry elsewhere, was served. The processor records the GOAWAY and refuses such a stream with RST_STREAM(REFUSED_STREAM) before its handler runs, having decoded its headers so the HPACK state stays in step. - A response still waiting for the client's WINDOW_UPDATE when the handler had returned looked settled, and was cut. The wait now includes streams with outbound DATA buffered (Processor.OutboundPending), on async and sync routes. - io_uring's wait gets the bound epoll's has since #807's round 2: a ctx without a deadline keeps it going until it is done, no longer than WriteTimeout. - engine.Engine.Shutdown's doc and Server.Shutdown's say what the engines do now. - Tests: no new connection is served 300 ms into a shutdown (std, epoll, io_uring, adaptive); streams opened after the GOAWAY are refused and not served; a 1 MiB response behind default 65535-byte windows arrives whole; a Shutdown(context.Background()) waits for a pool handler past the floor. --- engine/engine.go | 15 +- engine/epoll/loop.go | 51 ++++-- engine/iouring/worker.go | 97 +++++++--- internal/conn/h2.go | 16 +- protocol/h2/stream/processor.go | 46 ++++- server.go | 18 +- shutdown_h2_pool_linux_test.go | 310 +++++++++++++++++++++++++++++++- 7 files changed, 494 insertions(+), 59 deletions(-) diff --git a/engine/engine.go b/engine/engine.go index 4afa9380..3599f075 100644 --- a/engine/engine.go +++ b/engine/engine.go @@ -15,11 +15,16 @@ type Engine interface { // Listen starts the engine and blocks until ctx is canceled or a fatal // error occurs. The engine begins accepting connections on the configured address. Listen(ctx context.Context) error - // Shutdown gracefully drains in-flight connections, bounded by ctx: when - // ctx expires, Shutdown returns its error. On epoll and io_uring the - // drain runs in Listen once Listen's ctx is cancelled, and Shutdown itself - // does nothing. No engine closes a connection whose handler is still - // running when ctx expires; the handler runs to completion (celeris#753). + // Shutdown gracefully drains in-flight connections, bounded by ctx. On + // std Shutdown is the drain: when ctx expires first, it returns ctx's + // error. On epoll and io_uring the drain runs in Listen once Listen's + // ctx is cancelled, and Shutdown hands it ctx as its budget and returns + // nil at once (celeris#759, celeris#760); adaptive hands ctx to its + // sub-engines, then cancels its own Listen and waits for it, bounded by + // ctx. An HTTP/1 handler runs to completion on every engine whatever + // ctx (celeris#753); a handler of an HTTP/2 stream on the shared worker + // pool can still be running when the native engines close its + // connection at the end of the budget. Shutdown(ctx context.Context) error // Metrics returns a point-in-time snapshot of engine performance counters. Metrics() EngineMetrics diff --git a/engine/epoll/loop.go b/engine/epoll/loop.go index f403b472..f1ad6abf 100644 --- a/engine/epoll/loop.go +++ b/engine/epoll/loop.go @@ -482,9 +482,15 @@ func (l *Loop) run(ctx context.Context) { l.cachedNow = time.Now().UnixNano() for { - if ctx.Err() != nil && l.h2PoolSettled() { - l.shutdown() - return + if ctx.Err() != nil { + // Accept nothing more (celeris#759): the loop may keep turning + // below, for the HTTP/2 conns it has, and a conn accepted now + // would be served and then cut at the budget. + l.stopAccepting() + if l.h2PoolSettled() { + l.shutdown() + return + } } // Cache the atomic load: ACTIVE→LINGERING→DRAINING and @@ -507,8 +513,9 @@ func (l *Loop) run(ctx context.Context) { // false so a subsequent Pause observes a fresh signal. l.listenFDClosed.Store(paused && l.listenFD < 0) - // SUSPENDED → ACTIVE: re-create listen socket after ResumeAccept. - if l.listenFD < 0 && !paused { + // SUSPENDED → ACTIVE: re-create listen socket after ResumeAccept; + // never once shutdown has begun (stopAccepting). + if l.listenFD < 0 && !paused && ctx.Err() == nil { fd, err := createListenSocket(l.cfg.Addr, !l.cfg.DisableDeferAccept) l.deferCapable = !l.cfg.DisableDeferAccept if err != nil { @@ -3692,12 +3699,14 @@ func (l *Loop) shutdown() { // drains; shutdown cancelled such streams and closed their conns under their // handlers, so the client got unexpected EOF, and the hooks ran before the // handlers had finished. So the loop keeps turning, reading and writing as -// usual, until no HTTP/2 conn has a pool handler running or a response still -// in its write queue. Every HTTP/2 conn is sent GOAWAY first (again for a -// conn that arrives meanwhile), so its client opens no new stream on it, as -// net/http's graceful shutdown does. The wait is bounded like the send drain -// (sendDrainWait): the budget of the last Engine.Shutdown, and never less -// than shutdownSendDrainFloor. Loop thread. +// usual on the conns it has (it accepts no new one: stopAccepting), until no +// HTTP/2 conn has a pool handler running, a response still in its write +// queue, or response DATA waiting for the client's WINDOW_UPDATE. Every +// HTTP/2 conn is sent GOAWAY first, so its client opens no new stream on it, +// as net/http's graceful shutdown does, and a stream it opens anyway is +// refused. The wait is bounded like the send drain (sendDrainWait): the +// budget of the last Engine.Shutdown, and never less than +// shutdownSendDrainFloor. Loop thread. func (l *Loop) h2PoolSettled() bool { if len(l.h2Conns) == 0 { return true @@ -3724,7 +3733,7 @@ func (l *Loop) h2PoolSettled() bool { l.markDirty(cs) // the dirty pass flushes the GOAWAY } } - if cs.h2State.PoolHandlersRunning() || cs.h2State.WriteQueuePending() { + if cs.h2State.PoolHandlersRunning() || cs.h2State.WriteQueuePending() || cs.h2State.OutboundPending() { busy = true } } @@ -3735,6 +3744,24 @@ func (l *Loop) h2PoolSettled() bool { return !more } +// stopAccepting takes the listener out of the epoll set and closes it, once +// the loop's context is cancelled (celeris#759). The loop may go on turning +// after that, for as long as its HTTP/2 conns keep it (h2PoolSettled), and it +// used to accept and serve new connections meanwhile, which were then cut at +// the budget; net/http's Shutdown closes its listeners first. What is still +// in the kernel's accept queue is reset, as the close in shutdown did. +// Idempotent. Loop thread. +func (l *Loop) stopAccepting() { + if l.listenFD < 0 { + return + } + _ = unix.EpollCtl(l.epollFD, unix.EPOLL_CTL_DEL, l.listenFD, nil) + _ = unix.Close(l.listenFD) + l.listenFD = -1 + l.listenHot = false + l.lingerUntil = 0 +} + // drainSends is shutdown's send drain (celeris#760). It flushes every live // conn with response bytes still queued, and waits for their sockets to take // more, until nothing is queued or the drain's time is up (sendDrainWait). A diff --git a/engine/iouring/worker.go b/engine/iouring/worker.go index 5f99052a..8c612fbb 100644 --- a/engine/iouring/worker.go +++ b/engine/iouring/worker.go @@ -1172,16 +1172,18 @@ func (w *Worker) run(ctx context.Context) { // SQEs and reaps their completions through the normal path — // until no send is queued or in flight, bounded by // shutdownSendDrainNanos so a peer that stopped reading - // cannot hold shutdown open. The loop keeps accepting for - // that window — it is the ordinary iteration — which is the - // graceful side of the trade: a connection that arrives - // inside it is answered rather than reset. + // cannot hold shutdown open. It accepts nothing meanwhile + // (celeris#759): a connection accepted now would be served + // and then cut, and the wait for HTTP/2 below can last the + // whole budget; net/http's Shutdown closes its listeners + // first. // // First, though, the HTTP/2 stream handlers on the shared // worker pool (celeris#759): they run off this loop, and their // responses come back through the conns' write queues, which // only this loop drains. The send drain's clock starts once they // are done. + w.stopAccepting(ctx) if !w.h2PoolSettled() { w.shutdownDrainDeadline = 0 } else { @@ -1218,8 +1220,9 @@ func (w *Worker) run(ctx context.Context) { // false so a subsequent Pause observes a fresh signal. w.listenFDClosed.Store(paused && w.listenFD < 0) - // SUSPENDED → ACTIVE: re-create listen socket after ResumeAccept. - if w.listenFD < 0 && !paused { + // SUSPENDED → ACTIVE: re-create listen socket after ResumeAccept; + // never once shutdown has begun (stopAccepting). + if w.listenFD < 0 && !paused && ctx.Err() == nil { fd, err := createListenSocket(w.cfg.Addr, !w.cfg.DisableDeferAccept) w.deferCapable = !w.cfg.DisableDeferAccept if err != nil { @@ -2326,6 +2329,27 @@ func (w *Worker) closeListenerAfterDrain(ctx context.Context) { // worker does not own. lfd := w.listenFD w.listenFD = -1 + w.cancelAccept(ctx, lfd) + // The completions above only cover handshakes an accept had already + // reached. Anything still in the kernel accept queue would be aborted + // by the close below, after its client had possibly sent a request; + // accept it now instead and serve it like any other connection + // (celeris#662). + w.acceptQueuedOnPause(ctx, lfd) + _ = unix.Close(lfd) + w.lingerUntil = 0 + deferlinger.NoteClose() + w.listenFDClosed.Store(true) + w.pause.Notify() +} + +// cancelAccept cancels the accept armed on the listen socket lfd, which the +// caller has already taken out of w.listenFD, and handles the completions it +// produces before the caller closes lfd. The cancel releases the kernel's +// io_uring reference to the socket's file, so the close takes it out of the +// SO_REUSEPORT group at once; unix.Close alone left a phantom socket that +// intercepted connections. +func (w *Worker) cancelAccept(ctx context.Context, lfd int) { if sqe := w.ring.GetSQE(); sqe != nil { prepCancelFDSkipSuccess(sqe, lfd) setSQEUserData(sqe, 0) @@ -2344,17 +2368,24 @@ func (w *Worker) closeListenerAfterDrain(ctx context.Context) { } w.ring.EndCQ(cqH) } - // The completions above only cover handshakes an accept had already - // reached. Anything still in the kernel accept queue would be aborted - // by the close below, after its client had possibly sent a request; - // accept it now instead and serve it like any other connection - // (celeris#662). - w.acceptQueuedOnPause(ctx, lfd) +} + +// stopAccepting cancels the accept and closes the listener, once the +// worker's context is cancelled (celeris#759). The worker goes on turning +// after that, for its HTTP/2 conns (h2PoolSettled) and its send drain, and it +// used to accept and serve new connections meanwhile, which were then cut at +// the budget; net/http's Shutdown closes its listeners first. What is still +// in the kernel's accept queue is reset, as the close in shutdown did. +// Idempotent. Worker thread. +func (w *Worker) stopAccepting(ctx context.Context) { + if w.listenFD < 0 { + return + } + lfd := w.listenFD + w.listenFD = -1 // before the cancel's completions: handleAccept re-arms on a listenFD >= 0 + w.cancelAccept(ctx, lfd) _ = unix.Close(lfd) w.lingerUntil = 0 - deferlinger.NoteClose() - w.listenFDClosed.Store(true) - w.pause.Notify() } // acceptQueuedOnPause accepts every connection still waiting in the listen @@ -6063,8 +6094,8 @@ func (w *Worker) checkTimeouts() { // h2PoolDrainFloor is the least time the worker, its context cancelled, // waits for the HTTP/2 stream handlers on the shared worker pool before it -// shuts down (celeris#759); the budget of the last Engine.Shutdown extends it -// to that budget's deadline. h2PoolDrainPoll caps one wait meanwhile. +// shuts down (celeris#759); a live budget of the last Engine.Shutdown extends +// it (h2PoolSettled). h2PoolDrainPoll caps one wait meanwhile. const ( h2PoolDrainFloor = 250 * time.Millisecond h2PoolDrainPoll = 10 * time.Millisecond @@ -6077,12 +6108,14 @@ const ( // which only this loop drains; shutdown cancelled such streams and closed // their conns under their handlers, so the client got unexpected EOF, and // the hooks ran before the handlers had finished. So the loop keeps turning, -// reading and writing as usual, until no HTTP/2 conn has a pool handler -// running or a response still in its write queue. Every HTTP/2 conn is sent -// GOAWAY first (again for a conn that arrives meanwhile), so its client opens -// no new stream on it, as net/http's graceful shutdown does. The wait ends at -// the deadline of the budget the last Engine.Shutdown handed over, and never -// before h2PoolDrainFloor. Worker thread. +// reading and writing as usual on the conns it has (it accepts no new one: +// stopAccepting), until no HTTP/2 conn has a pool handler running, a response +// still in its write queue, or response DATA waiting for the client's +// WINDOW_UPDATE. Every HTTP/2 conn is sent GOAWAY first, so its client opens +// no new stream on it, as net/http's graceful shutdown does, and a stream it +// opens anyway is refused. The wait is bounded by the budget the last +// Engine.Shutdown handed over, as epoll's is, and never ends before +// h2PoolDrainFloor. Worker thread. func (w *Worker) h2PoolSettled() bool { if len(w.h2Conns) == 0 { return true @@ -6110,21 +6143,29 @@ func (w *Worker) h2PoolSettled() bool { mu.Unlock() } } - if cs.h2State.PoolHandlersRunning() || cs.h2State.WriteQueuePending() { + if cs.h2State.PoolHandlersRunning() || cs.h2State.WriteQueuePending() || cs.h2State.OutboundPending() { busy = true } } if !busy { return true } + // Bounded as epoll's send drain is (epoll's Loop.sendDrainWait): while + // the budget is live, until its deadline or, for a ctx without one, + // until it is done, no longer than WriteTimeout, and never less than the + // floor, which is all a done budget, or none, gets. end := w.h2DrainStart + int64(h2PoolDrainFloor) if w.drainBudget != nil { - // A budget that is done, at its deadline or by a cancel before it, - // leaves the floor. if p := w.drainBudget.Load(); p != nil && (*p).Err() == nil { - if d, has := (*p).Deadline(); has && d.UnixNano() > end { - end = d.UnixNano() + d, bounded := (*p).Deadline() + ext := d.UnixNano() + if wt := int64(w.cfg.WriteTimeout); wt > 0 && (!bounded || w.h2DrainStart+wt < ext) { + ext, bounded = w.h2DrainStart+wt, true + } + if !bounded { + return false // until the budget is done } + end = max(end, ext) } } return now > end diff --git a/internal/conn/h2.go b/internal/conn/h2.go index 0158d8de..474b0fd9 100644 --- a/internal/conn/h2.go +++ b/internal/conn/h2.go @@ -781,12 +781,22 @@ func (s *H2State) PoolHandlersRunning() bool { return s.processor.PoolHandlersRunning() } +// OutboundPending reports whether a stream of this connection has response +// DATA buffered for the client's WINDOW_UPDATE: its handler has returned and +// the rest of its response goes out as the client grants window +// (celeris#759). Not for the hot path. +func (s *H2State) OutboundPending() bool { + return s.processor.OutboundPending() +} + // GoAway starts a graceful close of the connection (celeris#759): it sends // GOAWAY(NO_ERROR) naming the last stream the client has opened, so the // client opens no new stream on the connection, and leaves the streams in -// flight to finish. It reports false, sending nothing, while the server's -// preface has not gone out yet. Called on the engine's event loop thread, -// like ProcessH2. +// flight to finish. A stream the client opens above the one named anyway is +// refused, not served (REFUSED_STREAM): the client counts it as not +// processed and may retry it elsewhere. It reports false, sending nothing, +// while the server's preface has not gone out yet. Called on the engine's +// event loop thread, like ProcessH2. func (s *H2State) GoAway(write func([]byte)) bool { s.mu.Lock() defer s.mu.Unlock() diff --git a/protocol/h2/stream/processor.go b/protocol/h2/stream/processor.go index 749ff875..716c3bdc 100644 --- a/protocol/h2/stream/processor.go +++ b/protocol/h2/stream/processor.go @@ -178,6 +178,15 @@ type Processor struct { // see them. Incremented in runHandler before Submit, decremented after // executeHandler has returned. poolRunning atomic.Int32 + + // goAwaySent records that a GOAWAY has gone out, and goAwayLastID the + // last stream it named (celeris#759). A stream the client opens above + // it is not served (runHandler): its client counts it as not processed + // and may retry it on another connection (RFC 9113 §6.8). Written by + // SendGoAway, read by runHandler, both on the frame-processing path + // (event loop under H2State.mu). + goAwaySent bool + goAwayLastID uint32 } // PoolHandlersRunning reports whether a stream of this connection has a @@ -187,6 +196,26 @@ func (p *Processor) PoolHandlersRunning() bool { return p.poolRunning.Load() > 0 } +// OutboundPending reports whether a stream of this connection has response +// DATA buffered, waiting for the client's WINDOW_UPDATE: its handler has +// returned, and the rest of its response goes out only as the client grants +// window (celeris#759). The native engines' graceful shutdown waits for it, +// as for the pool handlers. Takes the manager's and each stream's lock; not +// for the hot path. +func (p *Processor) OutboundPending() bool { + p.manager.mu.RLock() + defer p.manager.mu.RUnlock() + for _, s := range p.manager.streams { + s.mu.RLock() + pending := s.OutboundBuffer != nil && s.OutboundBuffer.Len() > 0 + s.mu.RUnlock() + if pending { + return true + } + } + return false +} + // rstRateLimit and rstBurstLimit bound RST_STREAM arrivals. An honest // client resets at most a handful of streams per second (client abort // of a pipelined request, EARLY_HINTS race, etc.). 100 resets per @@ -613,6 +642,17 @@ func (p *Processor) runHandler(stream *Stream) { if p.handler == nil { return } + // A stream opened above the last stream a GOAWAY named: its client + // counts it as not processed and may retry it elsewhere, so its handler + // must not run (celeris#759). REFUSED_STREAM says exactly that (RFC 9113 + // §8.7). Its headers were decoded all the same, so the HPACK state stays + // in step with the client's. + if p.goAwaySent && stream.ID > p.goAwayLastID { + _ = p.sendRSTStreamAndMarkClosed(stream.ID, http2.ErrCodeRefusedStream) + stream.SetState(StateClosed) + p.manager.DeleteStream(stream.ID) + return + } if p.canRunInline(stream) { p.executeHandlerInline(stream) @@ -1622,8 +1662,12 @@ func (p *Processor) GoAwayErr(lastStreamID uint32, code http2.ErrCode, debug []b return err } -// SendGoAway sends a GOAWAY frame. +// SendGoAway sends a GOAWAY frame, and records it: no stream above +// lastStreamID is served from then on (runHandler). func (p *Processor) SendGoAway(lastStreamID uint32, code http2.ErrCode, debugData []byte) error { + if !p.goAwaySent || lastStreamID < p.goAwayLastID { + p.goAwaySent, p.goAwayLastID = true, lastStreamID + } if p.connWriter != nil { return p.connWriter.SendGoAway(lastStreamID, code, debugData) } diff --git a/server.go b/server.go index 4c6c8c78..bf4a90cc 100644 --- a/server.go +++ b/server.go @@ -523,14 +523,18 @@ func (s *Server) cancelListen() { // call it from another goroutine (celeris#703). The Start* call that the // server was started with returns only after Shutdown has returned. // -// The drain waits for every HTTP/1.1 request and every HTTP/2 stream. An -// HTTP/2 stream on an async route (marked Async, or promoted to async under +// The drain waits for every HTTP/1.1 request and every HTTP/2 stream. No +// engine accepts a new connection once the shutdown has begun. An HTTP/2 +// stream on an async route (marked Async, or promoted to async under // [Config.AsyncHandlers]) runs on the shared HTTP/2 worker pool: epoll, -// io_uring and adaptive send each HTTP/2 connection GOAWAY and serve it until -// those handlers have returned and their responses have gone out, while ctx -// is live (until its deadline or, for a ctx without one, until it is done, -// but no longer than [Config.WriteTimeout]) and never for less than 250 ms; -// std waits for its h2c streams' handlers, bounded by ctx (celeris#759). Once +// io_uring and adaptive send each HTTP/2 connection GOAWAY, refuse +// (REFUSED_STREAM) a stream its client opens after it, and serve the +// connection until those handlers have returned and their responses have +// gone out, response DATA waiting for the client's WINDOW_UPDATE included, +// while ctx is live (until its deadline or, for a ctx without one, until it +// is done, but no longer than [Config.WriteTimeout]) and never for less than +// 250 ms; std waits for its h2c streams' handlers, bounded by ctx +// (celeris#759). Once // the handlers have returned, the native engines send what the sockets have // not taken yet before they close the connections: epoll (and adaptive while // it runs epoll) for as long, and never for less than 250 ms (celeris#760); diff --git a/shutdown_h2_pool_linux_test.go b/shutdown_h2_pool_linux_test.go index 4822b1a0..77a51713 100644 --- a/shutdown_h2_pool_linux_test.go +++ b/shutdown_h2_pool_linux_test.go @@ -3,13 +3,17 @@ package celeris_test import ( + "bufio" "bytes" "context" "errors" "io" "net" "net/http" + "strconv" "strings" + "sync" + "sync/atomic" "testing" "time" @@ -30,14 +34,18 @@ import ( // // The order is forced: the handler is released only once the client has read // the GOAWAY, or after releaseAfter if none comes, so a shutdown that closes -// at once, or sends no GOAWAY, fails every time. +// at once, or sends no GOAWAY, fails every time. "Shutdown-background" shuts +// down with context.Background(), a ctx with no deadline, and releases the +// handler holdNoDeadline after the GOAWAY: past the wait's 250 ms floor, which +// is all a ctx without a deadline got. func TestShutdownSendsH2GoAwayThenFinishesStreams(t *testing.T) { const releaseAfter = 2 * time.Second + const holdNoDeadline = 600 * time.Millisecond for _, e := range []struct { name string eng celeris.EngineType }{{"epoll", celeris.Epoll}, {"io_uring", celeris.IOUring}, {"adaptive", celeris.Adaptive}} { - for _, mode := range []string{"Shutdown", "cancel"} { + for _, mode := range []string{"Shutdown", "Shutdown-background", "cancel"} { t.Run(e.name+"/"+mode, func(t *testing.T) { entered := make(chan struct{}) release := make(chan struct{}) @@ -80,7 +88,11 @@ func TestShutdownSendsH2GoAwayThenFinishesStreams(t *testing.T) { events = append(events, "GOAWAY") if !released && releaseTimer.Stop() { released = true - close(release) + if mode == "Shutdown-background" { + time.AfterFunc(holdNoDeadline, func() { close(release) }) + } else { + close(release) + } } case *http2.HeadersFrame: if f.StreamID == 1 { @@ -219,6 +231,10 @@ func startH2PoolServer759(t *testing.T, eng celeris.EngineType, budget time.Dura } } +// beginShutdown starts the shutdown: a cancel of StartWithContext's context +// ("cancel", whose budget is the server's ShutdownTimeout), a Shutdown with +// context.Background(), which has no deadline ("Shutdown-background"), or a +// Shutdown with a budget of its own. func (srv *h2PoolServer759) beginShutdown(mode string, budget time.Duration) { if mode == "cancel" { srv.cancel() @@ -227,6 +243,9 @@ func (srv *h2PoolServer759) beginShutdown(mode string, budget time.Duration) { srv.shutErr = make(chan error, 1) go func() { ctx, cancel := context.WithTimeout(context.Background(), budget) + if mode == "Shutdown-background" { + ctx = context.Background() + } defer cancel() srv.shutErr <- srv.s.Shutdown(ctx) }() @@ -294,3 +313,288 @@ func describeH2ReadEnd759(err error) string { } return err.Error() } + +// frames759 reads frames from fr until the connection ends, idle passes +// without one, or on returns true, answering SETTINGS and PING and handing +// every frame to on; it returns how the read ended. +func frames759(c net.Conn, fr *http2.Framer, idle time.Duration, on func(http2.Frame) bool) string { + for { + _ = c.SetReadDeadline(time.Now().Add(idle)) + f, err := fr.ReadFrame() + if err != nil { + return describeH2ReadEnd759(err) + } + switch f := f.(type) { + case *http2.SettingsFrame: + if !f.IsAck() { + _ = fr.WriteSettingsAck() + } + case *http2.PingFrame: + if !f.IsAck() { + _ = fr.WritePing(true, f.Data) + } + } + if on(f) { + return "stopped" + } + } +} + +// engines759 are every engine, std included. +var engines759 = []struct { + name string + eng celeris.EngineType +}{{"std", celeris.Std}, {"epoll", celeris.Epoll}, {"io_uring", celeris.IOUring}, {"adaptive", celeris.Adaptive}} + +// TestShutdownAcceptsNoNewConnection: while a graceful shutdown waits for an +// HTTP/2 stream on the worker pool (celeris#759), the native engines kept +// their listeners and accepted and served new connections, HTTP/1.1 and h2c, +// for as long as the wait lasted (up to the whole budget), then cut them at +// its end. net/http's Shutdown closes its listeners first, and so must they: +// a connection attempted 300 ms into the shutdown is not served. +func TestShutdownAcceptsNoNewConnection(t *testing.T) { + for _, e := range engines759 { + t.Run(e.name, func(t *testing.T) { + entered := make(chan struct{}) + release := make(chan struct{}) + srv := startH2PoolServer759(t, e.eng, 5*time.Second, func(s *celeris.Server) { + s.GET("/held", func(c *celeris.Context) error { + close(entered) + <-release + return c.String(http.StatusOK, "done") + }).Async() + }) + c, fr := dialH2759(t, srv.addr) + writeH2Get759(t, fr, 1, "/held") + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("the held handler did not start within 5s") + } + heldDone := make(chan string, 1) + go func() { heldDone <- frames759(c, fr, 10*time.Second, func(http2.Frame) bool { return false }) }() + srv.beginShutdown("Shutdown", 5*time.Second) + time.Sleep(300 * time.Millisecond) + h1 := tryH1Ping759(srv.addr) + h2c := tryH2cPing759(srv.addr) + close(release) + srv.waitStart(t, 10*time.Second) + _ = c.Close() // std leaves an h2c connection open past the shutdown + <-heldDone + if h1 == "served" || h2c == "served" { + t.Fatalf("%s: 300 ms into the shutdown a new connection was served: HTTP/1.1 %s, h2c %s", e.name, h1, h2c) + } + t.Logf("%s: new HTTP/1.1 connection: %s; new h2c connection: %s", e.name, h1, h2c) + }) + } +} + +// tryH1Ping759 dials addr and sends GET /ping over HTTP/1.1; it returns +// "served" if a response came, else how the attempt ended. +func tryH1Ping759(addr string) string { + c, err := net.DialTimeout("tcp", addr, time.Second) + if err != nil { + return "dial: " + err.Error() + } + defer func() { _ = c.Close() }() + _ = c.SetDeadline(time.Now().Add(time.Second)) + if _, err := io.WriteString(c, "GET /ping HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil { + return "write: " + err.Error() + } + if _, err := http.ReadResponse(bufio.NewReader(c), nil); err != nil { + return "read: " + describeH2ReadEnd759(err) + } + return "served" +} + +// tryH2cPing759 dials addr and asks for /ping on h2c stream 1; it returns +// "served" if response HEADERS came for it, else how the attempt ended. +func tryH2cPing759(addr string) string { + c, err := net.DialTimeout("tcp", addr, time.Second) + if err != nil { + return "dial: " + err.Error() + } + defer func() { _ = c.Close() }() + if _, err := io.WriteString(c, http2.ClientPreface); err != nil { + return "write: " + err.Error() + } + fr := http2.NewFramer(c, c) + _ = fr.WriteSettings() + var hb bytes.Buffer + enc := hpack.NewEncoder(&hb) + for _, f := range []hpack.HeaderField{{Name: ":method", Value: "GET"}, {Name: ":scheme", Value: "http"}, {Name: ":authority", Value: "x"}, {Name: ":path", Value: "/ping"}} { + _ = enc.WriteField(f) + } + _ = fr.WriteHeaders(http2.HeadersFrameParam{StreamID: 1, BlockFragment: hb.Bytes(), EndStream: true, EndHeaders: true}) + res := "" + end := frames759(c, fr, time.Second, func(f http2.Frame) bool { + switch f := f.(type) { + case *http2.HeadersFrame: + if f.StreamID == 1 { + res = "served" + } + case *http2.RSTStreamFrame: + res = "RST_STREAM " + f.ErrCode.String() + } + return res != "" + }) + if res != "" { + return res + } + return "read: " + end +} + +// TestShutdownRefusesStreamsAfterGoAway: once the graceful shutdown has sent +// GOAWAY naming stream 1, a stream the client opens anyway (3, on an async +// route; 5, on a sync one) must not be served (celeris#759). Its client +// counts it as not processed and may retry it on another connection (RFC +// 9113 §6.8), so serving it could run a request twice. It is refused with +// RST_STREAM(REFUSED_STREAM), and stream 1 still finishes. std does not send +// h2c connections GOAWAY at all; that is not asserted here. +func TestShutdownRefusesStreamsAfterGoAway(t *testing.T) { + for _, e := range engines759[1:] { + t.Run(e.name, func(t *testing.T) { + entered := make(chan struct{}) + release := make(chan struct{}) + var served atomic.Int32 + srv := startH2PoolServer759(t, e.eng, 5*time.Second, func(s *celeris.Server) { + s.GET("/held", func(c *celeris.Context) error { + close(entered) + <-release + return c.String(http.StatusOK, "done") + }).Async() + s.GET("/quick", func(c *celeris.Context) error { + served.Add(1) + return c.String(http.StatusOK, "quick") + }).Async() + s.GET("/quicksync", func(c *celeris.Context) error { + served.Add(1) + return c.String(http.StatusOK, "quick") + }) + }) + c, fr := dialH2759(t, srv.addr) + writeH2Get759(t, fr, 1, "/held") + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("the held handler did not start within 5s") + } + srv.beginShutdown("Shutdown", 5*time.Second) + var events []string + rst := map[uint32]http2.ErrCode{} + sent := false + var releaseOnce sync.Once + end := frames759(c, fr, 5*time.Second, func(f http2.Frame) bool { + switch f := f.(type) { + case *http2.GoAwayFrame: + events = append(events, "GOAWAY(last="+strconv.Itoa(int(f.LastStreamID))+")") + if !sent { + sent = true + writeH2Get759(t, fr, 3, "/quick") + writeH2Get759(t, fr, 5, "/quicksync") + // Stream 1 goes on once the refusals had time to come. + time.AfterFunc(500*time.Millisecond, func() { releaseOnce.Do(func() { close(release) }) }) + } + case *http2.HeadersFrame: + events = append(events, "HEADERS(s"+strconv.Itoa(int(f.StreamID))+")") + case *http2.DataFrame: + events = append(events, "DATA(s"+strconv.Itoa(int(f.StreamID))+")") + case *http2.RSTStreamFrame: + rst[f.StreamID] = f.ErrCode + events = append(events, "RST_STREAM(s"+strconv.Itoa(int(f.StreamID))+","+f.ErrCode.String()+")") + } + return false + }) + releaseOnce.Do(func() { close(release) }) + srv.waitStart(t, 10*time.Second) + trace := strings.Join(events, ", ") + ", read: " + end + if !sent { + t.Fatalf("%s: no GOAWAY; frames: %s", e.name, trace) + } + if n := served.Load(); n != 0 || strings.Contains(trace, "HEADERS(s3)") || strings.Contains(trace, "HEADERS(s5)") { + t.Fatalf("%s: %d stream(s) opened after the GOAWAY were served; frames: %s", e.name, n, trace) + } + if rst[3] != http2.ErrCodeRefusedStream || rst[5] != http2.ErrCodeRefusedStream { + t.Fatalf("%s: streams 3 and 5, opened after the GOAWAY, were not refused (REFUSED_STREAM); frames: %s", e.name, trace) + } + if !strings.Contains(trace, "DATA(s1)") { + t.Fatalf("%s: stream 1 did not finish; frames: %s", e.name, trace) + } + }) + } +} + +// TestShutdownWaitsForFlowControlledH2Response: a response whose DATA waits +// for the client's WINDOW_UPDATE when the shutdown begins must still go out +// whole (celeris#759). The client keeps the default 65535-byte windows and +// opens them 400 ms into the shutdown, past the wait's 250 ms floor; the +// native engines took a handler that had returned for a settled stream and +// closed the connection with the rest of the body unsent. +func TestShutdownWaitsForFlowControlledH2Response(t *testing.T) { + const size = 1 << 20 + body := bytes.Repeat([]byte("z"), size) + for _, e := range engines759 { + for _, route := range []string{"sync", "async-route"} { + t.Run(e.name+"/"+route, func(t *testing.T) { + entered := make(chan struct{}) + release := make(chan struct{}) + srv := startH2PoolServer759(t, e.eng, 5*time.Second, func(s *celeris.Server) { + r := s.GET("/big", func(c *celeris.Context) error { + if route == "async-route" { + close(entered) + <-release + } + return c.Blob(http.StatusOK, "application/octet-stream", body) + }) + if route == "async-route" { + r.Async() + } + }) + c, fr := dialH2759(t, srv.addr) + writeH2Get759(t, fr, 1, "/big") + n, ended := 0, false + var events []string + done := make(chan string, 1) + go func() { + done <- frames759(c, fr, 5*time.Second, func(f http2.Frame) bool { + switch f := f.(type) { + case *http2.GoAwayFrame: + events = append(events, "GOAWAY") + case *http2.DataFrame: + if f.StreamID == 1 { + n += len(f.Data()) + ended = ended || f.StreamEnded() + } + case *http2.RSTStreamFrame: + events = append(events, "RST_STREAM "+f.ErrCode.String()) + } + return ended + }) + }() + if route == "async-route" { + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("the handler did not start within 5s") + } + } else { + time.Sleep(300 * time.Millisecond) // the handler has run; the rest waits for the window + } + start := time.Now() + srv.beginShutdown("Shutdown", 5*time.Second) + if route == "async-route" { + time.Sleep(100 * time.Millisecond) + close(release) + } + time.Sleep(time.Until(start.Add(400 * time.Millisecond))) + _ = fr.WriteWindowUpdate(0, size) + _ = fr.WriteWindowUpdate(1, size) + end := <-done + srv.waitStart(t, 10*time.Second) + if n != size || !ended { + t.Fatalf("%s/%s: stream 1 got %d of %d body bytes, ended=%v, then %s; frames: %s", e.name, route, n, size, ended, end, strings.Join(events, ", ")) + } + }) + } + } +} From 4bc182055f2f6e5a74caf5b4212dce678e8c1ad8 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Mon, 28 Sep 2026 11:54:27 +0200 Subject: [PATCH 3/7] docs(h2): write down OutboundPending's lock order (celeris#759) --- protocol/h2/stream/processor.go | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/protocol/h2/stream/processor.go b/protocol/h2/stream/processor.go index 716c3bdc..250bd175 100644 --- a/protocol/h2/stream/processor.go +++ b/protocol/h2/stream/processor.go @@ -200,8 +200,14 @@ func (p *Processor) PoolHandlersRunning() bool { // DATA buffered, waiting for the client's WINDOW_UPDATE: its handler has // returned, and the rest of its response goes out only as the client grants // window (celeris#759). The native engines' graceful shutdown waits for it, -// as for the pool handlers. Takes the manager's and each stream's lock; not -// for the hot path. +// as for the pool handlers. Not for the hot path. +// +// Lock order: the manager's mu (read), then each stream's mu (read), so that +// no stream is released (RemoveStreamFromMap and DeleteStream take the +// manager's mu for writing) while it is read. Nothing takes a stream's mu and +// then the manager's: flushStreamOutbound reserves windows under the stream's +// mu with atomics only, and flushConnWindowStalledStreams releases the +// manager's mu before it takes a stream's. func (p *Processor) OutboundPending() bool { p.manager.mu.RLock() defer p.manager.mu.RUnlock() From 5ededef58cbd35f692f40d85fcfaf3bd6d12d4c7 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Mon, 28 Sep 2026 11:56:01 +0200 Subject: [PATCH 4/7] fix(iouring): stop accepting only for the HTTP/2 wait, so a shutdown without one leaves the driver conns' queued ops to shutdown (celeris#759) --- engine/iouring/worker.go | 31 ++++++++++++++++++------------- 1 file changed, 18 insertions(+), 13 deletions(-) diff --git a/engine/iouring/worker.go b/engine/iouring/worker.go index 8c612fbb..31d41ea4 100644 --- a/engine/iouring/worker.go +++ b/engine/iouring/worker.go @@ -1172,19 +1172,21 @@ func (w *Worker) run(ctx context.Context) { // SQEs and reaps their completions through the normal path — // until no send is queued or in flight, bounded by // shutdownSendDrainNanos so a peer that stopped reading - // cannot hold shutdown open. It accepts nothing meanwhile - // (celeris#759): a connection accepted now would be served - // and then cut, and the wait for HTTP/2 below can last the - // whole budget; net/http's Shutdown closes its listeners - // first. + // cannot hold shutdown open. The loop keeps accepting for + // that window — it is the ordinary iteration — which is the + // graceful side of the trade: a connection that arrives + // inside it is answered rather than reset. // // First, though, the HTTP/2 stream handlers on the shared // worker pool (celeris#759): they run off this loop, and their // responses come back through the conns' write queues, which // only this loop drains. The send drain's clock starts once they - // are done. - w.stopAccepting(ctx) + // are done. That wait can last the whole budget, so it accepts + // nothing (stopAccepting): a connection accepted in it was + // served and then cut at the budget. net/http's Shutdown closes + // its listeners first. if !w.h2PoolSettled() { + w.stopAccepting(ctx) w.shutdownDrainDeadline = 0 } else { if w.shutdownDrainDeadline == 0 { @@ -2371,12 +2373,15 @@ func (w *Worker) cancelAccept(ctx context.Context, lfd int) { } // stopAccepting cancels the accept and closes the listener, once the -// worker's context is cancelled (celeris#759). The worker goes on turning -// after that, for its HTTP/2 conns (h2PoolSettled) and its send drain, and it -// used to accept and serve new connections meanwhile, which were then cut at -// the budget; net/http's Shutdown closes its listeners first. What is still -// in the kernel's accept queue is reset, as the close in shutdown did. -// Idempotent. Worker thread. +// worker's context is cancelled and it waits for the HTTP/2 stream handlers +// on the shared worker pool (celeris#759). That wait can last the whole +// budget, and the worker accepted and served new connections meanwhile, +// which were then cut at its end; net/http's Shutdown closes its listeners +// first. What is still in the kernel's accept queue is reset, as the close +// in shutdown did. Only then, not on every shutdown: the cancel submits the +// SQ ring and handles its completions, and a shutdown with no HTTP/2 wait +// goes straight on, leaving the driver conns' queued ops to shutdown +// (TestDriverShutdownReleasesDescriptors). Idempotent. Worker thread. func (w *Worker) stopAccepting(ctx context.Context) { if w.listenFD < 0 { return From 369df9ef20b4046f5e2ac50dc55ba2370ddc9cb9 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Mon, 28 Sep 2026 11:56:50 +0200 Subject: [PATCH 5/7] test: serialize the h2 test client's framer writes (celeris#759) --- shutdown_h2_pool_linux_test.go | 23 +++++++++++++++++------ 1 file changed, 17 insertions(+), 6 deletions(-) diff --git a/shutdown_h2_pool_linux_test.go b/shutdown_h2_pool_linux_test.go index 77a51713..cf6d63db 100644 --- a/shutdown_h2_pool_linux_test.go +++ b/shutdown_h2_pool_linux_test.go @@ -316,8 +316,10 @@ func describeH2ReadEnd759(err error) string { // frames759 reads frames from fr until the connection ends, idle passes // without one, or on returns true, answering SETTINGS and PING and handing -// every frame to on; it returns how the read ended. -func frames759(c net.Conn, fr *http2.Framer, idle time.Duration, on func(http2.Frame) bool) string { +// every frame to on; it returns how the read ended. Its writes take wmu, which +// a caller writing on fr from another goroutine holds too (a Framer is not +// safe for concurrent writes). +func frames759(c net.Conn, fr *http2.Framer, wmu *sync.Mutex, idle time.Duration, on func(http2.Frame) bool) string { for { _ = c.SetReadDeadline(time.Now().Add(idle)) f, err := fr.ReadFrame() @@ -327,11 +329,15 @@ func frames759(c net.Conn, fr *http2.Framer, idle time.Duration, on func(http2.F switch f := f.(type) { case *http2.SettingsFrame: if !f.IsAck() { + wmu.Lock() _ = fr.WriteSettingsAck() + wmu.Unlock() } case *http2.PingFrame: if !f.IsAck() { + wmu.Lock() _ = fr.WritePing(true, f.Data) + wmu.Unlock() } } if on(f) { @@ -372,7 +378,9 @@ func TestShutdownAcceptsNoNewConnection(t *testing.T) { t.Fatal("the held handler did not start within 5s") } heldDone := make(chan string, 1) - go func() { heldDone <- frames759(c, fr, 10*time.Second, func(http2.Frame) bool { return false }) }() + go func() { + heldDone <- frames759(c, fr, new(sync.Mutex), 10*time.Second, func(http2.Frame) bool { return false }) + }() srv.beginShutdown("Shutdown", 5*time.Second) time.Sleep(300 * time.Millisecond) h1 := tryH1Ping759(srv.addr) @@ -427,7 +435,7 @@ func tryH2cPing759(addr string) string { } _ = fr.WriteHeaders(http2.HeadersFrameParam{StreamID: 1, BlockFragment: hb.Bytes(), EndStream: true, EndHeaders: true}) res := "" - end := frames759(c, fr, time.Second, func(f http2.Frame) bool { + end := frames759(c, fr, new(sync.Mutex), time.Second, func(f http2.Frame) bool { switch f := f.(type) { case *http2.HeadersFrame: if f.StreamID == 1 { @@ -484,7 +492,7 @@ func TestShutdownRefusesStreamsAfterGoAway(t *testing.T) { rst := map[uint32]http2.ErrCode{} sent := false var releaseOnce sync.Once - end := frames759(c, fr, 5*time.Second, func(f http2.Frame) bool { + end := frames759(c, fr, new(sync.Mutex), 5*time.Second, func(f http2.Frame) bool { switch f := f.(type) { case *http2.GoAwayFrame: events = append(events, "GOAWAY(last="+strconv.Itoa(int(f.LastStreamID))+")") @@ -554,9 +562,10 @@ func TestShutdownWaitsForFlowControlledH2Response(t *testing.T) { writeH2Get759(t, fr, 1, "/big") n, ended := 0, false var events []string + var wmu sync.Mutex done := make(chan string, 1) go func() { - done <- frames759(c, fr, 5*time.Second, func(f http2.Frame) bool { + done <- frames759(c, fr, &wmu, 5*time.Second, func(f http2.Frame) bool { switch f := f.(type) { case *http2.GoAwayFrame: events = append(events, "GOAWAY") @@ -587,8 +596,10 @@ func TestShutdownWaitsForFlowControlledH2Response(t *testing.T) { close(release) } time.Sleep(time.Until(start.Add(400 * time.Millisecond))) + wmu.Lock() _ = fr.WriteWindowUpdate(0, size) _ = fr.WriteWindowUpdate(1, size) + wmu.Unlock() end := <-done srv.waitStart(t, 10*time.Second) if n != size || !ended { From 3f759376ade1e8cb16e8b9c39e2092171c5b5cb5 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Mon, 28 Sep 2026 11:57:17 +0200 Subject: [PATCH 6/7] docs(server): say which engines stop accepting when (celeris#759) --- server.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/server.go b/server.go index bf4a90cc..db478082 100644 --- a/server.go +++ b/server.go @@ -523,9 +523,10 @@ func (s *Server) cancelListen() { // call it from another goroutine (celeris#703). The Start* call that the // server was started with returns only after Shutdown has returned. // -// The drain waits for every HTTP/1.1 request and every HTTP/2 stream. No -// engine accepts a new connection once the shutdown has begun. An HTTP/2 -// stream on an async route (marked Async, or promoted to async under +// The drain waits for every HTTP/1.1 request and every HTTP/2 stream. std +// and epoll accept no new connection once the shutdown has begun, and +// io_uring none while it waits for HTTP/2 handlers (its 250 ms send drain +// keeps accepting, celeris#595). An HTTP/2 stream on an async route (marked Async, or promoted to async under // [Config.AsyncHandlers]) runs on the shared HTTP/2 worker pool: epoll, // io_uring and adaptive send each HTTP/2 connection GOAWAY, refuse // (REFUSED_STREAM) a stream its client opens after it, and serve the From 0df4ac23aaa049ed005536339e9f1be02bb4bea9 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 29 Sep 2026 14:46:22 +0200 Subject: [PATCH 7/7] fix(h2): leave a stream refused above the GOAWAY alone once its RST_STREAM has put it back in the pool (celeris#759) runHandler's refusal wrote the stream's state and deleted it by its ID after sendRSTStreamAndMarkClosed, which had already deleted it and put it back in the stream pool: another connection that took it from the pool meanwhile got its stream closed, and this connection lost the stream with that stream's ID. The refusal now deletes the stream only when the RST_STREAM could not be written. TestRefusedStreamIsLeftAloneOnceReleased fails on the old code (the released object's state is Closed, not the pool's Idle). --- protocol/h2/stream/goaway_refuse_test.go | 76 ++++++++++++++++++++++++ protocol/h2/stream/processor.go | 12 ++-- 2 files changed, 84 insertions(+), 4 deletions(-) create mode 100644 protocol/h2/stream/goaway_refuse_test.go diff --git a/protocol/h2/stream/goaway_refuse_test.go b/protocol/h2/stream/goaway_refuse_test.go new file mode 100644 index 00000000..7481135a --- /dev/null +++ b/protocol/h2/stream/goaway_refuse_test.go @@ -0,0 +1,76 @@ +package stream + +import ( + "context" + "errors" + "sync" + "testing" + + "golang.org/x/net/http2" +) + +// rstRecordingConn is a testResponseWriter that records the RST_STREAM +// frames written through it, and fails them with err when it is set. +type rstRecordingConn struct { + *testResponseWriter + mu sync.Mutex + rst []rstStreamRecord + err error +} + +func (c *rstRecordingConn) WriteRSTStreamPriority(streamID uint32, code http2.ErrCode) error { + c.mu.Lock() + defer c.mu.Unlock() + c.rst = append(c.rst, rstStreamRecord{streamID: streamID, code: code}) + return c.err +} + +// TestRefusedStreamIsLeftAloneOnceReleased: runHandler refuses a stream the +// client opened above the last stream a GOAWAY named (celeris#759), and the +// refusal deletes the stream, which puts it back in the stream pool. Nothing +// may touch it after that: another connection can have taken it from the +// pool by then, and a write to it closed that connection's stream, or, with +// its new ID, deleted the stream of this connection that had the same ID. +// The object's state after the call shows such a write: resetAndPool left +// it at ID 0 and StateIdle. +func TestRefusedStreamIsLeftAloneOnceReleased(t *testing.T) { + for _, tc := range []struct { + name string + err error + }{ + {"rst-sent", nil}, + {"rst-failed", errors.New("write failed")}, + } { + t.Run(tc.name, func(t *testing.T) { + conn := &rstRecordingConn{testResponseWriter: newTestResponseWriter(), err: tc.err} + ran := make(chan struct{}, 1) + p := NewProcessor(HandlerFunc(func(context.Context, *Stream) error { + ran <- struct{}{} + return nil + }), newTestFrameWriter(), conn) + if err := p.SendGoAway(1, http2.ErrCodeNo, nil); err != nil { + t.Fatalf("SendGoAway: %v", err) + } + s := p.manager.CreateStream(3) + p.runHandler(s) + + if _, ok := p.manager.GetStream(3); ok { + t.Fatal("refused stream 3 is still in the manager") + } + conn.mu.Lock() + rst := append([]rstStreamRecord(nil), conn.rst...) + conn.mu.Unlock() + if len(rst) != 1 || rst[0].streamID != 3 || rst[0].code != http2.ErrCodeRefusedStream { + t.Fatalf("RST_STREAM frames %+v, want one REFUSED_STREAM for stream 3", rst) + } + select { + case <-ran: + t.Fatal("the handler of refused stream 3 ran") + default: + } + if id, st := s.ID, State(s.state.Load()); id != 0 || st != StateIdle { + t.Fatalf("stream 3's object after its release: ID %d, state %v; want ID 0, %v: it was written after it went back to the pool", id, st, StateIdle) + } + }) + } +} diff --git a/protocol/h2/stream/processor.go b/protocol/h2/stream/processor.go index 250bd175..7dad1d99 100644 --- a/protocol/h2/stream/processor.go +++ b/protocol/h2/stream/processor.go @@ -652,11 +652,15 @@ func (p *Processor) runHandler(stream *Stream) { // counts it as not processed and may retry it elsewhere, so its handler // must not run (celeris#759). REFUSED_STREAM says exactly that (RFC 9113 // §8.7). Its headers were decoded all the same, so the HPACK state stays - // in step with the client's. + // in step with the client's. Once the RST_STREAM is written, + // sendRSTStreamAndMarkClosed has deleted the stream, which puts it back + // in the stream pool, so stream is not touched after the call: another + // connection can have it by then. if p.goAwaySent && stream.ID > p.goAwayLastID { - _ = p.sendRSTStreamAndMarkClosed(stream.ID, http2.ErrCodeRefusedStream) - stream.SetState(StateClosed) - p.manager.DeleteStream(stream.ID) + id := stream.ID + if err := p.sendRSTStreamAndMarkClosed(id, http2.ErrCodeRefusedStream); err != nil { + p.manager.DeleteStream(id) + } return }