From c5289d3d31561425237d292d20f8d7550f7a42c2 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Sun, 18 Oct 2026 12:32:17 +0200 Subject: [PATCH 01/14] sse, websocket, epoll, iouring: tests for a detached conn whose peer takes none of its queued bytes (v1.6.0 polish items) An SSE client and a WebSocket peer that read nothing, on epoll, io_uring and adaptive (std as the control), a client that reads slowly, and the sweep of each engine against a loopback TCP pair. The non-reading cases fail on main: a detached conn is exempt from WriteTimeout. Refs #888 item 36 Refs #888 item 37 Refs #884 item 66 --- .../epoll/detach_stall_888_linux_test.go | 255 +++++++++++++ .../complete_send_bench_888_linux_test.go | 28 ++ .../iouring/detach_stall_888_linux_test.go | 223 +++++++++++ middleware/sse/sse_disconnect_linux_test.go | 21 +- .../sse/sse_nonreading_888_linux_test.go | 352 ++++++++++++++++++ middleware/websocket/engine_linux_test.go | 9 +- .../websocket/nonreading_888_linux_test.go | 118 ++++++ 7 files changed, 998 insertions(+), 8 deletions(-) create mode 100644 internal/engine/epoll/detach_stall_888_linux_test.go create mode 100644 internal/engine/iouring/complete_send_bench_888_linux_test.go create mode 100644 internal/engine/iouring/detach_stall_888_linux_test.go create mode 100644 middleware/sse/sse_nonreading_888_linux_test.go create mode 100644 middleware/websocket/nonreading_888_linux_test.go diff --git a/internal/engine/epoll/detach_stall_888_linux_test.go b/internal/engine/epoll/detach_stall_888_linux_test.go new file mode 100644 index 000000000..6f7733244 --- /dev/null +++ b/internal/engine/epoll/detach_stall_888_linux_test.go @@ -0,0 +1,255 @@ +//go:build linux + +package epoll + +import ( + "net" + "sync" + "syscall" + "testing" + "time" + + "golang.org/x/sys/unix" + + "github.com/goceleris/celeris/internal/conn" +) + +// A truly-detached conn (WS/SSE) whose peer takes none of the bytes queued for +// it is closed WriteTimeout after the sweep found them queued (celeris#888 +// items 36 and 37, celeris#884 item 66): checkTimeouts skipped a detached conn +// altogether, and a write on a native engine only queues, so a client that +// stayed connected and read nothing held its stream for ever. The clock is +// "the peer took none", not the age of the stream or the last read. These +// tests drive the loop primitives against a loopback TCP pair, because the +// evidence the sweep reads (tcp_info bytes_acked) exists on a TCP socket only. + +const stall888Bound = 300 * time.Millisecond + +// stall888Conn is a detached conn on the server end of a loopback TCP pair +// whose client end has a small receive buffer, registered with l. +type stall888Conn struct { + cs *connState + peer net.Conn +} + +func newStall888Conn(t *testing.T, l *Loop) *stall888Conn { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Skipf("loopback unavailable: %v", err) + } + t.Cleanup(func() { _ = ln.Close() }) + d := net.Dialer{Timeout: 2 * time.Second, Control: func(_, _ string, c syscall.RawConn) error { + return c.Control(func(fd uintptr) { + _ = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUF, 4096) + }) + }} + peer, err := d.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + t.Cleanup(func() { _ = peer.Close() }) + srv, err := ln.Accept() + if err != nil { + t.Fatalf("accept: %v", err) + } + f, err := srv.(*net.TCPConn).File() + if err != nil { + t.Fatalf("File: %v", err) + } + fd, err := unix.Dup(int(f.Fd())) + _ = f.Close() + _ = srv.Close() + if err != nil { + t.Fatalf("dup: %v", err) + } + if fd >= connTableSize { + _ = unix.Close(fd) + t.Skip("fd out of table range") + } + if err := unix.SetNonblock(fd, true); err != nil { + t.Fatalf("set nonblock: %v", err) + } + cs := &connState{fd: fd, liveIdx: -1} + cs.h1State = conn.NewH1State() + cs.h1State.Detached.Store(true) + cs.detachMu = &sync.Mutex{} + registerForOut885(t, l, cs) + l.conns[fd] = cs + l.addLiveConn(cs) + l.connCount++ + l.activeConns.Add(1) + l.detachedCount = 1 + t.Cleanup(func() { + if l.conns[fd] == cs { + l.closeConn(fd) + } + }) + return &stall888Conn{cs: cs, peer: peer} +} + +// queue appends n bytes to the conn's write buffer and flushes, as the guarded +// writeFn and the dirty pass do, so that what the socket does not take stays +// queued. +func (c *stall888Conn) queue(t *testing.T, l *Loop, n int) { + t.Helper() + cs := c.cs + cs.detachMu.Lock() + cs.writeBuf = append(cs.writeBuf, make([]byte, n)...) + cs.pendingBytes = csPendingBytes(cs) + if err := l.flushWrites(cs, true); err != nil { + cs.detachMu.Unlock() + t.Fatalf("flush: %v", err) + } + cs.pendingBytes = csPendingBytes(cs) + cs.detachMu.Unlock() + if !cs.dirty { + l.markDirty(cs) + } +} + +// stall888Loop is a reap loop whose WriteTimeout is the bound. +func stall888Loop(t *testing.T, wt time.Duration) *Loop { + l := newReapLoop(t) + l.cfg.WriteTimeout = wt + return l +} + +// fillAndQueue fills the socket's buffers until a write parks, then queues a +// further chunk behind it. +func (c *stall888Conn) fillAndQueue(t *testing.T, l *Loop) { + t.Helper() + fillSendBuffer(t, c.cs.fd) + c.queue(t, l, 64<<10) + if !csWritePending(c.cs) { + t.Fatalf("the socket took the queued bytes; backpressure not reproduced") + } +} + +// sweepUntil sweeps every 10 ms until the conn is gone or the limit passes, +// and returns how long it took. +func sweepUntil(l *Loop, fd int, cs *connState, limit time.Duration, between func()) (closedAfter time.Duration, closed bool) { + start := time.Now() + for time.Since(start) < limit { + l.checkTimeouts() + if l.conns[fd] != cs { + return time.Since(start), true + } + if between != nil { + between() + } + time.Sleep(10 * time.Millisecond) + } + return time.Since(start), false +} + +// TestDetachedPeerThatTakesNothingIsClosed888: no idle deadline, no read, and +// a peer that reads nothing: closed after WriteTimeout, not before. +func TestDetachedPeerThatTakesNothingIsClosed888(t *testing.T) { + l := stall888Loop(t, stall888Bound) + c := newStall888Conn(t, l) + c.fillAndQueue(t, l) + // A read of the connection (a WebSocket ping, an SSE client's request) is + // not evidence that the peer takes what is queued: the clock is not keyed + // on it. + c.cs.lastActivity = time.Now().UnixNano() + + el, closed := sweepUntil(l, c.cs.fd, c.cs, 5*time.Second, func() { c.cs.lastActivity = time.Now().UnixNano() }) + if !closed { + t.Fatalf("a detached conn whose peer takes nothing was not closed within %v (WriteTimeout %v)", el, stall888Bound) + } + if el < stall888Bound { + t.Errorf("closed after %v, before WriteTimeout %v", el, stall888Bound) + } + t.Logf("closed %v after the first sweep (WriteTimeout %v)", el.Round(time.Millisecond), stall888Bound) +} + +// TestDetachedStallIsBoundedWhateverTheIdleDeadlineSays888: a WebSocket peer +// that sends keeps pushing the middleware's idle deadline out; the stall bound +// does not depend on it. +func TestDetachedStallIsBoundedWhateverTheIdleDeadlineSays888(t *testing.T) { + l := stall888Loop(t, stall888Bound) + c := newStall888Conn(t, l) + c.fillAndQueue(t, l) + push := func() { c.cs.h1State.IdleDeadlineNs.Store(time.Now().Add(time.Hour).UnixNano()) } + push() + if el, closed := sweepUntil(l, c.cs.fd, c.cs, 5*time.Second, push); !closed { + t.Fatalf("a stalled conn with its idle deadline pushed out was not closed within %v", el) + } +} + +// TestDetachedReadingPeerIsNotClosed888: a peer that takes bytes, slowly, +// keeps its conn for several WriteTimeouts, with bytes queued for it the whole +// time. A clock keyed on the last read of the connection, or on its age, would +// close it. +func TestDetachedReadingPeerIsNotClosed888(t *testing.T) { + l := stall888Loop(t, stall888Bound) + c := newStall888Conn(t, l) + c.fillAndQueue(t, l) + c.cs.lastActivity = time.Now().Add(-time.Hour).UnixNano() // the request is long ago + + buf := make([]byte, 4096) + tick := 0 + run := 5 * stall888Bound + el, closed := sweepUntil(l, c.cs.fd, c.cs, run, func() { + tick++ + if tick%5 == 0 { // every 50 ms: a little + _ = c.peer.SetReadDeadline(time.Now().Add(5 * time.Millisecond)) + _, _ = c.peer.Read(buf) + } + flushDirtyOnce(l) + if csWritePending(c.cs) && len(c.cs.writeBuf)-c.cs.writePos < 64<<10 { + c.queue(t, l, 64<<10) // the producer keeps queueing + } + }) + if closed { + t.Fatalf("a peer that kept reading was closed after %v (WriteTimeout %v)", el, stall888Bound) + } + if !csWritePending(c.cs) { + t.Fatalf("nothing was queued at the end: the control did not keep bytes queued") + } +} + +// TestDetachedStallNeedsAWriteTimeout888: WriteTimeout 0 (none, or -1) is no +// bound. +func TestDetachedStallNeedsAWriteTimeout888(t *testing.T) { + l := stall888Loop(t, 0) + c := newStall888Conn(t, l) + c.fillAndQueue(t, l) + if el, closed := sweepUntil(l, c.cs.fd, c.cs, 3*stall888Bound, nil); closed { + t.Fatalf("closed after %v with WriteTimeout 0", el) + } + if c.cs.closeSince != 0 { + t.Errorf("a stall clock runs with WriteTimeout 0: closeSince=%d", c.cs.closeSince) + } +} + +// TestDetachedStallClockEndsWithTheQueue888: a conn whose queue drained ends +// its clock, so the next queue is not charged for the time before it. +func TestDetachedStallClockEndsWithTheQueue888(t *testing.T) { + l := stall888Loop(t, stall888Bound) + c := newStall888Conn(t, l) + c.fillAndQueue(t, l) + l.checkTimeouts() + if c.cs.closeSince == 0 { + t.Fatalf("no stall clock after a sweep that found bytes queued") + } + // The peer reads it all. + buf := make([]byte, 64<<10) + deadline := time.Now().Add(5 * time.Second) + for csWritePending(c.cs) && time.Now().Before(deadline) { + _ = c.peer.SetReadDeadline(time.Now().Add(20 * time.Millisecond)) + _, _ = c.peer.Read(buf) + flushDirtyOnce(l) + } + if csWritePending(c.cs) { + t.Fatalf("the queue did not drain") + } + l.checkTimeouts() + if c.cs.closeSince != 0 { + t.Errorf("closeSince=%d with nothing queued, want 0", c.cs.closeSince) + } + if l.conns[c.cs.fd] != c.cs { + t.Fatalf("a drained conn was closed") + } +} diff --git a/internal/engine/iouring/complete_send_bench_888_linux_test.go b/internal/engine/iouring/complete_send_bench_888_linux_test.go new file mode 100644 index 000000000..ac77de76e --- /dev/null +++ b/internal/engine/iouring/complete_send_bench_888_linux_test.go @@ -0,0 +1,28 @@ +//go:build linux + +package iouring + +import ( + "testing" + "time" +) + +// BenchmarkCompleteSend888 is completeSend's all-sent path for an ordinary +// (not detached) conn, the common one: a response that went out whole. The +// stall clock of a detached conn (celeris#888) adds a load and a store to it; +// this is the figure to compare against an A/A twin (RULE 74). +func BenchmarkCompleteSend888(b *testing.B) { + w := &Worker{} + cs := &connState{fd: 7, liveIdx: -1, detected: true} + buf := make([]byte, 128) + now := time.Now().UnixNano() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + cs.sendBuf = buf + cs.sending = true + if w.completeSend(cs, cs.fd, len(buf), now, false) { + b.Fatal("closeAfter") + } + } +} diff --git a/internal/engine/iouring/detach_stall_888_linux_test.go b/internal/engine/iouring/detach_stall_888_linux_test.go new file mode 100644 index 000000000..4982a574d --- /dev/null +++ b/internal/engine/iouring/detach_stall_888_linux_test.go @@ -0,0 +1,223 @@ +//go:build linux + +package iouring + +import ( + "net" + "sync" + "sync/atomic" + "syscall" + "testing" + "time" + + "golang.org/x/sys/unix" + + "github.com/goceleris/celeris/internal/conn" + "github.com/goceleris/celeris/internal/engine/internal/errclass" + "github.com/goceleris/celeris/internal/resource" +) + +// A truly-detached conn (WS/SSE) whose peer takes none of the bytes queued for +// it is closed WriteTimeout after the sweep found them queued (celeris#888 +// items 36 and 37, celeris#884 item 66): checkTimeouts skipped a detached conn +// whatever it had queued, and a write on a native engine only queues, so a +// client that stayed connected and read nothing held its stream for ever. The +// clock is "the peer took none of it", not the age of the stream or the last +// read of the connection. The conn is driven through checkTimeouts and +// completeSend against a loopback TCP pair, since the sweep reads the peer's +// acknowledged byte count from a TCP socket. + +const stall888Bound = 300 * time.Millisecond + +type stall888 struct { + w *Worker + cs *connState + peer net.Conn + closeCB atomic.Int32 // OnDetachClose calls +} + +func newStall888(t *testing.T, wt time.Duration) *stall888 { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Skipf("loopback unavailable: %v", err) + } + t.Cleanup(func() { _ = ln.Close() }) + d := net.Dialer{Timeout: 2 * time.Second, Control: func(_, _ string, c syscall.RawConn) error { + return c.Control(func(fd uintptr) { + _ = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUF, 4096) + }) + }} + peer, err := d.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + t.Cleanup(func() { _ = peer.Close() }) + srv, err := ln.Accept() + if err != nil { + t.Fatalf("accept: %v", err) + } + f, err := srv.(*net.TCPConn).File() + if err != nil { + t.Fatalf("File: %v", err) + } + fd, err := unix.Dup(int(f.Fd())) + _ = f.Close() + _ = srv.Close() + if err != nil { + t.Fatalf("dup: %v", err) + } + t.Cleanup(func() { _ = unix.Close(fd) }) + if err := unix.SetNonblock(fd, true); err != nil { + t.Fatalf("set nonblock: %v", err) + } + s := &stall888{peer: peer} + s.w = &Worker{ + conns: make([]*connState, fd+16), + liveConns: make([]int, 0, 4), + errs: &errclass.Counters{}, + activeConns: &atomic.Int64{}, + closeCount: &atomic.Uint64{}, + cfg: resource.Config{WriteTimeout: wt}, + } + cs := &connState{fd: fd, liveIdx: -1, detected: true} + cs.h1State = conn.NewH1State() + cs.h1State.Detached.Store(true) + cs.h1State.WSReady.Store(true) + cs.h1State.OnDetachClose = func() { s.closeCB.Add(1) } + cs.detachMu = &sync.Mutex{} + // A SEND the peer is not taking: 1 MiB in flight, nothing taken yet. + cs.sendBuf = make([]byte, 1<<20) + cs.sending = true + s.cs = cs + s.w.conns[fd] = cs + s.w.addLiveConn(cs) + s.w.connCount++ + s.w.activeConns.Add(1) + s.w.maxFD = fd + return s +} + +// sweepUntilClosing sweeps every 10 ms until the conn is closing (closeConn +// deferred its descriptor behind the SEND in flight) or the limit passes. +func (s *stall888) sweepUntilClosing(limit time.Duration, between func()) (after time.Duration, closing bool) { + start := time.Now() + for time.Since(start) < limit { + s.w.checkTimeouts() + if s.cs.closing { + return time.Since(start), true + } + if between != nil { + between() + } + time.Sleep(10 * time.Millisecond) + } + return time.Since(start), false +} + +// TestDetachedPeerThatTakesNothingIsClosed888: no idle deadline, and a peer +// that takes nothing: closed after WriteTimeout, not before, and the +// middleware is told (OnDetachClose). +func TestDetachedPeerThatTakesNothingIsClosed888(t *testing.T) { + s := newStall888(t, stall888Bound) + // The last read of the connection is fresh and stays so: a WebSocket ping, + // an SSE client's request. The clock is not keyed on it. + s.cs.lastActivity = time.Now().UnixNano() + el, closing := s.sweepUntilClosing(5*time.Second, func() { s.cs.lastActivity = time.Now().UnixNano() }) + if !closing { + t.Fatalf("a detached conn whose peer takes nothing was not closed within %v (WriteTimeout %v)", el, stall888Bound) + } + if el < stall888Bound { + t.Errorf("closed after %v, before WriteTimeout %v", el, stall888Bound) + } + if s.closeCB.Load() != 1 { + t.Errorf("OnDetachClose ran %d times, want 1: the middleware was not told", s.closeCB.Load()) + } +} + +// TestDetachedStallIsBoundedWhateverTheIdleDeadlineSays888: a WebSocket peer +// that sends keeps pushing the middleware's idle deadline out; the stall bound +// does not depend on it. +func TestDetachedStallIsBoundedWhateverTheIdleDeadlineSays888(t *testing.T) { + s := newStall888(t, stall888Bound) + push := func() { s.cs.h1State.IdleDeadlineNs.Store(time.Now().Add(time.Hour).UnixNano()) } + push() + if el, closing := s.sweepUntilClosing(5*time.Second, push); !closing { + t.Fatalf("a stalled conn with its idle deadline pushed out was not closed within %v", el) + } +} + +// TestDetachedPeerThatTakesBytesIsNotClosed888: the clock restarts when a SEND +// took bytes (completeSend) and when the peer's acknowledged count moved. A +// conn whose peer keeps taking some is not closed for several WriteTimeouts. +func TestDetachedPeerThatTakesBytesIsNotClosed888(t *testing.T) { + for _, mode := range []string{"send-completions", "acked-bytes"} { + t.Run(mode, func(t *testing.T) { + s := newStall888(t, stall888Bound) + s.cs.lastActivity = time.Now().Add(-time.Hour).UnixNano() // the request is long ago + // Fill the socket so that acknowledged bytes are the peer's reads. + filled := fillSocket888(t, s.cs.fd) + t.Logf("socket took %d bytes before parking", filled) + buf := make([]byte, 4096) + tick := 0 + run := 5 * stall888Bound + el, closing := s.sweepUntilClosing(run, func() { + tick++ + if tick%5 != 0 { // every 50 ms + return + } + switch mode { + case "acked-bytes": + // The peer reads a little; no completion reaches the worker + // (an inline write of the stream's goroutine, or one very + // large SEND, completes later). + _ = s.peer.SetReadDeadline(time.Now().Add(5 * time.Millisecond)) + _, _ = s.peer.Read(buf) + case "send-completions": + // A SEND that took bytes completed and another is queued + // behind it (zcNotifPending stands in for the re-send, which + // needs a ring). + s.cs.detachMu.Lock() + s.cs.zcNotifPending = true + s.cs.sendBuf = make([]byte, 1<<20) + s.cs.sending = true + s.w.completeSend(s.cs, s.cs.fd, 4096, time.Now().UnixNano(), false) + s.cs.sending = true + s.cs.zcNotifPending = false + s.cs.detachMu.Unlock() + } + }) + if closing { + t.Fatalf("a peer that kept taking bytes was closed after %v (WriteTimeout %v)", el, stall888Bound) + } + }) + } +} + +// fillSocket888 writes into fd until the kernel refuses more. +func fillSocket888(t *testing.T, fd int) int { + t.Helper() + chunk := make([]byte, 32<<10) + total := 0 + for total < 64<<20 { + n, err := unix.Write(fd, chunk) + if err == unix.EAGAIN { + return total + } + if err != nil { + t.Fatalf("fill write: %v", err) + } + total += n + } + t.Fatalf("socket never filled") + return total +} + +// TestDetachedStallNeedsAWriteTimeout888: WriteTimeout 0 (none, or -1) is no +// bound. +func TestDetachedStallNeedsAWriteTimeout888(t *testing.T) { + s := newStall888(t, 0) + if el, closing := s.sweepUntilClosing(3*stall888Bound, nil); closing { + t.Fatalf("closed after %v with WriteTimeout 0", el) + } +} diff --git a/middleware/sse/sse_disconnect_linux_test.go b/middleware/sse/sse_disconnect_linux_test.go index 7bbc6ef1e..4994c8128 100644 --- a/middleware/sse/sse_disconnect_linux_test.go +++ b/middleware/sse/sse_disconnect_linux_test.go @@ -89,12 +89,23 @@ func sseDisconnectCells(t *testing.T) []sseCell { // celeris#498 is about. The H1 cells leave it off: there the per-conn // dispatch goroutine is chosen by Config.AsyncHandlers. func startSSEServer(tb testing.TB, engine celeris.EngineType, async, routeAsync bool, h celeris.HandlerFunc) (string, func()) { + tb.Helper() + return startSSEServerCfg(tb, celeris.Config{ + Engine: engine, + AsyncHandlers: async, + ShutdownTimeout: 2 * time.Second, + }, routeAsync, h) +} + +// startSSEServerCfg is startSSEServer with the server's Config given whole, +// for a test that sets timeouts (celeris#888 item 36). +func startSSEServerCfg(tb testing.TB, cfg celeris.Config, routeAsync bool, h celeris.HandlerFunc) (string, func()) { tb.Helper() began := time.Now() deadline := began.Add(startRetryWindow) retries := 0 for { - addr, shutdown, err := trySSEServer(tb, engine, async, routeAsync, h) + addr, shutdown, err := trySSEServer(tb, cfg, routeAsync, h) if err == nil { if retries > 0 { tb.Logf("startSSEServer: started after %d retries in %v", retries, time.Since(began).Round(time.Millisecond)) @@ -120,13 +131,9 @@ func startSSEServer(tb testing.TB, engine celeris.EngineType, async, routeAsync // listener, so a retry never reuses a listener that a failed Start closed. // It returns the error instead of failing the test, so the caller can // classify it (classifyStart). -func trySSEServer(tb testing.TB, engine celeris.EngineType, async, routeAsync bool, h celeris.HandlerFunc) (string, func(), error) { +func trySSEServer(tb testing.TB, cfg celeris.Config, routeAsync bool, h celeris.HandlerFunc) (string, func(), error) { tb.Helper() - s := celeris.New(celeris.Config{ - Engine: engine, - AsyncHandlers: async, - ShutdownTimeout: 2 * time.Second, - }) + s := celeris.New(cfg) r := s.GET("/events", h) if routeAsync { r.Async() diff --git a/middleware/sse/sse_nonreading_888_linux_test.go b/middleware/sse/sse_nonreading_888_linux_test.go new file mode 100644 index 000000000..9bba6db8a --- /dev/null +++ b/middleware/sse/sse_nonreading_888_linux_test.go @@ -0,0 +1,352 @@ +//go:build linux + +package sse_test + +import ( + "net" + "os" + "strconv" + "strings" + "syscall" + "testing" + "time" + + "golang.org/x/sys/unix" + + "github.com/goceleris/celeris" + "github.com/goceleris/celeris/middleware/sse" +) + +// A client that stays connected and reads nothing of an SSE stream +// (celeris#888 items 36 and 37, celeris#884 item 66). +// +// On epoll and io_uring a detached connection was exempt from WriteTimeout +// (checkTimeouts skipped it), and Client.Send never parks there: the stream's +// drain goroutine empties the 4-slot queue into the engine's own buffer, so +// ClientPolicyClose was never reached either. Nothing ended the stream of a +// client that took nothing, short of the 64 MiB cap on what the engine holds +// for it. WriteTimeout now bounds it as "no progress for WriteTimeout" (the +// peer takes no byte of what the engine has queued for it), and the close +// reaches the stream through SetWSDetachClose, so OnDisconnect runs about +// WriteTimeout after the stream stalled, as it does on std. + +const nonReadWriteTimeout888 = time.Second + +// nonReadCell888 is one engine configuration of the matrix. +type nonReadCell888 struct { + name string + engine celeris.EngineType + async bool +} + +func nonReadCells888(t *testing.T) []nonReadCell888 { + t.Helper() + var cells []nonReadCell888 + hasURing := false + for _, kind := range sseNativeEngineKinds(t) { + if kind == celeris.IOUring { + hasURing = true + } + cells = append(cells, + nonReadCell888{kind.String() + "/async", kind, true}, + nonReadCell888{kind.String() + "/sync", kind, false}, + ) + } + if hasURing { + cells = append(cells, nonReadCell888{"adaptive/async", celeris.Adaptive, true}) + } + // std is the control: its writes are bounded by WriteTimeout one at a time. + return append(cells, nonReadCell888{"std/async", celeris.Std, true}) +} + +// producer888 shapes what the handler sends: one event of burstBytes at once, +// which fills the socket buffers (a few MB on a loopback connection whose peer +// has a 4 KiB receive buffer) and leaves the rest queued in the engine, then, +// after settle, trickleBytes every trickleEvery until the client is gone. One +// event, not many, and a pause before the next: while the engine moves a burst +// of MB into the socket, it holds the connection's write lock for each partial +// send, the drain goroutine waits for it, and events sent meanwhile overflow the +// 4-slot queue: ClientPolicyClose would end the stream for that reason, +// whatever the engine did about the peer. The totals stay far +// below the 64 MiB the engine holds for a detached conn (celeris#1067): that +// cap closes the conn too, and a test that reached it would pass without the +// stall bound. +// +// With fastFor set, the burst is skipped and fastBytes every fastEvery is sent +// for fastFor first (the shape of a chat or ticker stream whose peer stopped +// reading: many small events, until the socket is full), then the trickle. +type producer888 struct { + burstBytes int + settle time.Duration + fastBytes int + fastEvery time.Duration + fastFor time.Duration + trickleBytes int + trickleEvery time.Duration + + // writeTimeout is the server's WriteTimeout (nonReadWriteTimeout888 when + // zero). noHeaderTimer sets ReadHeaderTimeout -1, which takes away the + // epoll engine's 25 ms timerfd: with a detached conn the sweep then + // depends on the loop's own iterations. + writeTimeout time.Duration + noHeaderTimer bool +} + +func (p producer888) wt() time.Duration { + if p.writeTimeout > 0 { + return p.writeTimeout + } + return nonReadWriteTimeout888 +} + +// burstFor888 is a burst that exceeds the send queue of a loopback connection +// whose peer has a 4 KiB receive buffer: twice the largest send buffer the +// kernel autotunes to (tcp_wmem), at least 8 MiB. It skips the test where that +// is over 24 MiB (a tuned host), since the engine's own cap is 64 MiB. +func burstFor888(tb testing.TB) int { + tb.Helper() + wmem := 4 << 20 + if b, err := os.ReadFile("/proc/sys/net/ipv4/tcp_wmem"); err == nil { + if f := strings.Fields(string(b)); len(f) == 3 { + if n, err := strconv.Atoi(f[2]); err == nil { + wmem = n + } + } + } + n := max(2*wmem, 8<<20) + if n > 24<<20 { + tb.Skipf("tcp_wmem allows a send buffer of %d MiB: a burst that fills it nears the engine's 64 MiB cap", wmem>>20) + } + return n +} + +// serve888 runs an SSE stream that produces by p until the client is gone, +// and reports when OnDisconnect ran. MaxQueueDepth 4 with ClientPolicyClose is +// the shape the finding names: the policy that was never reached. +func serve888(tb testing.TB, cell nonReadCell888, p producer888) (addr string, disconnected <-chan time.Time, shutdown func()) { + tb.Helper() + done := make(chan time.Time, 1) + var burst string + if p.fastFor == 0 { + burst = strings.Repeat("x", p.burstBytes) + } + trickle := strings.Repeat("y", p.trickleBytes) + handler := sse.New(sse.Config{ + HeartbeatInterval: -1, + MaxQueueDepth: 4, + OnSlowClient: func(*sse.Client, sse.Event) sse.ClientPolicy { return sse.ClientPolicyClose }, + OnDisconnect: func(*celeris.Context, *sse.Client) { + select { + case done <- time.Now(): + default: + } + }, + Handler: func(client *sse.Client) { + ctx := client.Context() + if p.fastFor > 0 { + fast := strings.Repeat("z", p.fastBytes) + tk := time.NewTicker(p.fastEvery) + defer tk.Stop() + for end := time.Now().Add(p.fastFor); time.Now().Before(end); { + select { + case <-ctx.Done(): + return + case <-tk.C: + if err := client.Send(sse.Event{Event: "fast", Data: fast}); err != nil { + return + } + } + } + } else if err := client.Send(sse.Event{Event: "burst", Data: burst}); err != nil { + return + } + select { + case <-ctx.Done(): + return + case <-time.After(p.settle): + } + tick := time.NewTicker(p.trickleEvery) + defer tick.Stop() + for { + select { + case <-ctx.Done(): + return + case <-tick.C: + if err := client.Send(sse.Event{Event: "tick", Data: trickle}); err != nil { + return + } + } + } + }, + }) + cfg := celeris.Config{ + Engine: cell.engine, + AsyncHandlers: cell.async, + WriteTimeout: p.wt(), + ShutdownTimeout: 2 * time.Second, + } + if p.noHeaderTimer { + cfg.ReadHeaderTimeout = -1 + } + addr, shutdown = startSSEServerCfg(tb, cfg, false, handler) + return addr, done, shutdown +} + +// dialSmallRcvBuf888 dials addr with a 4 KiB receive buffer set before the +// connect, so that the server's send queue to this client fills within the +// first few hundred KiB, sends the SSE request and reads exactly the response +// head, byte by byte. It returns the conn with nothing more read. +func dialSmallRcvBuf888(tb testing.TB, addr string) net.Conn { + tb.Helper() + d := net.Dialer{ + Timeout: 2 * time.Second, + Control: func(_, _ string, c syscall.RawConn) error { + var serr error + if err := c.Control(func(fd uintptr) { + serr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUF, 4096) + }); err != nil { + return err + } + return serr + }, + } + conn, err := d.Dial("tcp", addr) + if err != nil { + tb.Fatalf("dial: %v", err) + } + tb.Cleanup(func() { _ = conn.Close() }) + _ = conn.SetDeadline(time.Now().Add(10 * time.Second)) + req := "GET /events HTTP/1.1\r\nHost: " + addr + "\r\nAccept: text/event-stream\r\nConnection: keep-alive\r\n\r\n" + if _, err := conn.Write([]byte(req)); err != nil { + tb.Fatalf("write request: %v", err) + } + var head []byte + one := make([]byte, 1) + for !strings.HasSuffix(string(head), "\r\n\r\n") { + if _, err := conn.Read(one); err != nil { + tb.Fatalf("read head after %q: %v", head, err) + } + head = append(head, one[0]) + } + if !strings.HasPrefix(string(head), "HTTP/1.1 200") { + tb.Fatalf("response head %q, want 200", head) + } + _ = conn.SetDeadline(time.Time{}) + return conn +} + +// TestNonReadingClientEndsStream888 pins the finding: a client that took the +// head and reads nothing more ends its stream about WriteTimeout after the +// engine's queue for it stalled, and not before WriteTimeout. The lower bound +// is what tells this close from an early one (a cut at accept or at the first +// event); the producers here stay far below the 64 MiB cap (celeris#1067) for +// the whole run, so it is not the cap that ends the stream either. +func TestNonReadingClientEndsStream888(t *testing.T) { + prods := []struct { + name string + p producer888 + }{ + // The burst fills the socket and queues the rest in the engine; then + // a trickle of 1 KB every 5 ms. + {"steady", producer888{burstBytes: burstFor888(t), settle: 700 * time.Millisecond, trickleBytes: 1000, trickleEvery: 5 * time.Millisecond}}, + // The same, then one event every 300 ms: a sweep must still reach the + // conn when almost nothing wakes the worker. + {"sparse", producer888{burstBytes: burstFor888(t), settle: 700 * time.Millisecond, trickleBytes: 1000, trickleEvery: 300 * time.Millisecond}}, + // No burst: 4 KB events every 0.5 ms for 1.2 s fill the socket with + // small sends (an io_uring SEND of under 64 KiB is not one the worker + // is told to expect to wait for), then one event every 300 ms. + // WriteTimeout 2 s puts the close after the fast phase. + {"small-events", producer888{fastBytes: 4 << 10, fastEvery: 500 * time.Microsecond, fastFor: 1200 * time.Millisecond, settle: 100 * time.Millisecond, + trickleBytes: 1000, trickleEvery: 300 * time.Millisecond, writeTimeout: 2 * time.Second}}, + // The same without the epoll timerfd (ReadHeaderTimeout -1), so nothing + // but the loop's own iterations reaches the sweep. + {"small-events-no-header-timer", producer888{fastBytes: 4 << 10, fastEvery: 500 * time.Microsecond, fastFor: 1200 * time.Millisecond, settle: 100 * time.Millisecond, + trickleBytes: 1000, trickleEvery: 300 * time.Millisecond, writeTimeout: 2 * time.Second, noHeaderTimer: true}}, + } + for _, cell := range nonReadCells888(t) { + for _, pr := range prods { + t.Run(cell.name+"/"+pr.name, func(t *testing.T) { + addr, disconnected, shutdown := serve888(t, cell, pr.p) + defer shutdown() + began := time.Now() + _ = dialSmallRcvBuf888(t, addr) + const slack = 5 * time.Second + wt := pr.p.wt() + select { + case at := <-disconnected: + el := at.Sub(began) + t.Logf("OnDisconnect %v after connect (WriteTimeout %v)", el.Round(time.Millisecond), wt) + // std ends a stream whose write is blocked as soon as the + // queue behind it is full (ClientPolicyClose), earlier than + // WriteTimeout: the lower bound is for the native engines. + if cell.engine != celeris.Std && el < wt { + t.Errorf("OnDisconnect ran %v after connect, before WriteTimeout %v: the stream was ended by something other than the stall bound", el, wt) + } + case <-time.After(wt + slack): + t.Fatalf("OnDisconnect never ran %v after connect (WriteTimeout %v): a client that reads nothing holds its stream for ever (celeris#888 item 36)", + wt+slack, wt) + } + }) + } + } +} + +// TestSlowReadingClientKeepsStream888 is the control that the bound is not +// "the stream is older than WriteTimeout": a client that reads steadily, but +// slower than the producer writes, keeps a backlog queued for the whole run +// (the stall bound's trigger is on all that time) and its stream must live for +// several WriteTimeouts. Bytes the client takes restart the clock; a clock +// keyed on the last read of the connection (the request, which is long ago) +// would end it. +func TestSlowReadingClientKeepsStream888(t *testing.T) { + for _, cell := range nonReadCells888(t) { + if cell.engine == celeris.Std { + continue // std closes a stream whose queue is full (ClientPolicyClose) at once + } + t.Run(cell.name, func(t *testing.T) { + // The burst fills the socket and queues the rest in the engine; + // then 400 KB/s into a client that takes about 100 KB/s: bytes are + // queued for it the whole run, and the backlog (a few MB) stays far + // below 64 MiB. + addr, disconnected, shutdown := serve888(t, cell, producer888{burstBytes: burstFor888(t), settle: 700 * time.Millisecond, trickleBytes: 4 << 10, trickleEvery: 10 * time.Millisecond}) + defer shutdown() + conn := dialSmallRcvBuf888(t, addr) + stop := make(chan struct{}) + readDone := make(chan int64, 1) + go func() { + var total int64 + buf := make([]byte, 4096) + tk := time.NewTicker(40 * time.Millisecond) + defer tk.Stop() + for { + select { + case <-stop: + readDone <- total + return + case <-tk.C: + } + _ = conn.SetReadDeadline(time.Now().Add(time.Second)) + n, err := conn.Read(buf) + total += int64(n) + if err != nil { + readDone <- total + return + } + } + }() + const run = 5 * nonReadWriteTimeout888 + select { + case <-disconnected: + close(stop) + t.Fatalf("the stream of a client that kept reading (%d bytes taken) was ended before %v (WriteTimeout %v)", + <-readDone, run, nonReadWriteTimeout888) + case <-time.After(run): + } + close(stop) + if got := <-readDone; got < 100<<10 { + t.Fatalf("the client took only %d bytes in %v: the control did not read", got, run) + } + t.Logf("the stream lived %v with a reading client", run) + }) + } +} diff --git a/middleware/websocket/engine_linux_test.go b/middleware/websocket/engine_linux_test.go index a8ba0e027..eba882477 100644 --- a/middleware/websocket/engine_linux_test.go +++ b/middleware/websocket/engine_linux_test.go @@ -62,7 +62,14 @@ func startNativeServer(tb testing.TB, kind celeris.EngineType, cfg Config) (stri // survives shutdown, so EngineInfo() still answers). func startNativeServerWithHandle(tb testing.TB, kind celeris.EngineType, cfg Config) (string, func(), *celeris.Server) { tb.Helper() - s := celeris.New(celeris.Config{Engine: kind}) + return startNativeServerCfg(tb, celeris.Config{Engine: kind}, cfg) +} + +// startNativeServerCfg is startNativeServerWithHandle with the server's Config +// given whole, for a test that sets timeouts (celeris#888 items 36 and 37). +func startNativeServerCfg(tb testing.TB, ccfg celeris.Config, cfg Config) (string, func(), *celeris.Server) { + tb.Helper() + s := celeris.New(ccfg) s.GET("/ws", New(cfg)) ln, err := net.Listen("tcp", "127.0.0.1:0") diff --git a/middleware/websocket/nonreading_888_linux_test.go b/middleware/websocket/nonreading_888_linux_test.go new file mode 100644 index 000000000..6a53515a0 --- /dev/null +++ b/middleware/websocket/nonreading_888_linux_test.go @@ -0,0 +1,118 @@ +//go:build linux + +package websocket + +import ( + "bufio" + "net" + "strings" + "syscall" + "testing" + "time" + + "golang.org/x/sys/unix" + + "github.com/goceleris/celeris" +) + +// A WebSocket peer that reads nothing of what it is sent gets the same bound as +// an SSE client (celeris#888 items 36 and 37): the native engines apply +// WriteTimeout to a detached connection as "no progress for WriteTimeout", +// whatever the middleware's idle deadline says. A peer that sends pings keeps +// pushing that deadline out, which is the peer this test is. + +const wsStall888Timeout = time.Second + +// dialSmallRcvBuf888 dials addr with a 4 KiB receive buffer set before the +// connect, and upgrades. +func dialSmallRcvBuf888(tb testing.TB, addr string) *testWSClient { + tb.Helper() + d := net.Dialer{Timeout: 2 * time.Second, Control: func(_, _ string, c syscall.RawConn) error { + return c.Control(func(fd uintptr) { + _ = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUF, 4096) + }) + }} + conn, err := d.Dial("tcp", addr) + if err != nil { + tb.Fatal(err) + } + tb.Cleanup(func() { _ = conn.Close() }) + c := &testWSClient{conn: conn, br: bufio.NewReaderSize(conn, 16), bw: bufio.NewWriter(conn)} + req := "GET /ws HTTP/1.1\r\nHost: localhost\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 13\r\n\r\n" + _, _ = c.bw.WriteString(req) + _ = c.bw.Flush() + line, err := c.br.ReadString('\n') + if err != nil || !strings.Contains(line, "101") { + tb.Fatalf("upgrade: %q, %v", line, err) + } + for { + if line, err = c.br.ReadString('\n'); err != nil { + tb.Fatal(err) + } else if line == "\r\n" { + return c + } + } +} + +func TestNonReadingPeerEndsConn888(t *testing.T) { + for _, kind := range engineKinds(t) { + t.Run(kind.String(), func(t *testing.T) { + wrote := make(chan error, 1) + burst := make([]byte, 12<<20) + addr, shutdown, _ := startNativeServerCfg(t, celeris.Config{Engine: kind, WriteTimeout: wsStall888Timeout}, + Config{ + // The idle deadline is in the future and pushed out by each + // ping: nothing but the stall bound can end this conn. + IdleTimeout: time.Hour, + Handler: func(c *Conn) { + go func() { // the read side, so the pings are seen + for { + if _, _, err := c.ReadMessage(); err != nil { + return + } + } + }() + if err := c.WriteMessage(BinaryMessage, burst); err != nil { + wrote <- err + return + } + for { + time.Sleep(20 * time.Millisecond) + if err := c.WriteMessage(BinaryMessage, burst[:1000]); err != nil { + wrote <- err + return + } + } + }, + }) + defer shutdown() + + began := time.Now() + client := dialSmallRcvBuf888(t, addr) + // Pings, and no reads. + stop := make(chan struct{}) + defer close(stop) + go func() { + for { + select { + case <-stop: + return + case <-time.After(100 * time.Millisecond): + _ = client.writeClientFrame(true, OpPing, []byte("p")) + } + } + }() + select { + case err := <-wrote: + el := time.Since(began) + t.Logf("the handler's write failed %v after connect (WriteTimeout %v): %v", el.Round(time.Millisecond), wsStall888Timeout, err) + if el < wsStall888Timeout { + t.Errorf("ended %v after connect, before WriteTimeout %v", el, wsStall888Timeout) + } + case <-time.After(wsStall888Timeout + 6*time.Second): + t.Fatalf("a WebSocket peer that sends pings and reads nothing was not closed within %v (WriteTimeout %v): celeris#888 item 36", + wsStall888Timeout+6*time.Second, wsStall888Timeout) + } + }) + } +} From 9915418479465bbca0457c4158839383221618df Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Sun, 18 Oct 2026 12:32:17 +0200 Subject: [PATCH 02/14] epoll, iouring: a detached conn whose peer takes none of its queued bytes for WriteTimeout is closed (v1.6.0 polish items) A truly-detached conn (WebSocket, SSE) was exempt from WriteTimeout, and a write on a native engine only queues, so a client that stayed connected and read nothing held its stream for ever (the SSE queue never filled, so ClientPolicyClose was never reached). The sweep now closes a detached conn that has bytes queued and whose peer took none for WriteTimeout, measured on the bytes the peer acknowledges and the SENDs that completed, never on the last read of the connection. The close reaches the middleware as any engine close does, so an SSE stream's context ends and OnDisconnect runs. Refs #888 item 36 Refs #888 item 37 Refs #884 item 66 --- config.go | 15 ++- internal/engine/epoll/conn.go | 7 +- .../detach_stall_nontcp_888_linux_test.go | 63 +++++++++++ internal/engine/epoll/loop.go | 85 +++++++++++++++ internal/engine/iouring/conn.go | 20 +++- .../detach_stall_clock_888_linux_test.go | 44 ++++++++ internal/engine/iouring/worker.go | 100 ++++++++++++++++++ middleware/sse/doc.go | 26 +++-- middleware/websocket/config.go | 21 ++-- 9 files changed, 357 insertions(+), 24 deletions(-) create mode 100644 internal/engine/epoll/detach_stall_nontcp_888_linux_test.go create mode 100644 internal/engine/iouring/detach_stall_clock_888_linux_test.go diff --git a/config.go b/config.go index 7c300e5d1..ac369131b 100644 --- a/config.go +++ b/config.go @@ -133,10 +133,17 @@ type Config struct { // is not cut at ReadTimeout or WriteTimeout as a whole on any engine, so // an SSE stream stays open for as long as its client keeps reading, // without setting WriteTimeout to -1. The native engines (epoll, io_uring, - // adaptive) apply neither timeout to a detached connection. On the std - // engine net/http arms WriteTimeout as one deadline for the whole - // response and Detach clears it for that request (ReadTimeout never cuts - // a stream there: net/http has stopped it by the time the handler runs). + // adaptive) apply ReadTimeout and IdleTimeout to a detached connection + // not at all, and WriteTimeout as "no progress for WriteTimeout": a + // detached connection with bytes queued for its peer, which takes none of + // them for WriteTimeout, is closed (the stream's context ends and its + // OnDisconnect runs), whatever the middleware's idle deadline or the last + // read of the connection. A peer that reads, however slowly, and a + // connection with nothing queued are not closed; -1 sets no such bound. + // On the std engine net/http arms WriteTimeout as one deadline for the + // whole response and Detach clears it for that request (ReadTimeout never + // cuts a stream there: net/http has stopped it by the time the handler + // runs). // // On std, WriteTimeout then bounds a detached stream's writes one at a // time: a Write or Flush that has not finished within WriteTimeout fails diff --git a/internal/engine/epoll/conn.go b/internal/engine/epoll/conn.go index 9250e97a4..b7786084c 100644 --- a/internal/engine/epoll/conn.go +++ b/internal/engine/epoll/conn.go @@ -280,7 +280,12 @@ type connState struct { // there is one) and closeSampled when that was. checkTimeouts reaps a conn // whose closeSince is older than closingDrainBound, and measures nothing // else on it: lastActivity is the last READ, and a conn that is closing is - // no longer read. 0 when no deferred close is under way. Worker-thread- + // no longer read. 0 when no deferred close is under way. A truly-detached + // conn (WS/SSE) is never in that state (closeWhenFlushed closes it at + // once), and uses the same fields as the clock of a stall: how long its peer + // has taken none of the bytes queued for it, bounded by WriteTimeout + // (detachedStalled, celeris#888 items 36 and 37), started by the sweep that + // finds bytes queued and cleared by the one that finds none. Worker-thread- // only; reset on release. closeSince int64 closePending int diff --git a/internal/engine/epoll/detach_stall_nontcp_888_linux_test.go b/internal/engine/epoll/detach_stall_nontcp_888_linux_test.go new file mode 100644 index 000000000..e21415303 --- /dev/null +++ b/internal/engine/epoll/detach_stall_nontcp_888_linux_test.go @@ -0,0 +1,63 @@ +//go:build linux + +package epoll + +import ( + "sync" + "testing" + + "golang.org/x/sys/unix" + + "github.com/goceleris/celeris/internal/conn" +) + +// TestDetachedStallNeedsAnAckedCount888: the stall bound reads the peer's +// acknowledged byte count (tcp_info), which a socket that is not TCP does not +// report. The stream's own goroutine flushes with no lock the loop's clocks are +// under, so without that count a peer that reads cannot be told from one that +// does not, and the conn is not closed (no bound, as before celeris#888 items +// 36 and 37) rather than on no evidence. +func TestDetachedStallNeedsAnAckedCount888(t *testing.T) { + l := stall888Loop(t, stall888Bound) + pair, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM, 0) + if err != nil { + t.Skipf("socketpair unavailable: %v", err) + } + local, peer := pair[0], pair[1] + t.Cleanup(func() { _ = unix.Close(peer) }) + if local >= connTableSize { + _ = unix.Close(local) + t.Skip("fd out of table range") + } + if err := unix.SetNonblock(local, true); err != nil { + t.Fatalf("set nonblock: %v", err) + } + cs := &connState{fd: local, liveIdx: -1} + cs.h1State = conn.NewH1State() + cs.h1State.Detached.Store(true) + cs.detachMu = &sync.Mutex{} + registerForOut885(t, l, cs) + l.conns[local] = cs + l.addLiveConn(cs) + l.connCount++ + l.activeConns.Add(1) + l.detachedCount = 1 + t.Cleanup(func() { + if l.conns[local] == cs { + l.closeConn(local) + } + }) + + fillSendBuffer(t, local) + cs.writeBuf = append(cs.writeBuf, make([]byte, 64<<10)...) + cs.pendingBytes = len(cs.writeBuf) + if err := l.flushWrites(cs, true); err != nil { + t.Fatalf("flush: %v", err) + } + if !csWritePending(cs) { + t.Fatalf("the socket took the queued bytes; backpressure not reproduced") + } + if el, closed := sweepUntil(l, local, cs, 3*stall888Bound, nil); closed { + t.Fatalf("a conn on a socket with no acknowledged count was closed after %v (WriteTimeout %v)", el, stall888Bound) + } +} diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go index c36d84247..840f9b9cc 100644 --- a/internal/engine/epoll/loop.go +++ b/internal/engine/epoll/loop.go @@ -4181,6 +4181,20 @@ func (l *Loop) checkTimeouts() { // extends it): forget any grace stamp so a later expiry // starts with a full drain window. cs.drainDeadline = 0 + // WriteTimeout is the one engine timeout a detached stream + // keeps: "no progress for WriteTimeout" on what is queued for + // the peer (celeris#888 items 36 and 37). Here, whatever the + // middleware's idle deadline says: a WebSocket peer that sends + // pings pushes it out for ever and reads nothing of what it is + // sent. The expired case below is the middleware ending the + // stream and has its own bound. + if snap.writePending { + if l.detachedStalled(cs, now) { + l.closeConn(fd) + } + } else if cs.closeSince != 0 { + cs.endDetachedStall() + } continue } // The reap doubles as "this stream is over, drop the conn": @@ -4276,6 +4290,77 @@ func (l *Loop) checkTimeouts() { } } +// detachedStalled is checkTimeouts' look at a truly-detached conn (WS/SSE) +// that has bytes queued for its peer: whether the peer has taken none of them +// for WriteTimeout (celeris#888 items 36 and 37, celeris#884 item 66). The +// middleware owns the stream's lifecycle, so no ReadTimeout, IdleTimeout or +// absolute WriteTimeout applies to it; but a client that stays connected and +// reads nothing never fails a write on a native engine (the write only queues), +// and nothing else ended its stream short of the 64 MiB cap on what the engine +// holds for it. The close reaches the middleware as any engine close does +// (OnDetachClose, so an SSE stream's context ends and OnDisconnect runs). +// +// The clock is closeSince, the one a conn whose close is deferred behind its +// response has (a detached conn is never in that state: closeWhenFlushed closes +// it at once, and a half-close is the middleware's to see), started at the +// first sweep that finds bytes queued and restarted only by evidence that the +// peer takes them: the bytes it has acknowledged moved (noteClosingAcked, a +// sample every quarter of the bound, the first being the baseline, so a peer +// that took bytes in the first quarter and then none is closed up to a quarter +// of the bound earlier than WriteTimeout after its last byte). Not lastActivity, which is the last read: +// a WebSocket peer that sends and reads nothing, or an SSE client's request, +// says nothing about what it takes. The acknowledged count sees the bytes that +// the stream's own goroutine writes through the guarded hook (it flushes with +// no lock the loop's clocks are under) as well as the loop's flushes, and costs +// nothing per write. The sweep does the sampling after it released detachMu, +// and the clock is the loop thread's alone, as for a closing conn. +// +// A socket that does not report the acknowledged count has no bound (see the +// check below). +// +// WriteTimeout 0 (no timeout, or -1) is no bound: unlike a closing conn, which +// has no handler to wait for and gets closingDrainFloor, a stream is the +// middleware's to end, and a server that turned the write timeout off asked for +// a client to be held for as long as it stays. A conn with nothing queued ends +// the clock (endDetachedStall). Loop thread. +func (l *Loop) detachedStalled(cs *connState, now int64) bool { + bound := int64(l.cfg.WriteTimeout) + if bound <= 0 { + return false + } + if cs.closeSince == 0 { + // A new stall. No sample yet: a conn whose queue is there at one + // sweep and gone at the next (a hot broadcaster's) would pay a + // getsockopt per sweep for an episode that ends before it matters. + // The first sample, the baseline, is due a quarter of the bound from + // now. + cs.closeSince = now + cs.closeSampled = now + cs.closeAckedKnown = false + return false + } + if now-cs.closeSampled >= bound/4 || (now-cs.closeSince > bound && cs.closeSampled <= cs.closeSince+bound) { + l.noteClosingAcked(cs, now) + } + if cs.closeSampled > cs.closeSince && !cs.closeAckedKnown { + // A sample was taken since the stall began and the socket gave no + // acknowledged count (not a TCP socket, or a kernel before Linux 4.1): + // the stream's own goroutine flushes with no lock the loop's clocks are + // under, and nothing else sees the bytes it writes, so a peer that + // reads cannot be told from one that does not. No bound, as before this + // one. + return false + } + return now-cs.closeSince > bound +} + +// endDetachedStall ends the stall clock of a detached conn that has nothing +// queued (detachedStalled). Loop thread. +func (cs *connState) endDetachedStall() { + cs.closeSince = 0 + cs.closeAckedKnown = false +} + // sweepWriteClock is checkTimeouts' look at a live conn that holds a response // the kernel has not taken whole (celeris#885 items 35 and 45): it has the // write clock (writeSince), or is on the dirty list. WriteTimeout is "no diff --git a/internal/engine/iouring/conn.go b/internal/engine/iouring/conn.go index 7a60c718e..387d3c718 100644 --- a/internal/engine/iouring/conn.go +++ b/internal/engine/iouring/conn.go @@ -268,6 +268,21 @@ type connState struct { lastActivity int64 // nanosecond timestamp of last I/O activity (for timeout checks) + // stallSince is the clock of a truly-detached conn (WS/SSE) that has bytes + // queued for its peer (celeris#888 items 36 and 37, celeris#884 item 66): + // when the sweep last saw the peer take some of them, in nanoseconds, or + // first saw them queued. checkTimeouts closes the conn once it is older + // than WriteTimeout (detachedStalled). Restamped by completeSend when a + // SEND took bytes and by the sweep when the peer's acknowledged count moved + // (closeAcked); 0 when nothing is queued (completeSend, and the sweep that + // finds the queue empty, clear it) or the conn is not detached. Not + // lastActivity, which is the last read: a stream's client sends one request, + // or a WebSocket peer sends pings and reads nothing, and says nothing about + // what it takes. Worker-thread-only (the stream's goroutine, which flushes + // inline under detachMu, is seen through the acknowledged count); reset on + // release. Declared beside lastActivity: completeSend writes both. + stallSince int64 + // closeAcked, closeSampled and closeAckedKnown are the sweep's second kind // of evidence that a closing conn's peer is taking its response // (celeris#884 item 52): the peer's acknowledged byte count (tcp_info @@ -275,7 +290,9 @@ type connState struct { // closeAcked holds a sample. checkTimeouts samples a closing conn at most // every closingDrainBound/4 (noteClosingAcked). The first kind is // completeSend's restamp of lastActivity, the drain's clock, which moves - // once per SEND completion. Worker-thread-only; reset on release. + // once per SEND completion. The sweep samples a detached conn that has + // bytes queued the same way (stallSince, above). Worker-thread-only; reset + // on release. closeAcked uint64 closeSampled int64 closeAckedKnown bool @@ -693,6 +710,7 @@ func releaseConnState(cs *connState) { cs.closeAcked = 0 cs.closeSampled = 0 cs.closeAckedKnown = false + cs.stallSince = 0 cs.recvStallSince = 0 cs.linkArmedAt = 0 cs.lastRecvCQE = 0 diff --git a/internal/engine/iouring/detach_stall_clock_888_linux_test.go b/internal/engine/iouring/detach_stall_clock_888_linux_test.go new file mode 100644 index 000000000..a1636aa28 --- /dev/null +++ b/internal/engine/iouring/detach_stall_clock_888_linux_test.go @@ -0,0 +1,44 @@ +//go:build linux + +package iouring + +import ( + "testing" + "time" +) + +// TestDetachedStallClockEndsWithTheQueue888: a conn whose queue drained ends +// its clock, in completeSend (all sent) and in the sweep that finds nothing +// queued, so the next queue is not charged for the time before it. +func TestDetachedStallClockEndsWithTheQueue888(t *testing.T) { + s := newStall888(t, stall888Bound) + s.w.checkTimeouts() + if s.cs.stallSince == 0 { + t.Fatalf("no stall clock after a sweep that found bytes queued") + } + + // The last SEND completes whole. + s.cs.detachMu.Lock() + n := len(s.cs.sendBuf) + s.w.completeSend(s.cs, s.cs.fd, n, time.Now().UnixNano(), false) + s.cs.detachMu.Unlock() + if s.cs.stallSince != 0 { + t.Errorf("stallSince=%d after the queue drained in completeSend, want 0", s.cs.stallSince) + } + + // A queue the sweep finds gone, with the clock left running: the stream's + // goroutine drained it inline. + s.cs.sendBuf, s.cs.sending = s.cs.sendBuf[:1], true + s.w.checkTimeouts() + if s.cs.stallSince == 0 { + t.Fatalf("no stall clock after a sweep that found bytes queued") + } + s.cs.sendBuf, s.cs.sending = s.cs.sendBuf[:0], false + s.w.checkTimeouts() + if s.cs.stallSince != 0 { + t.Errorf("stallSince=%d after a sweep that found nothing queued, want 0", s.cs.stallSince) + } + if s.cs.closing { + t.Errorf("a drained conn was closed") + } +} diff --git a/internal/engine/iouring/worker.go b/internal/engine/iouring/worker.go index 89df5273c..5e4e0ccf6 100644 --- a/internal/engine/iouring/worker.go +++ b/internal/engine/iouring/worker.go @@ -411,6 +411,78 @@ func (w *Worker) noteClosingAcked(cs *connState, now int64) { } } +// detachedStalled is checkTimeouts' look at a truly-detached conn (WS/SSE) +// that has bytes queued for its peer: whether the peer has taken none of them +// for WriteTimeout (celeris#888 items 36 and 37, celeris#884 item 66). The +// middleware owns the stream's lifecycle, so no ReadTimeout, IdleTimeout or +// absolute WriteTimeout applies to it; but a client that stays connected and +// reads nothing never fails a write on a native engine (the write only queues), +// and nothing else ended its stream short of the cap on what the engine holds +// for it. The close reaches the middleware as any engine close does +// (OnDetachClose: an SSE stream's context ends and OnDisconnect runs). +// +// The clock is cs.stallSince, started at the first sweep that finds bytes +// queued and restarted by evidence that the peer takes them: a SEND that +// completed with bytes (completeSend), or the peer's acknowledged byte count +// (tcp_info bytes_acked) having moved since the last sample, which sees what +// the stream's own goroutine writes inline as well, and a SEND too large for +// the socket queue, which completes only as it drains. A sample is due every +// quarter of the bound, and once more when the bound has passed since the last +// one was taken before it, as for a closing conn (noteClosingAcked); the first +// is the baseline, so a peer that took bytes in the first quarter and then none +// is closed up to a quarter of the bound earlier than WriteTimeout after its +// last byte (a SEND completion restamps the clock at once). Not for a +// fixed-file conn, whose cs.fd is a slot of the ring's table: it has no inline +// writes either, so its SEND completions are all there is to see. +// +// WriteTimeout 0 (no timeout, or -1) is no bound: a stream is the middleware's +// to end, and a server that turned the write timeout off asked for a client to +// be held for as long as it stays. Worker thread, with detachMu not held. +func (w *Worker) detachedStalled(cs *connState, now int64) bool { + bound := int64(w.cfg.WriteTimeout) + if bound <= 0 { + return false + } + if cs.stallSince == 0 { + // A new stall. No sample yet: a conn whose queue is there at one + // sweep and gone at the next (a hot broadcaster's) would pay a + // getsockopt per sweep for an episode that ends before it matters. + // The first sample, the baseline, is due a quarter of the bound from + // now. + cs.stallSince = now + cs.closeSampled = now + cs.closeAckedKnown = false + return false + } + if !cs.fixedFile && (now-cs.closeSampled >= bound/4 || + (now-cs.stallSince > bound && cs.closeSampled <= cs.stallSince+bound)) { + w.sampleDetachedAcked(cs, now) + } + return now-cs.stallSince > bound +} + +// sampleDetachedAcked takes detachedStalled's sample of the socket: the +// acknowledged byte count of a conn that moved since the last one restarts the +// stall clock. The first sample of a stall is its baseline and counts as no +// progress; a descriptor that is not a TCP socket, or a kernel that does not +// report the count (before Linux 4.1), yields no evidence, which leaves the +// SEND completions' rule alone. Worker thread. +func (w *Worker) sampleDetachedAcked(cs *connState, now int64) { + cs.closeSampled = now + if cs.fixedFile { + return + } + ti, err := unix.GetsockoptTCPInfo(cs.fd, unix.IPPROTO_TCP, unix.TCP_INFO) + if err != nil { + return + } + known, prev := cs.closeAckedKnown, cs.closeAcked + cs.closeAcked, cs.closeAckedKnown = ti.Bytes_acked, true + if known && ti.Bytes_acked > prev { + cs.stallSince = now + } +} + // Worker is an io_uring event-loop worker pinned to a single OS thread. type Worker struct { id int @@ -4627,6 +4699,11 @@ func (w *Worker) completeSend(cs *connState, fd int, sent int, now int64, fromZC // (covers regular SEND and the SEND_ZC NOTIF path, both of which // reach completeSend with the byte count). w.bytesWrittenBatch += uint64(sent) + if cs.stallSince != 0 && sent > 0 { + // A detached conn whose peer is being watched for a stall + // (detachedStalled): the SEND took bytes, so the peer is reading. + cs.stallSince = now + } if cs.closing && sent > 0 { // The closing drain's clock (closingDrainBound) measures how long // the peer has taken nothing, not how long the drain has run: a @@ -4679,6 +4756,7 @@ func (w *Worker) completeSend(cs *connState, fd int, sent int, now int64, fromZC // All data sent — re-arm recv if needed, remove from dirty list. if len(cs.sendBuf) == 0 && len(cs.writeBuf) == 0 { + cs.stallSince = 0 // nothing is queued: the next queue starts its own clock if cs.needsRecv && !cs.recvPaused { if w.prepareRecv(cs, cs.buf) { cs.needsRecv = false @@ -7038,6 +7116,11 @@ type h1DeadlineSnapshot struct { detached bool idleDL int64 hdrDL int64 + // writePending: the conn has bytes queued for its peer or a SEND in flight + // (connSendPending). Read only when detached (the stall bound, + // detachedStalled): the callers that decide on a header deadline do not + // pay for it. + writePending bool } // snapshotH1Deadlines copies those fields out of cs.h1State under cs.detachMu @@ -7089,6 +7172,9 @@ func snapshotH1Deadlines(cs *connState) (snap h1DeadlineSnapshot, ok bool) { snap.detached = h1.Detached.Load() snap.idleDL = h1.IdleDeadlineNs.Load() snap.hdrDL = h1.HeaderDeadlineNs.Load() + if snap.detached { + snap.writePending = connSendPending(cs) + } } return snap, true } @@ -7273,6 +7359,20 @@ func (w *Worker) checkTimeouts() { if h1Detached { if idleDL > 0 && now > idleDL { w.closeConn(fd) + } else if snap.writePending { + // WriteTimeout is the one engine timeout a detached stream + // keeps: "no progress for WriteTimeout" on what is queued for + // the peer, whatever the middleware's idle deadline says (a + // WebSocket peer that sends pings pushes it out for ever and + // reads nothing of what it is sent; celeris#888 items 36 and + // 37). closeConn signals the middleware (OnDetachClose) and + // defers the descriptor's close behind the queue, as for an + // idle-deadline close. + if w.detachedStalled(cs, now) { + w.closeConn(fd) + } + } else if cs.stallSince != 0 { + cs.stallSince = 0 } continue } diff --git a/middleware/sse/doc.go b/middleware/sse/doc.go index 225932eee..6d5994497 100644 --- a/middleware/sse/doc.go +++ b/middleware/sse/doc.go @@ -24,11 +24,11 @@ // default; configure with [Config.HeartbeatInterval]. The middleware works // with all celeris engines and handles stream lifecycle internally. // -// An SSE stream is a detached stream, so the server's ReadTimeout and -// WriteTimeout do not end it as a whole on any engine (celeris#855): it stays -// open for as long as the client keeps reading, without setting WriteTimeout -// to -1. A client that stays connected and stops reading is dealt with per -// engine: +// An SSE stream is a detached stream, so the server's ReadTimeout does not end +// it as a whole on any engine and neither does WriteTimeout (celeris#855): it +// stays open for as long as the client keeps reading, without setting +// WriteTimeout to -1. A client that stays connected and stops reading is dealt +// with per engine: // // - std: each write of the stream is bounded by WriteTimeout. A write that // has not finished within WriteTimeout fails, so [Client.Send] returns an @@ -38,12 +38,16 @@ // that write, so returning or [ClientPolicyClose] cannot end the stream // sooner. When Shutdown's budget runs out, the writes of the streams still // running fail too. -// - epoll, io_uring and adaptive: [Client.Send] was measured not to wait for -// a client that reads nothing, so a handler that returns (its own bound, -// or its context) ends the stream, and Shutdown ends it. Neither -// WriteTimeout nor [ClientPolicyClose] ends the connection of such a -// client (a known gap, tracked with the engines' polish items, celeris#884 -// and celeris#885). +// - epoll, io_uring and adaptive: [Client.Send] does not wait for a client +// that reads nothing (the write only queues the event in the engine), so +// a write error and [ClientPolicyClose] are not what ends the stream of +// such a client. WriteTimeout does, as "no progress for WriteTimeout": a +// client that takes none of the bytes the engine has queued for it for +// WriteTimeout is disconnected, which cancels [Client.Context], and +// [Config.OnDisconnect] runs. A client that keeps reading, however +// slowly, is not disconnected, and neither is one with nothing queued for +// it (WriteTimeout -1 sets no such bound). A handler that returns (its +// own bound, or its context) ends the stream sooner, and so does Shutdown. // // A HEAD request to an SSE route (answered by its GET route) gets the stream's // response headers and no body. [Config.OnConnect] runs for it, and a rejection diff --git a/middleware/websocket/config.go b/middleware/websocket/config.go index 4c7b3c10a..b44cee541 100644 --- a/middleware/websocket/config.go +++ b/middleware/websocket/config.go @@ -108,14 +108,21 @@ type Config struct { // enforced via the engine's idle sweep using SetWSIdleDeadline. // // Zero means no idle timeout, and on the native engines that means the - // connection has NO liveness bound at all (celeris#524). Once a + // connection has NO idleness bound at all (celeris#524). Once a // connection detaches, the engine deliberately stops applying its own - // ReadTimeout / IdleTimeout / WriteTimeout — the middleware owns the I/O - // lifecycle from that point — and reaps a detached connection only when - // the middleware has published a deadline. This field is what publishes - // one. Leave it zero and nothing will ever close an idle connection, - // which is correct for a long-lived WebSocket and is why zero is the - // default. + // ReadTimeout / IdleTimeout — the middleware owns the I/O lifecycle from + // that point — and reaps a detached connection for idleness only when the + // middleware has published a deadline. This field is what publishes one. + // Leave it zero and nothing will ever close an idle connection, which is + // correct for a long-lived WebSocket and is why zero is the default. + // + // One server timeout does apply to a detached connection on the native + // engines: WriteTimeout, as "no progress for WriteTimeout" (celeris#888 + // items 36 and 37). A peer that takes none of the bytes the engine has + // queued for it for WriteTimeout is closed, whatever it sends meanwhile + // and whatever this field says. A peer that reads, however slowly, and a + // connection with nothing queued are not closed; the server's + // WriteTimeout -1 sets no such bound. // // Worth setting anyway if you want a backstop. A connection that is // WEDGED rather than merely idle is indistinguishable from a healthy From 6f1374f7d37b0bbee720418077ccfec70be83e66 Mon Sep 17 00:00:00 2001 From: x Date: Sun, 18 Oct 2026 13:16:25 +0200 Subject: [PATCH 03/14] epoll, sse: noteClosingAcked returns nothing (unparam); the small-events non-reading cells drop on a full queue until WriteTimeout (v1.6.0 polish items) Refs #884 item 66 Refs #888 items 36, 37 --- internal/engine/epoll/loop.go | 11 ++++--- .../sse/sse_nonreading_888_linux_test.go | 30 +++++++++++++++++-- 2 files changed, 32 insertions(+), 9 deletions(-) diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go index 840f9b9cc..7995fcfd2 100644 --- a/internal/engine/epoll/loop.go +++ b/internal/engine/epoll/loop.go @@ -2120,9 +2120,8 @@ func (cs *connState) endWriteSlow() { // noteClosingAcked is the sweep's evidence of progress on a closing conn: the // number of bytes the peer has acknowledged (tcp_info bytes_acked, which only // grows) moved since the last sample. It restarts the drain clock the same way -// a flush that sent bytes does (noteClosingProgress), and answers whether it -// did. The first sample of a conn is its baseline and counts as no progress, -// unless the conn is first met after its bound had already passed (a loop whose +// a flush that sent bytes does (noteClosingProgress). The first sample of a +// conn is its baseline and counts as no progress, unless the conn is first met after its bound had already passed (a loop whose // sweeps are further apart than a short WriteTimeout): there is no baseline to // compare with, and the conn is given one more bound rather than cut on no // evidence. @@ -2133,14 +2132,15 @@ func (cs *connState) endWriteSlow() { // descriptor that is not a TCP socket, or a kernel that does not report the // count (before Linux 4.1), yields no evidence, which leaves the userspace // queue's rule alone. Loop thread. -func (l *Loop) noteClosingAcked(cs *connState, now int64) (progressed bool) { +func (l *Loop) noteClosingAcked(cs *connState, now int64) { cs.closeSampled = now ti, err := unix.GetsockoptTCPInfo(cs.fd, unix.IPPROTO_TCP, unix.TCP_INFO) if err != nil { - return false + return } known, prev := cs.closeAckedKnown, cs.closeAcked cs.closeAcked, cs.closeAckedKnown = ti.Bytes_acked, true + var progressed bool if known { progressed = ti.Bytes_acked > prev } else { @@ -2149,7 +2149,6 @@ func (l *Loop) noteClosingAcked(cs *connState, now int64) (progressed bool) { if progressed { cs.closeSince = now } - return progressed } // closingDrainBound is how long a conn markClosing deferred may go without the diff --git a/middleware/sse/sse_nonreading_888_linux_test.go b/middleware/sse/sse_nonreading_888_linux_test.go index 9bba6db8a..172c549a7 100644 --- a/middleware/sse/sse_nonreading_888_linux_test.go +++ b/middleware/sse/sse_nonreading_888_linux_test.go @@ -7,6 +7,7 @@ import ( "os" "strconv" "strings" + "sync/atomic" "syscall" "testing" "time" @@ -90,6 +91,17 @@ type producer888 struct { // depends on the loop's own iterations. writeTimeout time.Duration noHeaderTimer bool + + // dropEarly makes a full queue drop the event, not close the stream, for + // the first WriteTimeout of the handler's run. The fast phase sends 2000 + // events a second; a drain goroutine that falls 4 events behind (the loop + // holds the connection's write lock for a partial send, or the runner is + // starved under -race and coverage) is a full queue, and + // ClientPolicyClose would end the stream at once, which says nothing about + // the stall bound this test measures (CI saw cuts at 0.5 s, when the + // socket fills). From WriteTimeout on the policy is Close, as in the other + // producers. + dropEarly bool } func (p producer888) wt() time.Duration { @@ -126,6 +138,8 @@ func burstFor888(tb testing.TB) int { func serve888(tb testing.TB, cell nonReadCell888, p producer888) (addr string, disconnected <-chan time.Time, shutdown func()) { tb.Helper() done := make(chan time.Time, 1) + var started, early atomic.Int64 // handler start (unix nanos), events dropped by dropEarly + wt := p.wt() var burst string if p.fastFor == 0 { burst = strings.Repeat("x", p.burstBytes) @@ -134,14 +148,24 @@ func serve888(tb testing.TB, cell nonReadCell888, p producer888) (addr string, d handler := sse.New(sse.Config{ HeartbeatInterval: -1, MaxQueueDepth: 4, - OnSlowClient: func(*sse.Client, sse.Event) sse.ClientPolicy { return sse.ClientPolicyClose }, + OnSlowClient: func(*sse.Client, sse.Event) sse.ClientPolicy { + if p.dropEarly && time.Since(time.Unix(0, started.Load())) < wt { + early.Add(1) + return sse.ClientPolicyDrop + } + return sse.ClientPolicyClose + }, OnDisconnect: func(*celeris.Context, *sse.Client) { + if n := early.Load(); n > 0 { + tb.Logf("%d events were dropped on a full queue before WriteTimeout", n) + } select { case done <- time.Now(): default: } }, Handler: func(client *sse.Client) { + started.Store(time.Now().UnixNano()) ctx := client.Context() if p.fastFor > 0 { fast := strings.Repeat("z", p.fastBytes) @@ -257,11 +281,11 @@ func TestNonReadingClientEndsStream888(t *testing.T) { // is told to expect to wait for), then one event every 300 ms. // WriteTimeout 2 s puts the close after the fast phase. {"small-events", producer888{fastBytes: 4 << 10, fastEvery: 500 * time.Microsecond, fastFor: 1200 * time.Millisecond, settle: 100 * time.Millisecond, - trickleBytes: 1000, trickleEvery: 300 * time.Millisecond, writeTimeout: 2 * time.Second}}, + trickleBytes: 1000, trickleEvery: 300 * time.Millisecond, writeTimeout: 2 * time.Second, dropEarly: true}}, // The same without the epoll timerfd (ReadHeaderTimeout -1), so nothing // but the loop's own iterations reaches the sweep. {"small-events-no-header-timer", producer888{fastBytes: 4 << 10, fastEvery: 500 * time.Microsecond, fastFor: 1200 * time.Millisecond, settle: 100 * time.Millisecond, - trickleBytes: 1000, trickleEvery: 300 * time.Millisecond, writeTimeout: 2 * time.Second, noHeaderTimer: true}}, + trickleBytes: 1000, trickleEvery: 300 * time.Millisecond, writeTimeout: 2 * time.Second, noHeaderTimer: true, dropEarly: true}}, } for _, cell := range nonReadCells888(t) { for _, pr := range prods { From 2fe07dd45be7ebb74a7dda6858830b3dc45e4e75 Mon Sep 17 00:00:00 2001 From: x Date: Sun, 18 Oct 2026 13:39:05 +0200 Subject: [PATCH 04/14] sse: the non-reading test reports the early-drop count from the test goroutine (v1.6.0 polish items) Refs #884 item 66 Refs #888 items 36, 37 --- .../sse/sse_nonreading_888_linux_test.go | 23 +++++++++++-------- 1 file changed, 14 insertions(+), 9 deletions(-) diff --git a/middleware/sse/sse_nonreading_888_linux_test.go b/middleware/sse/sse_nonreading_888_linux_test.go index 172c549a7..7e07f8eff 100644 --- a/middleware/sse/sse_nonreading_888_linux_test.go +++ b/middleware/sse/sse_nonreading_888_linux_test.go @@ -33,6 +33,14 @@ import ( const nonReadWriteTimeout888 = time.Second +// disconnect888 is what OnDisconnect reports: when it ran, and how many events +// dropEarly dropped on a full queue before that (logged by the test, never by +// OnDisconnect, which can run after the subtest has returned). +type disconnect888 struct { + at time.Time + early int64 +} + // nonReadCell888 is one engine configuration of the matrix. type nonReadCell888 struct { name string @@ -135,9 +143,9 @@ func burstFor888(tb testing.TB) int { // serve888 runs an SSE stream that produces by p until the client is gone, // and reports when OnDisconnect ran. MaxQueueDepth 4 with ClientPolicyClose is // the shape the finding names: the policy that was never reached. -func serve888(tb testing.TB, cell nonReadCell888, p producer888) (addr string, disconnected <-chan time.Time, shutdown func()) { +func serve888(tb testing.TB, cell nonReadCell888, p producer888) (addr string, disconnected <-chan disconnect888, shutdown func()) { tb.Helper() - done := make(chan time.Time, 1) + done := make(chan disconnect888, 1) var started, early atomic.Int64 // handler start (unix nanos), events dropped by dropEarly wt := p.wt() var burst string @@ -156,11 +164,8 @@ func serve888(tb testing.TB, cell nonReadCell888, p producer888) (addr string, d return sse.ClientPolicyClose }, OnDisconnect: func(*celeris.Context, *sse.Client) { - if n := early.Load(); n > 0 { - tb.Logf("%d events were dropped on a full queue before WriteTimeout", n) - } select { - case done <- time.Now(): + case done <- disconnect888{time.Now(), early.Load()}: default: } }, @@ -297,9 +302,9 @@ func TestNonReadingClientEndsStream888(t *testing.T) { const slack = 5 * time.Second wt := pr.p.wt() select { - case at := <-disconnected: - el := at.Sub(began) - t.Logf("OnDisconnect %v after connect (WriteTimeout %v)", el.Round(time.Millisecond), wt) + case d := <-disconnected: + el := d.at.Sub(began) + t.Logf("OnDisconnect %v after connect (WriteTimeout %v), %d events dropped on a full queue before WriteTimeout", el.Round(time.Millisecond), wt, d.early) // std ends a stream whose write is blocked as soon as the // queue behind it is full (ClientPolicyClose), earlier than // WriteTimeout: the lower bound is for the native engines. From ce4706d25b1fce2294990e88f5a49545d7d1b37d Mon Sep 17 00:00:00 2001 From: x Date: Sun, 18 Oct 2026 14:04:40 +0200 Subject: [PATCH 05/14] iouring: a benchmark of the checkTimeouts sweep, attached and detached-idle conns (RULE 74 for the detached stall bound) Refs #884 item 66 Refs #888 items 36, 37 --- .../checktimeouts_bench_888_linux_test.go | 59 +++++++++++++++++++ 1 file changed, 59 insertions(+) create mode 100644 internal/engine/iouring/checktimeouts_bench_888_linux_test.go diff --git a/internal/engine/iouring/checktimeouts_bench_888_linux_test.go b/internal/engine/iouring/checktimeouts_bench_888_linux_test.go new file mode 100644 index 000000000..22d2fdefb --- /dev/null +++ b/internal/engine/iouring/checktimeouts_bench_888_linux_test.go @@ -0,0 +1,59 @@ +//go:build linux + +package iouring + +import ( + "fmt" + "sync" + "sync/atomic" + "testing" + + "github.com/goceleris/celeris/internal/conn" + "github.com/goceleris/celeris/internal/engine/internal/errclass" + "github.com/goceleris/celeris/internal/resource" +) + +// BenchmarkCheckTimeouts888 is the cost of one checkTimeouts sweep over n live +// conns with no timeout configured and nothing queued (nothing closes), the +// steady state of a worker between sweeps. celeris#888 adds a branch for a +// detached conn (its stall clock) and one field to the snapshot of every conn; +// "attached" is the ordinary population, which only pays for the snapshot's +// size, and "detached-idle" is a population of streams with nothing queued, +// which also reads the send state under the snapshot's lock and finds nothing. +// One op is one whole sweep. Compare against an A/A twin of the same binary +// (RULE 74). +func BenchmarkCheckTimeouts888(b *testing.B) { + for _, mode := range []string{"attached", "detached-idle"} { + for _, n := range []int{256, 4096} { + b.Run(fmt.Sprintf("%s/%d", mode, n), func(b *testing.B) { + w := &Worker{ + conns: make([]*connState, n+16), + liveConns: make([]int, 0, n), + errs: &errclass.Counters{}, + activeConns: &atomic.Int64{}, + closeCount: &atomic.Uint64{}, + cfg: resource.Config{}, + } + for i := 0; i < n; i++ { + fd := i + 3 + cs := &connState{fd: fd, liveIdx: -1, detected: true} + cs.h1State = conn.NewH1State() + cs.detachMu = &sync.Mutex{} + if mode == "detached-idle" { + cs.h1State.Detached.Store(true) + cs.h1State.WSReady.Store(true) + } + w.conns[fd] = cs + w.addLiveConn(cs) + if fd > w.maxFD { + w.maxFD = fd + } + } + b.ReportAllocs() + for b.Loop() { + w.checkTimeouts() + } + }) + } + } +} From 55279c716f5526260f62ce64ea402911a046657f Mon Sep 17 00:00:00 2001 From: x Date: Sun, 18 Oct 2026 14:26:44 +0200 Subject: [PATCH 06/14] test/server: server1029 starts through the io_uring-ENOMEM retry (the Unit job's 8 MiB memlock, held by another package's rings) (v1.6.0 polish items) Refs #890 item 55 Refs #884 item 66 --- test/server/chunked_split_1029_linux_test.go | 36 ++++++-------------- 1 file changed, 11 insertions(+), 25 deletions(-) diff --git a/test/server/chunked_split_1029_linux_test.go b/test/server/chunked_split_1029_linux_test.go index b52dc2295..7af8230cb 100644 --- a/test/server/chunked_split_1029_linux_test.go +++ b/test/server/chunked_split_1029_linux_test.go @@ -5,7 +5,6 @@ package celeris_test import ( "bufio" "bytes" - "context" "fmt" "io" "net" @@ -358,33 +357,20 @@ func bodyServerRHT1029(t *testing.T, eng celeris.EngineType, rht time.Duration) } // server1029 serves /ping, a sync POST /up and an Async() POST /aup, both -// answering with the body length. +// answering with the body length. It starts through startServerConfig761, which +// retries an io_uring start that fails with ENOMEM: the ring is charged to the +// runner's 8 MiB RLIMIT_MEMLOCK, which another package's test binary running +// at the same time may hold (a start that did not retry failed the Unit job +// of celeris#1077 twice, while middleware/sse held rings for its io_uring +// cells). func server1029(t *testing.T, eng celeris.EngineType, cfg celeris.Config) string { t.Helper() - ln, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - t.Fatal(err) - } - addr := ln.Addr().String() - _ = ln.Close() - cfg.Addr, cfg.Engine, cfg.Protocol = addr, eng, celeris.Auto - s := celeris.New(cfg) - up := func(c *celeris.Context) error { return c.String(http.StatusOK, "n=%d", len(c.Body())) } - s.GET("/ping", func(c *celeris.Context) error { return c.String(http.StatusOK, "ok") }) - s.POST("/up", up) - s.POST("/aup", up).Async() - startDone := make(chan error, 1) - go func() { startDone <- s.Start() }() - if err := waitReady761(addr, startDone); err != nil { - t.Fatal(err) - } - t.Cleanup(func() { - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - _ = s.Shutdown(ctx) - <-startDone + cfg.Engine, cfg.Protocol = eng, celeris.Auto + return startServerConfig761(t, cfg, func(s *celeris.Server) { + up := func(c *celeris.Context) error { return c.String(http.StatusOK, "n=%d", len(c.Body())) } + s.POST("/up", up) + s.POST("/aup", up).Async() }) - return addr } // TestChunkedAsyncRoute1029: the same split bodies to a route marked Async(), From 7672253331e06f8025eb7c374663836e107e0ea4 Mon Sep 17 00:00:00 2001 From: x Date: Sun, 18 Oct 2026 15:03:26 +0200 Subject: [PATCH 07/14] iouring: the sweep reads a detached conn's send state in its own look, not in the snapshot the recv and dispatch paths share (RULE 74; v1.6.0 polish items) Refs #884 item 66 Refs #888 items 36, 37 --- internal/engine/iouring/worker.go | 65 ++++++++++++++++++++----------- 1 file changed, 43 insertions(+), 22 deletions(-) diff --git a/internal/engine/iouring/worker.go b/internal/engine/iouring/worker.go index 5e4e0ccf6..4e79e91f7 100644 --- a/internal/engine/iouring/worker.go +++ b/internal/engine/iouring/worker.go @@ -7116,11 +7116,6 @@ type h1DeadlineSnapshot struct { detached bool idleDL int64 hdrDL int64 - // writePending: the conn has bytes queued for its peer or a SEND in flight - // (connSendPending). Read only when detached (the stall bound, - // detachedStalled): the callers that decide on a header deadline do not - // pay for it. - writePending bool } // snapshotH1Deadlines copies those fields out of cs.h1State under cs.detachMu @@ -7172,13 +7167,51 @@ func snapshotH1Deadlines(cs *connState) (snap h1DeadlineSnapshot, ok bool) { snap.detached = h1.Detached.Load() snap.idleDL = h1.IdleDeadlineNs.Load() snap.hdrDL = h1.HeaderDeadlineNs.Load() - if snap.detached { - snap.writePending = connSendPending(cs) - } } return snap, true } +// detachedSendPending is checkTimeouts' look at whether a truly-detached conn +// has bytes queued for its peer or a SEND in flight (connSendPending), for the +// stall bound (detachedStalled). The queue is the stream's goroutine's under +// detachMu (its guarded inline writes), so it is read under the lock, with +// TryLock as snapshotH1Deadlines has it (celeris#593): known=false means the +// goroutine is inside a write right now, a conn that is not idle, and the sweep +// leaves it to the next. A separate look from the snapshot, which the recv and +// dispatch paths share, so that they and the sweep of the ordinary conns do not +// pay for a field only a detached conn needs (RULE 74, BenchmarkCheckTimeouts888). +func detachedSendPending(cs *connState) (pending, known bool) { + if mu := cs.detachMu; mu != nil { + if !mu.TryLock() { + return false, false + } + defer mu.Unlock() + } + return connSendPending(cs), true +} + +// sweepDetachedStall is checkTimeouts' stall look at a truly-detached conn +// whose idle deadline has not expired. WriteTimeout is the one engine timeout a +// detached stream keeps: "no progress for WriteTimeout" on what is queued for +// the peer, whatever the middleware's idle deadline says (a WebSocket peer that +// sends pings pushes it out for ever and reads nothing of what it is sent; +// celeris#888 items 36 and 37). closeConn signals the middleware +// (OnDetachClose) and defers the descriptor's close behind the queue, as for an +// idle-deadline close. Worker thread. +func (w *Worker) sweepDetachedStall(cs *connState, fd int, now int64) { + pending, known := detachedSendPending(cs) + switch { + case !known: + // Inside a write of the stream's goroutine: not idle. + case pending: + if w.detachedStalled(cs, now) { + w.closeConn(fd) + } + case cs.stallSince != 0: + cs.stallSince = 0 + } +} + // sweepWatchDue reports whether the loop owes a timeout sweep outside its // iteration gate (celeris#811): the dirty pass gave up a conn whose SEND is // outstanding (sendWatch), so no spin reaches the gate for its WriteTimeout, @@ -7359,20 +7392,8 @@ func (w *Worker) checkTimeouts() { if h1Detached { if idleDL > 0 && now > idleDL { w.closeConn(fd) - } else if snap.writePending { - // WriteTimeout is the one engine timeout a detached stream - // keeps: "no progress for WriteTimeout" on what is queued for - // the peer, whatever the middleware's idle deadline says (a - // WebSocket peer that sends pings pushes it out for ever and - // reads nothing of what it is sent; celeris#888 items 36 and - // 37). closeConn signals the middleware (OnDetachClose) and - // defers the descriptor's close behind the queue, as for an - // idle-deadline close. - if w.detachedStalled(cs, now) { - w.closeConn(fd) - } - } else if cs.stallSince != 0 { - cs.stallSince = 0 + } else { + w.sweepDetachedStall(cs, fd, now) } continue } From a06580e9e1ff5f31e9adac1163b6925b895d2d4d Mon Sep 17 00:00:00 2001 From: x Date: Sun, 18 Oct 2026 15:14:23 +0200 Subject: [PATCH 08/14] test/server: server1029's comment says only what the CI log shows (v1.6.0 polish items) Refs #890 item 55 --- test/server/chunked_split_1029_linux_test.go | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/test/server/chunked_split_1029_linux_test.go b/test/server/chunked_split_1029_linux_test.go index 7af8230cb..128d433c8 100644 --- a/test/server/chunked_split_1029_linux_test.go +++ b/test/server/chunked_split_1029_linux_test.go @@ -360,9 +360,8 @@ func bodyServerRHT1029(t *testing.T, eng celeris.EngineType, rht time.Duration) // answering with the body length. It starts through startServerConfig761, which // retries an io_uring start that fails with ENOMEM: the ring is charged to the // runner's 8 MiB RLIMIT_MEMLOCK, which another package's test binary running -// at the same time may hold (a start that did not retry failed the Unit job -// of celeris#1077 twice, while middleware/sse held rings for its io_uring -// cells). +// at the same time may hold (a start that did not retry failed the Unit steps +// job of celeris#1077 twice with "io_uring_setup: cannot allocate memory"). func server1029(t *testing.T, eng celeris.EngineType, cfg celeris.Config) string { t.Helper() cfg.Engine, cfg.Protocol = eng, celeris.Auto From 7c6fead12f9b809ae4947c13ee2544f7a0afe4e4 Mon Sep 17 00:00:00 2001 From: x Date: Sun, 18 Oct 2026 16:25:54 +0200 Subject: [PATCH 09/14] iouring: every SEND of a truly-detached conn is watched, the first stall sample met late is a baseline, a stall close skips the closing drain, the stall look has a rate limit (review round 1 of #1077; Refs #888 items 38, 39; Refs #884 item 69) --- .../checktimeouts_bench_888_linux_test.go | 29 +-- internal/engine/iouring/conn.go | 2 +- .../iouring/detach_stall_888_linux_test.go | 9 +- .../detach_stall_clock_888_linux_test.go | 6 +- .../detach_stall_round2_888_linux_test.go | 165 ++++++++++++++++++ internal/engine/iouring/worker.go | 103 +++++++++-- middleware/sse/sse_quiet_888_linux_test.go | 143 +++++++++++++++ 7 files changed, 425 insertions(+), 32 deletions(-) create mode 100644 internal/engine/iouring/detach_stall_round2_888_linux_test.go create mode 100644 middleware/sse/sse_quiet_888_linux_test.go diff --git a/internal/engine/iouring/checktimeouts_bench_888_linux_test.go b/internal/engine/iouring/checktimeouts_bench_888_linux_test.go index 22d2fdefb..4156a8ad7 100644 --- a/internal/engine/iouring/checktimeouts_bench_888_linux_test.go +++ b/internal/engine/iouring/checktimeouts_bench_888_linux_test.go @@ -7,6 +7,7 @@ import ( "sync" "sync/atomic" "testing" + "time" "github.com/goceleris/celeris/internal/conn" "github.com/goceleris/celeris/internal/engine/internal/errclass" @@ -14,16 +15,19 @@ import ( ) // BenchmarkCheckTimeouts888 is the cost of one checkTimeouts sweep over n live -// conns with no timeout configured and nothing queued (nothing closes), the -// steady state of a worker between sweeps. celeris#888 adds a branch for a -// detached conn (its stall clock) and one field to the snapshot of every conn; -// "attached" is the ordinary population, which only pays for the snapshot's -// size, and "detached-idle" is a population of streams with nothing queued, -// which also reads the send state under the snapshot's lock and finds nothing. -// One op is one whole sweep. Compare against an A/A twin of the same binary -// (RULE 74). +// conns, WriteTimeout at its default of 60 s and nothing queued (nothing +// closes), the steady state of a worker between sweeps. celeris#888 adds the +// stall look at a detached conn (sweepDetachedStall: a lock attempt on the +// conn's send state, and its clock). "attached" is the ordinary population, +// which pays for a branch it does not take; "detached-idle" is a population of +// streams with nothing queued, with the look due on every sweep (the worst +// case: lastStallLook is reset, as it would be after the floor of +// stallLookDue); "detached-idle-limited" is the same population in a run of +// sweeps back to back, where the floor (250 ms at this WriteTimeout) lets +// almost none take the look: the cost of the branch that skips it. One op is one whole sweep. Compare against an A/A twin of the same +// binary (RULE 74). func BenchmarkCheckTimeouts888(b *testing.B) { - for _, mode := range []string{"attached", "detached-idle"} { + for _, mode := range []string{"attached", "detached-idle", "detached-idle-limited"} { for _, n := range []int{256, 4096} { b.Run(fmt.Sprintf("%s/%d", mode, n), func(b *testing.B) { w := &Worker{ @@ -32,14 +36,14 @@ func BenchmarkCheckTimeouts888(b *testing.B) { errs: &errclass.Counters{}, activeConns: &atomic.Int64{}, closeCount: &atomic.Uint64{}, - cfg: resource.Config{}, + cfg: resource.Config{WriteTimeout: 60 * time.Second}, } for i := 0; i < n; i++ { fd := i + 3 cs := &connState{fd: fd, liveIdx: -1, detected: true} cs.h1State = conn.NewH1State() cs.detachMu = &sync.Mutex{} - if mode == "detached-idle" { + if mode != "attached" { cs.h1State.Detached.Store(true) cs.h1State.WSReady.Store(true) } @@ -51,6 +55,9 @@ func BenchmarkCheckTimeouts888(b *testing.B) { } b.ReportAllocs() for b.Loop() { + if mode == "detached-idle" { + w.lastStallLook = 0 + } w.checkTimeouts() } }) diff --git a/internal/engine/iouring/conn.go b/internal/engine/iouring/conn.go index 387d3c718..84053fb34 100644 --- a/internal/engine/iouring/conn.go +++ b/internal/engine/iouring/conn.go @@ -269,7 +269,7 @@ type connState struct { lastActivity int64 // nanosecond timestamp of last I/O activity (for timeout checks) // stallSince is the clock of a truly-detached conn (WS/SSE) that has bytes - // queued for its peer (celeris#888 items 36 and 37, celeris#884 item 66): + // queued for its peer (celeris#888 items 38 and 39, celeris#884 item 69): // when the sweep last saw the peer take some of them, in nanoseconds, or // first saw them queued. checkTimeouts closes the conn once it is older // than WriteTimeout (detachedStalled). Restamped by completeSend when a diff --git a/internal/engine/iouring/detach_stall_888_linux_test.go b/internal/engine/iouring/detach_stall_888_linux_test.go index 4982a574d..f94234794 100644 --- a/internal/engine/iouring/detach_stall_888_linux_test.go +++ b/internal/engine/iouring/detach_stall_888_linux_test.go @@ -19,7 +19,7 @@ import ( // A truly-detached conn (WS/SSE) whose peer takes none of the bytes queued for // it is closed WriteTimeout after the sweep found them queued (celeris#888 -// items 36 and 37, celeris#884 item 66): checkTimeouts skipped a detached conn +// items 38 and 39, celeris#884 item 69): checkTimeouts skipped a detached conn // whatever it had queued, and a write on a native engine only queues, so a // client that stayed connected and read nothing held its stream for ever. The // clock is "the peer took none of it", not the age of the stream or the last @@ -98,6 +98,13 @@ func newStall888(t *testing.T, wt time.Duration) *stall888 { return s } +// sweepNow is one timeout sweep that takes the stall look: the look has a rate +// limit (stallLookDue), which a test that sweeps back to back must not meet. +func (s *stall888) sweepNow() { + s.w.lastStallLook = 0 + s.w.checkTimeouts() +} + // sweepUntilClosing sweeps every 10 ms until the conn is closing (closeConn // deferred its descriptor behind the SEND in flight) or the limit passes. func (s *stall888) sweepUntilClosing(limit time.Duration, between func()) (after time.Duration, closing bool) { diff --git a/internal/engine/iouring/detach_stall_clock_888_linux_test.go b/internal/engine/iouring/detach_stall_clock_888_linux_test.go index a1636aa28..45531f662 100644 --- a/internal/engine/iouring/detach_stall_clock_888_linux_test.go +++ b/internal/engine/iouring/detach_stall_clock_888_linux_test.go @@ -12,7 +12,7 @@ import ( // queued, so the next queue is not charged for the time before it. func TestDetachedStallClockEndsWithTheQueue888(t *testing.T) { s := newStall888(t, stall888Bound) - s.w.checkTimeouts() + s.sweepNow() if s.cs.stallSince == 0 { t.Fatalf("no stall clock after a sweep that found bytes queued") } @@ -29,12 +29,12 @@ func TestDetachedStallClockEndsWithTheQueue888(t *testing.T) { // A queue the sweep finds gone, with the clock left running: the stream's // goroutine drained it inline. s.cs.sendBuf, s.cs.sending = s.cs.sendBuf[:1], true - s.w.checkTimeouts() + s.sweepNow() if s.cs.stallSince == 0 { t.Fatalf("no stall clock after a sweep that found bytes queued") } s.cs.sendBuf, s.cs.sending = s.cs.sendBuf[:0], false - s.w.checkTimeouts() + s.sweepNow() if s.cs.stallSince != 0 { t.Errorf("stallSince=%d after a sweep that found nothing queued, want 0", s.cs.stallSince) } diff --git a/internal/engine/iouring/detach_stall_round2_888_linux_test.go b/internal/engine/iouring/detach_stall_round2_888_linux_test.go new file mode 100644 index 000000000..f9cfd3257 --- /dev/null +++ b/internal/engine/iouring/detach_stall_round2_888_linux_test.go @@ -0,0 +1,165 @@ +//go:build linux + +package iouring + +import ( + "sync" + "testing" + "time" + + "github.com/goceleris/celeris/internal/conn" +) + +// Review round 1 of celeris#1077: the stall bound's reach (a SEND of any size +// of a detached conn is watched), the first sample of a stall met late, the +// look's rate limit and the drain after a stall close. + +// TestDetachedSendOfAnySizeIsWatched888: the SEND of the leftover of a +// detached stream's inline write is smaller than longSendBytes, and a producer +// that goes quiet after it leaves the worker with nothing to wake it. Its SEND +// must be marked (remainderWatch: the timed wait, sendWatch: the sweep) for the +// stall bound to run. Not for a conn that is not truly detached, or on a server +// without a WriteTimeout. +func TestDetachedSendOfAnySizeIsWatched888(t *testing.T) { + mk := func(wt time.Duration, detached, mu bool) (*Worker, *connState) { + w := &Worker{} + w.cfg.WriteTimeout = wt + cs := &connState{h1State: conn.NewH1State()} + cs.h1State.Detached.Store(detached) + if mu { + cs.detachMu = &sync.Mutex{} + } + return w, cs + } + cases := []struct { + name string + wt time.Duration + detached bool + mu bool + n int + want bool + }{ + {"detached-4KiB", time.Second, true, true, 4 << 10, true}, + {"detached-1-byte", time.Second, true, true, 1, true}, + {"detached-long", time.Second, true, true, longSendBytes, true}, + {"detached-no-write-timeout", 0, true, true, 4 << 10, false}, + {"detached-write-timeout-off", -1, true, true, 4 << 10, false}, + {"async-conn-not-detached", time.Second, false, true, 4 << 10, false}, + {"inline-conn", time.Second, false, false, 4 << 10, false}, + {"long-but-not-detached", time.Second, false, false, longSendBytes, true}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + w, cs := mk(c.wt, c.detached, c.mu) + w.markLongSend(cs, c.n) + if cs.remainderOut != c.want || w.remainderWatch != c.want || w.sendWatch != c.want { + t.Fatalf("marked=%v remainderWatch=%v sendWatch=%v, want %v", cs.remainderOut, w.remainderWatch, w.sendWatch, c.want) + } + if got := waitUntimed(true, true, false, w.remainderWatch); got == c.want { + t.Fatalf("waitUntimed = %v with remainderWatch %v: the worker waits with no timeout for a SEND it must look at", got, w.remainderWatch) + } + }) + } +} + +// TestDetachedSparseSweepsGiveAReadingPeerABound888: sweeps further apart than +// WriteTimeout (an idle worker's: about 0.8 s, 1.6 s, or the sendWatch's 1 to +// 2 s) meet a conn whose clock is already past the bound. The first sample of a +// stall is a baseline, not evidence: a peer that reads is not closed on it, as +// a closing conn's is not (noteClosingAcked). A peer that reads nothing is +// closed at the sample after, one bound later. +func TestDetachedSparseSweepsGiveAReadingPeerABound888(t *testing.T) { + for _, reads := range []bool{true, false} { + name := "peer-reads" + if !reads { + name = "peer-reads-nothing" + } + t.Run(name, func(t *testing.T) { + s := newStall888(t, stall888Bound) + s.cs.lastActivity = time.Now().Add(-time.Hour).UnixNano() + fillSocket888(t, s.cs.fd) + gap := stall888Bound * 3 / 2 + buf := make([]byte, 4096) + closedAt := -1 + for i := 0; i < 4 && closedAt < 0; i++ { + s.sweepNow() + if s.cs.closing { + closedAt = i + break + } + // The peer reads between the sweeps, in small steps: its + // acknowledged count grows the whole time. + for el := time.Duration(0); el < gap; el += 20 * time.Millisecond { + if reads { + _ = s.peer.SetReadDeadline(time.Now().Add(5 * time.Millisecond)) + _, _ = s.peer.Read(buf) + } + time.Sleep(20 * time.Millisecond) + } + } + switch { + case reads && closedAt >= 0: + t.Fatalf("a peer that read all along was closed at sweep %d (sweeps %v apart, WriteTimeout %v)", closedAt, gap, stall888Bound) + case !reads && closedAt < 0: + t.Fatalf("a peer that read nothing was not closed in 4 sweeps (sweeps %v apart, WriteTimeout %v)", gap, stall888Bound) + case !reads && closedAt < 2: + t.Fatalf("closed at sweep %d: the first sample after the bound is a baseline, the close is due at the next", closedAt) + } + }) + } +} + +// TestStallLookRateLimit888: the look at the detached conns takes place at most +// once per a sixteenth of the bound, within 25 ms to 250 ms, never without a +// WriteTimeout, and a clock that stepped back looks at once. +func TestStallLookRateLimit888(t *testing.T) { + ms := func(n int64) int64 { return n * int64(time.Millisecond) } + for _, c := range []struct { + wt time.Duration + floor int64 // ns + }{ + {300 * time.Millisecond, ms(25)}, + {time.Second, int64(time.Second) / 16}, + {60 * time.Second, ms(250)}, + } { + w := &Worker{} + w.cfg.WriteTimeout = c.wt + base := ms(1_000_000) + if !w.stallLookDue(base) { + t.Fatalf("WriteTimeout %v: the first sweep did not look", c.wt) + } + if w.stallLookDue(base + c.floor - 1) { + t.Errorf("WriteTimeout %v: looked again 1 ns before the floor of %v", c.wt, time.Duration(c.floor)) + } + if !w.stallLookDue(base + c.floor) { + t.Errorf("WriteTimeout %v: did not look at the floor of %v", c.wt, time.Duration(c.floor)) + } + if !w.stallLookDue(base - ms(1000)) { + t.Errorf("WriteTimeout %v: a clock that stepped back did not look", c.wt) + } + } + for _, wt := range []time.Duration{0, -1} { + w := &Worker{} + w.cfg.WriteTimeout = wt + if w.stallLookDue(ms(5_000_000)) { + t.Errorf("WriteTimeout %v: looked", wt) + } + } +} + +// TestStalledCloseSkipsTheClosingDrain888: the peer has just been shown to take +// nothing for WriteTimeout, so the drain that closeConn gives a conn with a +// SEND in flight (closingDrainBound, 5 s or more) is spent at once: the next +// sweep may reap it. A peer that reads in the meantime would restamp the clock +// (noteClosingAcked). +func TestStalledCloseSkipsTheClosingDrain888(t *testing.T) { + s := newStall888(t, stall888Bound) + s.cs.lastActivity = time.Now().UnixNano() + if _, closing := s.sweepUntilClosing(5*time.Second, nil); !closing { + t.Fatal("not closed") + } + if age := time.Now().UnixNano() - s.cs.lastActivity; age <= int64(s.w.closingDrainBound()) { + t.Fatalf("the closing drain's clock is %v old, the bound is %v: the stalled conn would hold its descriptor and queue for another bound", + time.Duration(age), s.w.closingDrainBound()) + } +} diff --git a/internal/engine/iouring/worker.go b/internal/engine/iouring/worker.go index 4e79e91f7..474a8ee94 100644 --- a/internal/engine/iouring/worker.go +++ b/internal/engine/iouring/worker.go @@ -320,8 +320,8 @@ const shutdownSendDrainNanos int64 = int64(250 * time.Millisecond) // last thing the worker submitted. A timed wait (mode 3b) is capped by // adaptiveTimeout, so the sweep (sweepWatchDue) is reached every second or // two. Mode 3a is lost only while such a SEND is outstanding (a SEND of a -// small response to a client that has stopped reading is not marked: see -// longSendBytes). Small enough to inline; a function so the rule can be +// small response to a client that has stopped reading is not marked, unless +// the conn is a truly-detached one: see longSendBytes). Small enough to inline; a function so the rule can be // tested. func waitUntimed(hasPending, sendsPending, draining, remainderWatch bool) bool { return hasPending && sendsPending && !draining && !remainderWatch @@ -388,9 +388,8 @@ func (w *Worker) closingDrainBound() int64 { // Not on the hot path: checkTimeouts calls it for a closing conn, at most every // quarter of the bound (closeSampled) and once more before it reaps one, and it // costs one getsockopt. A -// descriptor that is not a TCP socket, or a kernel that does not report the -// count (before Linux 4.1), yields no evidence, which leaves the SEND -// completions' rule alone. Not for a fixed-file conn, whose cs.fd is a slot of +// descriptor that is not a TCP socket yields no evidence, which leaves the +// SEND completions' rule alone. Not for a fixed-file conn, whose cs.fd is a slot of // the ring's table, not a descriptor (the caller skips it). Worker thread. func (w *Worker) noteClosingAcked(cs *connState, now int64) { cs.closeSampled = now @@ -413,7 +412,7 @@ func (w *Worker) noteClosingAcked(cs *connState, now int64) { // detachedStalled is checkTimeouts' look at a truly-detached conn (WS/SSE) // that has bytes queued for its peer: whether the peer has taken none of them -// for WriteTimeout (celeris#888 items 36 and 37, celeris#884 item 66). The +// for WriteTimeout (celeris#888 items 38 and 39, celeris#884 item 69). The // middleware owns the stream's lifecycle, so no ReadTimeout, IdleTimeout or // absolute WriteTimeout applies to it; but a client that stays connected and // reads nothing never fails a write on a native engine (the write only queues), @@ -464,9 +463,8 @@ func (w *Worker) detachedStalled(cs *connState, now int64) bool { // sampleDetachedAcked takes detachedStalled's sample of the socket: the // acknowledged byte count of a conn that moved since the last one restarts the // stall clock. The first sample of a stall is its baseline and counts as no -// progress; a descriptor that is not a TCP socket, or a kernel that does not -// report the count (before Linux 4.1), yields no evidence, which leaves the -// SEND completions' rule alone. Worker thread. +// progress; a descriptor that is not a TCP socket yields no evidence, which +// leaves the SEND completions' rule alone. Worker thread. func (w *Worker) sampleDetachedAcked(cs *connState, now int64) { cs.closeSampled = now if cs.fixedFile { @@ -478,7 +476,15 @@ func (w *Worker) sampleDetachedAcked(cs *connState, now int64) { } known, prev := cs.closeAckedKnown, cs.closeAcked cs.closeAcked, cs.closeAckedKnown = ti.Bytes_acked, true - if known && ti.Bytes_acked > prev { + switch { + case known && ti.Bytes_acked > prev: + cs.stallSince = now + case !known && now-cs.stallSince > int64(w.cfg.WriteTimeout): + // The first sample, met after the bound (sweeps further apart than + // WriteTimeout: an idle worker's, or a SEND completion that was late + // to restamp the clock): there is no baseline to compare with, and + // the conn is given one more bound, not cut on no evidence + // (noteClosingAcked does the same for a closing conn). cs.stallSince = now } } @@ -689,6 +695,9 @@ type Worker struct { // (celeris#884 item 62). Worker-owned. sweepMarks int lastSweep int64 + // lastStallLook is the clock of the last sweep that looked at the + // detached conns' stall (stallLookDue); worker-thread-only. + lastStallLook int64 // pendingRelease defers returning connState structs to the pool // until the kernel has drained every in-flight I/O SQE that may @@ -4756,7 +4765,9 @@ func (w *Worker) completeSend(cs *connState, fd int, sent int, now int64, fromZC // All data sent — re-arm recv if needed, remove from dirty list. if len(cs.sendBuf) == 0 && len(cs.writeBuf) == 0 { - cs.stallSince = 0 // nothing is queued: the next queue starts its own clock + if cs.stallSince != 0 { + cs.stallSince = 0 // nothing is queued: the next queue starts its own clock + } if cs.needsRecv && !cs.recvPaused { if w.prepareRecv(cs, cs.buf) { cs.needsRecv = false @@ -6757,7 +6768,9 @@ func (w *Worker) flushDirty() { // stalls with no progress at all. The figure is a choice, not a measurement: // the kernel's initial send buffer is 16 KiB. Smaller SENDs keep the untimed // wait (mode 3a), so a small SEND into a socket that is already full on an -// otherwise idle worker is still not looked at by a sweep. +// otherwise idle worker is still not looked at by a sweep (celeris#884 item +// 74), except for a truly-detached conn on a server with a WriteTimeout, whose +// every SEND is marked (watchesStall). const longSendBytes = 64 << 10 // markLongSend notes that the SEND being submitted for cs, n bytes, may be @@ -6766,12 +6779,31 @@ const longSendBytes = 64 << 10 // is, and sendWatch makes the sweep (sweepWatchDue) come for WriteTimeout // (celeris#884 items 62 and 64). Cleared by completeSend, re-derived by the // sweep. Worker thread; one compare for the common small response. +// +// A SEND of a truly-detached conn (WS/SSE) is marked whatever its size, while +// WriteTimeout bounds it (watchesStall): the stall bound (detachedStalled) +// exists for a peer that reads nothing, and the SEND of the leftover of the +// stream's own inline write is just such a SEND, of any size up to the socket's +// capacity. Unmarked, a producer that goes quiet after it leaves the worker in +// its untimed wait with nothing to run the sweep, and the peer is held for +// ever (celeris#888 items 38 and 39, celeris#884 item 69). func (w *Worker) markLongSend(cs *connState, n int) { - if n >= longSendBytes { + if n >= longSendBytes || w.watchesStall(cs) { w.markRemainder(cs) } } +// watchesStall reports whether cs is a truly-detached conn on a server with a +// WriteTimeout, the conns whose SENDs the stall bound looks at +// (detachedStalled). Async-mode HTTP/1 conns have a detachMu without being +// detached (sendCap): the ordinary write timeout bounds them. The compares are +// in the order of their cost: the pointer is nil for an inline conn. Worker +// thread. +func (w *Worker) watchesStall(cs *connState) bool { + return cs.detachMu != nil && w.cfg.WriteTimeout > 0 && + cs.h1State != nil && cs.h1State.Detached.Load() +} + // markRemainder notes that the SEND being submitted for cs may be outstanding // for as long as its peer holds out (see markLongSend), and counts the conn. // Worker thread. @@ -7195,7 +7227,7 @@ func detachedSendPending(cs *connState) (pending, known bool) { // detached stream keeps: "no progress for WriteTimeout" on what is queued for // the peer, whatever the middleware's idle deadline says (a WebSocket peer that // sends pings pushes it out for ever and reads nothing of what it is sent; -// celeris#888 items 36 and 37). closeConn signals the middleware +// celeris#888 items 38 and 39). closeConn signals the middleware // (OnDetachClose) and defers the descriptor's close behind the queue, as for an // idle-deadline close. Worker thread. func (w *Worker) sweepDetachedStall(cs *connState, fd int, now int64) { @@ -7205,13 +7237,51 @@ func (w *Worker) sweepDetachedStall(cs *connState, fd int, now int64) { // Inside a write of the stream's goroutine: not idle. case pending: if w.detachedStalled(cs, now) { - w.closeConn(fd) + w.closeStalled(fd, cs, now) } case cs.stallSince != 0: cs.stallSince = 0 } } +// stallLookDue reports whether this sweep takes the stall look at the detached +// conns (sweepDetachedStall): never when WriteTimeout is off (nothing to +// enforce), and at most once per a sixteenth of the bound, between 25 ms and +// 250 ms. The io_uring sweep has no wall-clock floor (it runs every 32 +// iterations while a conn is detached, hundreds of times a second on a busy +// worker), and the bound needs a resolution of a quarter of itself (the +// samples of detachedStalled), not of an iteration. The look costs a lock +// attempt per detached conn (BenchmarkCheckTimeouts888). Worker thread. +func (w *Worker) stallLookDue(now int64) bool { + bound := int64(w.cfg.WriteTimeout) + if bound <= 0 { + return false + } + floor := min(max(bound/16, int64(25*time.Millisecond)), int64(250*time.Millisecond)) + if now-w.lastStallLook < floor && now >= w.lastStallLook { + return false + } + w.lastStallLook = now + return true +} + +// closeStalled closes a detached conn whose peer took none of its queue for +// WriteTimeout (detachedStalled). closeConn defers the descriptor behind the +// SEND in flight and gives it a drain of closingDrainBound; the peer has just +// been shown to take nothing for WriteTimeout, so that bound is spent: the +// drain's clock is set back past it, and the next sweep reaps the conn unless +// its sample of the peer's acknowledged bytes shows it reading again +// (noteClosingAcked). Held for another bound, the conn would keep its +// descriptor, its kernel send buffer and up to maxSendQueueBytesDetached of +// queue for twice the WriteTimeout, which epoll does not (it closes at once). +// Worker thread. +func (w *Worker) closeStalled(fd int, cs *connState, now int64) { + w.closeConn(fd) + if w.conns[fd] == cs && cs.closing { + cs.lastActivity = now - w.closingDrainBound() - 1 + } +} + // sweepWatchDue reports whether the loop owes a timeout sweep outside its // iteration gate (celeris#811): the dirty pass gave up a conn whose SEND is // outstanding (sendWatch), so no spin reaches the gate for its WriteTimeout, @@ -7271,6 +7341,7 @@ func (w *Worker) checkTimeouts() { // already-visited slot) and (b) read a stale/zeroed tail slot as fd 0, // dereferencing w.conns[0]. Reverse-by-index visits each conn exactly // once even as entries are swap-removed from the tail (v1.5.0 review 1.9). + stallLook := w.stallLookDue(now) for i := len(w.liveConns) - 1; i >= 0; i-- { fd := w.liveConns[i] cs := w.conns[fd] @@ -7392,7 +7463,7 @@ func (w *Worker) checkTimeouts() { if h1Detached { if idleDL > 0 && now > idleDL { w.closeConn(fd) - } else { + } else if stallLook { w.sweepDetachedStall(cs, fd, now) } continue diff --git a/middleware/sse/sse_quiet_888_linux_test.go b/middleware/sse/sse_quiet_888_linux_test.go new file mode 100644 index 000000000..3e9bd6c6e --- /dev/null +++ b/middleware/sse/sse_quiet_888_linux_test.go @@ -0,0 +1,143 @@ +//go:build linux + +package sse_test + +import ( + "net" + "strconv" + "strings" + "syscall" + "testing" + "time" + + "golang.org/x/sys/unix" + + "github.com/goceleris/celeris" + "github.com/goceleris/celeris/middleware/sse" +) + +// A producer that goes quiet while its client reads nothing (celeris#888 items +// 38 and 39, celeris#884 item 69; review of #1077). One event larger than the +// socket takes leaves a remainder queued, and then nothing is written: no +// heartbeat, no further event. On io_uring the stream's inline write leaves a +// SEND of under 64 KiB for the worker, a size the worker did not expect to wait +// on its peer, so it waited for the completion with no timeout, no sweep ran +// and the client was held for ever. The test of the other producers never saw +// it, because each of them kept writing, and every write wakes the worker. + +const quietSndBuf888 = 64 << 10 + +// quietCapacity888 measures how many bytes one nonblocking write takes on a +// loopback pair shaped as the server's (send buffer quietSndBuf888, client +// receive buffer 4 KiB, client reads nothing). +func quietCapacity888(tb testing.TB) int { + tb.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + tb.Skipf("loopback unavailable: %v", err) + } + defer func() { _ = ln.Close() }() + d := net.Dialer{Control: func(_, _ string, c syscall.RawConn) error { + return c.Control(func(fd uintptr) { _ = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUF, 4096) }) + }} + cl, err := d.Dial("tcp", ln.Addr().String()) + if err != nil { + tb.Fatal(err) + } + defer func() { _ = cl.Close() }() + srv, err := ln.Accept() + if err != nil { + tb.Fatal(err) + } + defer func() { _ = srv.Close() }() + rc, err := srv.(*net.TCPConn).SyscallConn() + if err != nil { + tb.Fatal(err) + } + n := 0 + _ = rc.Control(func(fd uintptr) { + _ = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_SNDBUF, quietSndBuf888) + _ = unix.SetNonblock(int(fd), true) + n, _ = unix.Write(int(fd), make([]byte, 4<<20)) + }) + if n <= 0 { + tb.Skip("could not measure the socket's capacity") + } + return n +} + +// TestQuietNonReadingClientEndsStream888: one event of the socket's capacity +// plus delta, then silence, to a client that reads nothing. A remainder (delta +// above 0) is queued for a peer that takes none of it: the stream ends about +// WriteTimeout after connect, with the header timer or without it, and +// whatever the size of the remainder (4 KiB is under io_uring's long-SEND mark, +// 256 KiB over it). Delta -2048 is the control: the event fits, nothing is +// queued, and a client that took everything written is not closed. +func TestQuietNonReadingClientEndsStream888(t *testing.T) { + capN := quietCapacity888(t) + t.Logf("capacity of one write: %d bytes", capN) + const wt = nonReadWriteTimeout888 + for _, cell := range nonReadCells888(t) { + if cell.engine == celeris.Std { + continue // std: a write blocked on the peer is ended by its own deadline + } + for _, delta := range []int{-2048, 4 << 10, 40 << 10, 256 << 10} { + for _, noHdr := range []bool{false, true} { + if noHdr && delta != 4<<10 { + continue // the header timer matters for the smallest remainder only + } + name := cell.name + "/delta=" + strconv.Itoa(delta) + if noHdr { + name += "/no-header-timer" + } + t.Run(name, func(t *testing.T) { + done := make(chan time.Time, 1) + data := strings.Repeat("x", capN+delta) + h := sse.New(sse.Config{ + HeartbeatInterval: -1, + OnDisconnect: func(*celeris.Context, *sse.Client) { + select { + case done <- time.Now(): + default: + } + }, + Handler: func(client *sse.Client) { + if err := client.Send(sse.Event{Data: data}); err != nil { + return + } + <-client.Context().Done() + }, + }) + cfg := celeris.Config{Engine: cell.engine, AsyncHandlers: cell.async, WriteTimeout: wt, + ShutdownTimeout: 2 * time.Second, SocketSendBuf: quietSndBuf888} + if noHdr { + cfg.ReadHeaderTimeout = -1 + } + addr, shutdown := startSSEServerCfg(t, cfg, false, h) + defer shutdown() + began := time.Now() + _ = dialSmallRcvBuf888(t, addr) + if delta < 0 { + select { + case at := <-done: + t.Fatalf("a stream whose event fit the socket was ended %v after connect: nothing was queued", at.Sub(began)) + case <-time.After(3 * wt): + } + return + } + select { + case at := <-done: + el := at.Sub(began) + t.Logf("OnDisconnect %v after connect (WriteTimeout %v)", el.Round(time.Millisecond), wt) + if el < wt { + t.Errorf("OnDisconnect ran %v after connect, before WriteTimeout %v", el, wt) + } + case <-time.After(wt + 6*time.Second): + t.Fatalf("OnDisconnect never ran in %v (WriteTimeout %v): a quiet stream's client that reads nothing holds it for ever", + wt+6*time.Second, wt) + } + }) + } + } + } +} From fdfc9f88ba5422282b191b358e79ddbdc81ab762 Mon Sep 17 00:00:00 2001 From: x Date: Sun, 18 Oct 2026 16:25:54 +0200 Subject: [PATCH 10/14] epoll, sse, websocket: checklist item numbers, ClientPolicyClose can end a stalled stream earlier, no minimum rate, h2c not covered yet, tcp_info comments (review round 1 of #1077; Refs #888 items 38, 39; Refs #884 item 69) --- config.go | 4 +++- internal/engine/epoll/conn.go | 2 +- .../epoll/detach_stall_888_linux_test.go | 2 +- .../detach_stall_nontcp_888_linux_test.go | 2 +- internal/engine/epoll/loop.go | 11 +++++------ middleware/sse/doc.go | 18 +++++++++++------- middleware/sse/sse_disconnect_linux_test.go | 2 +- .../sse/sse_nonreading_888_linux_test.go | 4 ++-- middleware/websocket/config.go | 2 +- middleware/websocket/engine_linux_test.go | 2 +- .../websocket/nonreading_888_linux_test.go | 4 ++-- 11 files changed, 29 insertions(+), 24 deletions(-) diff --git a/config.go b/config.go index ac369131b..cf9167275 100644 --- a/config.go +++ b/config.go @@ -139,7 +139,9 @@ type Config struct { // them for WriteTimeout, is closed (the stream's context ends and its // OnDisconnect runs), whatever the middleware's idle deadline or the last // read of the connection. A peer that reads, however slowly, and a - // connection with nothing queued are not closed; -1 sets no such bound. + // connection with nothing queued are not closed (there is no minimum + // rate); -1 sets no such bound. An h2c stream is not covered yet + // (celeris#884 item 79). // On the std engine net/http arms WriteTimeout as one deadline for the // whole response and Detach clears it for that request (ReadTimeout never // cuts a stream there: net/http has stopped it by the time the handler diff --git a/internal/engine/epoll/conn.go b/internal/engine/epoll/conn.go index b7786084c..f51157015 100644 --- a/internal/engine/epoll/conn.go +++ b/internal/engine/epoll/conn.go @@ -284,7 +284,7 @@ type connState struct { // conn (WS/SSE) is never in that state (closeWhenFlushed closes it at // once), and uses the same fields as the clock of a stall: how long its peer // has taken none of the bytes queued for it, bounded by WriteTimeout - // (detachedStalled, celeris#888 items 36 and 37), started by the sweep that + // (detachedStalled, celeris#888 items 38 and 39), started by the sweep that // finds bytes queued and cleared by the one that finds none. Worker-thread- // only; reset on release. closeSince int64 diff --git a/internal/engine/epoll/detach_stall_888_linux_test.go b/internal/engine/epoll/detach_stall_888_linux_test.go index 6f7733244..34a0d664f 100644 --- a/internal/engine/epoll/detach_stall_888_linux_test.go +++ b/internal/engine/epoll/detach_stall_888_linux_test.go @@ -16,7 +16,7 @@ import ( // A truly-detached conn (WS/SSE) whose peer takes none of the bytes queued for // it is closed WriteTimeout after the sweep found them queued (celeris#888 -// items 36 and 37, celeris#884 item 66): checkTimeouts skipped a detached conn +// items 38 and 39, celeris#884 item 69): checkTimeouts skipped a detached conn // altogether, and a write on a native engine only queues, so a client that // stayed connected and read nothing held its stream for ever. The clock is // "the peer took none", not the age of the stream or the last read. These diff --git a/internal/engine/epoll/detach_stall_nontcp_888_linux_test.go b/internal/engine/epoll/detach_stall_nontcp_888_linux_test.go index e21415303..f7b1e2df5 100644 --- a/internal/engine/epoll/detach_stall_nontcp_888_linux_test.go +++ b/internal/engine/epoll/detach_stall_nontcp_888_linux_test.go @@ -16,7 +16,7 @@ import ( // report. The stream's own goroutine flushes with no lock the loop's clocks are // under, so without that count a peer that reads cannot be told from one that // does not, and the conn is not closed (no bound, as before celeris#888 items -// 36 and 37) rather than on no evidence. +// 38 and 39) rather than on no evidence. func TestDetachedStallNeedsAnAckedCount888(t *testing.T) { l := stall888Loop(t, stall888Bound) pair, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM, 0) diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go index 7995fcfd2..e7c4bf4c0 100644 --- a/internal/engine/epoll/loop.go +++ b/internal/engine/epoll/loop.go @@ -2129,9 +2129,8 @@ func (cs *connState) endWriteSlow() { // Not on the hot path: checkTimeouts calls it for a closing conn, at most // every quarter of the bound (closeSampled) and once more before it reaps one, // and it costs one getsockopt. A -// descriptor that is not a TCP socket, or a kernel that does not report the -// count (before Linux 4.1), yields no evidence, which leaves the userspace -// queue's rule alone. Loop thread. +// descriptor that is not a TCP socket yields no evidence, which leaves the +// userspace queue's rule alone. Loop thread. func (l *Loop) noteClosingAcked(cs *connState, now int64) { cs.closeSampled = now ti, err := unix.GetsockoptTCPInfo(cs.fd, unix.IPPROTO_TCP, unix.TCP_INFO) @@ -4182,7 +4181,7 @@ func (l *Loop) checkTimeouts() { cs.drainDeadline = 0 // WriteTimeout is the one engine timeout a detached stream // keeps: "no progress for WriteTimeout" on what is queued for - // the peer (celeris#888 items 36 and 37). Here, whatever the + // the peer (celeris#888 items 38 and 39). Here, whatever the // middleware's idle deadline says: a WebSocket peer that sends // pings pushes it out for ever and reads nothing of what it is // sent. The expired case below is the middleware ending the @@ -4291,7 +4290,7 @@ func (l *Loop) checkTimeouts() { // detachedStalled is checkTimeouts' look at a truly-detached conn (WS/SSE) // that has bytes queued for its peer: whether the peer has taken none of them -// for WriteTimeout (celeris#888 items 36 and 37, celeris#884 item 66). The +// for WriteTimeout (celeris#888 items 38 and 39, celeris#884 item 69). The // middleware owns the stream's lifecycle, so no ReadTimeout, IdleTimeout or // absolute WriteTimeout applies to it; but a client that stays connected and // reads nothing never fails a write on a native engine (the write only queues), @@ -4343,7 +4342,7 @@ func (l *Loop) detachedStalled(cs *connState, now int64) bool { } if cs.closeSampled > cs.closeSince && !cs.closeAckedKnown { // A sample was taken since the stall began and the socket gave no - // acknowledged count (not a TCP socket, or a kernel before Linux 4.1): + // acknowledged count (not a TCP socket): // the stream's own goroutine flushes with no lock the loop's clocks are // under, and nothing else sees the bytes it writes, so a peer that // reads cannot be told from one that does not. No bound, as before this diff --git a/middleware/sse/doc.go b/middleware/sse/doc.go index 6d5994497..06364691b 100644 --- a/middleware/sse/doc.go +++ b/middleware/sse/doc.go @@ -40,14 +40,18 @@ // running fail too. // - epoll, io_uring and adaptive: [Client.Send] does not wait for a client // that reads nothing (the write only queues the event in the engine), so -// a write error and [ClientPolicyClose] are not what ends the stream of -// such a client. WriteTimeout does, as "no progress for WriteTimeout": a -// client that takes none of the bytes the engine has queued for it for +// a write error is not what ends the stream of such a client, and +// [ClientPolicyClose] ends it only earlier, when the queue overflows (a +// drain goroutine that lags behind the producer). WriteTimeout is the +// bound when it does not, as "no progress for WriteTimeout": a client +// that takes none of the bytes the engine has queued for it for // WriteTimeout is disconnected, which cancels [Client.Context], and -// [Config.OnDisconnect] runs. A client that keeps reading, however -// slowly, is not disconnected, and neither is one with nothing queued for -// it (WriteTimeout -1 sets no such bound). A handler that returns (its -// own bound, or its context) ends the stream sooner, and so does Shutdown. +// [Config.OnDisconnect] runs. A client that keeps reading, however slowly +// (there is no minimum rate), is not disconnected, and neither is one +// with nothing queued for it (WriteTimeout -1 sets no such bound). A +// handler that returns (its own bound, or its context) ends the stream +// sooner, and so does Shutdown. This is for HTTP/1.1 streams: an h2c +// stream is not covered yet (celeris#884 item 79). // // A HEAD request to an SSE route (answered by its GET route) gets the stream's // response headers and no body. [Config.OnConnect] runs for it, and a rejection diff --git a/middleware/sse/sse_disconnect_linux_test.go b/middleware/sse/sse_disconnect_linux_test.go index 4994c8128..860139024 100644 --- a/middleware/sse/sse_disconnect_linux_test.go +++ b/middleware/sse/sse_disconnect_linux_test.go @@ -98,7 +98,7 @@ func startSSEServer(tb testing.TB, engine celeris.EngineType, async, routeAsync } // startSSEServerCfg is startSSEServer with the server's Config given whole, -// for a test that sets timeouts (celeris#888 item 36). +// for a test that sets timeouts (celeris#888 item 38). func startSSEServerCfg(tb testing.TB, cfg celeris.Config, routeAsync bool, h celeris.HandlerFunc) (string, func()) { tb.Helper() began := time.Now() diff --git a/middleware/sse/sse_nonreading_888_linux_test.go b/middleware/sse/sse_nonreading_888_linux_test.go index 7e07f8eff..5b1225301 100644 --- a/middleware/sse/sse_nonreading_888_linux_test.go +++ b/middleware/sse/sse_nonreading_888_linux_test.go @@ -19,7 +19,7 @@ import ( ) // A client that stays connected and reads nothing of an SSE stream -// (celeris#888 items 36 and 37, celeris#884 item 66). +// (celeris#888 items 38 and 39, celeris#884 item 69). // // On epoll and io_uring a detached connection was exempt from WriteTimeout // (checkTimeouts skipped it), and Client.Send never parks there: the stream's @@ -312,7 +312,7 @@ func TestNonReadingClientEndsStream888(t *testing.T) { t.Errorf("OnDisconnect ran %v after connect, before WriteTimeout %v: the stream was ended by something other than the stall bound", el, wt) } case <-time.After(wt + slack): - t.Fatalf("OnDisconnect never ran %v after connect (WriteTimeout %v): a client that reads nothing holds its stream for ever (celeris#888 item 36)", + t.Fatalf("OnDisconnect never ran %v after connect (WriteTimeout %v): a client that reads nothing holds its stream for ever (celeris#888 item 38)", wt+slack, wt) } }) diff --git a/middleware/websocket/config.go b/middleware/websocket/config.go index b44cee541..bc1dada8e 100644 --- a/middleware/websocket/config.go +++ b/middleware/websocket/config.go @@ -118,7 +118,7 @@ type Config struct { // // One server timeout does apply to a detached connection on the native // engines: WriteTimeout, as "no progress for WriteTimeout" (celeris#888 - // items 36 and 37). A peer that takes none of the bytes the engine has + // items 38 and 39). A peer that takes none of the bytes the engine has // queued for it for WriteTimeout is closed, whatever it sends meanwhile // and whatever this field says. A peer that reads, however slowly, and a // connection with nothing queued are not closed; the server's diff --git a/middleware/websocket/engine_linux_test.go b/middleware/websocket/engine_linux_test.go index eba882477..74e99c02a 100644 --- a/middleware/websocket/engine_linux_test.go +++ b/middleware/websocket/engine_linux_test.go @@ -66,7 +66,7 @@ func startNativeServerWithHandle(tb testing.TB, kind celeris.EngineType, cfg Con } // startNativeServerCfg is startNativeServerWithHandle with the server's Config -// given whole, for a test that sets timeouts (celeris#888 items 36 and 37). +// given whole, for a test that sets timeouts (celeris#888 items 38 and 39). func startNativeServerCfg(tb testing.TB, ccfg celeris.Config, cfg Config) (string, func(), *celeris.Server) { tb.Helper() s := celeris.New(ccfg) diff --git a/middleware/websocket/nonreading_888_linux_test.go b/middleware/websocket/nonreading_888_linux_test.go index 6a53515a0..8c2516a52 100644 --- a/middleware/websocket/nonreading_888_linux_test.go +++ b/middleware/websocket/nonreading_888_linux_test.go @@ -16,7 +16,7 @@ import ( ) // A WebSocket peer that reads nothing of what it is sent gets the same bound as -// an SSE client (celeris#888 items 36 and 37): the native engines apply +// an SSE client (celeris#888 items 38 and 39): the native engines apply // WriteTimeout to a detached connection as "no progress for WriteTimeout", // whatever the middleware's idle deadline says. A peer that sends pings keeps // pushing that deadline out, which is the peer this test is. @@ -110,7 +110,7 @@ func TestNonReadingPeerEndsConn888(t *testing.T) { t.Errorf("ended %v after connect, before WriteTimeout %v", el, wsStall888Timeout) } case <-time.After(wsStall888Timeout + 6*time.Second): - t.Fatalf("a WebSocket peer that sends pings and reads nothing was not closed within %v (WriteTimeout %v): celeris#888 item 36", + t.Fatalf("a WebSocket peer that sends pings and reads nothing was not closed within %v (WriteTimeout %v): celeris#888 item 38", wsStall888Timeout+6*time.Second, wsStall888Timeout) } }) From e314576efa1a393f3bc572710738f30cb5608567 Mon Sep 17 00:00:00 2001 From: x Date: Sun, 18 Oct 2026 16:26:33 +0200 Subject: [PATCH 11/14] iouring: a benchmark of markLongSend on the send path; the completeSend benchmark's comment says a load and a branch (review round 1 of #1077; Refs #888 items 38, 39) --- .../complete_send_bench_888_linux_test.go | 3 +- .../mark_long_send_bench_888_linux_test.go | 39 +++++++++++++++++++ 2 files changed, 41 insertions(+), 1 deletion(-) create mode 100644 internal/engine/iouring/mark_long_send_bench_888_linux_test.go diff --git a/internal/engine/iouring/complete_send_bench_888_linux_test.go b/internal/engine/iouring/complete_send_bench_888_linux_test.go index ac77de76e..dd7ebf93e 100644 --- a/internal/engine/iouring/complete_send_bench_888_linux_test.go +++ b/internal/engine/iouring/complete_send_bench_888_linux_test.go @@ -9,7 +9,8 @@ import ( // BenchmarkCompleteSend888 is completeSend's all-sent path for an ordinary // (not detached) conn, the common one: a response that went out whole. The -// stall clock of a detached conn (celeris#888) adds a load and a store to it; +// stall clock of a detached conn (celeris#888) adds a load and a branch to it (the +// store only when the clock is set); // this is the figure to compare against an A/A twin (RULE 74). func BenchmarkCompleteSend888(b *testing.B) { w := &Worker{} diff --git a/internal/engine/iouring/mark_long_send_bench_888_linux_test.go b/internal/engine/iouring/mark_long_send_bench_888_linux_test.go new file mode 100644 index 000000000..80042eacc --- /dev/null +++ b/internal/engine/iouring/mark_long_send_bench_888_linux_test.go @@ -0,0 +1,39 @@ +//go:build linux + +package iouring + +import ( + "sync" + "testing" + "time" + + "github.com/goceleris/celeris/internal/conn" +) + +// BenchmarkMarkLongSend888 is markLongSend for a small SEND, on every send's +// path (flushSend): celeris#888 marks the SENDs of a truly-detached conn +// whatever their size (watchesStall), and the ordinary conn pays the compares +// that tell it apart: a nil detachMu for an inline conn, the WriteTimeout and +// the h1State's Detached flag for an async-mode conn. "detached" is the marked +// case, which also counts the conn (markRemainder; reset every op, as the SEND's +// completion does). The figure to compare against an A/A twin (RULE 74). +func BenchmarkMarkLongSend888(b *testing.B) { + for _, mode := range []string{"inline", "async", "detached"} { + b.Run(mode, func(b *testing.B) { + w := &Worker{} + w.cfg.WriteTimeout = 60 * time.Second + cs := &connState{fd: 7, liveIdx: -1, detected: true, h1State: conn.NewH1State()} + if mode != "inline" { + cs.detachMu = &sync.Mutex{} + } + cs.h1State.Detached.Store(mode == "detached") + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + w.markLongSend(cs, 512) + cs.remainderOut = false + w.remainderN = 0 + } + }) + } +} From a7700fe95fac49fdd7e65cc1f579be5c07d05b41 Mon Sep 17 00:00:00 2001 From: x Date: Sun, 18 Oct 2026 16:37:34 +0200 Subject: [PATCH 12/14] config, sse: the disconnect comes WriteTimeout after the last byte the client took, plus the engine's sweep interval (review round 1 of #1077; Refs #888 items 38, 39) --- config.go | 6 ++++-- middleware/sse/doc.go | 7 +++++-- 2 files changed, 9 insertions(+), 4 deletions(-) diff --git a/config.go b/config.go index cf9167275..f0a997c28 100644 --- a/config.go +++ b/config.go @@ -140,8 +140,10 @@ type Config struct { // OnDisconnect runs), whatever the middleware's idle deadline or the last // read of the connection. A peer that reads, however slowly, and a // connection with nothing queued are not closed (there is no minimum - // rate); -1 sets no such bound. An h2c stream is not covered yet - // (celeris#884 item 79). + // rate); -1 sets no such bound. The close comes WriteTimeout after the + // peer last took a byte, plus the engine's sweep interval (about 50 ms on + // epoll, up to 2 s on io_uring; a WriteTimeout shorter than that can + // double). An h2c stream is not covered yet (celeris#884). // On the std engine net/http arms WriteTimeout as one deadline for the // whole response and Detach clears it for that request (ReadTimeout never // cuts a stream there: net/http has stopped it by the time the handler diff --git a/middleware/sse/doc.go b/middleware/sse/doc.go index 06364691b..475fa610c 100644 --- a/middleware/sse/doc.go +++ b/middleware/sse/doc.go @@ -50,8 +50,11 @@ // (there is no minimum rate), is not disconnected, and neither is one // with nothing queued for it (WriteTimeout -1 sets no such bound). A // handler that returns (its own bound, or its context) ends the stream -// sooner, and so does Shutdown. This is for HTTP/1.1 streams: an h2c -// stream is not covered yet (celeris#884 item 79). +// sooner, and so does Shutdown. The disconnect comes WriteTimeout after +// the client last took a byte, plus the engine's sweep interval (about 50 +// ms on epoll, up to 2 s on io_uring; a WriteTimeout shorter than that +// can double). This is for HTTP/1.1 streams: an h2c stream is not covered +// yet (celeris#884). // // A HEAD request to an SSE route (answered by its GET route) gets the stream's // response headers and no body. [Config.OnConnect] runs for it, and a rejection From f417d899a1212305978dd66ddb1ecb63167b6751 Mon Sep 17 00:00:00 2001 From: x Date: Sun, 18 Oct 2026 16:45:16 +0200 Subject: [PATCH 13/14] iouring: the drain-clock test compares closingDrainBound as it is (unconvert; review round 1 of #1077) --- internal/engine/iouring/detach_stall_round2_888_linux_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/engine/iouring/detach_stall_round2_888_linux_test.go b/internal/engine/iouring/detach_stall_round2_888_linux_test.go index f9cfd3257..c25a22bac 100644 --- a/internal/engine/iouring/detach_stall_round2_888_linux_test.go +++ b/internal/engine/iouring/detach_stall_round2_888_linux_test.go @@ -158,7 +158,7 @@ func TestStalledCloseSkipsTheClosingDrain888(t *testing.T) { if _, closing := s.sweepUntilClosing(5*time.Second, nil); !closing { t.Fatal("not closed") } - if age := time.Now().UnixNano() - s.cs.lastActivity; age <= int64(s.w.closingDrainBound()) { + if age := time.Now().UnixNano() - s.cs.lastActivity; age <= s.w.closingDrainBound() { t.Fatalf("the closing drain's clock is %v old, the bound is %v: the stalled conn would hold its descriptor and queue for another bound", time.Duration(age), s.w.closingDrainBound()) } From fe4327277eb564f95fe927514678a9b2672edf23 Mon Sep 17 00:00:00 2001 From: x Date: Sun, 18 Oct 2026 17:06:03 +0200 Subject: [PATCH 14/14] config, sse, iouring: the stall close comes between 3/4 and 5/4 of WriteTimeout after the last byte plus the sweep interval; a stalled conn's closing sample keeps the stall's baseline (review round 1 of #1077; Refs #888 items 38, 39) --- config.go | 6 ++-- .../detach_stall_round2_888_linux_test.go | 33 ++++++++++++++++--- middleware/sse/doc.go | 10 +++--- 3 files changed, 38 insertions(+), 11 deletions(-) diff --git a/config.go b/config.go index f0a997c28..31210fcb4 100644 --- a/config.go +++ b/config.go @@ -140,8 +140,10 @@ type Config struct { // OnDisconnect runs), whatever the middleware's idle deadline or the last // read of the connection. A peer that reads, however slowly, and a // connection with nothing queued are not closed (there is no minimum - // rate); -1 sets no such bound. The close comes WriteTimeout after the - // peer last took a byte, plus the engine's sweep interval (about 50 ms on + // rate); -1 sets no such bound. The close comes between about 3/4 + // and 5/4 of WriteTimeout after the peer last took a byte (its progress + // is sampled every quarter of WriteTimeout; a SEND completion restarts the + // clock at once), plus the engine's sweep interval (up to about 0.1 s on // epoll, up to 2 s on io_uring; a WriteTimeout shorter than that can // double). An h2c stream is not covered yet (celeris#884). // On the std engine net/http arms WriteTimeout as one deadline for the diff --git a/internal/engine/iouring/detach_stall_round2_888_linux_test.go b/internal/engine/iouring/detach_stall_round2_888_linux_test.go index c25a22bac..3c821f255 100644 --- a/internal/engine/iouring/detach_stall_round2_888_linux_test.go +++ b/internal/engine/iouring/detach_stall_round2_888_linux_test.go @@ -150,16 +150,39 @@ func TestStallLookRateLimit888(t *testing.T) { // TestStalledCloseSkipsTheClosingDrain888: the peer has just been shown to take // nothing for WriteTimeout, so the drain that closeConn gives a conn with a // SEND in flight (closingDrainBound, 5 s or more) is spent at once: the next -// sweep may reap it. A peer that reads in the meantime would restamp the clock -// (noteClosingAcked). +// sweep may reap it. The closing sample that comes first (noteClosingAcked) +// must not give it a fresh drain while the peer stays silent, because the +// stall's sample is its baseline; a peer that reads in the meantime restamps +// the clock. func TestStalledCloseSkipsTheClosingDrain888(t *testing.T) { s := newStall888(t, stall888Bound) s.cs.lastActivity = time.Now().UnixNano() + fillSocket888(t, s.cs.fd) if _, closing := s.sweepUntilClosing(5*time.Second, nil); !closing { t.Fatal("not closed") } - if age := time.Now().UnixNano() - s.cs.lastActivity; age <= s.w.closingDrainBound() { - t.Fatalf("the closing drain's clock is %v old, the bound is %v: the stalled conn would hold its descriptor and queue for another bound", - time.Duration(age), s.w.closingDrainBound()) + bound := s.w.closingDrainBound() + spent := func(when string) { + t.Helper() + if age := time.Now().UnixNano() - s.cs.lastActivity; age <= bound { + t.Fatalf("%s: the closing drain's clock is %v old, the bound is %v: the stalled conn would hold its descriptor and queue for another bound", + when, time.Duration(age), time.Duration(bound)) + } + } + spent("after the stall close") + // The closing sample, a quarter of the bound later, with the peer silent. + later := time.Now().UnixNano() + bound/4 + int64(time.Millisecond) + s.w.noteClosingAcked(s.cs, later) + spent("after a closing sample of a silent peer") + // The peer reads: its acknowledged count moves, and the drain goes on. + buf := make([]byte, 64<<10) + for i := 0; i < 5; i++ { + _ = s.peer.SetReadDeadline(time.Now().Add(50 * time.Millisecond)) + _, _ = s.peer.Read(buf) + } + time.Sleep(50 * time.Millisecond) + s.w.noteClosingAcked(s.cs, later+int64(time.Second)) + if s.cs.lastActivity != later+int64(time.Second) { + t.Fatalf("a peer that read after the stall close did not restamp the drain clock (lastActivity %d, want %d)", s.cs.lastActivity, later+int64(time.Second)) } } diff --git a/middleware/sse/doc.go b/middleware/sse/doc.go index 475fa610c..5ac50f61e 100644 --- a/middleware/sse/doc.go +++ b/middleware/sse/doc.go @@ -50,10 +50,12 @@ // (there is no minimum rate), is not disconnected, and neither is one // with nothing queued for it (WriteTimeout -1 sets no such bound). A // handler that returns (its own bound, or its context) ends the stream -// sooner, and so does Shutdown. The disconnect comes WriteTimeout after -// the client last took a byte, plus the engine's sweep interval (about 50 -// ms on epoll, up to 2 s on io_uring; a WriteTimeout shorter than that -// can double). This is for HTTP/1.1 streams: an h2c stream is not covered +// sooner, and so does Shutdown. The disconnect comes between about 3/4 +// and 5/4 of WriteTimeout after the client last took a byte (its progress +// is sampled every quarter of WriteTimeout; a SEND completion restarts the +// clock at once), plus the engine's sweep interval (up to about 0.1 s on +// epoll, up to 2 s on io_uring; a WriteTimeout shorter than that can +// double). This is for HTTP/1.1 streams: an h2c stream is not covered // yet (celeris#884). // // A HEAD request to an SSE route (answered by its GET route) gets the stream's