iouring: the shutdown gate closes idle and ended HTTP/1 conns and answers requests in progress, ring-wait ceiling, shutdown trials (v1.6.0, Refs #884)

internal/engine/epoll/poolwait_gate_885.go

@@ -43,7 +43,8 @@ import (
 // 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).
+// a follow-up (io_uring's gate, shutdown_gate_884.go, looks inside such a conn
+// under the goroutine's own locks and closes it when idle).
 
 // 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
@@ -58,10 +59,10 @@ 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
+// detected yet, which io_uring's gateReadsOn mirrors: 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
 // while it is HTTP/1: its state is not the loop's to read. Loop thread.
 func refusesRecv(cs *connState) bool {

internal/engine/iouring/fd_lifetime.go

@@ -740,7 +740,15 @@ func (w *Worker) endOwedOpsAtShutdown() {
 					// (a pause, a resume and a shutdown at once; cancelAccept
 					// reaps inline, which keeps that window narrow) is not
 					// taken for it unless a re-created socket reused the
-					// number (appended to #884).
+					// number, which it usually does. Known limit (celeris#884
+					// item 63, decided 2026-10-12 as a comment, not a fix): the
+					// wait then ends early and the port stays bound for the
+					// ring teardown's ~2 ms. Counting terminal accept CQEs
+					// against the cancel's result (the number it cancelled,
+					// listenCancelAnswer's positive count) would close it, but
+					// needs the order of the cancel's CQE and the accept's
+					// terminal CQE pinned on every kernel CI runs. Narrow, not
+					// reproduced; cancelAccept has the same limit.
 					if !cqeHasMore(c.Flags) && decodeFD(ud) == listenFDAtCancel {
 						acceptOwed = false
 					}

internal/engine/iouring/park_retracts_residue_test.go

@@ -23,9 +23,19 @@ package iouring
 // close owed does not park, so the next iteration's sweep() retracts. The arm
 // goes red with both removed, not with the retraction alone (checked in
 // celeris#885 item 11); the retraction is what a close with nothing in flight
-// would need.
+// would need. TestParkedWorkerRetractsResidueOfACloseWithNothingInFlight884 is
+// that close (celeris#884 item 80): the conn's recv is ended and reaped inside
+// the sweep's own pass, so its close lands after sweep() with nothing owed to
+// the kernel, the gate does not hold the worker, and only the retraction at
+// the park clears the gauges (red with it removed). What this proves: the
+// retraction is needed for that state. What it does not: that production
+// reaches that state (no traffic path was found that leaves the last conn with
+// no op in flight after sweep(); the state is manufactured in the hook), so the
+// item's other outcome, calling the retraction dead behind the closeFDOwed gate,
+// is not shown either.
 
 import (
+	"context"
 	"io"
 	"log/slog"
 	"net"
@@ -100,7 +110,12 @@ func parkWaitIou(d time.Duration, f func() bool) bool {
 // checkTimeouts (postSweep) or by the client, right after PauseAccept
 // (!postSweep), and asserts that a worker that has parked holding nothing
 // publishes nothing. The header timeout is 60 s in both arms: far from either.
-func parkResidueIou(t *testing.T, postSweep bool) {
+func parkResidueIou(t *testing.T, postSweep bool) { parkResidueIouMode(t, postSweep, false) }
+
+// parkResidueIouMode is parkResidueIou with a third kind of close: nothingOwed
+// replaces ReadTimeout's close by one made after sweep() with no op owed to the
+// kernel (see afterSweepNothingOwed884).
+func parkResidueIouMode(t *testing.T, postSweep, nothingOwed bool) {
 	var disc atomic.Int64
 	e, addr := startParkEngine711(t, fdlHandler{}, func(c *resource.Config) {
 		c.OnDisconnect = func(string) { disc.Add(1) }
@@ -110,7 +125,7 @@ func parkResidueIou(t *testing.T, postSweep bool) {
 		// listener, the only kind that parks.
 		c.DisableDeferAccept = true
 		c.ReadHeaderTimeout = 60 * time.Second // neither the header timer nor its checkTimeouts close can come first
-		if postSweep {
+		if postSweep && !nothingOwed {
 			c.ReadTimeout = 300 * time.Millisecond
 			c.IdleTimeout = 10 * time.Minute
 		}
@@ -156,21 +171,30 @@ func parkResidueIou(t *testing.T, postSweep bool) {
 		t.Fatalf("celeris711 PREMISE: the sweep never published a residue of 1 (sum=%d adopted=%d)",
 			residueSum(e.Metrics()), tgt.adopted.Load())
 	}
+	var manufactured atomic.Int32
+	if nothingOwed {
+		hook := afterSweepNothingOwed884(&manufactured)
+		testHookCheckTimeouts.Store(&hook)
+		t.Cleanup(func() { testHookCheckTimeouts.Store(nil) })
+	}
 	if err := e.PauseAccept(); err != nil {
 		t.Fatalf("pause: %v", err)
 	}
 	if n := e.Metrics().ActiveConnections; n != 1 {
 		t.Fatalf("celeris711 PREMISE: the connection closed (active=%d) before PauseAccept returned, so not "+
 			"on a worker without a listener", n)
 	}
-	if !postSweep {
+	if !postSweep && !nothingOwed {
 		// The client leaves, far before any deadline: the worker reads the
 		// EOF from a CQE, and CQEs are dispatched before sweep().
 		_ = c.Close()
 	}
 	if !parkWaitIou(10*time.Second, func() bool { return e.Metrics().ActiveConnections == 0 }) {
 		t.Fatalf("celeris711 PREMISE: the connection was never closed")
 	}
+	if nothingOwed && manufactured.Load() != 1 {
+		t.Fatalf("celeris884 PREMISE: the sweep hook ended the conn's recv %d times, want 1", manufactured.Load())
+	}
 	if !parkWaitIou(3*time.Second, allParked) {
 		t.Fatalf("celeris711 PREMISE: not every worker parked")
 	}
@@ -202,3 +226,45 @@ func TestParkedWorkerRetractsResidueOfAPostSweepClose(t *testing.T) { parkResidu
 // connection closed by the client's EOF, a CQE dispatched before sweep(),
 // which retracted before this fix as well.
 func TestParkedWorkerRetractsResidueOfAPreSweepClose(t *testing.T) { parkResidueIou(t, false) }
+
+// afterSweepNothingOwed884 is a checkTimeouts hook (it runs on the worker's
+// thread, after sweep() in the same pass) that gives a worker with one live
+// conn, no listener and a recv armed on it the close celeris#884 item 80 asks
+// for: it cancels the recv, submits, and reaps the cancelled recv's CQE the way
+// the loop does, which closes the conn (a failed recv closes it) with nothing
+// left in flight. The close is therefore after sweep() and owes the kernel
+// nothing, so closeFDOwed stays 0 and the worker parks in the same pass; once.
+func afterSweepNothingOwed884(done *atomic.Int32) func(*Worker) {
+	return func(w *Worker) {
+		if done.Load() != 0 || w.listenFD >= 0 || !w.acceptPaused.Load() || len(w.liveConns) != 1 {
+			return
+		}
+		cs := w.conns[w.liveConns[0]]
+		if cs == nil || !cs.recvArmed || cs.closing {
+			return
+		}
+		fd := cs.fd
+		if w.cancelConnOps(fd, cs) {
+			return // the cancel was not placed; the next sweep tries again
+		}
+		done.Store(1)
+		for end := time.Now().Add(2 * time.Second); w.conns[fd] == cs && time.Now().Before(end); {
+			if err := w.ring.SubmitAndWaitTimeout(10 * time.Millisecond); err != nil && !ringWaitRetryable(err) {
+				return
+			}
+			head, tail := w.ring.BeginCQ()
+			for ; head != tail; head++ {
+				w.processCQE(context.Background(), w.ring.cqeAt(head), time.Now().UnixNano())
+			}
+			w.ring.EndCQ(head)
+		}
+	}
+}
+
+// TestParkedWorkerRetractsResidueOfACloseWithNothingInFlight884 is the arm item
+// 80 asks for: the last conn closed after sweep() with nothing owed to the
+// kernel, which the closeFDOwed gate does not hold off the park. It goes red
+// with the park's sweepRetract removed.
+func TestParkedWorkerRetractsResidueOfACloseWithNothingInFlight884(t *testing.T) {
+	parkResidueIouMode(t, true, true)
+}

internal/engine/iouring/ring_wait_ceiling_884_linux_test.go

@@ -0,0 +1,189 @@
+//go:build linux
+
+package iouring
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"log/slog"
+	"net"
+	"net/http"
+	"testing"
+	"time"
+
+	"golang.org/x/sys/unix"
+)
+
+// celeris#884 item 72: ringWaitContinues had no ceiling. A ring wait that
+// answers a retryable error (EBUSY, EAGAIN) for good, with nothing in the CQ
+// ring to reap, is a ring that is failing, not one whose CQ ring is full; the
+// loop retried it for ever (a 50 us sleep per pass), where before PR #1026 it
+// shut the worker down on the first error. The ceiling restores that end for a
+// run of about a second of retries that reaped nothing, and nothing shorter or
+// less stuck: a retry that finds a completion in the CQ ring (the loop's own
+// pass reaps it: the overloaded case the retry exists for) or one that follows
+// a wait that succeeded is not part of the run.
+
+// ceilingErr884 is the error of a stuck ring.
+func ceilingErr884() error { return fmt.Errorf("io_uring_enter submit+wait: %w", unix.EBUSY) }
+
+// TestRingWaitContinuesGivesUpOnAStuckRing884: retryable errors, one after the
+// other, on a CQ ring with nothing in it end the worker after about a second.
+func TestRingWaitContinuesGivesUpOnAStuckRing884(t *testing.T) {
+	ring := newTestRing(t)
+	logs := &warnLog884{}
+	w := &Worker{ring: ring, id: 3, logger: slog.New(logs)}
+
+	start := time.Now()
+	retries := 0
+	var gaveUpAfter time.Duration
+	for time.Since(start) < 5*time.Second {
+		if !w.ringWaitContinues(waitModeTimed, ceilingErr884()) {
+			gaveUpAfter = time.Since(start)
+			break
+		}
+		retries++
+	}
+	t.Logf("celeris884: %d consecutive EBUSY waits with nothing to reap; gave up after %v", retries, gaveUpAfter)
+	if gaveUpAfter == 0 {
+		t.Fatalf("%d consecutive EBUSY waits over %v, with nothing in the CQ ring to reap, and the worker is still retrying: a failing ring is never given up on",
+			retries, time.Since(start).Round(time.Millisecond))
+	}
+	if gaveUpAfter < 500*time.Millisecond {
+		t.Errorf("the worker gave up on a ring after %v of retries (%d): too soon for a wait that only meets a full CQ ring", gaveUpAfter, retries)
+	}
+	if gaveUpAfter > 3*time.Second {
+		t.Errorf("the worker gave up on a ring after %v of retries: the ceiling is about a second", gaveUpAfter)
+	}
+	var errs int
+	logs.mu.Lock()
+	for _, r := range logs.recs {
+		if r.Level >= slog.LevelError {
+			errs++
+		}
+	}
+	logs.mu.Unlock()
+	if errs == 0 {
+		t.Error("the worker stopped on a failing ring and logged no error: its operator would see a stopped worker with no reason")
+	}
+}
+
+// TestRingWaitContinuesNeverGivesUpWhileItReaps884: the same error with a
+// completion waiting in the CQ ring at each retry is the loop being behind its
+// ring, not a failing ring: the loop's pass reaps, and the retry goes on for as
+// long as that lasts, well past the ceiling.
+func TestRingWaitContinuesNeverGivesUpWhileItReaps884(t *testing.T) {
+	ring := newTestRing(t)
+	w := &Worker{ring: ring, id: 3, logger: slog.New(&warnLog884{})}
+	postNOP884(t, ring)
+	start := time.Now()
+	n := 0
+	for time.Since(start) < 2500*time.Millisecond {
+		if !w.ringWaitContinues(waitModeSubmitSends, ceilingErr884()) {
+			t.Fatalf("the worker gave up after %v and %d retries, each of which found a completion in the CQ ring to reap",
+				time.Since(start).Round(time.Millisecond), n)
+		}
+		n++
+		time.Sleep(time.Millisecond)
+	}
+}
+
+// TestRingWaitContinuesCountsOnlyConsecutiveRetries884: retries that come
+// apart, each after waits that succeeded, are not a stuck ring: a worker that
+// meets an EBUSY now and then, with nothing to reap each time, runs on. The
+// loop pass count (tickCounter) moves once per pass, so a retry that follows
+// the last by more than one pass has a wait that succeeded between them: the
+// test models it by advancing the count, with the time between the retries a
+// stuck run's own (50 us sleeps) so that only the count tells the two apart.
+func TestRingWaitContinuesCountsOnlyConsecutiveRetries884(t *testing.T) {
+	ring := newTestRing(t)
+	w := &Worker{ring: ring, id: 3, logger: slog.New(&warnLog884{})}
+	start := time.Now()
+	n := 0
+	for time.Since(start) < 2500*time.Millisecond {
+		if !w.ringWaitContinues(waitModeTimed, ceilingErr884()) {
+			t.Fatalf("the worker gave up after %v and %d retries, each apart from the last by a pass whose wait succeeded",
+				time.Since(start).Round(time.Millisecond), n)
+		}
+		n++
+		w.tickCounter += 2 // the retried pass, then a pass whose wait succeeded
+	}
+	t.Logf("celeris884: %d retries, every other pass, over %v: the worker runs on", n, time.Since(start).Round(time.Millisecond))
+}
+
+// TestRunStopsOnAStuckRingWait884: the engine's own arm. One worker's waits
+// answer EBUSY for good (the seam) on an engine with nothing to do, so there is
+// nothing to reap; it stops within a few seconds, as it does for any other ring
+// failure (TestRunStopsOnARingWaitFailure884), and the other worker is left to
+// serve.
+func TestRunStopsOnAStuckRingWait884(t *testing.T) {
+	inj := &ebusyInject884{err: ebusyErr884(), n: 1 << 40, firstOnly: true}
+	inj.only.Store(-1)
+	setRingWaitSeam884(t, inj.seam)
+	e, addr := startFDLEngine(t, transplantTestHandler{}, nil)
+
+	time.Sleep(200 * time.Millisecond)
+	armedAt := time.Now()
+	inj.armed.Store(true)
+	var dead *Worker
+	for deadline := time.Now().Add(6 * time.Second); time.Now().Before(deadline) && dead == nil; time.Sleep(10 * time.Millisecond) {
+		for _, w := range e.workers {
+			if w.exited.Load() {
+				dead = w
+			}
+		}
+	}
+	t.Logf("celeris884 stuck ring: injected=%d%s", inj.total(-1), inj.summary())
+	if inj.total(-1) == 0 {
+		t.Fatal("the injected EBUSY never reached a wait: the test would prove nothing")
+	}
+	if dead == nil {
+		t.Fatalf("a worker whose every ring wait answered EBUSY for %v, with nothing to reap, was still running: the retry has no ceiling", time.Since(armedAt).Round(time.Millisecond))
+	}
+	t.Logf("celeris884 stuck ring: worker %d stopped %v after the injection began", dead.id, time.Since(armedAt).Round(time.Millisecond))
+	if dead != inj.chosen.Load() {
+		t.Errorf("worker %d stopped, but the EBUSY was answered to worker %d", dead.id, inj.chosen.Load().id)
+	}
+	for _, w := range e.workers {
+		if w != dead && w.exited.Load() {
+			t.Errorf("worker %d stopped though only worker %d met the failure", w.id, dead.id)
+		}
+	}
+	if len(e.workers) < 2 {
+		t.Log("one worker (the memlock cap): no survivor to ask")
+		return
+	}
+	c, err := net.DialTimeout("tcp", addr, time.Second)
+	if err != nil {
+		t.Fatalf("dial after the stuck worker stopped: %v", err)
+	}
+	defer func() { _ = c.Close() }()
+	if _, err := io.WriteString(c, "GET / HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil {
+		t.Fatal(err)
+	}
+	_ = c.SetReadDeadline(time.Now().Add(3 * time.Second))
+	if _, err := http.ReadResponse(bufio.NewReader(c), nil); err != nil {
+		t.Errorf("the surviving worker did not answer: %v", err)
+	}
+}
+
+// TestRingWaitContinuesDoesNotCarryARunAcrossALongPass884 is the regression test
+// of the review's m1 (PR #1123, round 1): a retried wait, then one loop pass that
+// lasts longer than the ceiling (a park in SUSPENDED, or a slow callback in the
+// gate pass: the pass moves tickCounter by one and takes over a second), then one
+// more retried wait. The two retries are one pass apart, but they are not
+// consecutive: the run's allowed spacing is a few passes' time, and a gap longer
+// than that ends it. Before, the worker stopped on the second retry of its life.
+func TestRingWaitContinuesDoesNotCarryARunAcrossALongPass884(t *testing.T) {
+	ring := newTestRing(t)
+	w := &Worker{ring: ring, id: 3, logger: slog.New(&warnLog884{})}
+	if !w.ringWaitContinues(waitModeTimed, ceilingErr884()) {
+		t.Fatal("the first retry gave up")
+	}
+	w.tickCounter++ // the pass that parks
+	time.Sleep(1100 * time.Millisecond)
+	if !w.ringWaitContinues(waitModeTimed, ceilingErr884()) {
+		t.Fatal("the worker gave up on its second retried wait, after one long pass between two isolated retries: a run is carried across a gap")
+	}
+}