engines: OnDetach rollback first, io_uring pause wake, zoned bind address, one-dump NIC table, MaxEvents by allowed nodes, upgrade-carry count, backlog hook (v1.6.0, Refs #885 #886 #889 #890)

.github/workflows/ci.yml

@@ -1861,10 +1861,12 @@ jobs:
       # name (its allow-list) and this step runs them with memlock raised and
       # CELERIS_REQUIRE_IOURING_WORKERS=1, which turns the skip into a failure.
       # The pause/resume pair of the `unit` job (one funded worker there) runs
-      # here with the two it asks for, and so does celeris#886's resume test
+      # here with the two it asks for, and so do celeris#886's resume test
       # (TestResumeAcceptReachesAWorkerThatCannotPark886: at 8 MiB it has one
       # worker and cannot tell a resume that signals every worker from one that
-      # signals the first). Exactly seven top-level PASS lines and no SKIP line: a rename or a
+      # signals the first) and its pause test
+      # (TestBeginPauseAcceptReachesAnIdleWorker886, the same for the pause).
+      # Exactly eight top-level PASS lines and no SKIP line: a rename or a
       # quarantine selects nothing silently, and a skip is not coverage.
       - name: celeris#895/#896 multi-worker listener tests (memlock raised, skipping forbidden)
         if: ${{ !cancelled() }}
@@ -1875,7 +1877,7 @@ jobs:
           set -o pipefail
           sudo prlimit --pid "$$" --memlock=unlimited:unlimited
           echo "memlock (KiB): $(ulimit -l)"
-          names='TestPort0EveryWorkerBindsTheSamePort895|TestListenReleasesThePortWithManyWorkers896|TestFailedStartReleasesThePort896|TestPort0PauseResumeWithManyWorkers895|TestPort0ResumeKeepsThePort895|TestPort0PausedPortCannotBeTaken895|TestResumeAcceptReachesAWorkerThatCannotPark886'
+          names='TestPort0EveryWorkerBindsTheSamePort895|TestListenReleasesThePortWithManyWorkers896|TestFailedStartReleasesThePort896|TestPort0PauseResumeWithManyWorkers895|TestPort0ResumeKeepsThePort895|TestPort0PausedPortCannotBeTaken895|TestResumeAcceptReachesAWorkerThatCannotPark886|TestBeginPauseAcceptReachesAnIdleWorker886'
           go test -race -count=1 -timeout=600s -v -run "^(${names})\$" \
             ./internal/engine/iouring/ 2>&1 | tee /tmp/iouring895.log
           want=$(( $(tr '|' '\n' <<<"$names" | wc -l) ))

internal/conn/backlog_seam_890.go

@@ -0,0 +1,29 @@
+package conn
+
+import "sync/atomic"
+
+// BacklogRefusedHook is a test seam (celeris#890 item 22). It is nil in
+// production. When a request finds the connection's unsent responses over the
+// engine's back-pressure limit (H1State.WriteBacklogged) and is therefore not
+// served (ErrWriteBacklog), a test that installed it is told which connection
+// by its peer address (H1State.RemoteAddr). That is the server-side event a
+// test of the limit has to wait for, instead of a fixed sleep after the
+// kernel's send queue has stopped growing: once it has fired the engine has
+// decided to close the connection when what is queued has gone out, whatever
+// the client does next.
+//
+// It is read only on the refusal, never on the path of a request that is
+// served, so it costs a request nothing.
+var BacklogRefusedHook atomic.Pointer[func(remoteAddr string)]
+
+// backlogRefused tells the test seam that a request was refused. Out of line
+// and marked noinline so that the branch it is called from, taken only by a
+// peer that stopped reading, adds nothing to handleH1Request's frame or to the
+// code of a request that is served.
+//
+//go:noinline
+func backlogRefused(state *H1State) {
+	if f := BacklogRefusedHook.Load(); f != nil {
+		(*f)(state.RemoteAddr)
+	}
+}

internal/conn/h1.go

@@ -1438,6 +1438,7 @@ func handleH1Request(ctx context.Context, state *H1State, body []byte,
 	handler stream.Handler, write func([]byte)) error {
 
 	if state.WriteBacklogged != nil && state.WriteBacklogged() {
+		backlogRefused(state)
 		return ErrWriteBacklog
 	}
 

internal/conn/h1_backlog_bench_890_test.go

@@ -0,0 +1,57 @@
+package conn
+
+import (
+	"context"
+	"testing"
+)
+
+// BenchmarkProcessH1Backlogged890 is the per-request path of ProcessH1 with the
+// engines' back-pressure query installed (H1State.WriteBacklogged, answering
+// false: the request is served), the path that celeris#890 item 22's hook sits
+// beside. The hook is called only when the query answers true, so this is the
+// figure a change to that branch must leave alone.
+func BenchmarkProcessH1Backlogged890(b *testing.B) {
+	st := NewH1State()
+	st.WriteBacklogged = func() bool { return false }
+	w := func([]byte) {}
+	ctx := context.Background()
+	req := []byte("GET /x HTTP/1.1\r\nHost: a\r\n\r\n")
+	b.SetBytes(int64(len(req)))
+	b.ReportAllocs()
+	for b.Loop() {
+		if err := ProcessH1(ctx, req, st, nopHandler1029{}, w); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+// TestBacklogRefusedHookFiresOnlyOnRefusal890: the hook is told the peer address
+// when a request finds the connection backlogged, and is not called for a
+// request that is served.
+func TestBacklogRefusedHookFiresOnlyOnRefusal890(t *testing.T) {
+	var got []string
+	f := func(remote string) { got = append(got, remote) }
+	BacklogRefusedHook.Store(&f)
+	t.Cleanup(func() { BacklogRefusedHook.Store(nil) })
+
+	st := NewH1State()
+	st.RemoteAddr = "192.0.2.7:4242"
+	backlogged := false
+	st.WriteBacklogged = func() bool { return backlogged }
+	w := func([]byte) {}
+	req := []byte("GET /x HTTP/1.1\r\nHost: a\r\n\r\n")
+
+	if err := ProcessH1(context.Background(), req, st, nopHandler1029{}, w); err != nil {
+		t.Fatalf("a request that is served: %v", err)
+	}
+	if len(got) != 0 {
+		t.Fatalf("the hook fired %v for a request that was served", got)
+	}
+	backlogged = true
+	if err := ProcessH1(context.Background(), req, st, nopHandler1029{}, w); err != ErrWriteBacklog {
+		t.Fatalf("a backlogged request: %v, want ErrWriteBacklog", err)
+	}
+	if len(got) != 1 || got[0] != "192.0.2.7:4242" {
+		t.Fatalf("the hook fired %v, want exactly the peer address once", got)
+	}
+}

internal/engine/epoll/abort_seam_844.go

@@ -26,6 +26,13 @@ var (
 	// recovers the panic, as the router does, then answers in).
 	detachPreFlagHook atomic.Pointer[func()]
 
+	// detachPostCountHook runs in OnDetach right after the detach count is made
+	// (sync mode) or asyncDetachPending is set (async mode), before the guarded
+	// writeFn is built: the span between the snapshot the rollback restores and
+	// the place the rollback used to be registered, which a fault there left
+	// uncovered (#889 item 48).
+	detachPostCountHook atomic.Pointer[func()]
+
 	// detachPostInstallHook runs in OnDetach of a conn that is not on the
 	// promoted async path, once everything the hook installs is in place (the
 	// guarded writeFn, the adapter's, RawWriteFn, pause/resume): the point a

internal/engine/epoll/adopt.go

@@ -248,12 +248,13 @@ func (l *Loop) attachFD(ctx context.Context, fd int, carry engine.Carryover, now
 		}
 		perr := conn.ProcessH1(cs.ctx, carry.Buffered, cs.h1State, l.handler, cs.writeFn)
 		// The replay is the recv event of this conn's first bytes and counts
-		// one, whatever the verdict (celeris#886). drainRead counts a recv
-		// after ProcessH1 too, but returns before it on an h2c upgrade and
-		// never counts a direct-body recv: a carry that is an upgrade request
-		// counts one here and its recv would have counted none. Not when this
-		// loop read and counted the bytes itself (a reclaim: recvCounted).
-		if !recvCounted {
+		// one, whatever the verdict but one (celeris#886). drainRead counts a
+		// recv after ProcessH1 too, but returns before it on an h2c upgrade
+		// and never counts a direct-body recv: a carry that is an upgrade
+		// request counts none here either, as its recv would have counted none
+		// (item 23: both paths now count the upgrade request as none). Not when
+		// this loop read and counted the bytes itself (a reclaim: recvCounted).
+		if !recvCounted && !errors.Is(perr, conn.ErrUpgradeH2C) {
 			l.reqBatch++
 		}
 		if perr != nil {
@@ -276,13 +277,13 @@ func (l *Loop) attachFD(ctx context.Context, fd int, carry engine.Carryover, now
 func (l *Loop) replayCarriedAsync(cs *connState, fd int, data []byte, recvCounted bool) {
 	cs.h1State.InlineMode = true
 	perr := conn.ProcessH1(cs.ctx, data, cs.h1State, l.handler, cs.writeFn)
-	// One request counted for the replay, whichever way it ends below: it is
-	// the recv event of this conn's first bytes (the promotion included, as
-	// in drainRead, celeris#626; celeris#886). Unlike drainRead it also counts
-	// a carry that ends in an h2c upgrade, which that path returns before
-	// counting. Not when this loop counted the recv already (a reclaim,
-	// recvCounted).
-	if !recvCounted {
+	// One request counted for the replay, whichever way it ends below but an
+	// h2c upgrade: it is the recv event of this conn's first bytes (the
+	// promotion included, as in drainRead, celeris#626; celeris#886). Like
+	// drainRead, which returns on the upgrade verdict before it counts, it
+	// counts a carry that ends in an h2c upgrade as none (item 23). Not when
+	// this loop counted the recv already (a reclaim, recvCounted).
+	if !recvCounted && !errors.Is(perr, conn.ErrUpgradeH2C) {
 		l.reqBatch++
 	}
 	// A handler that hijacked inline has had cs released inside the Hijack

internal/engine/epoll/adopt_request_count_886_linux_test.go

@@ -4,6 +4,7 @@ package epoll
 
 import (
 	"bufio"
+	"strings"
 	"testing"
 	"time"
 
@@ -18,8 +19,9 @@ import (
 // replayed in place of a recv, and the replay is that recv event. The rule
 // these tests pin, for every replay path (sync, async served inline, async
 // promoted, a replay that ends on a verdict): a replay of a non-empty carry
-// counts one, and an adoption with an empty carry counts none, as no recv
-// has happened. Each test adopts one conn onto a fresh engine, so the count
+// counts one, except one that is an h2c upgrade request (none, as on the
+// recv path, item 23), and an adoption with an empty carry counts none, as no
+// recv has happened. Each test adopts one conn onto a fresh engine, so the count
 // after it is the count of that adoption.
 
 // expectCount886 waits until RequestCount reaches want and then holds for
@@ -57,9 +59,21 @@ func adoptCount886(t *testing.T, e *Engine, carried string) (clientReader *bufio
 		_ = unix.Close(fd)
 		t.Fatalf("AdoptConn: %v", err)
 	}
+	_ = client.SetReadDeadline(time.Now().Add(10 * time.Second))
 	return bufio.NewReader(client)
 }
 
+// expectUpgradeAnswered886 reads the 101 of an h2c upgrade off the client: the
+// replay has then run to its verdict, so a count it was going to make has been
+// made. Without it expectCount886 can return, with 0, before the replay ran.
+func expectUpgradeAnswered886(t *testing.T, br *bufio.Reader) {
+	t.Helper()
+	status, err := br.ReadString('\n')
+	if err != nil || !strings.Contains(status, " 101 ") {
+		t.Fatalf("the upgrade request was answered %q (err %v), want a 101", status, err)
+	}
+}
+
 func TestEpollAdoptRequestCountSync886(t *testing.T) {
 	e, _ := start908(t, false, false)
 	adoptCount886(t, e, carryGood908)
@@ -103,20 +117,22 @@ func TestEpollAdoptRequestCountEmptyCarry886(t *testing.T) {
 	expectCount886(t, e, 0)
 }
 
-// A carry that is an h2c upgrade request counts one like any other non-empty
-// carry. The recv path does not: it returns on the upgrade verdict before it
-// counts, so a conn that upgraded on a recv has counted none (celeris#886
-// item 5, open question in the PR that added these). The tests pin the
-// replay's answer so that it cannot drift unnoticed, whichever way the
-// question is settled.
+// A carry that is an h2c upgrade request counts none, as the recv path counts
+// none for the request that upgrades the conn (it returns on the upgrade
+// verdict before it counts) and as an HTTP/2 conn counts no request on
+// either path (celeris#886 item 23: the replay counted one, the recv path
+// none, and the rule picked is the recv path's, so that BytesPerReq does not
+// move by one request per upgraded conn depending on how the conn arrived).
 func TestEpollAdoptRequestCountUpgradeCarry886(t *testing.T) {
 	e, _ := start908(t, false, true) // EnableH2Upgrade: the carry is a real upgrade
-	adoptCount886(t, e, h2cUpgradeHead908())
-	expectCount886(t, e, 1)
+	br := adoptCount886(t, e, h2cUpgradeHead908())
+	expectUpgradeAnswered886(t, br)
+	expectCount886(t, e, 0)
 }
 
 func TestEpollAdoptRequestCountUpgradeCarryAsync886(t *testing.T) {
 	e, _ := start908(t, true, true) // EnableH2Upgrade: the carry is a real upgrade
-	adoptCount886(t, e, h2cUpgradeHead908())
-	expectCount886(t, e, 1)
+	br := adoptCount886(t, e, h2cUpgradeHead908())
+	expectUpgradeAnswered886(t, br)
+	expectCount886(t, e, 0)
 }

internal/engine/epoll/detach_rollback_889_linux_test.go

@@ -56,17 +56,25 @@ func (r *rig889) out() string {
 	return strings.Join(r.wrote, "")
 }
 
-// Two points the fault is injected at, both after cs.writeFn = guarded: early
-// (the state the hook publishes through is gone, so UpdateWriteFn faults) and
-// late (everything is installed).
+// Three points the fault is injected at. count: right after the count is made
+// and before anything is installed (detachPostCountHook; the span between the
+// snapshot and the place the rollback used to be registered, #889 item 48).
+// early and late are both after cs.writeFn = guarded: early (the state the
+// hook publishes through is gone, so UpdateWriteFn faults) and late
+// (everything is installed).
 const (
+	faultCount889 = "count"
 	faultEarly889 = "early"
 	faultLate889  = "late"
 )
 
 func (r *rig889) panicking(t *testing.T, where string, hook func()) {
 	t.Helper()
 	switch where {
+	case faultCount889:
+		f := func() { panic("engine fault right after the count") }
+		detachPostCountHook.Store(&f)
+		defer detachPostCountHook.Store(nil)
 	case faultEarly889:
 		r.cs.h1State = nil
 		defer func() { r.cs.h1State = r.h1 }()
@@ -106,7 +114,7 @@ func (r *rig889) lockedWrite(fn func([]byte), s string) {
 
 func TestOnDetachPanicLeavesTheConnAsIfNeverRan889(t *testing.T) {
 	for _, async := range []bool{false, true} {
-		for _, where := range []string{faultEarly889, faultLate889} {
+		for _, where := range []string{faultCount889, faultEarly889, faultLate889} {
 			name := where + "/sync"
 			if async {
 				name = where + "/async-unpromoted"

internal/engine/epoll/engine.go

@@ -150,7 +150,7 @@ func (e *Engine) Listen(ctx context.Context) error {
 
 	resolved := e.cfg.Resources.Resolve()
 
-	topo := platform.DetectNUMA()
+	topo := detectNUMA()
 	// celeris#909: the CPU of loop i is the i-th member of the process's
 	// allowed set, and on a big.LITTLE host never a little CPU. The plan sets
 	// the loop count (celeris#1036): on a big.LITTLE host it holds one loop
@@ -159,12 +159,7 @@ func (e *Engine) Listen(ctx context.Context) error {
 	cpus := plan.CPUs
 	resolved.Workers = len(cpus)
 
-	if topo.NumNodes > 1 {
-		resolved.MaxEvents = resolved.MaxEvents / topo.NumNodes
-		if resolved.MaxEvents < 64 {
-			resolved.MaxEvents = 64
-		}
-	}
+	resolved.MaxEvents = splitMaxEvents(resolved.MaxEvents, topo, plan.Allowed)
 
 	e.mu.Lock()
 	e.loops = make([]*Loop, resolved.Workers)
@@ -342,7 +337,10 @@ func (e *Engine) Type() engine.EngineType {
 
 // BeginPauseAccept starts pausing accept and returns without waiting
 // (celeris#662). Each loop sees the flag at the top of its next iteration,
-// a few milliseconds later at most while it listens. Then, on its own
+// a few milliseconds later at most while it listens (adaptiveTimeoutMs caps
+// the wait at 4 ms; measured 0.8 to 5.9 ms from the call to the closed
+// listener, celeris#886 item 24), which is why it does not signal the wake
+// eventfd, as io_uring's does for its 100 ms ring wait. Then, on its own
 // thread, it clears TCP_DEFER_ACCEPT on its listener, keeps accepting and
 // serving for the linger (about 1.5 s from that clear), and closes the
 // listener once it has drained the accept queue. A connection whose

internal/engine/epoll/h1verdict.go

@@ -67,7 +67,22 @@ func (l *Loop) finishH1Upgrade(cs *connState, fd int, writeFn func([]byte)) bool
 // out (closeWhenFlushed, celeris#761). The lock is released before the close,
 // which takes it again.
 //
-// internal/engine/iouring's endH1Verdict does the same.
+// internal/engine/iouring's endH1Verdict does the same, except for OnError on a
+// conn that has no detachMu: this one calls it whenever one is registered,
+// io_uring's only inside its `if mu := cs.detachMu; mu != nil`, so a sync conn
+// with a callback and no detach gets it here and loses it there (celeris#886
+// item 9). The recv path and the transplant replay both end here, so the rule
+// is the same for both on each engine. A sync conn holds a callback without a
+// detachMu when a handler registers one (Context.SetWSErrorHandler) and does
+// not Detach, and when a Detach faulted and was rolled back (the rollback
+// leaves the callback). The middleware callbacks (websocket, sse) are registered
+// just before the Detach, so they are in that state only after a faulted
+// Detach, and then only end their own reader or stream, which has not started.
+// Pinned on this side by TestEpollRecvPathOnErrorSync908 and
+// TestEpollAdoptCarriedOnErrorSync908, and on io_uring's by
+// TestIouringRecvPathOnErrorSyncNoDetachMuIsDropped886 and
+// TestIouringAdoptCarriedOnErrorSyncNoDetachMuIsDropped886. Which rule is right
+// is an open decision.
 func (l *Loop) endH1Verdict(cs *connState, err error) {
 	if mu := cs.detachMu; mu != nil {
 		mu.Lock()