From 36358f6b09940843bda8627cea07c495528602af Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 20:17:39 +0100 Subject: [PATCH 1/5] tests: a loop with no HTTP/2 conn holds the FIN of its idle HTTP/1 conns behind another conn's send drain (red on main; Refs #885, Refs #1104) --- ...tdown_idle_before_drain_1104_linux_test.go | 96 +++++++++++++++++++ 1 file changed, 96 insertions(+) create mode 100644 internal/engine/epoll/shutdown_idle_before_drain_1104_linux_test.go 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 new file mode 100644 index 000000000..8d395d209 --- /dev/null +++ b/internal/engine/epoll/shutdown_idle_before_drain_1104_linux_test.go @@ -0,0 +1,96 @@ +//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#1104 (the flakes lane; TestShutdownPoolWaitGateRequestsInProgressAnd +// IdleConns885 failed once in CI with an idle keep-alive conn that had no FIN +// 1.5 s into the shutdown). A loop with no HTTP/2 conn has nothing to wait for +// (h2PoolSettled is true at once), so it never raises the pool wait's gate and +// goes straight into shutdown(): phase 2b is a send drain that polls every conn +// with a response still queued until its socket takes it, up to the budget of +// the Shutdown (20 s there), and phase 3, which closes the conns, comes after. +// A client that is slow to read one download so held the FIN of every idle +// keep-alive conn on the same loop for as long as the drain lasted, where the +// loops that have an HTTP/2 conn close theirs 250 ms in (the gate), and where +// net/http's Shutdown closes its idle conns at once. + +// idleConn1104 adds an HTTP/1 conn that holds nothing (a keep-alive conn after +// its response) to the loop of r, and returns the peer's descriptor. +func (r *drainRig874) idleConn1104(t *testing.T) (fd, peer int) { + t.Helper() + pair, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM|unix.SOCK_NONBLOCK, 0) + if err != nil { + t.Fatalf("socketpair: %v", err) + } + t.Cleanup(func() { _ = unix.Close(pair[1]) }) + if pair[0] >= len(r.l.conns) { + t.Fatalf("socketpair fd %d outside the test conn table", pair[0]) + } + cs := acquireConnState(t.Context(), pair[0], 4096, false) + cs.protocol = engine.HTTP1 + cs.detected = true + cs.h1State = conn.NewH1State() + cs.remoteAddr = "127.0.0.1:10" + r.l.conns[pair[0]] = cs + r.l.addLiveConn(cs) + r.l.connCount++ + r.l.activeConns.Add(1) + return pair[0], pair[1] +} + +func TestShutdownClosesAnIdleConnWhileAnotherConnsSendDrainWaits1104(t *testing.T) { + const queued = 2 << 20 + r := newDrainRig874(t, queued) // a conn whose peer reads nothing + _, idlePeer := r.idleConn1104(t) + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + t.Cleanup(cancel) + var budget atomic.Pointer[context.Context] + c := ctx + budget.Store(&c) + r.l.drainBudget = &budget + + shutdownDone := make(chan struct{}) + go func() { + defer close(shutdownDone) + r.l.shutdown() + }() + + // The idle conn's FIN, with the other conn's peer still not reading. + begin := time.Now() + fds := []unix.PollFd{{Fd: int32(idlePeer), Events: unix.POLLIN}} + var fin bool + for time.Since(begin) < 1500*time.Millisecond && !fin { + if n, _ := unix.Poll(fds, 50); n > 0 { + var b [1]byte + if m, err := unix.Read(idlePeer, b[:]); m == 0 && err == nil { + fin = true + } + } + } + took := time.Since(begin).Round(time.Millisecond) + // Let the drain finish: the stalled conn's peer reads now. + r.startReader() + select { + case <-shutdownDone: + case <-time.After(20 * time.Second): + t.Fatal("shutdown did not return once the stalled conn's peer read") + } + r.wait(t) + if !fin { + t.Fatalf("the idle conn had no FIN %v into the shutdown, while the send drain of another conn of the loop waited for its peer (celeris#1104)", took) + } + t.Logf("the idle conn had its FIN %v into the shutdown", took) +} From a98536480a3eb8387d831f303fe35c1e2ec53875 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 21:42:06 +0100 Subject: [PATCH 2/5] epoll: shutdown closes the idle HTTP/1 conns of a loop before anything in it waits, as the pool wait's gate does for a loop with HTTP/2 conns (v1.6.0, Refs #885 #1104) --- internal/engine/epoll/loop.go | 19 ++++ internal/engine/epoll/poolwait_gate_885.go | 7 +- ...tdown_idle_before_drain_1104_linux_test.go | 96 +++++++++++++++++++ 3 files changed, 119 insertions(+), 3 deletions(-) diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go index 4e39b6c8b..a654d503e 100644 --- a/internal/engine/epoll/loop.go +++ b/internal/engine/epoll/loop.go @@ -5306,6 +5306,25 @@ func (l *Loop) shutdown() { l.closeAdoptQueue() l.shutdownDrivers() + // Phase 0 (celeris#1104): close the HTTP/1 conns that hold nothing, now, + // ahead of everything below, which waits: phase 2 for the async goroutines, + // the send drain for the peers of the conns with a response queued, up to the + // Shutdown's budget, and phase 3, which closes the rest, comes after them. An + // idle keep-alive conn, or one that was accepted and sent nothing, waited that + // long for its FIN behind the slow download of another conn of the loop. A + // loop with an HTTP/2 conn to wait for has closed these 250 ms into the wait + // (poolWaitGate); one with none comes straight here (h2PoolSettled), and + // net/http's Shutdown closes its idle conns at once. The pass is the gate's, on + // conns in the state it was written for (nothing torn down yet): a conn with a + // response queued or unacked, a request in progress, a dispatch goroutine + // running a handler, or a detached stream not ended yet is left to the drain + // and to phase 3. Its cap per pass is for a loop that has other conns to + // serve; this one has none, so it goes on until a pass closes fewer. + now := time.Now().UnixNano() + for l.closeIdleH1(false, now) >= gateClosesPerPass { + now = time.Now().UnixNano() + } + // Phase 1: signal every async dispatch / detached goroutine to stop and // tear down protocol state — but do NOT close any fd and do NOT release // any connState yet. asyncClosed + the cond Broadcast MUST happen before diff --git a/internal/engine/epoll/poolwait_gate_885.go b/internal/engine/epoll/poolwait_gate_885.go index a8de5f5d5..90f49abe7 100644 --- a/internal/engine/epoll/poolwait_gate_885.go +++ b/internal/engine/epoll/poolwait_gate_885.go @@ -141,9 +141,9 @@ func (l *Loop) poolWaitGate() { // the pass would otherwise close, ahead of the cap and the send-side tests so // that a conn those leave for the next pass is not met unasked; later passes // do not ask: what has arrived since is a request that begins after the gate, -// refused as before, and its conn is closed. now is the pass's UnixNano. Loop -// thread. -func (l *Loop) closeIdleH1(first bool, now int64) { +// refused as before, and its conn is closed. now is the pass's UnixNano. It +// returns the number of conns it closed. Loop thread. +func (l *Loop) closeIdleH1(first bool, now int64) int { closed := 0 for i := len(l.liveConns) - 1; i >= 0; i-- { cs := l.liveConns[i] @@ -212,6 +212,7 @@ func (l *Loop) closeIdleH1(first bool, now int64) { l.closeConn(cs.fd) closed++ } + return closed } // testHookGateBeforeClose, when set, runs on the loop thread in closeIdleH1 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 8d395d209..0229553bc 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 @@ -94,3 +94,99 @@ func TestShutdownClosesAnIdleConnWhileAnotherConnsSendDrainWaits1104(t *testing. } t.Logf("the idle conn had its FIN %v into the shutdown", took) } + +// The pass runs ahead of phase 1, on conns that have served a request: their +// stream and Context are pooled (CloseH1) and closeConn tears them down once, +// the conn counted closed once and its OnDisconnect fired once, whether it is +// a conn the loop owns or one with a dispatch goroutine parked after its +// request. +func TestShutdownIdlePassTearsAConnThatServedARequestDownOnce1104(t *testing.T) { + for _, async := range []bool{false, true} { + name := "sync" + if async { + name = "async" + } + t.Run(name, func(t *testing.T) { + var disconnects atomic.Int64 + l := shutdownLoop863(t, func(string) { disconnects.Add(1) }) + l.async = async + pair, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM|unix.SOCK_NONBLOCK, 0) + if err != nil { + t.Fatalf("socketpair: %v", err) + } + fd, peer := pair[0], pair[1] + t.Cleanup(func() { _ = unix.Close(peer) }) + if fd >= len(l.conns) { + t.Fatalf("socketpair fd %d outside the test conn table", fd) + } + cs := acquireConnState(t.Context(), fd, 4096, async) + cs.protocol = engine.HTTP1 + cs.detected = true + cs.remoteAddr = "127.0.0.1:9" + cs.writeFn = l.makeWriteFn(cs) + l.conns[fd] = cs + l.addLiveConn(cs) + l.connCount++ + l.activeConns.Add(1) + l.initProtocol(cs) + const req = "GET /x HTTP/1.1\r\nHost: x\r\n\r\n" + if async { + // A dispatch goroutine answers the request and parks. + cs.asyncInMu.Lock() + cs.asyncInBuf = append(cs.asyncInBuf, req...) + cs.asyncRun = true + cs.asyncInMu.Unlock() + l.asyncWG.Add(1) + go l.runAsyncHandler(cs) + for dl := time.Now().Add(5 * time.Second); ; time.Sleep(time.Millisecond) { + cs.asyncInMu.Lock() + parked := cs.asyncParked && cs.asyncRun && len(cs.asyncInBuf) == 0 + cs.asyncInMu.Unlock() + if parked { + break + } + if time.Now().After(dl) { + t.Fatal("the dispatch goroutine never parked after its request") + } + } + } else { + if _, err := unix.Write(peer, []byte(req)); err != nil { + t.Fatalf("write request: %v", err) + } + l.drainRead(fd, time.Now().UnixNano()) + if l.conns[fd] == nil { + t.Fatal("the conn was closed by its own request") + } + } + // The response went out; the peer takes it, so the send queue is empty. + buf := make([]byte, 4096) + for dl := time.Now().Add(5 * time.Second); ; time.Sleep(time.Millisecond) { + if n, err := unix.Read(peer, buf); err == nil && n > 0 { + break + } + if time.Now().After(dl) { + t.Fatal("no response") + } + } + done := make(chan struct{}) + go func() { defer close(done); l.shutdown() }() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("shutdown did not return") + } + if got := disconnects.Load(); got != 1 { + t.Errorf("OnDisconnect ran %d times for one conn, want 1", got) + } + if got := l.closeCount.Load(); got != 1 { + t.Errorf("closeCount %d for one conn, want 1", got) + } + if got := l.activeConns.Load(); got != 0 { + t.Errorf("activeConns %d after shutdown, want 0", got) + } + if n, err := unix.Read(peer, make([]byte, 1)); n != 0 || err != nil { + t.Errorf("the peer read n=%d err=%v after shutdown, want a FIN (0, nil)", n, err) + } + }) + } +} From c826265f4d0e874e60f4ee400d71a6fbee3dcbd8 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 20:25:38 +0100 Subject: [PATCH 3/5] tests: the shutdown gate test reads every idle conn's FIN at the same time and reports each one's own wait (Refs #885, Refs #1104) --- test/server/shutdown_h2_pool_linux_test.go | 45 +++++++++++++++++----- 1 file changed, 35 insertions(+), 10 deletions(-) diff --git a/test/server/shutdown_h2_pool_linux_test.go b/test/server/shutdown_h2_pool_linux_test.go index 9ccfc1c01..721695acc 100644 --- a/test/server/shutdown_h2_pool_linux_test.go +++ b/test/server/shutdown_h2_pool_linux_test.go @@ -1291,17 +1291,31 @@ func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) { begin := time.Now() srv.beginShutdown("Shutdown", 20*time.Second) + // Every conn is read at the same time, so that each one's wait for + // its FIN is its own: read one after the other, the first conn that + // timed out left the others a deadline already past, and Go's read + // of a conn whose deadline has passed fails before it looks at the + // socket, FIN or not: one late conn was reported as five. + type finResult struct { + name string + err error + at time.Duration + } + var finWG sync.WaitGroup + finRes := make(chan finResult, len(idle)+len(silent)) expectFIN := func(name string, c net.Conn, br *bufio.Reader) { - _ = c.SetReadDeadline(begin.Add(1500 * time.Millisecond)) - var err error - if br != nil { - _, err = br.ReadByte() - } else { - _, err = c.Read(make([]byte, 1)) - } - if err != io.EOF { - t.Errorf("%s: read %v at %v, want a FIN (io.EOF) once the gate has passed (celeris#885 item 49)", name, err, time.Since(begin).Round(time.Millisecond)) - } + finWG.Add(1) + go func() { + defer finWG.Done() + _ = c.SetReadDeadline(begin.Add(1500 * time.Millisecond)) + var err error + if br != nil { + _, err = br.ReadByte() + } else { + _, err = c.Read(make([]byte, 1)) + } + finRes <- finResult{name, err, time.Since(begin)} + }() } for i, c := range idle { expectFIN("idle keep-alive "+strconv.Itoa(i), c, idleBr[i]) @@ -1309,6 +1323,17 @@ func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) { for i, c := range silent { expectFIN("connection with no byte sent "+strconv.Itoa(i), c, nil) } + finWG.Wait() + close(finRes) + var latest time.Duration + for r := range finRes { + if r.err != io.EOF { + t.Errorf("%s: read %v at %v, want a FIN (io.EOF) once the gate has passed (celeris#885 item 49)", r.name, r.err, r.at.Round(time.Millisecond)) + } else if r.at > latest { + 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 { From 393c82efa833dbe4b15e2201b9b848e1c2db9627 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 20:39:34 +0100 Subject: [PATCH 4/5] tests: the shutdown gate test holds the wait with 64 HTTP/2 conns, so that no loop is left without one (Refs #885, Refs #1104) --- test/server/shutdown_h2_pool_linux_test.go | 43 ++++++++++++++++------ 1 file changed, 32 insertions(+), 11 deletions(-) diff --git a/test/server/shutdown_h2_pool_linux_test.go b/test/server/shutdown_h2_pool_linux_test.go index 721695acc..0c330a9da 100644 --- a/test/server/shutdown_h2_pool_linux_test.go +++ b/test/server/shutdown_h2_pool_linux_test.go @@ -1142,11 +1142,18 @@ func TestShutdownPoolWaitGateClosesIdleH1AndEndsDetached885(t *testing.T) { } } +// The HTTP/2 conns that hold the pool wait of the test below: h2Big1104 asking +// for 16 MiB and the rest for 512 KiB. +const ( + h2Conns1104 = 64 + h2Big1104 = 24 +) + // TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885 pins what the pool // wait's gate does to the HTTP/1 conns it stops reading, on a loop with no Async // route (the default, synchronous one) and on one that has an Async route (a -// dispatch goroutine for the conns it promotes), with the wait held by 24 HTTP/2 -// responses of 16 MiB that nobody gives window to (celeris#885 item 49, Opus +// dispatch goroutine for the conns it promotes), with the wait held by HTTP/2 +// responses (24 of 16 MiB and 40 of 512 KiB) that nobody gives window to (celeris#885 item 49, Opus // review r2): // // - an idle keep-alive conn, and one that was accepted and has sent nothing @@ -1168,8 +1175,10 @@ func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) { for _, arm := range []string{"sync", "async-route"} { t.Run(arm, func(t *testing.T) { big := strings.Repeat("y", 16<<20) + hold := big[:512<<10] // more than a stream's initial window: it holds the wait as /big does srv := startH2PoolServer759(t, celeris.Epoll, 20*time.Second, func(s *celeris.Server) { s.GET("/big", func(c *celeris.Context) error { return c.String(http.StatusOK, "%s", big) }) + s.GET("/hold", func(c *celeris.Context) error { return c.String(http.StatusOK, "%s", hold) }) s.POST("/up", func(c *celeris.Context) error { return c.String(http.StatusOK, "got %d", len(c.Body())) }) if arm == "async-route" { s.GET("/aping", func(c *celeris.Context) error { return c.String(http.StatusOK, "a") }).Async() @@ -1179,12 +1188,24 @@ func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) { // probe of startH2PoolServer759 (its keep-alive conn stays open in the // default transport's pool, and it polled /ping one or more times). base := settledEngineMetrics1104(srv) - // The wait is held: GET /big on 24 HTTP/2 conns, no WINDOW_UPDATE. + // The wait is held: GET /big on 24 HTTP/2 conns, and GET /hold on 40 more, + // no WINDOW_UPDATE. The gate belongs to a loop that waits for HTTP/2 + // conns; a loop with none shuts down at once and does neither (it cuts a + // request in progress and the conn of an Async route as it cuts any, and + // used to hold its idle conns behind another conn's send drain: + // celeris#1104). The server has one loop per CPU and the kernel spreads + // the conns over them by their addresses, so the HTTP/2 conns are many + // enough that no loop is left without one: 4 loops, (3/4)^64 each, and + // 16 loops (15/16)^64 each, 16 of them: 0.3%. var h2conns []net.Conn - for range 24 { + for i := range h2Conns1104 { c, fr := dialH2759(t, srv.addr) h2conns = append(h2conns, c) - writeH2Get759(t, fr, 1, "/big") + path := "/big" + if i >= h2Big1104 { + path = "/hold" + } + writeH2Get759(t, fr, 1, path) } time.Sleep(300 * time.Millisecond) @@ -1198,7 +1219,7 @@ func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) { if _, err := io.WriteString(c, "GET "+path+" HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil { t.Fatal(err) } - _ = c.SetReadDeadline(time.Now().Add(10 * time.Second)) // 24 HTTP/2 conns of 16 MiB are running under -race + _ = c.SetReadDeadline(time.Now().Add(10 * time.Second)) // 64 HTTP/2 conns are running under -race resp, err := http.ReadResponse(br, nil) if err != nil { t.Fatalf("%s before the shutdown: %v", path, err) @@ -1261,12 +1282,12 @@ func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) { // The gate's first pass runs 250 ms after the shutdown began, and a // conn the loop has accepted but not read yet, or a request whose // bytes it has not read, is closed by it as idle (the loop of a runner - // busy with 24 HTTP/2 downloads under -race and -cover can be that far + // busy with the HTTP/2 downloads under -race and -cover can be that far // behind: celeris#1104). So begin only once the engine counts, on top // of the baseline taken before the first dial, every conn this test - // opened (24 HTTP/2, the keep-alive ones, the silent ones, the download, + // opened (the HTTP/2 ones, the keep-alive ones, the silent ones, the download, // the upload on the sync arm, the promoted conn on the async arm) and - // every request it sent (24 HTTP/2 GETs, one ping per keep-alive conn, + // every request it sent (the HTTP/2 GETs, one ping per keep-alive conn, // the download's GET, and on the sync arm the upload, which the loop // counts once it has read its headers and first 64 KiB, or on the // async arm the promoted conn's ping). The conn count is exact: +34 in @@ -1276,8 +1297,8 @@ func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) { // keep-alive step, before the download and the upload are sent, so it is // neither of them; its source was not found, which is why this wait is // `>=` for requests and exact for conns. - wantConns := uint64(24 + len(idle) + len(silent) + 1) // + the download - wantReqs := uint64(24 + len(idle) + 1) // + the download's GET + wantConns := uint64(h2Conns1104 + len(idle) + len(silent) + 1) // + the download + wantReqs := uint64(h2Conns1104 + len(idle) + 1) // + the download's GET if arm == "sync" { wantConns++ // the upload wantReqs++ // the upload's request, counted when its first bytes are read From cccbce40b131ff4cd45bccd1730a83b734e6ac13 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 22:45:03 +0100 Subject: [PATCH 5/5] epoll: shutdown's idle pass has a bound of its own; its unit test models a promoted conn as production does and pins what the pass leaves (Refs #885, Refs #1104, Refs #1165) --- internal/engine/epoll/loop.go | 7 ++- ...tdown_idle_before_drain_1104_linux_test.go | 56 ++++++++++++++----- 2 files changed, 48 insertions(+), 15 deletions(-) diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go index a654d503e..fd1456b94 100644 --- a/internal/engine/epoll/loop.go +++ b/internal/engine/epoll/loop.go @@ -5320,8 +5320,13 @@ func (l *Loop) shutdown() { // running a handler, or a detached stream not ended yet is left to the drain // and to phase 3. Its cap per pass is for a loop that has other conns to // serve; this one has none, so it goes on until a pass closes fewer. + // A pass counts the closeConn calls it made, and closeConn may leave a conn + // in place (a dispatch goroutine running a handler owes the close), so the + // loop has a bound of its own: a pass that closes the cap is followed by + // another only while the passes made could still have closed every conn + // that was live on entry. It cannot spin. now := time.Now().UnixNano() - for l.closeIdleH1(false, now) >= gateClosesPerPass { + for passes := len(l.liveConns)/gateClosesPerPass + 1; passes > 0 && l.closeIdleH1(false, now) >= gateClosesPerPass; passes-- { now = time.Now().UnixNano() } 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 0229553bc..34df986d8 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 @@ -96,15 +96,24 @@ func TestShutdownClosesAnIdleConnWhileAnotherConnsSendDrainWaits1104(t *testing. } // The pass runs ahead of phase 1, on conns that have served a request: their -// stream and Context are pooled (CloseH1) and closeConn tears them down once, -// the conn counted closed once and its OnDisconnect fired once, whether it is -// a conn the loop owns or one with a dispatch goroutine parked after its -// request. +// stream and Context are pooled (CloseH1). +// +// - sync: a conn the loop owns is torn down once by closeConn: counted closed +// once, OnDisconnect once, a FIN. +// - 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. func TestShutdownIdlePassTearsAConnThatServedARequestDownOnce1104(t *testing.T) { - for _, async := range []bool{false, true} { + for _, promoted := range []bool{false, true} { + async := promoted name := "sync" - if async { - name = "async" + if promoted { + name = "promoted" } t.Run(name, func(t *testing.T) { var disconnects atomic.Int64 @@ -131,7 +140,11 @@ func TestShutdownIdlePassTearsAConnThatServedARequestDownOnce1104(t *testing.T) l.initProtocol(cs) const req = "GET /x HTTP/1.1\r\nHost: x\r\n\r\n" if async { - // A dispatch goroutine answers the request and parks. + // A dispatch goroutine answers the request and parks, as the loop + // starts one: with the conn marked promoted and counted. + cs.asyncPromoted = true + l.asyncPromoted = &atomic.Uint64{} // the cumulative promotion counter + l.asyncPromoted.Add(1) cs.asyncInMu.Lock() cs.asyncInBuf = append(cs.asyncInBuf, req...) cs.asyncRun = true @@ -168,6 +181,16 @@ func TestShutdownIdlePassTearsAConnThatServedARequestDownOnce1104(t *testing.T) t.Fatal("no response") } } + 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) + } + 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", + l.conns[fd] != nil, l.activeConns.Load(), disconnects.Load()) + } + } done := make(chan struct{}) go func() { defer close(done); l.shutdown() }() select { @@ -175,14 +198,19 @@ func TestShutdownIdlePassTearsAConnThatServedARequestDownOnce1104(t *testing.T) case <-time.After(10 * time.Second): t.Fatal("shutdown did not return") } - if got := disconnects.Load(); got != 1 { - t.Errorf("OnDisconnect ran %d times for one conn, want 1", got) + 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) } - if got := l.closeCount.Load(); got != 1 { - t.Errorf("closeCount %d for one conn, want 1", got) + if got := l.closeCount.Load(); got != uint64(wantCalls) { + t.Errorf("closeCount %d for one conn, want %d", got, wantCalls) } - if got := l.activeConns.Load(); got != 0 { - t.Errorf("activeConns %d after shutdown, want 0", got) + if got := l.activeConns.Load(); got != wantActive { + t.Errorf("activeConns %d after shutdown, want %d", got, wantActive) } if n, err := unix.Read(peer, make([]byte, 1)); n != 0 || err != nil { t.Errorf("the peer read n=%d err=%v after shutdown, want a FIN (0, nil)", n, err)