From f2518b76ffb98da305b9e81064ecd667470a8a79 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Mon, 28 Sep 2026 08:24:56 +0200 Subject: [PATCH 1/2] fix(epoll): shutdown sends what the sockets have not taken yet before it closes the conns (celeris#760) epoll's Loop.shutdown closed every connection as soon as the handlers had returned (phase 3), whatever was still queued on it. A response larger than the socket buffers, to a client that reads more slowly than the loop shuts down, lost its tail: the client got EOF in the middle of the body. io_uring drains its sends first (celeris#595), and std drains through net/http. adaptive had the same defect while it ran epoll. Loop.shutdown now runs a send drain (drainSends) between joining the async dispatch goroutines (phase 2) and closing the fds (phase 3): it flushes every conn with bytes queued and polls their sockets for POLLOUT until nothing is queued, or the drain's time is up. That time is the later of 250 ms (io_uring's shutdownSendDrainNanos) and the deadline of the budget the last Engine.Shutdown call handed over: epoll's Engine.Shutdown, a no-op before, now records its ctx for the loops, and Server.Shutdown calls it before it cancels Listen's context (a cancel of StartWithContext's context reaches the loops first, and the watcher's Shutdown follows within the 250 ms). adaptive's Shutdown hands ctx to its sub-engines before it cancels them, since it cancels first. A client that never reads therefore holds the shutdown for the budget, no longer. Tests: TestShutdownSendsTheWholeResponse (std, epoll, adaptive; handler on the worker and on a dispatch goroutine, direct Shutdown and cancel): a 3 MiB response to a client with a 64 KiB receive buffer that starts reading 1 s after the handler returns, 200 ms into a shutdown with a 30 s budget, must arrive whole. TestShutdownSendDrainIsBounded (epoll, adaptive): with a client that never reads, the Start call returns within the budget and the connection is closed. io_uring's own drain returns about 10 s late in that case (celeris#806). Fixes #760 --- adaptive/engine.go | 14 ++ engine/epoll/engine.go | 29 ++- engine/epoll/loop.go | 97 +++++++++ server.go | 10 +- shutdown_send_drain_linux_test.go | 318 ++++++++++++++++++++++++++++++ 5 files changed, 454 insertions(+), 14 deletions(-) create mode 100644 shutdown_send_drain_linux_test.go diff --git a/adaptive/engine.go b/adaptive/engine.go index d4acf0e0..63ac9fa9 100644 --- a/adaptive/engine.go +++ b/adaptive/engine.go @@ -962,6 +962,20 @@ func (e *Engine) Shutdown(ctx context.Context) error { done := e.listenDone e.listenMu.Unlock() + // Hand ctx to the sub-engines before the cancel below stops them: + // epoll's Shutdown makes it the budget of the send drain its loops run + // as they stop (celeris#760), and after the cancel it would come too + // late. Both sub-engines' Shutdown does nothing else, and they are + // called again, as before, once Listen has returned. + e.mu.Lock() + subs := [2]engine.Engine{e.primary, e.secondary} + e.mu.Unlock() + for _, sub := range subs { + if sub != nil { + _ = sub.Shutdown(ctx) + } + } + if cancel != nil { cancel() } diff --git a/engine/epoll/engine.go b/engine/epoll/engine.go index bb9a943d..1aaf1664 100644 --- a/engine/epoll/engine.go +++ b/engine/epoll/engine.go @@ -81,6 +81,11 @@ type Engine struct { // adoptRR round-robins io_uring→epoll transplant adoptions across loops (#383). adoptRR atomic.Uint64 + + // drainBudget is the ctx of the last Shutdown call: the budget of the + // send drain the loops run when Listen's context is cancelled + // (celeris#760; see Loop.drainSends). + drainBudget atomic.Pointer[context.Context] } // New creates a new epoll engine. @@ -148,6 +153,7 @@ func (e *Engine) Listen(ctx context.Context) error { l.transplantAdoptRefused = &e.metrics.transplantAdoptRefused l.sweepCnt = &e.metrics.sweep l.pause = &e.pause + l.drainBudget = &e.drainBudget e.loops[i] = l } e.mu.Unlock() @@ -202,19 +208,24 @@ func (e *Engine) Listen(ctx context.Context) error { return nil } -// Shutdown is a no-op for the epoll engine — graceful shutdown is +// Shutdown does not stop the epoll engine itself — graceful shutdown is // driven by context cancellation on Listen's parent context. The // Server calls Listen with its managed context and cancels it during // Server.Shutdown; the Listen goroutine returns after running -// Loop.shutdown (which closes connections and joins async dispatch -// goroutines via asyncWG). Server.Shutdown waits for that return before -// it runs the OnShutdown hooks (celeris#703). +// Loop.shutdown (which joins async dispatch goroutines via asyncWG, sends +// the responses still queued, and closes connections). Server.Shutdown +// waits for that return before it runs the OnShutdown hooks +// (celeris#703). // -// The context parameter is accepted for interface parity with engines -// that do run async drain operations on Shutdown (e.g. std's -// http.Server.Shutdown), and for future use if epoll Shutdown gains -// explicit drain semantics. -func (e *Engine) Shutdown(_ context.Context) error { +// What Shutdown does is hand ctx to the loops as the budget of that send +// drain (celeris#760): a response larger than the socket buffers is sent +// until ctx's deadline, and never for less than shutdownSendDrainFloor, +// before its conn is closed. Server.Shutdown calls it before it cancels +// Listen's context, and a cancel of StartWithContext's context reaches +// the loops first but the watcher's Shutdown follows at once, within the +// drain's floor. +func (e *Engine) Shutdown(ctx context.Context) error { + e.drainBudget.Store(&ctx) return nil } diff --git a/engine/epoll/loop.go b/engine/epoll/loop.go index 7b239f06..85ad4827 100644 --- a/engine/epoll/loop.go +++ b/engine/epoll/loop.go @@ -69,6 +69,17 @@ const maxEpollEvents = 2048 // connection cleanly (read returns 0 bytes / EOF). var errPeerClosed = fmt.Errorf("celeris: peer closed connection: %w", io.EOF) +// shutdownSendDrainFloor is the least time shutdown's send drain gives the +// responses still queued to reach the kernel before the conns are closed +// (celeris#760): io_uring's bound, shutdownSendDrainNanos (celeris#595). The +// budget of the Engine.Shutdown call that stopped the engine extends it to +// that budget's deadline; a peer that stops reading holds it no longer. +const shutdownSendDrainFloor = 250 * time.Millisecond + +// shutdownSendDrainPoll caps one wait of the drain for a writable socket, so +// the drain notices a budget that ends early (a cancelled Shutdown ctx). +const shutdownSendDrainPoll = 20 * time.Millisecond + // Loop is an epoll-based event loop worker. type Loop struct { id int @@ -180,6 +191,12 @@ type Loop struct { // connState with (celeris#624). runCtx context.Context + // drainBudget points at the engine's record of the budget of the last + // Engine.Shutdown call (celeris#760): shutdown's send drain may run + // until that ctx is done. nil, or no Shutdown yet, leaves the drain its + // floor, shutdownSendDrainFloor. + drainBudget *atomic.Pointer[context.Context] + // 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. @@ -3446,6 +3463,13 @@ func (l *Loop) shutdown() { // close fds or recycle connState below. l.asyncWG.Wait() + // Phase 2b (celeris#760): every response the handlers wrote is queued + // now; send what the sockets have not taken yet before phase 3 closes + // them, as io_uring does before its shutdown (celeris#595). Closing at + // once cut off the tail of any response larger than the socket buffers + // to a client that reads more slowly than the loop shuts down. + l.drainSends() + // Phase 3: now that the async dispatch goroutines have exited (phase 2), // close the fds and release the connState back to the pool. The pool // release stays gated on !detached for the same reason as closeConn: a @@ -3499,6 +3523,79 @@ func (l *Loop) shutdown() { l.closeEpollFD() } +// 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: the later of +// shutdownSendDrainFloor after it began and the deadline of the budget the +// last Engine.Shutdown call handed over (drainBudget), or, before that +// deadline, the moment that budget's ctx is done. A conn whose write fails +// is left to phase 3's close. The run loop is no longer turning: the conns +// are polled directly (poll(2), POLLOUT), not through the epoll set. +// +// Loop thread, after phase 2: no dispatch goroutine is left to write, and a +// detached conn's middleware finds detachClosed set by phase 1 and writes +// nothing more. detachMu is still taken around each flush, as every flush +// site does. +func (l *Loop) drainSends() { + start := time.Now() + var fds []unix.PollFd + for { + fds = fds[:0] + for i := len(l.liveConns) - 1; i >= 0; i-- { + cs := l.liveConns[i] + if cs.hijacked.Load() { + continue // the application's since the Hijack (celeris#668) + } + mu := cs.detachMu + if mu != nil { + mu.Lock() + } + pending := false + if csWritePending(cs) && l.flushWrites(cs, true) == nil { + pending = csWritePending(cs) + } + if mu != nil { + mu.Unlock() + } + if pending { + fds = append(fds, unix.PollFd{Fd: int32(cs.fd), Events: unix.POLLOUT}) + } + } + if len(fds) == 0 { + return + } + wait, ok := l.sendDrainWait(start) + if !ok { + return + } + if _, err := unix.Poll(fds, int(wait/time.Millisecond)+1); err != nil && err != unix.EINTR { + return + } + } +} + +// sendDrainWait reports how long drainSends may wait for a writable socket +// now, at most shutdownSendDrainPoll, and false once the drain's time is up +// (see drainSends). +func (l *Loop) sendDrainWait(start time.Time) (time.Duration, bool) { + now := time.Now() + end := start.Add(shutdownSendDrainFloor) + if l.drainBudget != nil { + // A budget that is done, at its deadline or by a cancel before it, + // leaves the floor. + if p := l.drainBudget.Load(); p != nil && (*p).Err() == nil { + if d, has := (*p).Deadline(); has && d.After(end) { + end = d + } + } + } + left := end.Sub(now) + if left <= 0 { + return 0, false + } + return min(left, shutdownSendDrainPoll), true +} + // createListenSocket binds and listens on addr. deferAccept asks for // TCP_DEFER_ACCEPT (resource.Config.DisableDeferAccept turns it off). A // pause clears the option on this socket and lingers before it closes it diff --git a/server.go b/server.go index 3e4c4d23..c86882d8 100644 --- a/server.go +++ b/server.go @@ -529,11 +529,11 @@ func (s *Server) cancelListen() { // 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). On epoll, and on adaptive while it runs -// epoll, the drain ends when the handlers have returned, and a connection is -// then closed without flushing what the socket has not taken yet: a response -// larger than the socket buffers, to a client that reads slowly, loses its -// tail (celeris#760). +// 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) until ctx's deadline, +// 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_send_drain_linux_test.go b/shutdown_send_drain_linux_test.go new file mode 100644 index 00000000..caa040f9 --- /dev/null +++ b/shutdown_send_drain_linux_test.go @@ -0,0 +1,318 @@ +//go:build linux + +package celeris_test + +import ( + "bufio" + "bytes" + "context" + "errors" + "fmt" + "io" + "net" + "net/http" + "strings" + "testing" + "time" + + "github.com/goceleris/celeris" +) + +// TestShutdownSendsTheWholeResponse pins celeris#760: epoll's shutdown closed +// every connection as soon as the handlers had returned, with whatever the +// socket had not taken yet still queued, so a response larger than the socket +// buffers, to a client that reads more slowly than the loop shuts down, lost +// its tail and the client got EOF in the middle of the body. io_uring drained +// its sends first (celeris#595) and std drains through net/http. +// +// The client has a 64 KiB receive buffer and starts reading readDelay after +// the handler returns, which is 200 ms into the shutdown, so the loop is +// certain to reach its shutdown with most of the 3 MiB still queued. The +// shutdown's budget is 30 s, so a drain bounded by it has all the time it +// needs; a drain that closes at once cuts the body short every time. Both +// ways of shutting down (a direct Shutdown and a cancel of StartWithContext's +// context) and both kinds of route (the handler on the worker, and on a +// dispatch goroutine) are covered. io_uring is not: its own drain (celeris#595) +// gives up after 250 ms whatever the budget, before this client reads, and +// whether the tail survives then depends on what the kernel has taken +// (celeris#806). +func TestShutdownSendsTheWholeResponse(t *testing.T) { + const size = 3 << 20 + const readDelay = time.Second + body := make([]byte, size) + for i := range body { + body[i] = byte(i*7 + i>>13) + } + for _, e := range []struct { + name string + eng celeris.EngineType + }{{"std", celeris.Std}, {"epoll", celeris.Epoll}, {"adaptive", celeris.Adaptive}} { + for _, route := range []string{"sync", "async-route"} { + for _, mode := range []string{"Shutdown", "cancel"} { + t.Run(e.name+"/"+route+"/"+mode, func(t *testing.T) { + desc := e.name + "/" + route + "/" + mode + entered := make(chan struct{}) + release := make(chan struct{}) + srv := startServer760(t, e.eng, 30*time.Second, func(s *celeris.Server) { + r := s.GET("/big", func(c *celeris.Context) error { + close(entered) + <-release + return c.Blob(http.StatusOK, "application/octet-stream", body) + }) + if route == "async-route" { + r.Async() + } + }) + c := dialSlowReader760(t, srv.addr) + if _, err := io.WriteString(c, "GET /big HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil { + t.Fatal(err) + } + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("handler did not start within 5s") + } + shutErr := srv.beginShutdown(mode, 30*time.Second) + time.Sleep(200 * time.Millisecond) + close(release) + time.Sleep(readDelay) + got, n, end := readResponse760(c) + if !bytes.Equal(got, body) { + t.Errorf("%s: the client got %d of %d body bytes (%d in all), then %s", desc, len(got), size, n, end) + } + if err := srv.waitShutdown(mode, shutErr, 20*time.Second); err != nil { + t.Errorf("%s: %v", desc, err) + } + }) + } + } + } +} + +// TestShutdownSendDrainIsBounded is the other half of the contract: a client +// that never reads must not hold epoll's shutdown past its budget. The drain +// gives up when the shutdown's budget runs out, never before 250 ms +// (io_uring's bound, celeris#595), and the Start call returns within bound of +// the budget. io_uring is not asserted: its own drain returns about 10 s late +// with a stalled send whatever the budget (celeris#806). std's handler writes +// the response itself and blocks in that write; net/http's WriteTimeout is +// its bound. +func TestShutdownSendDrainIsBounded(t *testing.T) { + // Larger than the socket buffers, so part of it stays queued while the + // client does not read, and smaller than the 4 MiB write cap, so the + // native engines queue it whole (celeris#761 is a different defect). + const size = 3 << 20 + const budget = 500 * time.Millisecond + const bound = 5 * time.Second + body := make([]byte, size) + for _, e := range []struct { + name string + eng celeris.EngineType + }{{"epoll", celeris.Epoll}, {"adaptive", celeris.Adaptive}} { + for _, mode := range []string{"Shutdown", "cancel"} { + t.Run(e.name+"/"+mode, func(t *testing.T) { + desc := e.name + "/" + mode + served := make(chan struct{}) + srv := startServer760(t, e.eng, budget, func(s *celeris.Server) { + s.GET("/big", func(c *celeris.Context) error { + defer close(served) // native engines: the body is queued, not written, here + return c.Blob(http.StatusOK, "application/octet-stream", body) + }) + }) + c := dialSlowReader760(t, srv.addr) + if _, err := io.WriteString(c, "GET /big HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil { + t.Fatal(err) + } + select { + case <-served: + case <-time.After(5 * time.Second): + t.Fatal("handler did not run within 5s") + } + time.Sleep(100 * time.Millisecond) // the response is queued behind a full socket + start := time.Now() + shutErr := srv.beginShutdown(mode, budget) + if err := srv.waitShutdown(mode, shutErr, budget+bound); err != nil { + t.Fatalf("%s: %v (a drain the stalled client can hold open past the budget)", desc, err) + } + t.Logf("%s: Start returned %v after the shutdown began", desc, time.Since(start).Round(time.Millisecond)) + // The engine has stopped, so the connection must be closed: + // the client, which has read nothing so far, reads what the + // kernel took and then EOF (or, where the kernel took it + // all, the whole body), never an open connection. + got, n, end := readResponse760(c) + if end != "EOF" && end != "the end of the body" { + t.Errorf("%s: after the shutdown the client got %d of %d body bytes (%d in all), then %s", desc, len(got), size, n, end) + } + }) + } + } +} + +type server760 struct { + s *celeris.Server + addr string + cancel context.CancelFunc + startDone chan error +} + +// startServer760 starts a server with routes on StartWithContext, with +// ShutdownTimeout budget, and waits until it answers /ping. An io_uring start +// that fails only with ENOMEM is retried, with a new server, for up to 30 s: +// the kernel charges ring memory to RLIMIT_MEMLOCK per UID and gives it back +// some milliseconds after a ring closes, so at the CI runner's 8 MiB a start +// made right after the previous server stopped can fail although nothing +// leaked (see startC714DetachServer). +func startServer760(t *testing.T, eng celeris.EngineType, budget time.Duration, routes func(*celeris.Server)) *server760 { + 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 := &server760{s: s, addr: addr, cancel: cancel, startDone: make(chan error, 1)} + go func() { srv.startDone <- s.StartWithContext(ctx) }() + err = waitReady760(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 <-srv.startDone: + 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) + } +} + +// waitReady760 polls /ping until it answers 200, or returns the error the +// start returned first. +func waitReady760(addr string, startDone <-chan error) error { + probe := &http.Client{Timeout: 300 * time.Millisecond} + for deadline := time.Now().Add(15 * time.Second); time.Now().Before(deadline); { + select { + case err := <-startDone: + if err == nil { + err = errors.New("StartWithContext returned nil before the server was ready") + } + return err + default: + } + if resp, err := probe.Get("http://" + addr + "/ping"); err == nil { + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() + if resp.StatusCode == http.StatusOK { + return nil + } + } + time.Sleep(20 * time.Millisecond) + } + return fmt.Errorf("no answer on /ping at %s within 15s", addr) +} + +// beginShutdown starts the shutdown: a direct Shutdown with a budget of its +// own, or a cancel of StartWithContext's context (whose budget is the +// server's ShutdownTimeout). The returned channel carries the direct +// Shutdown's error; it is nil for a cancel. +func (srv *server760) beginShutdown(mode string, budget time.Duration) chan error { + if mode == "cancel" { + srv.cancel() + return nil + } + ch := make(chan error, 1) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), budget) + defer cancel() + ch <- srv.s.Shutdown(ctx) + }() + return ch +} + +// waitShutdown waits up to limit for the Start call (and a direct Shutdown) +// to return. +func (srv *server760) waitShutdown(mode string, shutErr chan error, limit time.Duration) error { + deadline := time.After(limit) + if shutErr != nil { + select { + case <-shutErr: + case <-deadline: + return fmt.Errorf("Shutdown had not returned %v after it began", limit) + } + } + select { + case err := <-srv.startDone: + srv.startDone <- err // for the cleanup + return nil + case <-deadline: + return fmt.Errorf("StartWithContext had not returned %v after the shutdown (%s) began", limit, mode) + } +} + +// dialSlowReader760 dials a raw connection with a 64 KiB receive buffer, so +// the server's socket takes little of a large response until the client +// reads. +func dialSlowReader760(t *testing.T, addr string) net.Conn { + t.Helper() + c, err := net.Dial("tcp", addr) + if err != nil { + t.Fatal(err) + } + _ = c.(*net.TCPConn).SetReadBuffer(64 << 10) + t.Cleanup(func() { _ = c.Close() }) + return c +} + +// readResponse760 reads one HTTP/1.1 response from c, giving up after 5 s +// without a byte, and returns its body, the bytes received in all, and how +// the read ended. +func readResponse760(c net.Conn) (body []byte, total int, end string) { + cr := &countReader760{c: c} + resp, err := http.ReadResponse(bufio.NewReaderSize(cr, 64<<10), nil) + if err != nil { + return nil, cr.n, describeEnd760(err) + } + body, err = io.ReadAll(resp.Body) + return body, cr.n, describeEnd760(err) +} + +type countReader760 struct { + c net.Conn + n int +} + +func (r *countReader760) Read(p []byte) (int, error) { + _ = r.c.SetReadDeadline(time.Now().Add(5 * time.Second)) + n, err := r.c.Read(p) + r.n += n + return n, err +} + +func describeEnd760(err error) string { + var ne net.Error + switch { + case err == nil: + return "the end of the body" + case errors.As(err, &ne) && ne.Timeout(): + return "no byte for 5s, connection still open" + case errors.Is(err, io.EOF), errors.Is(err, io.ErrUnexpectedEOF): + return "EOF" + } + return err.Error() +} From 0cefbb0318a281d101e2eda71e0cd29e4aeb1ca7 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Mon, 28 Sep 2026 11:43:59 +0200 Subject: [PATCH 2/2] fix(epoll): a Shutdown ctx without a deadline keeps the send drain going until it is done, no longer than WriteTimeout (celeris#760) Round 2 of #807, from its review. sendDrainWait extended the drain only to ctx's deadline, so context.Background() or a WithCancel ctx, net/http's "wait as long as it takes", got the 250 ms floor and the tail of a response larger than the socket buffers was cut as before the fix. A live ctx without a deadline now keeps the drain going until it is done; a drain is never longer than WriteTimeout after it began (when set), the bound a live conn's stalled write gets, so a client that never reads cannot hold Shutdown(context.Background()) for ever. TestShutdownSendsTheWholeResponse covers Shutdown(context.Background()); TestShutdownSendDrainIsBounded covers a cancelled WithCancel ctx and a Background ctx bounded by WriteTimeout, and allows the budget + 1 s instead of + 5 s. --- engine/epoll/engine.go | 12 +++--- engine/epoll/loop.go | 40 ++++++++++++------- server.go | 7 ++-- shutdown_send_drain_linux_test.go | 64 ++++++++++++++++++++++--------- 4 files changed, 82 insertions(+), 41 deletions(-) diff --git a/engine/epoll/engine.go b/engine/epoll/engine.go index 1aaf1664..81a3d90c 100644 --- a/engine/epoll/engine.go +++ b/engine/epoll/engine.go @@ -219,11 +219,13 @@ func (e *Engine) Listen(ctx context.Context) error { // // What Shutdown does is hand ctx to the loops as the budget of that send // drain (celeris#760): a response larger than the socket buffers is sent -// until ctx's deadline, and never for less than shutdownSendDrainFloor, -// before its conn is closed. Server.Shutdown calls it before it cancels -// Listen's context, and a cancel of StartWithContext's context reaches -// the loops first but the watcher's Shutdown follows at once, within the -// drain's floor. +// while ctx is live, until its deadline or, for a ctx without one, until it +// is done, but no longer than the config's WriteTimeout, and never for less +// than shutdownSendDrainFloor, before its conn is closed (see +// Loop.sendDrainWait). Server.Shutdown calls it before it cancels Listen's +// context, and a cancel of StartWithContext's context reaches the loops +// first but the watcher's Shutdown follows at once, within the drain's +// floor. func (e *Engine) Shutdown(ctx context.Context) error { e.drainBudget.Store(&ctx) return nil diff --git a/engine/epoll/loop.go b/engine/epoll/loop.go index 8b01bef7..f15b2f2b 100644 --- a/engine/epoll/loop.go +++ b/engine/epoll/loop.go @@ -3539,12 +3539,10 @@ func (l *Loop) shutdown() { // 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: the later of -// shutdownSendDrainFloor after it began and the deadline of the budget the -// last Engine.Shutdown call handed over (drainBudget), or, before that -// deadline, the moment that budget's ctx is done. A conn whose write fails -// is left to phase 3's close. The run loop is no longer turning: the conns -// are polled directly (poll(2), POLLOUT), not through the epoll set. +// more, until nothing is queued or the drain's time is up (sendDrainWait). A +// conn whose write fails is left to phase 3's close. The run loop is no +// longer turning: the conns are polled directly (poll(2), POLLOUT), not +// through the epoll set. // // Loop thread, after phase 2: no dispatch goroutine is left to write, and a // detached conn's middleware finds detachClosed set by phase 1 and writes @@ -3589,21 +3587,35 @@ func (l *Loop) drainSends() { } // sendDrainWait reports how long drainSends may wait for a writable socket -// now, at most shutdownSendDrainPoll, and false once the drain's time is up -// (see drainSends). +// now, at most shutdownSendDrainPoll, and false once the drain's time is up. +// +// The drain, begun at start, runs while the budget the last Engine.Shutdown +// call handed over (drainBudget) is live: until its ctx's deadline, and for +// a ctx with none (context.Background, a WithCancel ctx: net/http's "wait as +// long as it takes") until the ctx is done. Either way no longer than +// WriteTimeout after the drain began, when that is set: the bound a live +// conn's stalled write gets, and net/http's, so a client that never reads +// cannot hold Shutdown(context.Background()) for ever. It never ends before +// shutdownSendDrainFloor, which is also all it gets once the budget is done +// (at its deadline, or cancelled before it) or when no Shutdown handed one +// over. func (l *Loop) sendDrainWait(start time.Time) (time.Duration, bool) { - now := time.Now() end := start.Add(shutdownSendDrainFloor) if l.drainBudget != nil { - // A budget that is done, at its deadline or by a cancel before it, - // leaves the floor. if p := l.drainBudget.Load(); p != nil && (*p).Err() == nil { - if d, has := (*p).Deadline(); has && d.After(end) { - end = d + ext, bounded := (*p).Deadline() + if wt := l.cfg.WriteTimeout; wt > 0 && (!bounded || start.Add(wt).Before(ext)) { + ext, bounded = start.Add(wt), true + } + if !bounded { + return shutdownSendDrainPoll, true // until the budget is done + } + if ext.After(end) { + end = ext } } } - left := end.Sub(now) + left := time.Until(end) if left <= 0 { return 0, false } diff --git a/server.go b/server.go index c86882d8..632aefd1 100644 --- a/server.go +++ b/server.go @@ -531,9 +531,10 @@ func (s *Server) cancelListen() { // 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) until ctx's deadline, -// and never for less than 250 ms (celeris#760); io_uring for 250 ms -// (celeris#806). +// 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 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_send_drain_linux_test.go b/shutdown_send_drain_linux_test.go index caa040f9..2d7038a3 100644 --- a/shutdown_send_drain_linux_test.go +++ b/shutdown_send_drain_linux_test.go @@ -26,19 +26,23 @@ import ( // its sends first (celeris#595) and std drains through net/http. // // The client has a 64 KiB receive buffer and starts reading readDelay after -// the handler returns, which is 200 ms into the shutdown, so the loop is +// the handler returns, which is 200 ms into the shutdown, well past the +// drain's 250 ms floor, so the loop is // certain to reach its shutdown with most of the 3 MiB still queued. The // shutdown's budget is 30 s, so a drain bounded by it has all the time it -// needs; a drain that closes at once cuts the body short every time. Both -// ways of shutting down (a direct Shutdown and a cancel of StartWithContext's -// context) and both kinds of route (the handler on the worker, and on a -// dispatch goroutine) are covered. io_uring is not: its own drain (celeris#595) +// needs; a drain that closes at once cuts the body short every time. The +// ways of shutting down are a direct Shutdown with that budget, a direct +// Shutdown(context.Background()), whose ctx has no deadline at all (net/http's +// "wait as long as it takes", which the drain first took for no budget and +// gave its 250 ms floor), and a cancel of StartWithContext's context; both +// kinds of route (the handler on the worker, and on a dispatch goroutine) are +// covered. io_uring is not: its own drain (celeris#595) // gives up after 250 ms whatever the budget, before this client reads, and // whether the tail survives then depends on what the kernel has taken // (celeris#806). func TestShutdownSendsTheWholeResponse(t *testing.T) { const size = 3 << 20 - const readDelay = time.Second + const readDelay = 500 * time.Millisecond body := make([]byte, size) for i := range body { body[i] = byte(i*7 + i>>13) @@ -48,12 +52,12 @@ func TestShutdownSendsTheWholeResponse(t *testing.T) { eng celeris.EngineType }{{"std", celeris.Std}, {"epoll", celeris.Epoll}, {"adaptive", celeris.Adaptive}} { for _, route := range []string{"sync", "async-route"} { - for _, mode := range []string{"Shutdown", "cancel"} { + for _, mode := range []string{"Shutdown", "Shutdown-background", "cancel"} { t.Run(e.name+"/"+route+"/"+mode, func(t *testing.T) { desc := e.name + "/" + route + "/" + mode entered := make(chan struct{}) release := make(chan struct{}) - srv := startServer760(t, e.eng, 30*time.Second, func(s *celeris.Server) { + srv := startServer760(t, e.eng, 30*time.Second, 0, func(s *celeris.Server) { r := s.GET("/big", func(c *celeris.Context) error { close(entered) <-release @@ -93,7 +97,12 @@ func TestShutdownSendsTheWholeResponse(t *testing.T) { // that never reads must not hold epoll's shutdown past its budget. The drain // gives up when the shutdown's budget runs out, never before 250 ms // (io_uring's bound, celeris#595), and the Start call returns within bound of -// the budget. io_uring is not asserted: its own drain returns about 10 s late +// the budget. A ctx with no deadline bounds the drain by its cancel +// ("Shutdown-withcancel", cancelled budget into the shutdown), and, when +// nothing cancels it (context.Background()), by the config's WriteTimeout, +// the bound a live conn's stalled write gets ("Shutdown-background", with +// WriteTimeout = budget): a client that never reads cannot hold that Shutdown +// for ever. io_uring is not asserted: its own drain returns about 10 s late // with a stalled send whatever the budget (celeris#806). std's handler writes // the response itself and blocks in that write; net/http's WriteTimeout is // its bound. @@ -103,17 +112,21 @@ func TestShutdownSendDrainIsBounded(t *testing.T) { // native engines queue it whole (celeris#761 is a different defect). const size = 3 << 20 const budget = 500 * time.Millisecond - const bound = 5 * time.Second + const bound = time.Second body := make([]byte, size) for _, e := range []struct { name string eng celeris.EngineType }{{"epoll", celeris.Epoll}, {"adaptive", celeris.Adaptive}} { - for _, mode := range []string{"Shutdown", "cancel"} { + for _, mode := range []string{"Shutdown", "Shutdown-withcancel", "Shutdown-background", "cancel"} { t.Run(e.name+"/"+mode, func(t *testing.T) { desc := e.name + "/" + mode served := make(chan struct{}) - srv := startServer760(t, e.eng, budget, func(s *celeris.Server) { + var writeTimeout time.Duration + if mode == "Shutdown-background" { + writeTimeout = budget + } + srv := startServer760(t, e.eng, budget, writeTimeout, func(s *celeris.Server) { s.GET("/big", func(c *celeris.Context) error { defer close(served) // native engines: the body is queued, not written, here return c.Blob(http.StatusOK, "application/octet-stream", body) @@ -156,13 +169,14 @@ type server760 struct { } // startServer760 starts a server with routes on StartWithContext, with -// ShutdownTimeout budget, and waits until it answers /ping. An io_uring start +// ShutdownTimeout budget (and WriteTimeout writeTimeout, 0 for the default), +// and waits until it answers /ping. An io_uring start // that fails only with ENOMEM is retried, with a new server, for up to 30 s: // the kernel charges ring memory to RLIMIT_MEMLOCK per UID and gives it back // some milliseconds after a ring closes, so at the CI runner's 8 MiB a start // made right after the previous server stopped can fail although nothing // leaked (see startC714DetachServer). -func startServer760(t *testing.T, eng celeris.EngineType, budget time.Duration, routes func(*celeris.Server)) *server760 { +func startServer760(t *testing.T, eng celeris.EngineType, budget, writeTimeout time.Duration, routes func(*celeris.Server)) *server760 { t.Helper() retryUntil := time.Now().Add(30 * time.Second) for tries := 1; ; tries++ { @@ -172,7 +186,7 @@ func startServer760(t *testing.T, eng celeris.EngineType, budget time.Duration, } addr := ln.Addr().String() _ = ln.Close() - s := celeris.New(celeris.Config{Engine: eng, Addr: addr, ShutdownTimeout: budget}) + s := celeris.New(celeris.Config{Engine: eng, Addr: addr, ShutdownTimeout: budget, WriteTimeout: writeTimeout}) s.GET("/ping", func(c *celeris.Context) error { return c.String(http.StatusOK, "ok") }) routes(s) ctx, cancel := context.WithCancel(context.Background()) @@ -228,9 +242,11 @@ func waitReady760(addr string, startDone <-chan error) error { } // beginShutdown starts the shutdown: a direct Shutdown with a budget of its -// own, or a cancel of StartWithContext's context (whose budget is the -// server's ShutdownTimeout). The returned channel carries the direct -// Shutdown's error; it is nil for a cancel. +// own ("Shutdown"), a direct Shutdown whose ctx has no deadline, never +// cancelled ("Shutdown-background") or cancelled budget into the shutdown +// ("Shutdown-withcancel"), or a cancel of StartWithContext's context (whose +// budget is the server's ShutdownTimeout). The returned channel carries the +// direct Shutdown's error; it is nil for a cancel. func (srv *server760) beginShutdown(mode string, budget time.Duration) chan error { if mode == "cancel" { srv.cancel() @@ -238,7 +254,17 @@ func (srv *server760) beginShutdown(mode string, budget time.Duration) chan erro } ch := make(chan error, 1) go func() { - ctx, cancel := context.WithTimeout(context.Background(), budget) + var ctx context.Context + var cancel context.CancelFunc + switch mode { + case "Shutdown-background": + ctx, cancel = context.Background(), func() {} + case "Shutdown-withcancel": + ctx, cancel = context.WithCancel(context.Background()) + time.AfterFunc(budget, cancel) + default: + ctx, cancel = context.WithTimeout(context.Background(), budget) + } defer cancel() ch <- srv.s.Shutdown(ctx) }()