epoll: the pool wait stops serving HTTP/1 and detached writes and closes idle keep-alive conns, queued driver rounds are not idle rounds (v1.6.0, Refs #885)

internal/conn/h1.go

@@ -347,6 +347,15 @@ type H1State struct {
 	// mid-body. Called on the goroutine running ProcessH1, under the
 	// engine's lock for the connection's buffers if it has one.
 	WriteBacklogged func() bool
+
+	// CloseAfter, set by an engine that is shutting down (epoll's pool-wait
+	// gate, celeris#885 item 49), makes every request handled from then on
+	// close-after: its response carries Connection: close and the connection
+	// ends once the response has gone, as net/http's response written after
+	// Shutdown began does, so a client does not write its next request to a
+	// connection that is about to be closed. Set and read on the goroutine that
+	// runs ProcessH1 (the loop thread for a conn the loop parses).
+	CloseAfter bool
 }
 
 // TakeBufferedBytes returns a copy of any bytes ProcessH1 stashed in the
@@ -1426,6 +1435,9 @@ func handleH1Request(ctx context.Context, state *H1State, body []byte,
 	}
 
 	req := &state.req
+	if state.CloseAfter {
+		req.KeepAlive = false
+	}
 	s := populateCachedStream(state, req, body)
 
 	// Reuse the connection-scoped response adapter — avoids a heap allocation

internal/engine/epoll/consecutive_empty_queue_885_linux_test.go

@@ -0,0 +1,93 @@
+//go:build linux
+
+package epoll
+
+import (
+	"testing"
+	"time"
+
+	"golang.org/x/sys/unix"
+
+	"github.com/goceleris/celeris/internal/resource"
+)
+
+// celeris#885 item 42: while driverReadQ is non-empty the run loop waits in
+// epoll_wait with timeout 0 (celeris#930), and the rounds that return no
+// events still counted toward consecutiveEmpty, the count adaptiveTimeoutMs
+// stretches the idle wait by (base*2 after 10, base*4 after 100). A flood of
+// driver bytes that takes a dozen rounds to read therefore ended with the loop
+// already counted as long idle, and its first idle waits started stretched.
+// No correctness impact (an event ends a wait at once); the test is on the
+// counter's effect: the wait the loop would make as the last of the queued
+// reads is delivered.
+
+// sndbuf885x42 is the send buffer asked for on the socketpair; the kernel caps
+// it at wmem_max, which is 212992 on a stock host and 4 MiB on some others.
+const sndbuf885x42 = 16 << 20
+
+// TestQueuedZeroTimeoutRoundsDoNotStretchTheIdleWait885 fills a socketpair
+// with as much as it takes (16 MiB asked for; a kernel with a small wmem_max
+// gives a few hundred KiB), so the loop reads it over several rounds: a turn is
+// driverReadBudget reads of driverReadBufSize (256 KiB), and every round after
+// the first is an epoll_wait(0) that collects nothing. The engine has
+// ReadHeaderTimeout -1, so no 25 ms timerfd can end the flood's counting between
+// two rounds. On the last read the count is 0 (round 1 had an event), and with
+// more than eleven turns the wait the loop would make next (adaptiveTimeoutMs)
+// is the base, 1 ms: the rounds a queue forced to zero are not idle rounds.
+func TestQueuedZeroTimeoutRoundsDoNotStretchTheIdleWait885(t *testing.T) {
+	eng, _ := startE2E885(t, false, newPathHandler885(1), func(c *resource.Config) { c.ReadHeaderTimeout = -1 })
+	l := eng.WorkerLoop(0).(*Loop)
+	if l.timerFD >= 0 {
+		t.Fatalf("apparatus: the loop has a sweep timerfd, which can fire between two rounds of the flood")
+	}
+	a, peer := socketpairNonblocking(t)
+	t.Cleanup(func() { _ = unix.Close(a); _ = unix.Close(peer) })
+	for _, fd := range []int{a, peer} {
+		if err := unix.SetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_SNDBUF, sndbuf885x42); err != nil {
+			t.Fatalf("SO_SNDBUF: %v", err)
+		}
+	}
+	chunk := make([]byte, 64<<10)
+	filled := 0
+	for {
+		n, err := unix.Write(peer, chunk)
+		if err != nil || n <= 0 {
+			break
+		}
+		filled += n
+	}
+	const turn = driverReadBudget * driverReadBufSize
+	if filled <= turn {
+		t.Skipf("apparatus: the socketpair took %d bytes, not more than one turn (%d)", filled, turn)
+	}
+
+	total, reads := 0, 0
+	last := make(chan [2]int, 1)
+	if err := l.RegisterConn(a, func(p []byte) {
+		total += len(p)
+		reads++
+		if total == filled {
+			// On the loop thread, the loop's own state: the counter as of
+			// the end of the previous round.
+			last <- [2]int{l.adaptiveTimeoutMs(1), int(l.consecutiveEmpty)}
+		}
+	}, func(error) {}); err != nil {
+		t.Fatalf("RegisterConn: %v", err)
+	}
+	t.Cleanup(func() { _ = l.UnregisterConn(a) })
+	select {
+	case got := <-last:
+		turns := (filled + turn - 1) / turn
+		t.Logf("celeris885 item 42: %d bytes (%d turns) read in %d reads; on the last, consecutiveEmpty=%d and the next idle wait would be %d ms",
+			filled, turns, reads, got[1], got[0])
+		if got[1] != 0 {
+			t.Errorf("after %d turns of queued reads (epoll_wait(0) rounds that collected nothing) consecutiveEmpty=%d, want 0: a round the queue "+
+				"forced to timeout 0 must not count as an idle round (celeris#885 item 42)", turns, got[1])
+		}
+		if turns > 11 && got[0] != 1 {
+			t.Errorf("after %d turns the idle wait is %d ms, want the base, 1 ms (celeris#885 item 42)", turns, got[0])
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatalf("the loop read %d of %d bytes in 10 s", total, filled)
+	}
+}

internal/engine/epoll/loop.go

@@ -214,6 +214,13 @@ type Loop struct {
 	// (celeris#759; h2PoolSettled). Zero until then. Loop thread.
 	h2DrainStart time.Time
 
+	// refuseRecv is set by the pool wait's gate (poolWaitGate, celeris#885
+	// item 49): from then on drainRead reads no more from an HTTP/1 conn.
+	// producersStopNext is the time (UnixNano) before which the gate does not
+	// walk the conns again. Loop thread.
+	refuseRecv        bool
+	producersStopNext int64
+
 	// transplantInFlight counts connections this loop has detached for a
 	// transplant whose hand-off is not finished yet — the deferred async
 	// path, where tryTransplant detaches and drainDetachQueue completes.
@@ -536,6 +543,10 @@ func (l *Loop) run(ctx context.Context) {
 				l.shutdown()
 				return
 			}
+			// Waiting for HTTP/2 pool handlers (h2DrainStart is set): from
+			// poolWaitGrace into the wait, serve no more HTTP/1 and let no
+			// detached stream write (celeris#885 item 49).
+			l.poolWaitGate()
 		}
 
 		// Cache the atomic load: ACTIVE→LINGERING→DRAINING and
@@ -771,10 +782,13 @@ func (l *Loop) run(ctx context.Context) {
 			l.acceptAll(ctx, time.Now().UnixNano())
 		}
 
-		// Adaptive timeout tracking.
+		// Adaptive timeout tracking. A round that collected nothing counts as
+		// an idle one only if the loop was willing to wait: a queued driver
+		// conn forced the timeout to 0, which says nothing about idleness
+		// (celeris#885 item 42).
 		if n > 0 {
 			l.consecutiveEmpty = 0
-		} else {
+		} else if owedDriverReads == 0 {
 			l.consecutiveEmpty++
 		}
 
@@ -1517,6 +1531,11 @@ func (l *Loop) drainRead(fd int, now int64) {
 	if cs == nil {
 		return
 	}
+	// Past the pool wait's gate an HTTP/1 conn is read no more (celeris#885
+	// item 49): one load and branch per read event.
+	if l.refuseRecv && refusesRecv(cs) {
+		return
+	}
 
 	for {
 		// Zero-copy body recv: when H1 is in a partial-body state we read

internal/engine/epoll/poolwait_gate_885.go

@@ -0,0 +1,225 @@
+//go:build linux
+
+package epoll
+
+import (
+	"time"
+
+	"golang.org/x/sys/unix"
+
+	"github.com/goceleris/celeris/internal/engine"
+)
+
+// The gate of the wait for HTTP/2 pool handlers (celeris#885 item 49), the
+// epoll twin of io_uring's (celeris#884 item 50). A loop whose context is
+// cancelled waits, as long as an HTTP/2 conn has a handler running on the
+// shared worker pool, a response queued or DATA waiting for a window
+// (h2PoolSettled), and the wait can last the whole budget of the shutdown. It
+// used to go on reading and answering the keep-alive requests of its HTTP/1
+// clients and let detached streams (Server-Sent Events, a WebSocket) write
+// meanwhile: the phases of shutdown that stop both (its phase 1 and the send
+// drain) run only after the wait. Measured with
+// TestShutdownPoolWaitStopsServingH1AndDetached884 on epoll: 16 of 16
+// requests answered and about 700 ticks written 1.5 s into the shutdown.
+//
+// The gate comes poolWaitGrace after the wait began, as the send drain's does,
+// so a request in flight when the shutdown began is still answered. From then
+// on the loop reads no more new requests from an HTTP/1 conn (refuseRecv) and
+// ends the writes of every detached stream (stopDetachedProducers). A request
+// already in progress on a conn the loop parses itself (headers in pieces, a
+// body arriving) is read on and answered, as std's Shutdown answers an active
+// conn, with Connection: close (H1State.CloseAfter, as net/http's response
+// written after Shutdown began). The HTTP/2 conns the wait is for are served on:
+// they have had GOAWAY since the wait began.
+//
+// An HTTP/1 conn that is idle, holds nothing queued and has no request in
+// progress is closed, as std's Shutdown closes its idle conns: its client fails
+// at once and retries elsewhere, where it would otherwise wait, with no answer,
+// for as long as the wait lasts (up to the budget of the shutdown) and then be
+// cut (closeIdleH1). So is a conn that was accepted and has sent nothing, and a
+// detached stream the gate has ended once what was queued for it has gone out
+// (its client gets a FIN and reconnects), whatever kind of route it is on. A
+// conn a dispatch goroutine owns (asyncPromoted: on a server with any Async
+// route every conn whose read leaves a partial request is promoted, whatever its
+// route, and a conn on an Async route is) is otherwise left to the end of the
+// wait, with a request in progress on it unread: the documented behaviour and
+// a follow-up shared with io_uring (which has no idle close at all).
+
+// poolWaitGrace is how long into the wait for the HTTP/2 pool handlers the
+// gate stays open: io_uring's shutdownSendDrainNanos, and this engine's
+// shutdownSendDrainFloor.
+const poolWaitGrace = shutdownSendDrainFloor
+
+// gateClosesPerPass caps the conns closeIdleH1 closes in one pass, so that a
+// loop with thousands of idle conns does not stop serving the HTTP/2 conns the
+// wait is for while it closes them all (several syscalls each): the next pass,
+// 10 ms on, goes on.
+const gateClosesPerPass = 64
+
+// refusesRecv reports whether a loop past the pool wait's gate reads no more
+// from cs: an HTTP/1 conn (detached or not) or one whose protocol is not
+// detected yet, which io_uring's refuseRecv test mirrors, with one exception
+// io_uring does not make: a request in progress is read on, and answered, as
+// std's Shutdown answers the request of an active conn (headers in pieces, a
+// body still arriving, the bytes of a protocol decision begun). A conn whose
+// dispatch goroutine owns its parse state (asyncPromoted) is refused as before:
+// its state is not the loop's to read. Loop thread.
+func refusesRecv(cs *connState) bool {
+	if !cs.detected {
+		return cs.detectN == 0
+	}
+	if cs.protocol != engine.HTTP1 {
+		return false
+	}
+	return cs.asyncPromoted || cs.h1State == nil || !cs.h1State.HasPendingData()
+}
+
+// poolWaitGate applies the gate once the loop has waited poolWaitGrace for the
+// HTTP/2 pool handlers (h2PoolSettled has set h2DrainStart). The wait is a
+// serving phase, so the walk over every conn, which ends the producers and
+// closes the idle HTTP/1 conns, is not paid on every pass: it runs at most once
+// per h2PoolDrainPollMs. Loop thread.
+func (l *Loop) poolWaitGate() {
+	now := time.Now()
+	if now.Sub(l.h2DrainStart) <= poolWaitGrace {
+		return
+	}
+	l.refuseRecv = true
+	if n := now.UnixNano(); n >= l.producersStopNext {
+		l.producersStopNext = n + int64(h2PoolDrainPollMs*time.Millisecond)
+		l.stopDetachedProducers()
+		l.closeIdleH1()
+	}
+}
+
+// closeIdleH1 closes the HTTP/1 conns the gate has stopped reading that hold
+// nothing: no response queued in the engine or the kernel's send queue (a close
+// with bytes unsent would reset them away), no request in progress, not a
+// detached stream still running (one the gate has ended is closed once it has
+// flushed, on a sync route or an Async one), no dispatch goroutine otherwise
+// (see the gate's comment), and a conn with no byte sent (not detected,
+// detectN 0). It also marks every conn it meets close-after (H1State.CloseAfter),
+// so that a request it still reads (one in progress) is answered with
+// Connection: close. A conn whose response is
+// still going out is met again by the next pass. closeConn removes the conn
+// from liveConns by swapping it with the last entry, so the walk goes from the
+// top down, as checkTimeouts does. Loop thread.
+func (l *Loop) closeIdleH1() {
+	closed := 0
+	for i := len(l.liveConns) - 1; i >= 0; i-- {
+		cs := l.liveConns[i]
+		if cs.hijacked.Load() {
+			continue
+		}
+		// A conn a dispatch goroutine owns is closed here only when the gate has
+		// ended its detached stream (an SSE or WebSocket handler on an Async
+		// route detaches on that goroutine): the goroutine is out of its handler,
+		// since detachClosed is set only under a detachMu the handler held, its
+		// writes no-op, and closeConn takes the lock as shutdown's own teardown
+		// does. The state is read under the lock.
+		ended := false
+		if cs.asyncPromoted {
+			mu := cs.detachMu
+			if mu == nil || !mu.TryLock() {
+				continue
+			}
+			ended = cs.detachClosed && cs.protocol == engine.HTTP1 && cs.h1State != nil && cs.h1State.Detached.Load()
+			mu.Unlock()
+			if !ended {
+				continue
+			}
+		}
+		if cs.detected {
+			if cs.protocol != engine.HTTP1 {
+				continue
+			}
+			if h := cs.h1State; h != nil && !ended {
+				// A request read after the gate is answered close-after (net/http
+				// does the same for a response written after Shutdown began).
+				h.CloseAfter = true
+				if h.HasPendingData() || (h.Detached.Load() && !cs.detachClosed) {
+					continue
+				}
+			}
+		} else if cs.detectN != 0 {
+			continue
+		}
+		if closed >= gateClosesPerPass || csWritePending(cs) || cs.dirty || cs.epollOut {
+			continue
+		}
+		if q, err := unix.IoctlGetInt(cs.fd, unix.TIOCOUTQ); err != nil || q != 0 {
+			continue
+		}
+		l.closeConn(cs.fd)
+		closed++
+	}
+}
+
+// stopDetachedProducers is the detached half of shutdown's phase 1, run early
+// by the pool wait's gate: a detached conn (Server-Sent Events, a WebSocket)
+// has a goroutine of its own that keeps writing, and with the wait lasting as
+// long as the budget a stream that never ends would write for all of it. The
+// stream is told as shutdown tells it: asyncClosed, which the write-error
+// query reads first, then, under detachMu, the parked read-end error, then
+// detachClosed (the guarded write hook no-ops from there) and OnDetachClose.
+// Only conns that have detached (h1State.Detached) and whose dispatch
+// goroutine is not inside the handler: the goroutine holds detachMu across
+// the handler, so the TryLock fails for a running async handler, which is
+// answered, up to the budget, as before. TryLock, never Lock, so a goroutine
+// parked inside a write cannot wedge the loop; a conn whose lock is busy is
+// tried again on the next pass of the pacing. A conn whose async Detach is
+// not yet finalised (asyncDetachPending) is left for that pass.
+//
+// What the producer's guarded write hook left queued is flushed, not
+// stranded: the inline write may have taken only part of the bytes, and
+// drainDetachQueue skips a detachClosed conn, so a conn with bytes queued and
+// neither EPOLLOUT armed nor on the dirty list is put on the list here. The
+// rest is shutdown's own, which every step repeats harmlessly (the parked
+// error and OnDetachClose are consumed once). Loop thread.
+func (l *Loop) stopDetachedProducers() {
+	for i := len(l.liveConns) - 1; i >= 0; i-- {
+		cs := l.liveConns[i]
+		mu := cs.detachMu
+		if mu == nil || cs.hijacked.Load() || cs.h1State == nil || !cs.h1State.Detached.Load() {
+			continue
+		}
+		if !mu.TryLock() {
+			continue
+		}
+		// Re-tested under the lock, as closeConn's ownership check is: an async
+		// Hijack takes the conn and closes its descriptor under this lock.
+		if cs.hijacked.Load() || (!cs.detachClosed && cs.asyncDetachPending) {
+			mu.Unlock()
+			continue
+		}
+		if !cs.detachClosed {
+			cs.asyncClosed.Store(true)
+			if cs.asyncCond.L != nil {
+				cs.asyncInMu.Lock()
+				cs.asyncCond.Broadcast()
+				cs.asyncInMu.Unlock()
+			}
+			if err := cs.closeErr; err != nil {
+				cs.closeErr = nil
+				if cs.h1State.OnError != nil {
+					cs.h1State.OnError(err)
+				}
+			}
+			cs.detachClosed = true
+			if cs.h1State.WSReady.Load() && cs.h1State.OnDetachClose != nil {
+				cs.h1State.OnDetachClose()
+				cs.h1State.OnDetachClose = nil
+			}
+		}
+		// What is queued must still reach the peer, and nothing but this pass
+		// will see to it once detachClosed is set: drainDetachQueue skips such
+		// a conn, so a remainder the producer's write queued for the loop, or a
+		// conn a relink gave up (its hand-back is skipped the same way) and
+		// that EPOLLOUT no longer watches, is stranded. Met again on every
+		// pass, for a conn that is given up after this one.
+		if csWritePending(cs) && !cs.epollOut && !cs.dirty {
+			l.markDirty(cs)
+		}
+		mu.Unlock()
+	}
+}