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/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..f1ad6abf 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. @@ -474,8 +483,14 @@ func (l *Loop) run(ctx context.Context) { for { if ctx.Err() != nil { - l.shutdown() - return + // 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 @@ -498,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 { @@ -548,6 +564,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 +3692,76 @@ 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 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 + } + 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() || cs.h2State.OutboundPending() { + busy = true + } + } + if !busy { + return true + } + _, more := l.sendDrainWait(l.h2DrainStart) + 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/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..31d41ea4 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,26 @@ 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. 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 { + w.shutdownDrainDeadline = time.Now().UnixNano() + shutdownSendDrainNanos + } + if !w.hasPendingSends() || time.Now().UnixNano() > w.shutdownDrainDeadline { + w.shutdown() + return + } } } @@ -1200,8 +1222,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 { @@ -2048,6 +2071,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) } @@ -2302,6 +2331,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) @@ -2320,17 +2370,27 @@ 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 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 + } + 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 @@ -6037,6 +6097,85 @@ 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); 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 +) + +// 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 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 + } + 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() || 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 { + if p := w.drainBudget.Load(); p != nil && (*p).Err() == nil { + 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 +} + // 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..474b0fd9 100644 --- a/internal/conn/h2.go +++ b/internal/conn/h2.go @@ -774,6 +774,40 @@ 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() +} + +// 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. 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() + 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/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/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..7dad1d99 100644 --- a/protocol/h2/stream/processor.go +++ b/protocol/h2/stream/processor.go @@ -169,6 +169,57 @@ 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 + + // 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 +// handler running on the shared worker pool (celeris#759). Safe from any +// goroutine. +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. 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() + 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 @@ -597,6 +648,21 @@ 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. 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 { + id := stream.ID + if err := p.sendRSTStreamAndMarkClosed(id, http2.ErrCodeRefusedStream); err != nil { + p.manager.DeleteStream(id) + } + return + } if p.canRunInline(stream) { p.executeHandlerInline(stream) @@ -609,6 +675,7 @@ func (p *Processor) runHandler(stream *Stream) { } stream.flags.Or(flagAsyncRunning) + p.poolRunning.Add(1) globalH2Pool.Submit(p, stream) } @@ -732,6 +799,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 @@ -1601,8 +1672,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 e5fa22d6..db478082 100644 --- a/server.go +++ b/server.go @@ -523,18 +523,23 @@ 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. 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 +// 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); +// 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..cf6d63db --- /dev/null +++ b/shutdown_h2_pool_linux_test.go @@ -0,0 +1,611 @@ +//go:build linux + +package celeris_test + +import ( + "bufio" + "bytes" + "context" + "errors" + "io" + "net" + "net/http" + "strconv" + "strings" + "sync" + "sync/atomic" + "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. "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", "Shutdown-background", "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 + if mode == "Shutdown-background" { + time.AfterFunc(holdNoDeadline, func() { close(release) }) + } else { + 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) + } +} + +// 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() + return + } + 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) + }() +} + +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() +} + +// 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. 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() + if err != nil { + return describeH2ReadEnd759(err) + } + 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) { + 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, 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) + 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, new(sync.Mutex), 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, 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))+")") + 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 + var wmu sync.Mutex + done := make(chan string, 1) + go func() { + done <- frames759(c, fr, &wmu, 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))) + wmu.Lock() + _ = fr.WriteWindowUpdate(0, size) + _ = fr.WriteWindowUpdate(1, size) + wmu.Unlock() + 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, ", ")) + } + }) + } + } +}