From d0048e83fb69056c64ed3ff8c7836736eca92c48 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Mon, 26 Oct 2026 18:41:49 +0100 Subject: [PATCH 1/2] epoll: failing tests for the pool-wait gate and the empty-round count (Refs #885 items 42, 49) Tests only, each failing on main as an assertion: the pool wait that keeps serving HTTP/1 and detached writes (epoll and adaptive-on-epoll arms of the shutdown test), idle keep-alive clients left waiting, a producer that is never told, and a queued driver flood that counts empty rounds as idle ones. --- .../consecutive_empty_queue_885_linux_test.go | 93 ++++ test/server/shutdown_h2_pool_linux_test.go | 430 ++++++++++++++++++ 2 files changed, 523 insertions(+) create mode 100644 internal/engine/epoll/consecutive_empty_queue_885_linux_test.go diff --git a/internal/engine/epoll/consecutive_empty_queue_885_linux_test.go b/internal/engine/epoll/consecutive_empty_queue_885_linux_test.go new file mode 100644 index 000000000..dd413f734 --- /dev/null +++ b/internal/engine/epoll/consecutive_empty_queue_885_linux_test.go @@ -0,0 +1,93 @@ +//go:build linux + +package epoll + +import ( + "testing" + "time" + + "golang.org/x/sys/unix" + + "github.com/goceleris/celeris/internal/resource" +) + +// celeris#885 item 42: while driverReadQ is non-empty the run loop waits in +// epoll_wait with timeout 0 (celeris#930), and the rounds that return no +// events still counted toward consecutiveEmpty, the count adaptiveTimeoutMs +// stretches the idle wait by (base*2 after 10, base*4 after 100). A flood of +// driver bytes that takes a dozen rounds to read therefore ended with the loop +// already counted as long idle, and its first idle waits started stretched. +// No correctness impact (an event ends a wait at once); the test is on the +// counter's effect: the wait the loop would make as the last of the queued +// reads is delivered. + +// sndbuf885x42 is the send buffer asked for on the socketpair; the kernel caps +// it at wmem_max, which is 212992 on a stock host and 4 MiB on some others. +const sndbuf885x42 = 16 << 20 + +// TestQueuedZeroTimeoutRoundsDoNotStretchTheIdleWait885 fills a socketpair +// with as much as it takes (16 MiB asked for; a kernel with a small wmem_max +// gives a few hundred KiB), so the loop reads it over several rounds: a turn is +// driverReadBudget reads of driverReadBufSize (256 KiB), and every round after +// the first is an epoll_wait(0) that collects nothing. The engine has +// ReadHeaderTimeout -1, so no 25 ms timerfd can end the flood's counting between +// two rounds. On the last read the count is 0 (round 1 had an event), and with +// more than eleven turns the wait the loop would make next (adaptiveTimeoutMs) +// is the base, 1 ms: the rounds a queue forced to zero are not idle rounds. +func TestQueuedZeroTimeoutRoundsDoNotStretchTheIdleWait885(t *testing.T) { + eng, _ := startE2E885(t, false, newPathHandler885(1), func(c *resource.Config) { c.ReadHeaderTimeout = -1 }) + l := eng.WorkerLoop(0).(*Loop) + if l.timerFD >= 0 { + t.Fatalf("apparatus: the loop has a sweep timerfd, which can fire between two rounds of the flood") + } + a, peer := socketpairNonblocking(t) + t.Cleanup(func() { _ = unix.Close(a); _ = unix.Close(peer) }) + for _, fd := range []int{a, peer} { + if err := unix.SetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_SNDBUF, sndbuf885x42); err != nil { + t.Fatalf("SO_SNDBUF: %v", err) + } + } + chunk := make([]byte, 64<<10) + filled := 0 + for { + n, err := unix.Write(peer, chunk) + if err != nil || n <= 0 { + break + } + filled += n + } + const turn = driverReadBudget * driverReadBufSize + if filled <= turn { + t.Skipf("apparatus: the socketpair took %d bytes, not more than one turn (%d)", filled, turn) + } + + total, reads := 0, 0 + last := make(chan [2]int, 1) + if err := l.RegisterConn(a, func(p []byte) { + total += len(p) + reads++ + if total == filled { + // On the loop thread, the loop's own state: the counter as of + // the end of the previous round. + last <- [2]int{l.adaptiveTimeoutMs(1), int(l.consecutiveEmpty)} + } + }, func(error) {}); err != nil { + t.Fatalf("RegisterConn: %v", err) + } + t.Cleanup(func() { _ = l.UnregisterConn(a) }) + select { + case got := <-last: + turns := (filled + turn - 1) / turn + t.Logf("celeris885 item 42: %d bytes (%d turns) read in %d reads; on the last, consecutiveEmpty=%d and the next idle wait would be %d ms", + filled, turns, reads, got[1], got[0]) + if got[1] != 0 { + t.Errorf("after %d turns of queued reads (epoll_wait(0) rounds that collected nothing) consecutiveEmpty=%d, want 0: a round the queue "+ + "forced to timeout 0 must not count as an idle round (celeris#885 item 42)", turns, got[1]) + } + if turns > 11 && got[0] != 1 { + t.Errorf("after %d turns the idle wait is %d ms, want the base, 1 ms (celeris#885 item 42)", turns, got[0]) + } + case <-time.After(10 * time.Second): + t.Fatalf("the loop read %d of %d bytes in 10 s", total, filled) + } +} diff --git a/test/server/shutdown_h2_pool_linux_test.go b/test/server/shutdown_h2_pool_linux_test.go index bd099d19d..40aa9c9d7 100644 --- a/test/server/shutdown_h2_pool_linux_test.go +++ b/test/server/shutdown_h2_pool_linux_test.go @@ -7,6 +7,7 @@ import ( "bytes" "context" "errors" + "fmt" "io" "net" "net/http" @@ -14,6 +15,7 @@ import ( "strings" "sync" "sync/atomic" + "syscall" "testing" "time" @@ -670,7 +672,9 @@ func TestShutdownPoolWaitStopsServingH1AndDetached884(t *testing.T) { eng celeris.EngineType setup func(*testing.T) }{ + {"epoll", celeris.Epoll, func(*testing.T) {}}, {"io_uring", celeris.IOUring, func(*testing.T) {}}, + {"adaptive-epoll", celeris.Adaptive, func(t *testing.T) { t.Setenv("CELERIS_ADAPTIVE_START", "epoll") }}, {"adaptive-iouring", celeris.Adaptive, func(t *testing.T) { t.Setenv("CELERIS_ADAPTIVE_START", "iouring") }}, } { t.Run(e.name, func(t *testing.T) { @@ -912,3 +916,429 @@ func TestShutdownPoolWaitStopsServingH1AndDetached884(t *testing.T) { }) } } + +// TestShutdownPoolWaitGateClosesIdleH1AndEndsDetached885 pins what the pool +// wait's gate does with the conns it stops reading (celeris#885 item 49), on +// the epoll engine: with 48 HTTP/2 streams held so that every loop is in the +// pool wait, +// +// - an idle HTTP/1 keep-alive conn is closed once the gate has passed, so +// its client fails at once and retries elsewhere, as it does against +// std's Shutdown, and does not wait for the end of the pool wait to be cut +// with no answer (the budget here is 20 s; the assertion is 2 s); +// - a detached stream's producer, ended by the gate, gets an error from its +// next Write (net.ErrClosed), not a silent drop; +// - what the producer had queued for a client that was not reading when the +// gate came is still flushed: the client reads it afterwards, every +// byte of the writes that reported success (to within the one chunk that +// may have raced the gate). +// +// The HTTP/1 request that the gate refuses is the previous test's subject. +func TestShutdownPoolWaitGateClosesIdleH1AndEndsDetached885(t *testing.T) { + const nH2, nIdle, nBulk, chunk = 48, 8, 4, 256 << 10 + // asyncBulk: the stream's route is an Async one, so its handler detaches on + // a dispatch goroutine and the conn is promoted (r3 M1: its client used to + // get no FIN). + for _, e := range []struct { + name string + eng celeris.EngineType + setup func(*testing.T) + asyncBulk bool + }{ + {"epoll", celeris.Epoll, func(*testing.T) {}, false}, + {"epoll-async-route", celeris.Epoll, func(*testing.T) {}, true}, + {"adaptive-epoll", celeris.Adaptive, func(t *testing.T) { t.Setenv("CELERIS_ADAPTIVE_START", "epoll") }, false}, + {"adaptive-epoll-async-route", celeris.Adaptive, func(t *testing.T) { t.Setenv("CELERIS_ADAPTIVE_START", "epoll") }, true}, + } { + t.Run(e.name, func(t *testing.T) { + e.setup(t) + entered := make(chan struct{}, nH2) + release := make(chan struct{}) + var releaseOnce sync.Once + releaseHeld := func() { releaseOnce.Do(func() { close(release) }) } + defer releaseHeld() + var okBytes [nBulk]atomic.Int64 + var firstErr [nBulk]atomic.Pointer[error] + srv := startH2PoolServer759(t, e.eng, 20*time.Second, func(s *celeris.Server) { + s.GET("/held", func(c *celeris.Context) error { + entered <- struct{}{} + <-release + return c.String(http.StatusOK, "done") + }).Async() + // The stream's index is in the path: the client conn it belongs to + // is not the order the server happens to accept them in. + bulkRoute := s.GET("/bulk/:i", func(c *celeris.Context) error { + i, convErr := strconv.Atoi(c.Param("i")) + done := c.Detach() + sw := c.StreamWriter() + if sw == nil || convErr != nil || i < 0 || i >= nBulk { + done() + return errors.New("no stream writer") + } + if err := sw.WriteHeader(http.StatusOK, [][2]string{{"content-type", "application/octet-stream"}}); err != nil { + done() + return err + } + go func() { + defer done() + buf := bytes.Repeat([]byte{'x'}, chunk) + for { + if _, err := sw.Write(buf); err != nil { + firstErr[i].Store(&err) + return + } + okBytes[i].Add(chunk) + time.Sleep(10 * time.Millisecond) + } + }() + return nil + }) + if e.asyncBulk { + bulkRoute.Async() + } + }) + + var h2wg sync.WaitGroup + for range nH2 { + c, fr := dialH2759(t, srv.addr) + writeH2Get759(t, fr, 1, "/held") + h2wg.Add(1) + go func() { + defer h2wg.Done() + frames759(c, fr, new(sync.Mutex), 15*time.Second, func(f http2.Frame) bool { + d, ok := f.(*http2.DataFrame) + return ok && d.StreamID == 1 && d.StreamEnded() + }) + }() + } + for i := range nH2 { + select { + case <-entered: + case <-time.After(10 * time.Second): + t.Fatalf("only %d of %d held handlers started within 10s", i, nH2) + } + } + + // Idle keep-alive clients, each answered once. + idle := make([]net.Conn, nIdle) + idleBr := make([]*bufio.Reader, nIdle) + for i := range idle { + c, err := net.Dial("tcp", srv.addr) + if err != nil { + t.Fatal(err) + } + defer func() { _ = c.Close() }() + idle[i], idleBr[i] = c, bufio.NewReader(c) + if _, err := io.WriteString(c, "GET /ping HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil { + t.Fatal(err) + } + _ = c.SetReadDeadline(time.Now().Add(2 * time.Second)) + resp, err := http.ReadResponse(idleBr[i], nil) + if err != nil { + t.Fatalf("keep-alive request before the shutdown: %v", err) + } + _, _ = io.Copy(io.Discard, resp.Body) + } + + // Detached streams to clients that read nothing until later: the + // producers' bytes pile up in the engine's queues. + bulk := make([]net.Conn, nBulk) + for i := range bulk { + c, err := net.Dial("tcp", srv.addr) + if err != nil { + t.Fatal(err) + } + defer func() { _ = c.Close() }() + bulk[i] = c + if _, err := io.WriteString(c, fmt.Sprintf("GET /bulk/%d HTTP/1.1\r\nHost: x\r\n\r\n", i)); err != nil { + t.Fatal(err) + } + } + for deadline := time.Now().Add(5 * time.Second); ; { + ready := true + for i := range okBytes { + ready = ready && okBytes[i].Load() >= 4*chunk + } + if ready { + break + } + if time.Now().After(deadline) { + t.Fatal("the producers wrote under 1 MiB each within 5s") + } + time.Sleep(10 * time.Millisecond) + } + + begin := time.Now() + srv.beginShutdown("Shutdown", 20*time.Second) + + // Idle HTTP/1: closed, not left to hang. + for i, c := range idle { + _ = c.SetReadDeadline(begin.Add(2500 * time.Millisecond)) + if _, err := idleBr[i].ReadByte(); err == nil { + t.Errorf("idle keep-alive client %d read a byte after the shutdown began", i) + } else if ne, ok := err.(net.Error); ok && ne.Timeout() { + t.Errorf("idle keep-alive client %d was still connected %v into the shutdown, in the wait for HTTP/2 handlers (budget 20s): "+ + "the gate left it waiting for a close with no answer (celeris#885 item 49)", i, time.Since(begin).Round(10*time.Millisecond)) + } + } + + // Producers ended with an error. + time.Sleep(time.Until(begin.Add(1500 * time.Millisecond))) + for i := range firstErr { + p := firstErr[i].Load() + if p == nil { + t.Errorf("producer %d: Write still succeeded 1.5 s into the shutdown, while HTTP/2 handlers were running", i) + } else if !errors.Is(*p, net.ErrClosed) { + t.Errorf("producer %d: Write ended with %v, want an error wrapping net.ErrClosed", i, *p) + } + } + + // What was queued is flushed: read it all now, every stream at the + // same time (a client that sat on a full window for seconds and + // then reads one stream after the other waits for the TCP persist + // timer of the others, which says nothing about the engine). + type streamResult struct { + got, ok int64 + err error + end error // what the read after the flushed bytes met + } + results := make([]streamResult, nBulk) + var rwg sync.WaitGroup + for i, c := range bulk { + ok := okBytes[i].Load() + rwg.Add(1) + go func() { + defer rwg.Done() + _ = c.SetReadDeadline(time.Now().Add(15 * time.Second)) + resp, err := http.ReadResponse(bufio.NewReader(c), nil) + if err != nil { + results[i] = streamResult{ok: ok, err: err} + return + } + var got int64 + buf := make([]byte, 64<<10) + for got < ok { + n, err := resp.Body.Read(buf) + got += int64(n) + if err != nil { + results[i] = streamResult{got: got, ok: ok, err: err} + return + } + } + // The stream is over: the gate ended it, and a client that gets no + // FIN would wait for the end of the wait (up to the budget) + // before it could reconnect elsewhere. + // (A write that raced the gate may have been queued and sent + // although its Write reported the close: read what is left, up + // to the end of the stream.) + _ = c.SetReadDeadline(time.Now().Add(2 * time.Second)) + var endErr error + for endErr == nil { + _, endErr = resp.Body.Read(buf) + } + results[i] = streamResult{got: got, ok: ok, end: endErr} + }() + } + rwg.Wait() + for i, r := range results { + t.Logf("celeris885 %s stream %d: %d of %d bytes written before the gate were received (%v)", e.name, i, r.got, r.ok, r.err) + if r.err == nil && !errors.Is(r.end, io.EOF) && !errors.Is(r.end, io.ErrUnexpectedEOF) { + t.Errorf("stream %d: after the flushed bytes the client read %v, 2 s on, want a FIN (EOF or unexpected EOF, not a reset or silence): the gate ended the stream but did not close its connection (celeris#885 item 49)", i, r.end) + } + if r.err != nil || r.got < r.ok-chunk { + t.Errorf("stream %d: the client received %d of the %d bytes whose writes had succeeded (%v): what the producer had queued was not flushed after the gate (celeris#885 item 49)", i, r.got, r.ok, r.err) + } + } + releaseHeld() + h2wg.Wait() + srv.waitStart(t, 25*time.Second) + }) + } +} + +// 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 +// review r2): +// +// - an idle keep-alive conn, and one that was accepted and has sent nothing +// (a browser's preconnect), get a FIN as soon as the gate has passed; +// - a download queued in the engine for a client reading slowly is delivered in +// full before its conn is closed; +// - a request in progress when the shutdown begins (the headers and the first +// 64 KiB of a 2 MiB body) is read on and answered, as std's Shutdown answers +// 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 +// stream on an Async route is closed once the gate has ended it: the other +// test.) +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) + 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.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() + } + }) + // The wait is held: GET /big on 24 HTTP/2 conns, no WINDOW_UPDATE. + var h2conns []net.Conn + for range 24 { + c, fr := dialH2759(t, srv.addr) + h2conns = append(h2conns, c) + writeH2Get759(t, fr, 1, "/big") + } + time.Sleep(300 * time.Millisecond) + + keepAlive := func(path string) (net.Conn, *bufio.Reader) { + c, err := net.Dial("tcp", srv.addr) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = c.Close() }) + br := bufio.NewReader(c) + 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 + resp, err := http.ReadResponse(br, nil) + if err != nil { + t.Fatalf("%s before the shutdown: %v", path, err) + } + _, _ = io.Copy(io.Discard, resp.Body) + return c, br + } + idle := make([]net.Conn, 6) + idleBr := make([]*bufio.Reader, 6) + for i := range idle { + idle[i], idleBr[i] = keepAlive("/ping") + } + silent := make([]net.Conn, 2) + for i := range silent { + c, err := net.Dial("tcp", srv.addr) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = c.Close() }) + silent[i] = c + } + var promoted net.Conn + var promotedBr *bufio.Reader + if arm == "async-route" { + promoted, promotedBr = keepAlive("/aping") + } + // A download queued in the engine: the client's receive buffer is small. + d := net.Dialer{Control: func(_, _ string, rc syscall.RawConn) error { + return rc.Control(func(fd uintptr) { + _ = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_RCVBUF, 32<<10) + }) + }} + dl, err := d.Dial("tcp", srv.addr) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = dl.Close() }) + if _, err := io.WriteString(dl, "GET /big HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil { + t.Fatal(err) + } + // A request in progress. + var up net.Conn + const upSize = 2 << 20 + if arm == "sync" { + up, err = net.Dial("tcp", srv.addr) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = up.Close() }) + 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) + } + } + // TCP_DEFER_ACCEPT hands the engine a conn with no byte after about a + // second: the silent conns are in its table by then. + time.Sleep(1300 * time.Millisecond) + + begin := time.Now() + srv.beginShutdown("Shutdown", 20*time.Second) + + 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)) + } + } + for i, c := range idle { + expectFIN("idle keep-alive "+strconv.Itoa(i), c, idleBr[i]) + } + for i, c := range silent { + expectFIN("connection with no byte sent "+strconv.Itoa(i), c, nil) + } + 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)) + } + } + if up != nil { + // The rest of the body arrives after the gate has passed. + _ = 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#885 item 49)", time.Since(begin).Round(time.Millisecond), err) + } else { + _ = up.SetReadDeadline(time.Now().Add(3 * time.Second)) + resp, err := http.ReadResponse(bufio.NewReader(up), nil) + if err != nil { + t.Errorf("the response to the upload: %v: a request in progress was not answered (celeris#885 item 49)", 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) + } + // Answered after the gate: close-after, as net/http's + // response written after Shutdown began, so that a client + // does not write its next request to a conn that is closing. + if !resp.Close { + t.Errorf("the response to the upload, written after the gate, has no Connection: close (r3 m1)") + } + _ = up.SetReadDeadline(time.Now().Add(2 * time.Second)) + if _, err := up.Read(make([]byte, 1)); err != io.EOF { + t.Errorf("after the response to the upload the conn read %v, want a FIN (io.EOF)", err) + } + } + } + } + // The queued download arrives whole, and then its conn is closed. + _ = dl.SetReadDeadline(time.Now().Add(15 * time.Second)) + resp, err := http.ReadResponse(bufio.NewReader(dl), nil) + if err != nil { + t.Fatalf("the download's head: %v", err) + } + if n, err := io.Copy(io.Discard, resp.Body); n != int64(len(big)) { + t.Errorf("the queued download delivered %d of %d bytes (%v): the gate cut a response that was still going out", n, len(big), err) + } + for _, c := range h2conns { // the HTTP/2 clients go: the wait ends + _ = c.Close() + } + srv.waitStart(t, 25*time.Second) + }) + } +} From ad7c63f8a744ab0d0d7750ecc1d5f0fa31136807 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Mon, 26 Oct 2026 18:41:49 +0100 Subject: [PATCH 2/2] epoll: the pool wait stops serving HTTP/1 and detached writes and closes idle keep-alive conns, queued driver rounds are not idle rounds (Refs #885 items 42, 49) Item 49: poolWaitGate, 250 ms into the wait for HTTP/2 pool handlers, refuses HTTP/1 reads (a request in progress on a conn the loop parses is read on and answered with Connection: close, as std's Shutdown does), ends detached producers (TryLock, hijack re-tested under the lock, a stranded remainder put on the dirty list) and closes the HTTP/1 conns that hold nothing: idle keep-alive conns, conns that sent no byte, and detached streams the gate has ended, on a sync or an Async route, once their queue has gone (at most 64 per 10 ms pass). Item 42: a round whose timeout a driver queue forced to 0 does not count toward consecutiveEmpty. --- internal/conn/h1.go | 12 ++ internal/engine/epoll/loop.go | 23 ++- internal/engine/epoll/poolwait_gate_885.go | 225 +++++++++++++++++++++ test/server/shutdown_h2_pool_linux_test.go | 15 +- 4 files changed, 266 insertions(+), 9 deletions(-) create mode 100644 internal/engine/epoll/poolwait_gate_885.go diff --git a/internal/conn/h1.go b/internal/conn/h1.go index 50f308520..da43569f5 100644 --- a/internal/conn/h1.go +++ b/internal/conn/h1.go @@ -347,6 +347,15 @@ type H1State struct { // mid-body. Called on the goroutine running ProcessH1, under the // engine's lock for the connection's buffers if it has one. WriteBacklogged func() bool + + // CloseAfter, set by an engine that is shutting down (epoll's pool-wait + // gate, celeris#885 item 49), makes every request handled from then on + // close-after: its response carries Connection: close and the connection + // ends once the response has gone, as net/http's response written after + // Shutdown began does, so a client does not write its next request to a + // connection that is about to be closed. Set and read on the goroutine that + // runs ProcessH1 (the loop thread for a conn the loop parses). + CloseAfter bool } // TakeBufferedBytes returns a copy of any bytes ProcessH1 stashed in the @@ -1426,6 +1435,9 @@ func handleH1Request(ctx context.Context, state *H1State, body []byte, } req := &state.req + if state.CloseAfter { + req.KeepAlive = false + } s := populateCachedStream(state, req, body) // Reuse the connection-scoped response adapter — avoids a heap allocation diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go index b93426e1e..bf0bf25a9 100644 --- a/internal/engine/epoll/loop.go +++ b/internal/engine/epoll/loop.go @@ -214,6 +214,13 @@ type Loop struct { // (celeris#759; h2PoolSettled). Zero until then. Loop thread. h2DrainStart time.Time + // refuseRecv is set by the pool wait's gate (poolWaitGate, celeris#885 + // item 49): from then on drainRead reads no more from an HTTP/1 conn. + // producersStopNext is the time (UnixNano) before which the gate does not + // walk the conns again. Loop thread. + refuseRecv bool + producersStopNext int64 + // 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. @@ -536,6 +543,10 @@ func (l *Loop) run(ctx context.Context) { l.shutdown() return } + // Waiting for HTTP/2 pool handlers (h2DrainStart is set): from + // poolWaitGrace into the wait, serve no more HTTP/1 and let no + // detached stream write (celeris#885 item 49). + l.poolWaitGate() } // Cache the atomic load: ACTIVE→LINGERING→DRAINING and @@ -771,10 +782,13 @@ func (l *Loop) run(ctx context.Context) { l.acceptAll(ctx, time.Now().UnixNano()) } - // Adaptive timeout tracking. + // Adaptive timeout tracking. A round that collected nothing counts as + // an idle one only if the loop was willing to wait: a queued driver + // conn forced the timeout to 0, which says nothing about idleness + // (celeris#885 item 42). if n > 0 { l.consecutiveEmpty = 0 - } else { + } else if owedDriverReads == 0 { l.consecutiveEmpty++ } @@ -1517,6 +1531,11 @@ func (l *Loop) drainRead(fd int, now int64) { if cs == nil { return } + // Past the pool wait's gate an HTTP/1 conn is read no more (celeris#885 + // item 49): one load and branch per read event. + if l.refuseRecv && refusesRecv(cs) { + return + } for { // Zero-copy body recv: when H1 is in a partial-body state we read diff --git a/internal/engine/epoll/poolwait_gate_885.go b/internal/engine/epoll/poolwait_gate_885.go new file mode 100644 index 000000000..03e59c0b1 --- /dev/null +++ b/internal/engine/epoll/poolwait_gate_885.go @@ -0,0 +1,225 @@ +//go:build linux + +package epoll + +import ( + "time" + + "golang.org/x/sys/unix" + + "github.com/goceleris/celeris/internal/engine" +) + +// The gate of the wait for HTTP/2 pool handlers (celeris#885 item 49), the +// epoll twin of io_uring's (celeris#884 item 50). A loop whose context is +// cancelled waits, as long as an HTTP/2 conn has a handler running on the +// shared worker pool, a response queued or DATA waiting for a window +// (h2PoolSettled), and the wait can last the whole budget of the shutdown. It +// used to go on reading and answering the keep-alive requests of its HTTP/1 +// clients and let detached streams (Server-Sent Events, a WebSocket) write +// meanwhile: the phases of shutdown that stop both (its phase 1 and the send +// drain) run only after the wait. Measured with +// TestShutdownPoolWaitStopsServingH1AndDetached884 on epoll: 16 of 16 +// requests answered and about 700 ticks written 1.5 s into the shutdown. +// +// The gate comes poolWaitGrace after the wait began, as the send drain's does, +// so a request in flight when the shutdown began is still answered. From then +// on the loop reads no more new requests from an HTTP/1 conn (refuseRecv) and +// ends the writes of every detached stream (stopDetachedProducers). A request +// already in progress on a conn the loop parses itself (headers in pieces, a +// body arriving) is read on and answered, as std's Shutdown answers an active +// conn, with Connection: close (H1State.CloseAfter, as net/http's response +// written after Shutdown began). The HTTP/2 conns the wait is for are served on: +// they have had GOAWAY since the wait began. +// +// An HTTP/1 conn that is idle, holds nothing queued and has no request in +// progress is closed, as std's Shutdown closes its idle conns: its client fails +// at once and retries elsewhere, where it would otherwise wait, with no answer, +// for as long as the wait lasts (up to the budget of the shutdown) and then be +// cut (closeIdleH1). So is a conn that was accepted and has sent nothing, and a +// detached stream the gate has ended once what was queued for it has gone out +// (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 shared with io_uring (which has no idle close at all). + +// 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 +// shutdownSendDrainFloor. +const poolWaitGrace = shutdownSendDrainFloor + +// gateClosesPerPass caps the conns closeIdleH1 closes in one pass, so that a +// loop with thousands of idle conns does not stop serving the HTTP/2 conns the +// wait is for while it closes them all (several syscalls each): the next pass, +// 10 ms on, goes on. +const gateClosesPerPass = 64 + +// refusesRecv reports whether a loop past the pool wait's gate reads no more +// from cs: an HTTP/1 conn (detached or not) or one whose protocol is not +// detected yet, which io_uring's refuseRecv test mirrors, with one exception +// io_uring does not make: 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: +// its state is not the loop's to read. Loop thread. +func refusesRecv(cs *connState) bool { + if !cs.detected { + return cs.detectN == 0 + } + if cs.protocol != engine.HTTP1 { + return false + } + return cs.asyncPromoted || cs.h1State == nil || !cs.h1State.HasPendingData() +} + +// poolWaitGate applies the gate once the loop has waited poolWaitGrace for the +// HTTP/2 pool handlers (h2PoolSettled has set h2DrainStart). The wait is a +// serving phase, so the walk over every conn, which ends the producers and +// closes the idle HTTP/1 conns, is not paid on every pass: it runs at most once +// per h2PoolDrainPollMs. Loop thread. +func (l *Loop) poolWaitGate() { + now := time.Now() + if now.Sub(l.h2DrainStart) <= poolWaitGrace { + return + } + l.refuseRecv = true + if n := now.UnixNano(); n >= l.producersStopNext { + l.producersStopNext = n + int64(h2PoolDrainPollMs*time.Millisecond) + l.stopDetachedProducers() + l.closeIdleH1() + } +} + +// closeIdleH1 closes the HTTP/1 conns the gate has stopped reading that hold +// nothing: no response queued in the engine or the kernel's send queue (a close +// with bytes unsent would reset them away), no request in progress, not a +// detached stream still running (one the gate has ended is closed once it has +// flushed, on a sync route or an Async one), no dispatch goroutine otherwise +// (see the gate's comment), and a conn with no byte sent (not detected, +// detectN 0). It also marks every conn it meets close-after (H1State.CloseAfter), +// so that a request it still reads (one in progress) is answered with +// Connection: close. A conn whose response is +// still going out is met again by the next pass. closeConn removes the conn +// from liveConns by swapping it with the last entry, so the walk goes from the +// top down, as checkTimeouts does. Loop thread. +func (l *Loop) closeIdleH1() { + closed := 0 + for i := len(l.liveConns) - 1; i >= 0; i-- { + cs := l.liveConns[i] + 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 + 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 { + 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) { + continue + } + } + } else if cs.detectN != 0 { + continue + } + if closed >= gateClosesPerPass || csWritePending(cs) || cs.dirty || cs.epollOut { + continue + } + if q, err := unix.IoctlGetInt(cs.fd, unix.TIOCOUTQ); err != nil || q != 0 { + continue + } + l.closeConn(cs.fd) + closed++ + } +} + +// stopDetachedProducers is the detached half of shutdown's phase 1, run early +// by the pool wait's gate: a detached conn (Server-Sent Events, a WebSocket) +// has a goroutine of its own that keeps writing, and with the wait lasting as +// long as the budget a stream that never ends would write for all of it. The +// stream is told as shutdown tells it: asyncClosed, which the write-error +// query reads first, then, under detachMu, the parked read-end error, then +// detachClosed (the guarded write hook no-ops from there) and OnDetachClose. +// Only conns that have detached (h1State.Detached) and whose dispatch +// goroutine is not inside the handler: the goroutine holds detachMu across +// the handler, so the TryLock fails for a running async handler, which is +// answered, up to the budget, as before. TryLock, never Lock, so a goroutine +// parked inside a write cannot wedge the loop; a conn whose lock is busy is +// tried again on the next pass of the pacing. A conn whose async Detach is +// not yet finalised (asyncDetachPending) is left for that pass. +// +// What the producer's guarded write hook left queued is flushed, not +// stranded: the inline write may have taken only part of the bytes, and +// drainDetachQueue skips a detachClosed conn, so a conn with bytes queued and +// neither EPOLLOUT armed nor on the dirty list is put on the list here. The +// rest is shutdown's own, which every step repeats harmlessly (the parked +// error and OnDetachClose are consumed once). Loop thread. +func (l *Loop) stopDetachedProducers() { + for i := len(l.liveConns) - 1; i >= 0; i-- { + cs := l.liveConns[i] + mu := cs.detachMu + if mu == nil || cs.hijacked.Load() || cs.h1State == nil || !cs.h1State.Detached.Load() { + continue + } + if !mu.TryLock() { + continue + } + // Re-tested under the lock, as closeConn's ownership check is: an async + // Hijack takes the conn and closes its descriptor under this lock. + if cs.hijacked.Load() || (!cs.detachClosed && cs.asyncDetachPending) { + mu.Unlock() + continue + } + if !cs.detachClosed { + cs.asyncClosed.Store(true) + if cs.asyncCond.L != nil { + cs.asyncInMu.Lock() + cs.asyncCond.Broadcast() + cs.asyncInMu.Unlock() + } + if err := cs.closeErr; err != nil { + cs.closeErr = nil + if cs.h1State.OnError != nil { + cs.h1State.OnError(err) + } + } + cs.detachClosed = true + if cs.h1State.WSReady.Load() && cs.h1State.OnDetachClose != nil { + cs.h1State.OnDetachClose() + cs.h1State.OnDetachClose = nil + } + } + // What is queued must still reach the peer, and nothing but this pass + // will see to it once detachClosed is set: drainDetachQueue skips such + // a conn, so a remainder the producer's write queued for the loop, or a + // conn a relink gave up (its hand-back is skipped the same way) and + // that EPOLLOUT no longer watches, is stranded. Met again on every + // pass, for a conn that is given up after this one. + if csWritePending(cs) && !cs.epollOut && !cs.dirty { + l.markDirty(cs) + } + mu.Unlock() + } +} diff --git a/test/server/shutdown_h2_pool_linux_test.go b/test/server/shutdown_h2_pool_linux_test.go index 40aa9c9d7..2a212847c 100644 --- a/test/server/shutdown_h2_pool_linux_test.go +++ b/test/server/shutdown_h2_pool_linux_test.go @@ -658,13 +658,14 @@ func TestShutdownWaitsForFlowControlledH2Response(t *testing.T) { // request's open questions record. The oracle looks 1.5 s in, well past the grace and the time a worker idle in its ring // wait takes to notice the cancel. // -// io_uring only: epoll's pool wait (Loop.run, h2PoolSettled) does not stop -// reading or writing either; its loops stop at their shutdown, after the pool -// settles (the same run asserts it: with epoll in the list this test fails on -// it, which the polish checklist of epoll, celeris#885, records; with epoll -// in the list the run measures 16 of 16 answered and 688 to 704 ticks in -// 900 ms). The adaptive engine is run on io_uring (CELERIS_ADAPTIVE_START), as -// the second arm: its gate is the io_uring worker's. +// Both native engines have the gate: io_uring's (celeris#884) and epoll's +// (celeris#885 item 49, poolWaitGate). Before epoll's, its pool wait did not +// stop reading or writing either (its loops stopped at their shutdown, after +// the pool settled): with epoll in the list the run measured 16 of 16 answered +// and 688 to 704 ticks in 900 ms. The adaptive engine is run on each engine +// (CELERIS_ADAPTIVE_START), as the further arms. Where io_uring is not +// available the io_uring arm cannot start and the adaptive-iouring arm runs +// on epoll. func TestShutdownPoolWaitStopsServingH1AndDetached884(t *testing.T) { const nH2 = 48 for _, e := range []struct {