diff --git a/config.go b/config.go index 863577382..4b340e4ce 100644 --- a/config.go +++ b/config.go @@ -159,10 +159,23 @@ 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 (there is no minimum + // 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 + // 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..f51157015 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 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 closePending int 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..34a0d664f --- /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 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 +// 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/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..f7b1e2df5 --- /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 +// 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) + 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..e7c4bf4c0 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. @@ -2130,17 +2129,17 @@ 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. -func (l *Loop) noteClosingAcked(cs *connState, now int64) (progressed bool) { +// 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) 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 +2148,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 @@ -4181,6 +4179,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 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 + // 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 +4288,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 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), +// 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): + // 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/checktimeouts_bench_888_linux_test.go b/internal/engine/iouring/checktimeouts_bench_888_linux_test.go new file mode 100644 index 000000000..4156a8ad7 --- /dev/null +++ b/internal/engine/iouring/checktimeouts_bench_888_linux_test.go @@ -0,0 +1,66 @@ +//go:build linux + +package iouring + +import ( + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "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, 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", "detached-idle-limited"} { + 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{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 != "attached" { + 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() { + if mode == "detached-idle" { + w.lastStallLook = 0 + } + w.checkTimeouts() + } + }) + } + } +} 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..dd7ebf93e --- /dev/null +++ b/internal/engine/iouring/complete_send_bench_888_linux_test.go @@ -0,0 +1,29 @@ +//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 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{} + 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/conn.go b/internal/engine/iouring/conn.go index 7a60c718e..84053fb34 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 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 + // 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_888_linux_test.go b/internal/engine/iouring/detach_stall_888_linux_test.go new file mode 100644 index 000000000..f94234794 --- /dev/null +++ b/internal/engine/iouring/detach_stall_888_linux_test.go @@ -0,0 +1,230 @@ +//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 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 +// 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 +} + +// 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) { + 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/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..45531f662 --- /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.sweepNow() + 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.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.sweepNow() + 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/detach_stall_round2_888_linux_test.go b/internal/engine/iouring/detach_stall_round2_888_linux_test.go new file mode 100644 index 000000000..3c821f255 --- /dev/null +++ b/internal/engine/iouring/detach_stall_round2_888_linux_test.go @@ -0,0 +1,188 @@ +//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. 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") + } + 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/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 + } + }) + } +} diff --git a/internal/engine/iouring/worker.go b/internal/engine/iouring/worker.go index 89df5273c..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 @@ -411,6 +410,85 @@ 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 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), +// 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 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 + 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 + } +} + // Worker is an io_uring event-loop worker pinned to a single OS thread. type Worker struct { id int @@ -617,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 @@ -4627,6 +4708,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 +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 { + 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 @@ -6679,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 @@ -6688,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. @@ -7093,6 +7203,85 @@ func snapshotH1Deadlines(cs *connState) (snap h1DeadlineSnapshot, ok bool) { 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 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) { + 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.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, @@ -7152,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] @@ -7273,6 +7463,8 @@ func (w *Worker) checkTimeouts() { if h1Detached { if idleDL > 0 && now > idleDL { w.closeConn(fd) + } else if stallLook { + w.sweepDetachedStall(cs, fd, now) } continue } diff --git a/middleware/sse/doc.go b/middleware/sse/doc.go index 225932eee..5ac50f61e 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,25 @@ // 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 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 +// (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 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 // 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 7bbc6ef1e..860139024 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 38). +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..5b1225301 --- /dev/null +++ b/middleware/sse/sse_nonreading_888_linux_test.go @@ -0,0 +1,381 @@ +//go:build linux + +package sse_test + +import ( + "net" + "os" + "strconv" + "strings" + "sync/atomic" + "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 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 +// 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 + +// 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 + 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 + + // 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 { + 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 disconnect888, shutdown func()) { + tb.Helper() + done := make(chan disconnect888, 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) + } + trickle := strings.Repeat("y", p.trickleBytes) + handler := sse.New(sse.Config{ + HeartbeatInterval: -1, + MaxQueueDepth: 4, + 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) { + select { + case done <- disconnect888{time.Now(), early.Load()}: + default: + } + }, + Handler: func(client *sse.Client) { + started.Store(time.Now().UnixNano()) + 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, 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, dropEarly: 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 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. + 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 38)", + 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/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) + } + }) + } + } + } +} diff --git a/middleware/websocket/config.go b/middleware/websocket/config.go index 4c7b3c10a..bc1dada8e 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 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 + // 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 diff --git a/middleware/websocket/engine_linux_test.go b/middleware/websocket/engine_linux_test.go index a8ba0e027..74e99c02a 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 38 and 39). +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..8c2516a52 --- /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 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. + +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 38", + wsStall888Timeout+6*time.Second, wsStall888Timeout) + } + }) + } +} diff --git a/test/server/chunked_split_1029_linux_test.go b/test/server/chunked_split_1029_linux_test.go index b52dc2295..128d433c8 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,19 @@ 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 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() - 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(),