From da86f84add739522704b7909e00ce13c0f380ab0 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 22:54:11 +0100 Subject: [PATCH 01/11] tests: the shutdown gate leaves an idle conn with a dispatch goroutine behind another conn's send drain (red; Refs #1165) --- .../epoll/gate_promoted_1165_linux_test.go | 157 ++++++++++++++++++ ...tdown_idle_before_drain_1104_linux_test.go | 22 +-- 2 files changed, 165 insertions(+), 14 deletions(-) create mode 100644 internal/engine/epoll/gate_promoted_1165_linux_test.go diff --git a/internal/engine/epoll/gate_promoted_1165_linux_test.go b/internal/engine/epoll/gate_promoted_1165_linux_test.go new file mode 100644 index 000000000..34000cc53 --- /dev/null +++ b/internal/engine/epoll/gate_promoted_1165_linux_test.go @@ -0,0 +1,157 @@ +//go:build linux + +package epoll + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "golang.org/x/sys/unix" + + "github.com/goceleris/celeris/internal/conn" + "github.com/goceleris/celeris/internal/engine" +) + +// celeris#1165. closeIdleH1 skipped every conn with a dispatch goroutine +// (asyncPromoted: sticky, set for every conn that ever served an Async route +// and, on a server with any Async route, for every conn whose request once +// arrived in pieces), so such an idle keep-alive conn was held behind another +// conn's send drain for the whole Shutdown budget on a loop with no HTTP/2 conn, +// and for the whole pool wait on one with. It now looks inside the conn as +// io_uring's gate does: closed only when its goroutine is parked with no input, +// nothing is in progress in its parser, nothing is detached or queued; the +// rules below are each pinned by a case that must NOT be closed. + +type promoted1165 struct { + l *Loop + cs *connState + fd int + peer int + discons atomic.Int64 +} + +// newPromoted1165 is a conn as production leaves it after its dispatch goroutine +// answered a request and parked: asyncPromoted, asyncRun and asyncParked set, +// no input, a parser at a request boundary, nothing queued. No goroutine runs: +// the state is what the gate reads. +func newPromoted1165(t *testing.T) *promoted1165 { + t.Helper() + r := &promoted1165{} + r.l = shutdownLoop863(t, func(string) { r.discons.Add(1) }) + r.l.asyncPromoted = &atomic.Uint64{} + pair, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM|unix.SOCK_NONBLOCK, 0) + if err != nil { + t.Fatalf("socketpair: %v", err) + } + r.fd, r.peer = pair[0], pair[1] + t.Cleanup(func() { _ = unix.Close(r.peer) }) + if r.fd >= len(r.l.conns) { + t.Fatalf("socketpair fd %d outside the test conn table", r.fd) + } + cs := acquireConnState(t.Context(), r.fd, 4096, true) + cs.protocol = engine.HTTP1 + cs.detected = true + cs.remoteAddr = "127.0.0.1:9" + cs.writeFn = r.l.makeWriteFn(cs) + r.l.conns[r.fd] = cs + r.l.addLiveConn(cs) + r.l.connCount++ + r.l.activeConns.Add(1) + r.l.initProtocol(cs) + cs.asyncPromoted = true + r.l.asyncPromoted.Add(1) + cs.asyncInMu.Lock() + cs.asyncRun, cs.asyncParked = true, true + cs.asyncInMu.Unlock() + r.cs = cs + return r +} + +func TestGateClosesAPromotedConnOnlyWhenItIsIdle1165(t *testing.T) { + cases := []struct { + name string + mutate func(r *promoted1165) + closed bool + }{ + {"parked-idle", func(*promoted1165) {}, true}, + {"goroutine-exited-idle", func(r *promoted1165) { + r.cs.asyncInMu.Lock() + r.cs.asyncRun, r.cs.asyncParked = false, false + r.cs.asyncInMu.Unlock() + }, true}, + {"input-waiting", func(r *promoted1165) { + r.cs.asyncInMu.Lock() + r.cs.asyncInBuf = append(r.cs.asyncInBuf, "GET /x HTTP/1.1\r\n"...) + r.cs.asyncInMu.Unlock() + }, false}, + {"inside-a-handler", func(r *promoted1165) { + r.cs.asyncInMu.Lock() + r.cs.asyncParked = false + r.cs.asyncInMu.Unlock() + }, false}, + {"detachMu-held", func(r *promoted1165) { r.cs.detachMu.Lock() }, false}, + {"request-in-progress", func(r *promoted1165) { + _ = conn.ProcessH1(context.Background(), []byte("GET /x HTTP/1.1\r\nHo"), r.cs.h1State, r.l.handler, func([]byte) {}) + if !r.cs.h1State.HasPendingData() { + panic("premise: the partial head is not pending data") + } + }, false}, + {"switched-to-h2", func(r *promoted1165) { + r.cs.asyncInMu.Lock() + r.cs.h2Switched = true + r.cs.asyncInMu.Unlock() + }, false}, + {"response-queued", func(r *promoted1165) { r.cs.writeBuf = append(r.cs.writeBuf, 'x') }, false}, + {"detached-stream-running", func(r *promoted1165) { r.cs.h1State.Detached.Store(true) }, false}, + {"detached-stream-ended", func(r *promoted1165) { + r.cs.h1State.Detached.Store(true) + r.cs.detachClosed = true + }, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + r := newPromoted1165(t) + tc.mutate(r) + locked := tc.name == "detachMu-held" + n := r.l.closeIdleH1(false, time.Now().UnixNano()) + if locked { + r.cs.detachMu.Unlock() + } + if tc.closed { + if n != 1 || r.l.conns[r.fd] != nil || r.discons.Load() != 1 || r.l.closeCount.Load() != 1 || r.l.activeConns.Load() != 0 { + t.Fatalf("closed=%d live=%v OnDisconnect=%d closeCount=%d activeConns=%d, want 1 closed once", + n, r.l.conns[r.fd] != nil, r.discons.Load(), r.l.closeCount.Load(), r.l.activeConns.Load()) + } + return + } + if n != 0 || r.l.conns[r.fd] == nil || r.discons.Load() != 0 || r.l.closeCount.Load() != 0 || r.l.activeConns.Load() != 1 { + t.Fatalf("closed=%d live=%v OnDisconnect=%d closeCount=%d activeConns=%d, want the conn left alone", + n, r.l.conns[r.fd] != nil, r.discons.Load(), r.l.closeCount.Load(), r.l.activeConns.Load()) + } + }) + } +} + +// The first pass holds, and does not close, a promoted conn whose next request +// is already in the kernel's receive queue (celeris#1104): it is read on, which +// for a promoted conn is a hand-over of the bytes to its goroutine. +func TestGateHoldsAPromotedConnWithAnUnreadRequest1165(t *testing.T) { + r := newPromoted1165(t) + if _, err := unix.Write(r.peer, []byte("GET /y HTTP/1.1\r\nHost: x\r\n\r\n")); err != nil { + t.Fatal(err) + } + if n := r.l.closeIdleH1(true, time.Now().UnixNano()); n != 0 { + t.Fatalf("the first pass closed %d conns, want the promoted conn with an unread request held", n) + } + if r.l.conns[r.fd] == nil || !r.cs.gateRead { + t.Fatalf("live=%v gateRead=%v, want live and marked to be read on", r.l.conns[r.fd] != nil, r.cs.gateRead) + } + r.cs.asyncInMu.Lock() + got := len(r.cs.asyncInBuf) + r.cs.asyncInMu.Unlock() + if got == 0 { + t.Fatal("the request was not handed to the dispatch goroutine's input") + } +} diff --git a/internal/engine/epoll/shutdown_idle_before_drain_1104_linux_test.go b/internal/engine/epoll/shutdown_idle_before_drain_1104_linux_test.go index 34df986d8..dd3243bc1 100644 --- a/internal/engine/epoll/shutdown_idle_before_drain_1104_linux_test.go +++ b/internal/engine/epoll/shutdown_idle_before_drain_1104_linux_test.go @@ -103,11 +103,9 @@ func TestShutdownClosesAnIdleConnWhileAnotherConnsSendDrainWaits1104(t *testing. // - promoted: a conn with a dispatch goroutine parked after its request, as // production makes one (asyncPromoted is set whenever the goroutine is // started: the flag is sticky, so every conn that ever served an Async route -// is one). The pass does NOT close it (closeIdleH1 skips every promoted conn: -// the follow-up celeris#1165, which gives it io_uring's look inside the -// goroutine's own locks), shutdown's later phases do, and nothing is counted -// for it there (phase 3 fires no OnDisconnect). This pins what the pass -// guarantees today, and is the case #1165 turns around. +// is one). The pass closes it as it closes the sync one, from outside the +// goroutine (celeris#1165): once, with one OnDisconnect, and the goroutine +// exits on the asyncClosed closeConn sets. func TestShutdownIdlePassTearsAConnThatServedARequestDownOnce1104(t *testing.T) { for _, promoted := range []bool{false, true} { async := promoted @@ -182,12 +180,12 @@ func TestShutdownIdlePassTearsAConnThatServedARequestDownOnce1104(t *testing.T) } } if promoted { - // The pass alone: it leaves the promoted conn live and uncounted. - if n := l.closeIdleH1(false, time.Now().UnixNano()); n != 0 { - t.Errorf("the pass closed %d promoted conns, want 0 (celeris#1165)", n) + // The pass alone closes the parked promoted conn (celeris#1165). + if n := l.closeIdleH1(false, time.Now().UnixNano()); n != 1 { + t.Errorf("the pass closed %d promoted conns, want 1", n) } - if l.conns[fd] == nil || l.activeConns.Load() != 1 || disconnects.Load() != 0 { - t.Errorf("after the pass the promoted conn is live=%v activeConns=%d OnDisconnect=%d, want live, 1, 0", + if l.conns[fd] != nil || l.activeConns.Load() != 0 || disconnects.Load() != 1 { + t.Errorf("after the pass the promoted conn is live=%v activeConns=%d OnDisconnect=%d, want closed, 0, 1", l.conns[fd] != nil, l.activeConns.Load(), disconnects.Load()) } } @@ -199,10 +197,6 @@ func TestShutdownIdlePassTearsAConnThatServedARequestDownOnce1104(t *testing.T) t.Fatal("shutdown did not return") } wantCalls, wantActive := int64(1), int64(0) - if promoted { - // Closed by phase 3, which counts nothing (pre-existing; celeris#1165). - wantCalls, wantActive = 0, 1 - } if got := disconnects.Load(); got != wantCalls { t.Errorf("OnDisconnect ran %d times for one conn, want %d", got, wantCalls) } From e3102a62ffc6a590fd102af84d9659401bb7c4c9 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 22:54:34 +0100 Subject: [PATCH 02/11] epoll: the shutdown gate looks inside a conn with a dispatch goroutine, as io_uring's does, and closes it when it is idle (v1.6.0, Refs #1165) --- .../epoll/gate_promoted_1165_linux_test.go | 3 + internal/engine/epoll/poolwait_gate_885.go | 88 +++++++++++++------ 2 files changed, 65 insertions(+), 26 deletions(-) diff --git a/internal/engine/epoll/gate_promoted_1165_linux_test.go b/internal/engine/epoll/gate_promoted_1165_linux_test.go index 34000cc53..3156e2daf 100644 --- a/internal/engine/epoll/gate_promoted_1165_linux_test.go +++ b/internal/engine/epoll/gate_promoted_1165_linux_test.go @@ -126,6 +126,9 @@ func TestGateClosesAPromotedConnOnlyWhenItIsIdle1165(t *testing.T) { } return } + if tc.name == "request-in-progress" && !r.cs.h1State.CloseAfter { + t.Fatal("a request in progress on a promoted conn is not marked close-after: it would be answered with keep-alive") + } if n != 0 || r.l.conns[r.fd] == nil || r.discons.Load() != 0 || r.l.closeCount.Load() != 0 || r.l.activeConns.Load() != 1 { t.Fatalf("closed=%d live=%v OnDisconnect=%d closeCount=%d activeConns=%d, want the conn left alone", n, r.l.conns[r.fd] != nil, r.discons.Load(), r.l.closeCount.Load(), r.l.activeConns.Load()) diff --git a/internal/engine/epoll/poolwait_gate_885.go b/internal/engine/epoll/poolwait_gate_885.go index 90f49abe7..c669b48b0 100644 --- a/internal/engine/epoll/poolwait_gate_885.go +++ b/internal/engine/epoll/poolwait_gate_885.go @@ -150,40 +150,43 @@ func (l *Loop) closeIdleH1(first bool, now int64) int { if cs.hijacked.Load() { continue } - // A conn a dispatch goroutine owns is closed here only when the gate has - // ended its detached stream (an SSE or WebSocket handler on an Async - // route detaches on that goroutine): the goroutine is out of its handler, - // since detachClosed is set only under a detachMu the handler held, its - // writes no-op, and closeConn takes the lock as shutdown's own teardown - // does. The state is read under the lock. - ended := false + // A conn a dispatch goroutine owns (asyncPromoted: every conn that ever served + // an Async route, and on a server with any Async route every conn whose + // request once arrived in pieces) is looked at as io_uring's gate looks at + // it (gateCloseable, celeris#884; here celeris#1165): not while the goroutine + // has input waiting or is inside a handler (promotedGateState), and then + // under detachMu, which a goroutine parked between requests does not hold. + // Closed when the gate has ended its detached stream, or when it is idle: + // no request in progress in its parser. The state is read under the lock; + // closeConn takes the lock itself, and a goroutine that is parked exits on + // the asyncClosed it sets. + ended, promotedIdle := false, false if cs.asyncPromoted { - mu := cs.detachMu - if mu == nil || !mu.TryLock() { - continue - } - ended = cs.detachClosed && cs.protocol == engine.HTTP1 && cs.h1State != nil && cs.h1State.Detached.Load() - mu.Unlock() - if !ended { + ended, promotedIdle = promotedGateState(cs) + if !ended && !promotedIdle { continue } } - if cs.detected { - if cs.protocol != engine.HTTP1 { - continue - } - if h := cs.h1State; h != nil && !ended { - // A request read after the gate is answered close-after (net/http - // does the same for a response written after Shutdown began). - h.CloseAfter = true - if h.HasPendingData() || (h.Detached.Load() && !cs.detachClosed) { + h1Detached := false + if !promotedIdle { + if cs.detected { + if cs.protocol != engine.HTTP1 { continue } + if h := cs.h1State; h != nil && !ended { + // A request read after the gate is answered close-after (net/http + // does the same for a response written after Shutdown began). + h.CloseAfter = true + if h.HasPendingData() || (h.Detached.Load() && !cs.detachClosed) { + continue + } + } + } else if cs.detectN != 0 { + continue } - } else if cs.detectN != 0 { - continue + h1Detached = cs.h1State != nil && cs.h1State.Detached.Load() } - if first && !ended && (cs.h1State == nil || !cs.h1State.Detached.Load()) { + if first && !ended && !h1Detached { // The kernel's receive queue, last before the tests that lead to // the close, so that the window between the question and close(2) // is the few microseconds of those tests (a request that arrives @@ -318,3 +321,36 @@ func (l *Loop) stopDetachedProducers() { mu.Unlock() } } + +// promotedGateState is closeIdleH1's look at a conn whose dispatch goroutine +// owns its parse state (asyncPromoted). ended: the gate has ended its detached +// stream and what was queued has been handed over, so the conn may be closed. +// idle: the goroutine is parked with no input waiting, nothing is detached, the +// conn is still HTTP/1 and its parser holds no part of a request; the conn is +// marked close-after, as the gate marks every conn it meets, since the loop +// cannot know that no request will follow before it closes the conn. Both false +// while the goroutine has work (dispatchHasWork) or holds detachMu inside a +// handler or a guarded write, and for a conn the goroutine switched to HTTP/2. +// Loop thread. +func promotedGateState(cs *connState) (ended, idle bool) { + cs.asyncInMu.Lock() + busy := len(cs.asyncInBuf) > 0 || (cs.asyncRun && !cs.asyncParked && !cs.asyncDetachUnlocked) + cs.asyncInMu.Unlock() + if busy { + return false, false + } + mu := cs.detachMu + if mu == nil || !mu.TryLock() { + return false, false + } + defer mu.Unlock() + h := cs.h1State + if h == nil || cs.protocol != engine.HTTP1 || cs.switchedToH2() { + return false, false + } + if h.Detached.Load() { + return cs.detachClosed, false + } + h.CloseAfter = true + return false, !h.HasPendingData() && h.AtRequestBoundary() +} From 67d62eb9da41b5baafc6fa560e20fb69361d6408 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 22:55:37 +0100 Subject: [PATCH 03/11] tests: a promoted conn with an h2c upgrade pending is not idle for the gate (mutant; Refs #1165) --- internal/engine/epoll/gate_promoted_1165_linux_test.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/internal/engine/epoll/gate_promoted_1165_linux_test.go b/internal/engine/epoll/gate_promoted_1165_linux_test.go index 3156e2daf..48d0518ee 100644 --- a/internal/engine/epoll/gate_promoted_1165_linux_test.go +++ b/internal/engine/epoll/gate_promoted_1165_linux_test.go @@ -98,6 +98,7 @@ func TestGateClosesAPromotedConnOnlyWhenItIsIdle1165(t *testing.T) { panic("premise: the partial head is not pending data") } }, false}, + {"h2c-upgrade-pending", func(r *promoted1165) { r.cs.h1State.UpgradeInfo = &conn.UpgradeInfo{} }, false}, {"switched-to-h2", func(r *promoted1165) { r.cs.asyncInMu.Lock() r.cs.h2Switched = true @@ -126,7 +127,7 @@ func TestGateClosesAPromotedConnOnlyWhenItIsIdle1165(t *testing.T) { } return } - if tc.name == "request-in-progress" && !r.cs.h1State.CloseAfter { + if (tc.name == "request-in-progress" || tc.name == "h2c-upgrade-pending") && !r.cs.h1State.CloseAfter { t.Fatal("a request in progress on a promoted conn is not marked close-after: it would be answered with keep-alive") } if n != 0 || r.l.conns[r.fd] == nil || r.discons.Load() != 0 || r.l.closeCount.Load() != 0 || r.l.activeConns.Load() != 1 { From 8f27797fa7acf879ab7e989122e152480b68cee5 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 22:58:41 +0100 Subject: [PATCH 04/11] tests: an HTTP/1-only server's shutdown must give idle conns their FIN and answer a request in progress, on every engine (red on epoll: the upload; Refs #1165, Refs #1166) --- .../shutdown_h1_gate_1165_linux_test.go | 213 ++++++++++++++++++ 1 file changed, 213 insertions(+) create mode 100644 test/server/shutdown_h1_gate_1165_linux_test.go diff --git a/test/server/shutdown_h1_gate_1165_linux_test.go b/test/server/shutdown_h1_gate_1165_linux_test.go new file mode 100644 index 000000000..65f651ec7 --- /dev/null +++ b/test/server/shutdown_h1_gate_1165_linux_test.go @@ -0,0 +1,213 @@ +//go:build linux + +package celeris_test + +import ( + "bufio" + "context" + "io" + "net" + "net/http" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/goceleris/celeris" +) + +// celeris#1165 and #1166 (found by the review of #1163): on a server whose +// clients are all HTTP/1, so that no loop holds an HTTP/2 conn, the shutdown +// must still do what net/http's Shutdown does and what the pool wait's gate +// does for a loop with an HTTP/2 conn: +// +// - every idle keep-alive conn gets its FIN at once, whatever it has served: +// a sync route, an Async route (a conn with a dispatch goroutine), or a sync +// route whose request arrived in pieces on a server that has an Async route +// (epoll promotes such a conn too), and whether or not other clients are +// slowly reading big responses on the same loop (a send drain waiting for +// them held these conns for the whole budget: #1165); +// - a request in progress when the shutdown begins (a slow upload) is read on +// and answered, with Connection: close, then its conn is closed (#1166). +// +// Workers 2 (the least a native engine takes) and the idle conns are spread over +// both; with downloads, six clients ask for 16 MiB and read nothing, so both +// loops hold one with p = 97%; without them no loop has anything to drain. +func TestShutdownIdleAndInProgressH1ConnsWithoutHTTP2_1165(t *testing.T) { + engines := []struct { + name string + eng celeris.EngineType + }{ + {"std", celeris.Std}, + {"epoll", celeris.Epoll}, + {"io_uring", celeris.IOUring}, + {"adaptive", celeris.Adaptive}, + } + for _, ec := range engines { + for _, downloads := range []bool{true, false} { + name := ec.name + "/no-downloads" + if downloads { + name = ec.name + "/slow-downloads" + } + t.Run(name, func(t *testing.T) { runShutdownH1Gate1165(t, ec.eng, downloads) }) + } + } +} + +func runShutdownH1Gate1165(t *testing.T, eng celeris.EngineType, downloads bool) { + big := strings.Repeat("y", 16<<20) + const upSize = 2 << 20 + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + s, addr, done := startPickedCtx1104(ctx, t, celeris.Config{Engine: eng, Workers: 2, ShutdownTimeout: 20 * time.Second}, func(s *celeris.Server) { + s.GET("/ping", func(c *celeris.Context) error { return c.String(http.StatusOK, "ok") }) + s.GET("/big", func(c *celeris.Context) error { return c.String(http.StatusOK, "%s", big) }) + s.GET("/aping", func(c *celeris.Context) error { return c.String(http.StatusOK, "a") }).Async() + s.POST("/up", func(c *celeris.Context) error { return c.String(http.StatusOK, "got %d", len(c.Body())) }) + }, func(_ *celeris.Server, addr string, done <-chan error) error { return waitDrainOrderReady(addr, done) }) + + type idleConn struct { + name string + c net.Conn + br *bufio.Reader + } + var idle []idleConn + serve := func(c net.Conn, br *bufio.Reader, what string) { + _ = c.SetReadDeadline(time.Now().Add(10 * time.Second)) + resp, err := http.ReadResponse(br, nil) + if err != nil { + t.Fatalf("%s before the shutdown: %v", what, err) + } + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() + } + dial := func() net.Conn { + c, err := net.Dial("tcp", addr) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = c.Close() }) + return c + } + for i := range 4 { + for _, path := range []string{"/ping", "/aping"} { + c := dial() + br := bufio.NewReader(c) + _, _ = io.WriteString(c, "GET "+path+" HTTP/1.1\r\nHost: x\r\n\r\n") + serve(c, br, path) + kind := "sync-route" + if path == "/aping" { + kind = "Async-route" + } + idle = append(idle, idleConn{kind + " " + strconv.Itoa(i), c, br}) + } + // A sync route whose head arrives in two writes 30 ms apart. + c := dial() + _ = c.(*net.TCPConn).SetNoDelay(true) + _, _ = io.WriteString(c, "GET /ping HTTP/1.1\r\nHo") + time.Sleep(30 * time.Millisecond) + _, _ = io.WriteString(c, "st: x\r\n\r\n") + br := bufio.NewReader(c) + serve(c, br, "split /ping") + idle = append(idle, idleConn{"split-head " + strconv.Itoa(i), c, br}) + } + var dls []net.Conn + if downloads { + for range 6 { + dl := dial() + _, _ = io.WriteString(dl, "GET /big HTTP/1.1\r\nHost: x\r\n\r\n") + dls = append(dls, dl) + } + } + // A request in progress: the head and the first 64 KiB of a 2 MiB body. + up := dial() + if _, err := io.WriteString(up, "POST /up HTTP/1.1\r\nHost: x\r\nContent-Length: "+strconv.Itoa(upSize)+"\r\n\r\n"); err != nil { + t.Fatal(err) + } + if _, err := up.Write(make([]byte, 64<<10)); err != nil { + t.Fatal(err) + } + time.Sleep(500 * time.Millisecond) // the downloads are in the engine; the upload's first bytes are read + + begin := time.Now() + shutErr := make(chan error, 1) + go func() { + sctx, scancel := context.WithTimeout(context.Background(), 20*time.Second) + defer scancel() + shutErr <- s.Shutdown(sctx) + }() + + // Every idle conn is read at the same time, each with its own wait. + const finBy = 1500 * time.Millisecond + type result struct { + name string + err error + at time.Duration + } + res := make(chan result, len(idle)) + var wg sync.WaitGroup + for _, k := range idle { + wg.Add(1) + go func() { + defer wg.Done() + _ = k.c.SetReadDeadline(begin.Add(finBy)) + _, err := k.br.ReadByte() + res <- result{k.name, err, time.Since(begin)} + }() + } + + // The rest of the upload, sent once the gate's grace (250 ms) has passed. + time.Sleep(400 * time.Millisecond) + _ = up.SetWriteDeadline(time.Now().Add(3 * time.Second)) + if _, err := up.Write(make([]byte, upSize-(64<<10))); err != nil { + t.Errorf("the rest of the upload, sent %v into the shutdown: %v: a request in progress was cut (celeris#1166)", time.Since(begin).Round(time.Millisecond), err) + } else { + _ = up.SetReadDeadline(time.Now().Add(5 * time.Second)) + upBr := bufio.NewReader(up) + resp, err := http.ReadResponse(upBr, nil) + if err != nil { + t.Errorf("the response to the upload: %v: a request in progress was not answered (celeris#1166)", err) + } else { + b, _ := io.ReadAll(resp.Body) + if want := "got " + strconv.Itoa(upSize); string(b) != want { + t.Errorf("the response to the upload: %q, want %q", b, want) + } + if !resp.Close { + t.Errorf("the response to the upload, written after the shutdown began, has no Connection: close") + } + if _, err := upBr.ReadByte(); err != io.EOF { + t.Errorf("after the response to the upload the conn read %v, want a FIN", err) + } + } + } + + wg.Wait() + close(res) + var late []string + for r := range res { + if r.err != io.EOF { + late = append(late, r.name+": "+r.err.Error()+" at "+r.at.Round(time.Millisecond).String()) + } + } + if len(late) > 0 { + t.Errorf("%d of %d idle conns had no FIN %v into the shutdown (celeris#1165):\n %s", len(late), len(idle), finBy, strings.Join(late, "\n ")) + } + // The downloaders go; the shutdown ends well before its budget. + for _, c := range dls { + _ = c.Close() + } + for _, k := range idle { + _ = k.c.Close() + } + select { + case <-shutErr: + case <-time.After(25 * time.Second): + t.Error("Shutdown did not return") + } + select { + case <-done: + case <-time.After(25 * time.Second): + t.Error("the server did not stop") + } +} From 8e90d4a1234be66b38ea4f6e398129b3cbc1094d Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 23:06:12 +0100 Subject: [PATCH 05/11] epoll: the shutdown waits, as for HTTP/2 pool handlers, for an HTTP/1 request in progress on a loop with no HTTP/2 conn, and the gate reads on a promoted conn that has one (v1.6.0, Refs #1166) --- .../epoll/gate_promoted_1165_linux_test.go | 121 ++++++++++++++++ internal/engine/epoll/loop.go | 6 +- internal/engine/epoll/poolwait_gate_885.go | 129 ++++++++++++++++-- 3 files changed, 244 insertions(+), 12 deletions(-) diff --git a/internal/engine/epoll/gate_promoted_1165_linux_test.go b/internal/engine/epoll/gate_promoted_1165_linux_test.go index 48d0518ee..1d228bc23 100644 --- a/internal/engine/epoll/gate_promoted_1165_linux_test.go +++ b/internal/engine/epoll/gate_promoted_1165_linux_test.go @@ -159,3 +159,124 @@ func TestGateHoldsAPromotedConnWithAnUnreadRequest1165(t *testing.T) { t.Fatal("the request was not handed to the dispatch goroutine's input") } } + +// celeris#1166. The gate reads on, and answers, a request in progress on a +// promoted conn too (a slow upload on a server with any Async route promotes the +// conn whose request arrives in pieces), and refuses what an idle one sends. +func TestRefusesRecvReadsOnAPromotedConnOnlyWhileARequestIsInProgress1166(t *testing.T) { + for _, tc := range []struct { + name string + mutate func(r *promoted1165) + refuse bool + }{ + {"idle", func(*promoted1165) {}, true}, + {"partial-request-in-the-parser", func(r *promoted1165) { + _ = conn.ProcessH1(context.Background(), []byte("POST /x HTTP/1.1\r\nHost: x\r\nContent-Length: 100\r\n\r\nabc"), r.cs.h1State, r.l.handler, func([]byte) {}) + }, false}, + {"input-waiting-for-the-goroutine", func(r *promoted1165) { + r.cs.asyncInMu.Lock() + r.cs.asyncInBuf = append(r.cs.asyncInBuf, 'x') + r.cs.asyncInMu.Unlock() + }, false}, + {"inside-a-handler", func(r *promoted1165) { + r.cs.asyncInMu.Lock() + r.cs.asyncParked = false + r.cs.asyncInMu.Unlock() + }, false}, + {"switched-to-h2", func(r *promoted1165) { + r.cs.asyncInMu.Lock() + r.cs.h2Switched = true + r.cs.asyncInMu.Unlock() + }, false}, + } { + t.Run(tc.name, func(t *testing.T) { + r := newPromoted1165(t) + tc.mutate(r) + if got := refusesRecv(r.cs); got != tc.refuse { + t.Fatalf("refusesRecv = %v, want %v", got, tc.refuse) + } + }) + } +} + +// celeris#1166. h1InProgress is what keeps a loop with no HTTP/2 conn waiting, +// after its context is cancelled, for the request a client is in the middle of +// sending (or the handler answering it); every case below is a state that must +// not end the wait, and the idle ones must. +func TestH1InProgressDecidesTheWaitOfALoopWithoutHTTP21166(t *testing.T) { + partial := func(r *promoted1165) { + _ = conn.ProcessH1(context.Background(), []byte("POST /x HTTP/1.1\r\nHost: x\r\nContent-Length: 100\r\n\r\nabc"), r.cs.h1State, r.l.handler, func([]byte) {}) + } + plain := func(f func(r *promoted1165)) func(r *promoted1165) { + return func(r *promoted1165) { r.cs.asyncPromoted = false; f(r) } + } + cases := []struct { + name string + mutate func(r *promoted1165) + inProgress bool + }{ + {"plain-idle", plain(func(*promoted1165) {}), false}, + {"promoted-idle", func(*promoted1165) {}, false}, + {"plain-partial-request", plain(partial), true}, + {"promoted-partial-request", partial, true}, + {"plain-h2c-upgrade-pending", plain(func(r *promoted1165) { r.cs.h1State.UpgradeInfo = &conn.UpgradeInfo{} }), true}, + {"undetected-with-some-bytes", plain(func(r *promoted1165) { r.cs.detected, r.cs.detectN = false, 3 }), true}, + {"undetected-with-none", plain(func(r *promoted1165) { r.cs.detected, r.cs.detectN = false, 0 }), false}, + {"promoted-input-waiting", func(r *promoted1165) { + r.cs.asyncInMu.Lock() + r.cs.asyncInBuf = append(r.cs.asyncInBuf, 'x') + r.cs.asyncInMu.Unlock() + }, true}, + {"promoted-inside-a-handler", func(r *promoted1165) { + r.cs.asyncInMu.Lock() + r.cs.asyncParked = false + r.cs.asyncInMu.Unlock() + }, true}, + {"promoted-parked-with-a-guarded-write", func(r *promoted1165) { r.cs.detachMu.Lock() }, false}, + {"plain-detached-stream", plain(func(r *promoted1165) { + partial(r) + r.cs.h1State.Detached.Store(true) + }), false}, + {"plain-http2", plain(func(r *promoted1165) { r.cs.protocol = engine.H2C }), false}, + {"promoted-switched-to-h2", func(r *promoted1165) { + partial(r) + r.cs.asyncInMu.Lock() + r.cs.h2Switched = true + r.cs.asyncInMu.Unlock() + }, false}, + {"hijacked", func(r *promoted1165) { + partial(r) + r.cs.hijacked.Store(true) + }, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + r := newPromoted1165(t) + tc.mutate(r) + locked := tc.name == "promoted-parked-with-a-guarded-write" + got := r.l.h1InProgress() + if locked { + r.cs.detachMu.Unlock() + } + if got != tc.inProgress { + t.Fatalf("h1InProgress = %v, want %v", got, tc.inProgress) + } + }) + } + t.Run("request-unread-in-the-kernel-first-call-only", func(t *testing.T) { + r := newPromoted1165(t) + r.cs.asyncPromoted = false + if _, err := unix.Write(r.peer, []byte("GET /y HTTP/1.1\r\nHost: x\r\n\r\n")); err != nil { + t.Fatal(err) + } + if !r.l.h1InProgress() { + t.Fatal("a request in the kernel's receive queue, not read yet, is not in progress on the first call") + } + // The pacing clock holds the answer between polls; a second asker after it + // does not ask the kernel again (the loop reads the bytes as events). + r.l.h1InProgressNext = 0 + if r.l.h1InProgress() { + t.Fatal("the second call asked the kernel again") + } + }) +} diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go index e6337f832..01702be59 100644 --- a/internal/engine/epoll/loop.go +++ b/internal/engine/epoll/loop.go @@ -223,6 +223,10 @@ type Loop struct { // walk the conns again. Loop thread. refuseRecv bool producersStopNext int64 + // h1InProgressNext and h1InProgressAsked pace and mark the shutdown's look + // for HTTP/1 requests in progress (poolWaitSettled, celeris#1166). + h1InProgressNext int64 + h1InProgressAsked bool // transplantInFlight counts connections this loop has detached for a // transplant whose hand-off is not finished yet — the deferred async @@ -556,7 +560,7 @@ func (l *Loop) run(ctx context.Context) { // 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() { + if l.poolWaitSettled() { l.shutdown() return } diff --git a/internal/engine/epoll/poolwait_gate_885.go b/internal/engine/epoll/poolwait_gate_885.go index c669b48b0..5f6124ea7 100644 --- a/internal/engine/epoll/poolwait_gate_885.go +++ b/internal/engine/epoll/poolwait_gate_885.go @@ -72,9 +72,10 @@ const gateClosesPerPass = 64 // detected yet, which io_uring's gateReadsOn mirrors: a request in progress is // read on, and answered, as std's Shutdown answers the request of an active // conn (headers in pieces, a body still arriving, the bytes of a protocol -// decision begun). A conn whose -// dispatch goroutine owns its parse state (asyncPromoted) is refused as before -// while it is HTTP/1: its state is not the loop's to read. Loop thread. +// decision begun). A conn whose dispatch goroutine owns its parse state +// (asyncPromoted) is looked at the same way, under its locks (celeris#1166): +// read on while a request is in progress on it, refused when it is idle. Loop +// thread. func refusesRecv(cs *connState) bool { if cs.gateRead { // The gate's first pass found the bytes of a request in the kernel's @@ -87,14 +88,19 @@ func refusesRecv(cs *connState) bool { } if cs.asyncPromoted { // protocol, h1State and h2State are the dispatch goroutine's, which - // writes them in switchToH2Local under detachMu: not read here. The - // conn is HTTP/1 until the goroutine sets h2Switched (under asyncInMu, - // which is held for an instant, never across a handler, so waiting for - // it cannot stall the pass), and an HTTP/1 conn it owns is refused, as - // before; one it has switched is HTTP/2 and is served (celeris#885 - // item 28). Not refused on asyncPromoted alone: the pool wait is for - // the HTTP/2 conns. - return !cs.switchedToH2() + // writes them in switchToH2Local under detachMu: not read here but + // under the lock (promotedRequestInProgress). The conn is HTTP/1 until the + // goroutine sets h2Switched (under asyncInMu, which is held for an + // instant, never across a handler, so waiting for it cannot stall the + // pass); one it has switched is HTTP/2 and is served (celeris#885 item + // 28). An HTTP/1 conn it owns is read on while a request is in progress + // on it (input waiting for the goroutine, a handler running, a parser + // holding part of a request), as io_uring's gateReadsOn does for a + // promoted conn (celeris#1166), and refused when it is idle. + if cs.switchedToH2() { + return false + } + return !promotedRequestInProgress(cs) } if cs.protocol != engine.HTTP1 { return false @@ -354,3 +360,104 @@ func promotedGateState(cs *connState) (ended, idle bool) { h.CloseAfter = true return false, !h.HasPendingData() && h.AtRequestBoundary() } + +// poolWaitSettled reports whether a loop whose context is cancelled may go on to +// shutdown(): the HTTP/2 pool handlers are done (h2PoolSettled) and no HTTP/1 +// request is in progress on it (celeris#1166). The gate answers a request in +// progress (a slow upload, headers in pieces, a body arriving, a request not +// read yet) with Connection: close, as std's Shutdown does; it ran only on a loop +// that had an HTTP/2 conn to wait for, and on every other loop shutdown() cut the +// request with its conn, unanswered. A loop with such a request now waits for +// it as it waits for the pool handlers, from h2DrainStart and for as long as +// sendDrainWait gives, and runs the same gate meanwhile (poolWaitGate). Loop +// thread. +func (l *Loop) poolWaitSettled() bool { + if !l.h2PoolSettled() { + return false + } + if !l.h1InProgress() { + return true + } + if l.h2DrainStart.IsZero() { + l.h2DrainStart = time.Now() + } + _, more := l.sendDrainWait(l.h2DrainStart) + return !more +} + +// h1InProgress reports whether an HTTP/1 conn of the loop has a request in +// progress: part of one in its parser (headers in pieces, a body arriving, an +// upgrade pending), the protocol decision of its first bytes begun, a dispatch +// goroutine with input waiting or inside a handler, and, on the first call of the +// shutdown, bytes in the kernel's receive queue of a conn that holds nothing (a +// request that reached the kernel before the cancel and that the loop has not +// read: the epoll event that reports it is still to come, and the gate's first +// pass holds it as celeris#1104 does). Detached streams and HTTP/2 conns are not +// requests in progress here (the gate ends the first, h2PoolSettled waits for the +// second). It is asked at most once per h2PoolDrainPollMs while it answers yes. +// Loop thread. +func (l *Loop) h1InProgress() bool { + now := time.Now().UnixNano() + if l.h1InProgressNext != 0 && now < l.h1InProgressNext { + return true + } + first := !l.h1InProgressAsked + l.h1InProgressAsked = true + for i := len(l.liveConns) - 1; i >= 0; i-- { + cs := l.liveConns[i] + if cs.hijacked.Load() { + continue + } + if cs.asyncPromoted { + if promotedRequestInProgress(cs) { + l.h1InProgressNext = now + int64(h2PoolDrainPollMs*time.Millisecond) + return true + } + continue + } + if cs.detected { + if cs.protocol != engine.HTTP1 { + continue + } + if h := cs.h1State; h != nil && !h.Detached.Load() && (h.HasPendingData() || !h.AtRequestBoundary()) { + l.h1InProgressNext = now + int64(h2PoolDrainPollMs*time.Millisecond) + return true + } + } else if cs.detectN != 0 { + l.h1InProgressNext = now + int64(h2PoolDrainPollMs*time.Millisecond) + return true + } + if first && !csWritePending(cs) && !(cs.h1State != nil && cs.h1State.Detached.Load()) { + if q, err := unix.IoctlGetInt(cs.fd, unix.TIOCINQ); err == nil && q != 0 { + l.h1InProgressNext = now + int64(h2PoolDrainPollMs*time.Millisecond) + return true + } + } + } + l.h1InProgressNext = 0 + return false +} + +// promotedRequestInProgress is h1InProgress's look at a conn whose dispatch +// goroutine owns its parse state: true while input waits for the goroutine, it is +// inside a handler (not one that has detached and streams inline), or its parser +// holds part of a request. A lock held by a goroutine that is parked is a guarded +// write of a stream, not a request. Loop thread. +func promotedRequestInProgress(cs *connState) bool { + cs.asyncInMu.Lock() + busy := len(cs.asyncInBuf) > 0 || (cs.asyncRun && !cs.asyncParked && !cs.asyncDetachUnlocked) + cs.asyncInMu.Unlock() + if busy { + return true + } + mu := cs.detachMu + if mu == nil || !mu.TryLock() { + return false + } + defer mu.Unlock() + h := cs.h1State + if h == nil || cs.protocol != engine.HTTP1 || cs.switchedToH2() || h.Detached.Load() { + return false + } + return h.HasPendingData() || !h.AtRequestBoundary() +} From 18318d30588d5847eec1ed0a1ad677542b17b0b6 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 23:07:23 +0100 Subject: [PATCH 06/11] tests: the wait for a request in progress is paced, bounded and ends the wait; the h2 conn is not one (mutants; Refs #1166) --- .../epoll/gate_promoted_1165_linux_test.go | 37 ++++++++++++++++++- 1 file changed, 36 insertions(+), 1 deletion(-) diff --git a/internal/engine/epoll/gate_promoted_1165_linux_test.go b/internal/engine/epoll/gate_promoted_1165_linux_test.go index 1d228bc23..4ac10c240 100644 --- a/internal/engine/epoll/gate_promoted_1165_linux_test.go +++ b/internal/engine/epoll/gate_promoted_1165_linux_test.go @@ -237,7 +237,10 @@ func TestH1InProgressDecidesTheWaitOfALoopWithoutHTTP21166(t *testing.T) { partial(r) r.cs.h1State.Detached.Store(true) }), false}, - {"plain-http2", plain(func(r *promoted1165) { r.cs.protocol = engine.H2C }), false}, + {"plain-http2", plain(func(r *promoted1165) { + partial(r) + r.cs.protocol = engine.H2C + }), false}, {"promoted-switched-to-h2", func(r *promoted1165) { partial(r) r.cs.asyncInMu.Lock() @@ -280,3 +283,35 @@ func TestH1InProgressDecidesTheWaitOfALoopWithoutHTTP21166(t *testing.T) { } }) } + +// The wait is paced and bounded: a yes holds until the next poll time, and the +// wait ends at sendDrainWait's end whatever the clients are doing. +func TestPoolWaitSettledWaitsForARequestInProgressAndIsBounded1166(t *testing.T) { + r := newPromoted1165(t) + r.cs.asyncPromoted = false + _ = conn.ProcessH1(context.Background(), []byte("POST /x HTTP/1.1\r\nHost: x\r\nContent-Length: 100\r\n\r\nabc"), r.cs.h1State, r.l.handler, func([]byte) {}) + if r.l.poolWaitSettled() { + t.Fatal("the loop is settled with a request in progress") + } + if r.l.h2DrainStart.IsZero() { + t.Fatal("the wait has no start: its bound would never come") + } + // The request ends; the pacing holds the answer until the next poll time. + r.cs.h1State = conn.NewH1State() + if r.l.poolWaitSettled() { + t.Fatal("the loop is settled before the pacing's next poll") + } + r.l.h1InProgressNext = 0 + if !r.l.poolWaitSettled() { + t.Fatal("the loop is not settled with nothing in progress") + } + // A request that never ends: the wait ends at its bound (no Shutdown budget: + // the floor, 250 ms, from the start). + r.cs.h1State = conn.NewH1State() + _ = conn.ProcessH1(context.Background(), []byte("POST /x HTTP/1.1\r\nHost: x\r\nContent-Length: 100\r\n\r\nabc"), r.cs.h1State, r.l.handler, func([]byte) {}) + r.l.h1InProgressNext = 0 + r.l.h2DrainStart = time.Now().Add(-2 * shutdownSendDrainFloor) + if !r.l.poolWaitSettled() { + t.Fatal("the wait for a request in progress is not bounded") + } +} From e2501e106b6da91ad1a48ec481549d39d2a57e26 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 23:08:49 +0100 Subject: [PATCH 07/11] io_uring: a worker with no sends pending waits for an HTTP/1 request in progress before it shuts down (v1.6.0, Refs #1166) --- internal/engine/epoll/poolwait_gate_885.go | 12 +++-- internal/engine/iouring/shutdown_gate_884.go | 46 +++++++++++++++++++- internal/engine/iouring/worker.go | 2 +- 3 files changed, 53 insertions(+), 7 deletions(-) diff --git a/internal/engine/epoll/poolwait_gate_885.go b/internal/engine/epoll/poolwait_gate_885.go index 5f6124ea7..5681b122e 100644 --- a/internal/engine/epoll/poolwait_gate_885.go +++ b/internal/engine/epoll/poolwait_gate_885.go @@ -51,10 +51,14 @@ import ( // (its client gets a FIN and reconnects), whatever kind of route it is on. A // conn a dispatch goroutine owns (asyncPromoted: on a server with any Async // route every conn whose read leaves a partial request is promoted, whatever its -// route, and a conn on an Async route is) is otherwise left to the end of the -// wait, with a request in progress on it unread: the documented behaviour and -// a follow-up (io_uring's gate, shutdown_gate_884.go, looks inside such a conn -// under the goroutine's own locks and closes it when idle). +// route, and a conn on an Async route is) is looked at under the goroutine's own +// locks, as io_uring's gate does (shutdown_gate_884.go; celeris#1165, #1166): +// closed when idle, read on and answered when a request is in progress on it. +// +// A loop with no HTTP/2 conn runs the same gate for as long as an HTTP/1 +// request is in progress on it (poolWaitSettled), and closes its idle conns at +// the start of shutdown (celeris#1104), so the gate's behaviour does not depend +// on whether the server has an HTTP/2 client. // poolWaitGrace is how long into the wait for the HTTP/2 pool handlers the // gate stays open: io_uring's shutdownSendDrainNanos, and this engine's diff --git a/internal/engine/iouring/shutdown_gate_884.go b/internal/engine/iouring/shutdown_gate_884.go index d8803be25..a7793fc01 100644 --- a/internal/engine/iouring/shutdown_gate_884.go +++ b/internal/engine/iouring/shutdown_gate_884.go @@ -42,8 +42,7 @@ import ( // asyncInMu says whether input is waiting for it or it is inside a handler, // and detachMu, taken with TryLock as every worker-thread site does (celeris#593), // is free only while the goroutine is between requests, so h1State is read under -// it. Epoll cannot yet look inside such a conn (its follow-up F2); this engine -// can. +// it. Epoll looks inside such a conn the same way (celeris#1165, #1166). // gateClosesPerPass caps the conns closeIdleH1 closes in one pass, so that a // worker with thousands of idle conns does not stop serving the conns the gate @@ -345,3 +344,46 @@ func (w *Worker) closeIdlePaced(now int64) { w.producersStopNext = now + int64(h2PoolDrainPoll) w.closeIdleH1() } + +// h1RequestInProgress reports whether an HTTP/1 conn of the worker has a request +// in progress: part of one in its parser, the bytes of a protocol decision +// begun, or a dispatch goroutine with input waiting or inside a handler +// (promotedInProgress). With sends pending the send drain keeps the worker +// turning and the gate reads such a request on and answers it; with none, the +// worker used to shut down at once and cut the request with its conn, unanswered +// (celeris#1166, as epoll's loop with no HTTP/2 conn did). It now waits for it, +// bounded as the drain is (drainEnd), with the same gate meanwhile. A request +// still in the kernel's queue or a recv CQE not reaped is not seen here (the +// gate's first pass holds those at the floor, when the drain is still on). +// Worker thread. +func (w *Worker) h1RequestInProgress() bool { + for _, fd := range w.liveConns { + cs := w.conns[fd] + if cs == nil || cs.closing { + continue + } + if !cs.detected { + if len(cs.detectAccum) > 0 { + return true + } + continue + } + if engine.Protocol(cs.protocol.Load()) != engine.HTTP1 { + continue + } + if cs.asyncPromoted.Load() { + if w.promotedInProgress(cs) { + return true + } + continue + } + h := cs.h1State + if h == nil || h.Detached.Load() { + continue + } + if cs.recvIntoBody || h.HasPendingData() || !h.AtRequestBoundary() { + return true + } + } + return false +} diff --git a/internal/engine/iouring/worker.go b/internal/engine/iouring/worker.go index 5f9b5b600..dac66cda5 100644 --- a/internal/engine/iouring/worker.go +++ b/internal/engine/iouring/worker.go @@ -1628,7 +1628,7 @@ func (w *Worker) run(ctx context.Context) { if w.shutdownDrainStart == 0 { w.shutdownDrainStart = now } - if !w.hasPendingSends() { + if !w.hasPendingSends() && !w.h1RequestInProgress() { w.shutdown() return } From 82f1b0a763b6dab469d5e62d6908e4c087ddc77c Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 23:16:12 +0100 Subject: [PATCH 08/11] tests: the shutdown gate test expects the idle Async-route conn to get its FIN, as the others do (Refs #1165) --- test/server/shutdown_h2_pool_linux_test.go | 33 +++++++++++++--------- 1 file changed, 19 insertions(+), 14 deletions(-) diff --git a/test/server/shutdown_h2_pool_linux_test.go b/test/server/shutdown_h2_pool_linux_test.go index 0c330a9da..34b8f5904 100644 --- a/test/server/shutdown_h2_pool_linux_test.go +++ b/test/server/shutdown_h2_pool_linux_test.go @@ -1165,10 +1165,11 @@ const ( // the request of an active conn, with Connection: close, then its conn is // closed. On a server with any Async route a conn whose read leaves a partial // request is promoted whatever its route, its state belongs to the dispatch -// goroutine, and the request is not read: not asserted (a known gap, noted in -// the pull request); -// - a conn on an Async route (promoted) that is idle is not closed by the gate: -// it is left to the end of the wait, which is what is pinned. (A detached +// goroutine; the gate reads such a request on and answers it since +// celeris#1166 (asserted by TestShutdownIdleAndInProgressH1ConnsWithoutHTTP2_1165); +// - a conn on an Async route (promoted) that is idle is closed by the gate +// as the others are: it is looked at under its dispatch goroutine's locks +// (celeris#1165; it used to be left to the end of the wait). (A detached // stream on an Async route is closed once the gate has ended it: the other // test.) func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) { @@ -1323,7 +1324,7 @@ func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) { at time.Duration } var finWG sync.WaitGroup - finRes := make(chan finResult, len(idle)+len(silent)) + finRes := make(chan finResult, len(idle)+len(silent)+1) expectFIN := func(name string, c net.Conn, br *bufio.Reader) { finWG.Add(1) go func() { @@ -1344,6 +1345,11 @@ func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) { for i, c := range silent { expectFIN("connection with no byte sent "+strconv.Itoa(i), c, nil) } + if promoted != nil { + // A conn that served an Async route has a dispatch goroutine; the + // gate looks inside it and closes it when it is idle (celeris#1165). + expectFIN("idle Async-route conn", promoted, promotedBr) + } finWG.Wait() close(finRes) var latest time.Duration @@ -1354,15 +1360,7 @@ func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) { latest = r.at } } - t.Logf("the last of the %d idle and silent conns had its FIN %v into the shutdown", len(idle)+len(silent), latest.Round(time.Millisecond)) - if promoted != nil { - _ = promoted.SetReadDeadline(begin.Add(2 * time.Second)) - if _, err := promotedBr.ReadByte(); err == nil { - t.Errorf("the conn an Async route promoted read a byte after the shutdown began") - } else if ne, ok := err.(net.Error); !ok || !ne.Timeout() { - t.Errorf("the conn an Async route promoted was closed (%v) at %v: it has a dispatch goroutine, and the gate leaves it to the end of the wait", err, time.Since(begin).Round(time.Millisecond)) - } - } + t.Logf("the last of the %d idle, silent and promoted conns had its FIN %v into the shutdown", len(idle)+len(silent)+btoi1165(promoted != nil), latest.Round(time.Millisecond)) if up != nil { // The rest of the body arrives after the gate has passed. _ = up.SetWriteDeadline(time.Now().Add(3 * time.Second)) @@ -1442,3 +1440,10 @@ func waitEngineBusy1104(t *testing.T, srv *h2PoolServer759, base celeris.EngineM t.Fatalf("the engine counted +%d conns and +%d requests over its baseline, want +%d and +%d before the shutdown begins", m.ActiveConnections-base.ActiveConnections, m.RequestCount-base.RequestCount, conns, reqs) } + +func btoi1165(b bool) int { + if b { + return 1 + } + return 0 +} From 8faabd7bffddb43df95a0e26f334d60fa639c2ec Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Thu, 29 Oct 2026 00:05:02 +0100 Subject: [PATCH 09/11] tests: the upload of the shutdown gate e2e test is completed after every idle conn has had its FIN, so the gate's pass has run (Refs #1166) --- .../shutdown_h1_gate_1165_linux_test.go | 29 ++++++++++--------- 1 file changed, 16 insertions(+), 13 deletions(-) diff --git a/test/server/shutdown_h1_gate_1165_linux_test.go b/test/server/shutdown_h1_gate_1165_linux_test.go index 65f651ec7..ea385fbfb 100644 --- a/test/server/shutdown_h1_gate_1165_linux_test.go +++ b/test/server/shutdown_h1_gate_1165_linux_test.go @@ -157,8 +157,22 @@ func runShutdownH1Gate1165(t *testing.T, eng celeris.EngineType, downloads bool) }() } - // The rest of the upload, sent once the gate's grace (250 ms) has passed. - time.Sleep(400 * time.Millisecond) + wg.Wait() + close(res) + var late []string + for r := range res { + if r.err != io.EOF { + late = append(late, r.name+": "+r.err.Error()+" at "+r.at.Round(time.Millisecond).String()) + } + } + if len(late) > 0 { + t.Errorf("%d of %d idle conns had no FIN %v into the shutdown (celeris#1165):\n %s", len(late), len(idle), finBy, strings.Join(late, "\n ")) + } + // The rest of the upload, sent once every idle conn has had its FIN: the + // gate's pass (grace 250 ms, then it marks each conn it meets close-after + // and closes the idle ones) has run on both loops by then, so the response is + // written after it, whatever the load. (A response written before the pass + // is a keep-alive one: the request was in flight when the shutdown began.) _ = up.SetWriteDeadline(time.Now().Add(3 * time.Second)) if _, err := up.Write(make([]byte, upSize-(64<<10))); err != nil { t.Errorf("the rest of the upload, sent %v into the shutdown: %v: a request in progress was cut (celeris#1166)", time.Since(begin).Round(time.Millisecond), err) @@ -182,17 +196,6 @@ func runShutdownH1Gate1165(t *testing.T, eng celeris.EngineType, downloads bool) } } - wg.Wait() - close(res) - var late []string - for r := range res { - if r.err != io.EOF { - late = append(late, r.name+": "+r.err.Error()+" at "+r.at.Round(time.Millisecond).String()) - } - } - if len(late) > 0 { - t.Errorf("%d of %d idle conns had no FIN %v into the shutdown (celeris#1165):\n %s", len(late), len(idle), finBy, strings.Join(late, "\n ")) - } // The downloaders go; the shutdown ends well before its budget. for _, c := range dls { _ = c.Close() From a8a6df90da24f14280e1ecd9344ab649e5cd8950 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Thu, 29 Oct 2026 00:52:54 +0100 Subject: [PATCH 10/11] epoll: the shutdown's look for a request in progress does not read a detached stream's write state (a data race with its writer); the e2e test waits for the engine to accept every conn (Refs #1165, Refs #1166) --- .../epoll/gate_promoted_1165_linux_test.go | 39 +++++++++++++++++++ internal/engine/epoll/poolwait_gate_885.go | 6 ++- .../shutdown_h1_gate_1165_linux_test.go | 29 +++++++++++++- 3 files changed, 72 insertions(+), 2 deletions(-) diff --git a/internal/engine/epoll/gate_promoted_1165_linux_test.go b/internal/engine/epoll/gate_promoted_1165_linux_test.go index 4ac10c240..02574ebe6 100644 --- a/internal/engine/epoll/gate_promoted_1165_linux_test.go +++ b/internal/engine/epoll/gate_promoted_1165_linux_test.go @@ -315,3 +315,42 @@ func TestPoolWaitSettledWaitsForARequestInProgressAndIsBounded1166(t *testing.T) t.Fatal("the wait for a request in progress is not bounded") } } + +// A detached stream's writer goroutine owns the conn's write state (it appends +// to writeBuf under detachMu through the guarded write hook); h1InProgress, on +// the loop thread, must not read that state (csWritePending) of a detached +// conn: CI's -race saw it in TestStalledSSEEndsWithShutdown855. The writer here +// does what the hook does, and the first call asks about the kernel's queue as +// the shutdown's first call does. Run it under -race. +func TestH1InProgressDoesNotReadADetachedStreamsWriteState1166(t *testing.T) { + r := newPromoted1165(t) + r.cs.asyncPromoted = false + r.cs.h1State.Detached.Store(true) + stop := make(chan struct{}) + done := make(chan struct{}) + go func() { + defer close(done) + for { + select { + case <-stop: + return + default: + } + r.cs.detachMu.Lock() + r.cs.writeBuf = append(r.cs.writeBuf[:0], 'x') + r.cs.writePos = 0 + r.cs.bodyBuf = nil + r.cs.detachMu.Unlock() + } + }() + for range 2000 { + r.l.h1InProgressAsked = false // the first call, which asks the kernel + r.l.h1InProgressNext = 0 + if r.l.h1InProgress() { + t.Error("a detached stream is a request in progress") + break + } + } + close(stop) + <-done +} diff --git a/internal/engine/epoll/poolwait_gate_885.go b/internal/engine/epoll/poolwait_gate_885.go index 5681b122e..f6012758e 100644 --- a/internal/engine/epoll/poolwait_gate_885.go +++ b/internal/engine/epoll/poolwait_gate_885.go @@ -431,7 +431,11 @@ func (l *Loop) h1InProgress() bool { l.h1InProgressNext = now + int64(h2PoolDrainPollMs*time.Millisecond) return true } - if first && !csWritePending(cs) && !(cs.h1State != nil && cs.h1State.Detached.Load()) { + // A detached stream is not asked: its writer goroutine owns the conn's + // write state (writeBuf and the fields csWritePending reads) under its own + // lock, which the loop does not take here, so the test for Detached comes + // before the one for the queue. + if detached := cs.h1State != nil && cs.h1State.Detached.Load(); first && !detached && !csWritePending(cs) { if q, err := unix.IoctlGetInt(cs.fd, unix.TIOCINQ); err == nil && q != 0 { l.h1InProgressNext = now + int64(h2PoolDrainPollMs*time.Millisecond) return true diff --git a/test/server/shutdown_h1_gate_1165_linux_test.go b/test/server/shutdown_h1_gate_1165_linux_test.go index ea385fbfb..1dde8d79f 100644 --- a/test/server/shutdown_h1_gate_1165_linux_test.go +++ b/test/server/shutdown_h1_gate_1165_linux_test.go @@ -67,6 +67,7 @@ func runShutdownH1Gate1165(t *testing.T, eng celeris.EngineType, downloads bool) s.POST("/up", func(c *celeris.Context) error { return c.String(http.StatusOK, "got %d", len(c.Body())) }) }, func(_ *celeris.Server, addr string, done <-chan error) error { return waitDrainOrderReady(addr, done) }) + base := settledEngineMetrics1104Server(s) type idleConn struct { name string c net.Conn @@ -128,7 +129,18 @@ func runShutdownH1Gate1165(t *testing.T, eng celeris.EngineType, downloads bool) if _, err := up.Write(make([]byte, 64<<10)); err != nil { t.Fatal(err) } - time.Sleep(500 * time.Millisecond) // the downloads are in the engine; the upload's first bytes are read + // The engine has accepted every conn before the shutdown begins: a conn still + // in the accept queue (a loop busy formatting six 16 MiB responses under -race + // has not got to the upload's) is reset by the listener's close, which says + // nothing about the gate. Then the downloads are in the engine and the upload's + // first bytes are read. + want := base + int64(len(idle)+len(dls)+1) + for dl := time.Now().Add(20 * time.Second); s.EngineInfo().Metrics.ActiveConnections < want; time.Sleep(10 * time.Millisecond) { + if time.Now().After(dl) { + t.Fatalf("the engine counts %d conns, want %d before the shutdown begins", s.EngineInfo().Metrics.ActiveConnections, want) + } + } + time.Sleep(300 * time.Millisecond) begin := time.Now() shutErr := make(chan error, 1) @@ -214,3 +226,18 @@ func runShutdownH1Gate1165(t *testing.T, eng celeris.EngineType, downloads bool) t.Error("the server did not stop") } } + +// settledEngineMetrics1104Server is the engine's open conns once they have not +// changed for 100 ms (the readiness probe's conn goes when its client does). +func settledEngineMetrics1104Server(s *celeris.Server) int64 { + prev := s.EngineInfo().Metrics.ActiveConnections + for range 100 { + time.Sleep(100 * time.Millisecond) + m := s.EngineInfo().Metrics.ActiveConnections + if m == prev { + return m + } + prev = m + } + return prev +} From a04e32e267c78f7d43cd7a9082100736062a13a3 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Thu, 29 Oct 2026 01:58:12 +0100 Subject: [PATCH 11/11] epoll: the shutdown's first look asks the kernel's queue of an idle promoted conn, under its lock; the tests pin the stream-has-taken-over rule (review r2 of #1168; Refs #1166) --- .../epoll/gate_promoted_1165_linux_test.go | 59 +++++++++++++++++++ internal/engine/epoll/poolwait_gate_885.go | 25 ++++++-- 2 files changed, 79 insertions(+), 5 deletions(-) diff --git a/internal/engine/epoll/gate_promoted_1165_linux_test.go b/internal/engine/epoll/gate_promoted_1165_linux_test.go index 02574ebe6..3a7a1676b 100644 --- a/internal/engine/epoll/gate_promoted_1165_linux_test.go +++ b/internal/engine/epoll/gate_promoted_1165_linux_test.go @@ -170,6 +170,15 @@ func TestRefusesRecvReadsOnAPromotedConnOnlyWhileARequestIsInProgress1166(t *tes refuse bool }{ {"idle", func(*promoted1165) {}, true}, + {"idle-with-a-request-arriving-after-the-gate", func(r *promoted1165) { + if _, err := unix.Write(r.peer, []byte("GET /y HTTP/1.1\r\nHost: x\r\n\r\n")); err != nil { + panic(err) + } + }, true}, + {"detached-stream-with-bytes-in-its-parser", func(r *promoted1165) { + _ = conn.ProcessH1(context.Background(), []byte("POST /x HTTP/1.1\r\nHost: x\r\nContent-Length: 100\r\n\r\nabc"), r.cs.h1State, r.l.handler, func([]byte) {}) + r.cs.h1State.Detached.Store(true) + }, true}, {"partial-request-in-the-parser", func(r *promoted1165) { _ = conn.ProcessH1(context.Background(), []byte("POST /x HTTP/1.1\r\nHost: x\r\nContent-Length: 100\r\n\r\nabc"), r.cs.h1State, r.l.handler, func([]byte) {}) }, false}, @@ -232,6 +241,10 @@ func TestH1InProgressDecidesTheWaitOfALoopWithoutHTTP21166(t *testing.T) { r.cs.asyncParked = false r.cs.asyncInMu.Unlock() }, true}, + {"promoted-detached-stream-with-bytes-in-its-parser", func(r *promoted1165) { + partial(r) + r.cs.h1State.Detached.Store(true) + }, false}, {"promoted-parked-with-a-guarded-write", func(r *promoted1165) { r.cs.detachMu.Lock() }, false}, {"plain-detached-stream", plain(func(r *promoted1165) { partial(r) @@ -354,3 +367,49 @@ func TestH1InProgressDoesNotReadADetachedStreamsWriteState1166(t *testing.T) { close(stop) <-done } + +// celeris#1166 (review r2 of #1168, MINOR-2). A request that reached the +// kernel before the cancel, on an idle PROMOTED conn, is as much a request in +// progress as on a plain conn: the first call asks the kernel's queue, inside +// the conn's lock and only with the goroutine parked and nothing queued (the +// plain conn's line would read state the dispatch goroutine owns). +func TestH1InProgressAsksTheKernelAboutAPromotedConn1166(t *testing.T) { + send := func(r *promoted1165) { + if _, err := unix.Write(r.peer, []byte("GET /y HTTP/1.1\r\nHost: x\r\n\r\n")); err != nil { + t.Fatal(err) + } + } + t.Run("unread-request-first-call", func(t *testing.T) { + r := newPromoted1165(t) + send(r) + if !r.l.h1InProgress() { + t.Fatal("a request in the kernel's queue of an idle promoted conn is not in progress: the loop would shut down and cut it") + } + r.l.h1InProgressNext = 0 + if r.l.h1InProgress() { + t.Fatal("the second call asked the kernel again") + } + }) + t.Run("nothing-unread", func(t *testing.T) { + r := newPromoted1165(t) + if r.l.h1InProgress() { + t.Fatal("an idle promoted conn with an empty queue is in progress") + } + }) + t.Run("response-queued-is-not-asked", func(t *testing.T) { + r := newPromoted1165(t) + send(r) + r.cs.writeBuf = append(r.cs.writeBuf, 'x') // a conn that holds a response is not asked, as a plain one + if r.l.h1InProgress() { + t.Fatal("a promoted conn with a response queued was asked about the kernel's queue") + } + }) + t.Run("detached-stream-is-not-asked", func(t *testing.T) { + r := newPromoted1165(t) + send(r) + r.cs.h1State.Detached.Store(true) + if r.l.h1InProgress() { + t.Fatal("a detached stream was asked about the kernel's queue") + } + }) +} diff --git a/internal/engine/epoll/poolwait_gate_885.go b/internal/engine/epoll/poolwait_gate_885.go index f6012758e..bc264e178 100644 --- a/internal/engine/epoll/poolwait_gate_885.go +++ b/internal/engine/epoll/poolwait_gate_885.go @@ -104,7 +104,7 @@ func refusesRecv(cs *connState) bool { if cs.switchedToH2() { return false } - return !promotedRequestInProgress(cs) + return !promotedRequestInProgress(cs, false) } if cs.protocol != engine.HTTP1 { return false @@ -413,7 +413,7 @@ func (l *Loop) h1InProgress() bool { continue } if cs.asyncPromoted { - if promotedRequestInProgress(cs) { + if promotedRequestInProgress(cs, first) { l.h1InProgressNext = now + int64(h2PoolDrainPollMs*time.Millisecond) return true } @@ -450,8 +450,10 @@ func (l *Loop) h1InProgress() bool { // goroutine owns its parse state: true while input waits for the goroutine, it is // inside a handler (not one that has detached and streams inline), or its parser // holds part of a request. A lock held by a goroutine that is parked is a guarded -// write of a stream, not a request. Loop thread. -func promotedRequestInProgress(cs *connState) bool { +// write of a stream, not a request. With askKernel (the shutdown's first look) +// the kernel's receive queue of an idle conn that holds nothing is asked too. +// Loop thread. +func promotedRequestInProgress(cs *connState, askKernel bool) bool { cs.asyncInMu.Lock() busy := len(cs.asyncInBuf) > 0 || (cs.asyncRun && !cs.asyncParked && !cs.asyncDetachUnlocked) cs.asyncInMu.Unlock() @@ -467,5 +469,18 @@ func promotedRequestInProgress(cs *connState) bool { if h == nil || cs.protocol != engine.HTTP1 || cs.switchedToH2() || h.Detached.Load() { return false } - return h.HasPendingData() || !h.AtRequestBoundary() + if h.HasPendingData() || !h.AtRequestBoundary() { + return true + } + if askKernel && !csWritePending(cs) { + // The goroutine is parked with no input (busy above) and the lock is + // ours, so nothing writes the state csWritePending reads: the question + // h1InProgress asks of a plain conn on its first call, here (review r2 of + // celeris#1168: a request in the kernel's queue of an idle promoted conn + // was cut by phase 0 while a plain conn's was waited for). + if q, err := unix.IoctlGetInt(cs.fd, unix.TIOCINQ); err == nil && q != 0 { + return true + } + } + return false }