diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ba2e16b03..0b13f30a5 100644 --- a/.github/workflows/ci.yml +++ b/.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) )) diff --git a/internal/conn/backlog_seam_890.go b/internal/conn/backlog_seam_890.go new file mode 100644 index 000000000..a81b1a533 --- /dev/null +++ b/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) + } +} diff --git a/internal/conn/h1.go b/internal/conn/h1.go index 7392dfdfe..96ecdd031 100644 --- a/internal/conn/h1.go +++ b/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 } diff --git a/internal/conn/h1_backlog_bench_890_test.go b/internal/conn/h1_backlog_bench_890_test.go new file mode 100644 index 000000000..aac9dca48 --- /dev/null +++ b/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) + } +} diff --git a/internal/engine/epoll/abort_seam_844.go b/internal/engine/epoll/abort_seam_844.go index e7dcaf7fb..cc4e7e4b5 100644 --- a/internal/engine/epoll/abort_seam_844.go +++ b/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 diff --git a/internal/engine/epoll/adopt.go b/internal/engine/epoll/adopt.go index d051d24e9..136e07053 100644 --- a/internal/engine/epoll/adopt.go +++ b/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 diff --git a/internal/engine/epoll/adopt_request_count_886_linux_test.go b/internal/engine/epoll/adopt_request_count_886_linux_test.go index b11b19300..571d77146 100644 --- a/internal/engine/epoll/adopt_request_count_886_linux_test.go +++ b/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) } diff --git a/internal/engine/epoll/detach_rollback_889_linux_test.go b/internal/engine/epoll/detach_rollback_889_linux_test.go index 7020fc09c..eb21e80b3 100644 --- a/internal/engine/epoll/detach_rollback_889_linux_test.go +++ b/internal/engine/epoll/detach_rollback_889_linux_test.go @@ -56,10 +56,14 @@ 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" ) @@ -67,6 +71,10 @@ const ( 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" diff --git a/internal/engine/epoll/engine.go b/internal/engine/epoll/engine.go index 1257cd16b..4ff50e0ee 100644 --- a/internal/engine/epoll/engine.go +++ b/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 diff --git a/internal/engine/epoll/h1verdict.go b/internal/engine/epoll/h1verdict.go index d848e192d..b67cd7eba 100644 --- a/internal/engine/epoll/h1verdict.go +++ b/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() diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go index 6bc92dfab..cd980e13d 100644 --- a/internal/engine/epoll/loop.go +++ b/internal/engine/epoll/loop.go @@ -9,9 +9,11 @@ import ( "io" "log/slog" "net" + "net/netip" "os" "runtime" "runtime/debug" + "strings" "sync" "sync/atomic" "time" @@ -2848,6 +2850,49 @@ func (l *Loop) initProtocol(cs *connState) { prevRaw, prevPause, prevResume = h1.RawWriteFn, h1.PauseRecv, h1.ResumeRecv } counted, finished := false, false + // unlockDetachMu: this goroutine holds detachMu (the promoted async path) and + // releases it on the way out; a conn that is not on that path is rolled back + // instead. Decided here, from the snapshot, so that the rollback below is + // registered before the first mutation and an edit that adds one between + // the snapshot and the install is inside it (#889 item 48). It is the gate + // the hook used after it had allocated detachMu: in sync mode l.async is false + // either way, and in async mode detachMu was allocated at acquireConnState, + // so prevMu is not nil. + unlockDetachMu := l.async && cs.asyncPromoted && prevMu != nil && !cs.asyncDetachUnlocked + if !unlockDetachMu { + // A conn that is not on the promoted async path goes on after a panic + // out of this hook: Context.Detach has not marked the Context detached, + // the router answers 500, and the next request is served on the same + // conn. Put back everything the hook changes from here on, so that it is + // exactly as if Detach had not run: a guarded writeFn left behind + // locks detachMu again on the first write of the H2C branch and of + // serveAsync, which hold it across theirs (the loop thread, or the + // dispatch goroutine, deadlocks on itself), a count left behind is + // made again by the next Detach, and closeConn undoes one. The + // promoted path closes the conn instead (the deferred hook below). + // Assignments only, under no lock: this goroutine owns the conn's + // state here, as it did when the hook installed it (#889 item 34). + defer func() { + if finished { + return + } + cs.writeFn = prevWrite + if h1 != nil { + h1.UpdateWriteFn(prevWrite) + h1.SetWriteErrQuery(cs) + h1.RawWriteFn, h1.PauseRecv, h1.ResumeRecv = prevRaw, prevPause, prevResume + } + cs.detachMu = prevMu + cs.asyncDetachPending = prevPending + if counted { + // Sync mode only. The next Detach counts again, and closeConn undoes the count + // of a conn with detachCounted set, so the sync path relies on this leaving it + // false (#889 item 48). + l.detachedCount-- + cs.detachCounted = false + } + }() + } var mu *sync.Mutex if cs.detachMu != nil { mu = cs.detachMu @@ -2868,6 +2913,9 @@ func (l *Loop) initProtocol(cs *connState) { } else { cs.asyncDetachPending = true } + if f := detachPostCountHook.Load(); f != nil { + (*f)() // test seam, #889 item 48 (between the snapshot and the rollback) + } orig := cs.writeFn guarded := func(data []byte) { mu.Lock() @@ -2977,43 +3025,13 @@ func (l *Loop) initProtocol(cs *connState) { // Context/stream this WS/SSE middleware goroutine is still using — // the loop.go:2465-vs-websocket.go:243 use-after-recycle that // nil-derefs WSRawWriteFn under a peer RST mid-upgrade. - unlockDetachMu := l.async && cs.asyncPromoted && cs.detachMu != nil && !cs.asyncDetachUnlocked + // unlockDetachMu was decided at the top of the hook, before the first mutation, so that the rollback is + // registered before any of them. // completed: this closure ran to its end. Nothing is owed when this // goroutine holds no lock here. completed := !unlockDetachMu // flagged: asyncDetachUnlocked has been published by this closure. flagged := false - if !unlockDetachMu { - // A conn that is not on the promoted async path goes on after a panic - // out of this hook: Context.Detach has not marked the Context detached, - // the router answers 500, and the next request is served on the same - // conn. Put back everything installed above and below, so that it is - // exactly as if Detach had not run: a guarded writeFn left behind - // locks detachMu again on the first write of the H2C branch and of - // serveAsync, which hold it across theirs (the loop thread, or the - // dispatch goroutine, deadlocks on itself), a count left behind is - // made again by the next Detach, and closeConn undoes one. The - // promoted path closes the conn instead (the deferred hook below). - // Assignments only, under no lock: this goroutine owns the conn's - // state here, as it did when the hook installed it (#889 item 34). - defer func() { - if finished { - return - } - cs.writeFn = prevWrite - if h1 != nil { - h1.UpdateWriteFn(prevWrite) - h1.SetWriteErrQuery(cs) - h1.RawWriteFn, h1.PauseRecv, h1.ResumeRecv = prevRaw, prevPause, prevResume - } - cs.detachMu = prevMu - cs.asyncDetachPending = prevPending - if counted { - l.detachedCount-- - cs.detachCounted = false - } - }() - } if unlockDetachMu { // Registered before guarded is installed (celeris#844): from the moment // cs.writeFn is guarded, a write through the response adapter locks @@ -5332,7 +5350,22 @@ func (l *Loop) flushBatchedCounters() { // as net/http's graceful shutdown does, and a stream it opens anyway is // refused. The wait is bounded like the send drain (sendDrainWait): the // budget of the last Engine.Shutdown, and never less than -// shutdownSendDrainFloor. Loop thread. +// shutdownSendDrainFloor. +// +// A GOAWAY the write hook refuses (the conn's backlog is over the HTTP/2 cap, +// 64 MiB, so its peer has stopped reading) is lost and sets writeRefused; this +// pass does not look at it (celeris#885 item 14, read from the code, not +// measured). The conn is not left open for it: the next drainH2Writes with a +// frame queued discards the queue and closes the conn once what is staged has +// gone out, a read of the conn that finds writeRefused closes it (drainRead), +// and a conn with none of those is not busy here, so the pass settles and +// shutdown closes it. What the lost GOAWAY costs is the client's graceful +// notice, from a peer that was not reading; what a conn that stays busy +// (OutboundPending, response DATA waiting for a WINDOW_UPDATE that will not +// come) costs is the wait up to the drain's bound, the same as for any conn +// whose peer is not taking its response. +// +// Loop thread. func (l *Loop) h2PoolSettled() bool { if len(l.h2Conns) == 0 { return true @@ -5602,6 +5635,21 @@ func parseAddr(addr string) (unix.Sockaddr, error) { // own rules: a byte over 255 ("300.1.1.1" bound 44.1.1.1), a leading zero // ("0127.0.0.1": octal in some parsers), a fifth byte, and the shorthands // inet_aton reads ("10", "127.1": "1.2.3" bound 1.2.3.0). + // + // A zoned IPv6 host ("fe80::1%eth0") is what net.Listen binds and net.ParseIP + // rejects: it is read with netip, which keeps the zone, and the zone becomes + // the socket address's scope id as net.Listen makes it (bindiag.ZoneID). An + // IPv4-mapped host is IPv4 whatever its zone, as for std (celeris#885 + // item 54). + var zone string + if strings.IndexByte(host, '%') >= 0 { + za, err := netip.ParseAddr(host) + if err != nil || !za.Is6() { + return nil, fmt.Errorf("invalid addr: %s", addr) + } + zone = za.Zone() + host = za.WithZone("").String() + } ip := net.ParseIP(host) if ip == nil { return nil, fmt.Errorf("invalid addr: %s", addr) @@ -5611,7 +5659,7 @@ func parseAddr(addr string) (unix.Sockaddr, error) { copy(sa.Addr[:], ip4) return sa, nil } - sa := &unix.SockaddrInet6{Port: port} + sa := &unix.SockaddrInet6{Port: port, ZoneId: bindiag.ZoneID(zone)} copy(sa.Addr[:], ip.To16()) return sa, nil } diff --git a/internal/engine/epoll/maxevents_nodes_885_linux_test.go b/internal/engine/epoll/maxevents_nodes_885_linux_test.go new file mode 100644 index 000000000..4cbfae1cb --- /dev/null +++ b/internal/engine/epoll/maxevents_nodes_885_linux_test.go @@ -0,0 +1,90 @@ +//go:build linux + +package epoll + +import ( + "testing" + + "github.com/goceleris/celeris/internal/platform" +) + +// celeris#885 item 31. Listen divided MaxEvents by the host's NUMA node count, +// not by the nodes the process may use: a process kept to one node (numactl +// --cpunodebind, a single-numa-node pod, a cpuset of one socket) has every loop +// on that node and still got MaxEvents / NumNodes. The split follows the nodes +// the plan's allowed CPU set meets. + +// nodes885 is a host of n nodes with one CPU each (node i holds CPU i). +func nodes885(n int) platform.NUMATopology { + t := platform.NUMATopology{NumNodes: n} + for i := range n { + t.NodeCPUs = append(t.NodeCPUs, []int{i}) + } + return t +} + +// seq885 is the CPUs 0..n-1. +func seq885(n int) []int { + s := make([]int, n) + for i := range s { + s[i] = i + } + return s +} + +func TestSplitMaxEventsFollowsTheNodesTheAllowedSetMeets885(t *testing.T) { + for _, tc := range []struct { + name string + topo platform.NUMATopology + allowed []int + want int + }{ + {"one-node-host", platform.NUMATopology{NumNodes: 1}, []int{0, 1}, 8192}, + {"restricted-to-one-of-eight", nodes885(8), []int{3}, 8192}, + {"restricted-to-two-of-eight", nodes885(8), []int{2, 5, 5}, 4096}, + {"all-eight", nodes885(8), seq885(8), 1024}, + {"unrestricted-no-plan-set", nodes885(8), nil, 1024}, // no allowed set to read: the host's count, as before + {"topology-without-cpu-lists", platform.NUMATopology{NumNodes: 4}, []int{0}, 2048}, + {"allowed-meets-no-listed-node", nodes885(4), []int{40}, 2048}, // unknown CPUs: the host's count + {"floor", nodes885(256), seq885(256), 64}, + } { + if got := splitMaxEvents(8192, tc.topo, tc.allowed); got != tc.want { + t.Errorf("%s: splitMaxEvents(8192, %d nodes, allowed %v) = %d, want %d", tc.name, tc.topo.NumNodes, tc.allowed, got, tc.want) + } + } +} + +// TestListenSplitsMaxEventsByTheNodesOfThePlan885 goes through Listen: a loop's +// event array is min(MaxEvents, 2048), so the host has to be a big one (8 +// nodes: 8192/8 = 1024) for the split to show in len(events). +func TestListenSplitsMaxEventsByTheNodesOfThePlan885(t *testing.T) { + oldN, oldP := detectNUMA, planWorkerCPUs + t.Cleanup(func() { detectNUMA, planWorkerCPUs = oldN, oldP }) + detectNUMA = func() platform.NUMATopology { return nodes885(8) } + for _, tc := range []struct { + name string + allowed []int + want int + }{ + {"process-kept-to-one-node", []int{0}, 2048}, + {"process-spans-all-nodes", seq885(8), 1024}, + } { + t.Run(tc.name, func(t *testing.T) { + planWorkerCPUs = func(n int, _ bool, _ string) platform.CPUPlan { + cpus := make([]int, n) + return platform.CPUPlan{CPUs: cpus, Requested: n, Allowed: tc.allowed} + } + r := start895(t, "127.0.0.1:0", 2, false) + r.e.mu.Lock() + defer r.e.mu.Unlock() + if len(r.e.loops) == 0 { + t.Fatal("no loops") + } + for i, l := range r.e.loops { + if got := len(l.events); got != tc.want { + t.Errorf("loop %d: len(events) = %d, want %d", i, got, tc.want) + } + } + }) + } +} diff --git a/internal/engine/epoll/parse_addr_mapped_engcfg_test.go b/internal/engine/epoll/parse_addr_mapped_engcfg_test.go index 52abd3430..29cb37b9d 100644 --- a/internal/engine/epoll/parse_addr_mapped_engcfg_test.go +++ b/internal/engine/epoll/parse_addr_mapped_engcfg_test.go @@ -47,9 +47,8 @@ func TestParseAddrReadsAnAddressAsNetParseIPDoes(t *testing.T) { {"0127.0.0.1:80", nil}, {"300.1.1.1:80", nil}, {"1.2.3.4.5:80", nil}, - // KNOWN GAP, not a goal: std binds a zoned address (net.ParseIP, which - // this reads, rejects the zone); tracked as checklist #885 item 54. - {"[fe80::1%eth0]:80", nil}, + // A zoned address is read by TestParseAddrReadsAZonedIPv6Host885 + // (checklist #885 item 54). {"localhost:80", nil}, } { got, err := parseAddr(tc.addr) diff --git a/internal/engine/epoll/parse_addr_zone_885_linux_test.go b/internal/engine/epoll/parse_addr_zone_885_linux_test.go new file mode 100644 index 000000000..ed09c31b6 --- /dev/null +++ b/internal/engine/epoll/parse_addr_zone_885_linux_test.go @@ -0,0 +1,159 @@ +//go:build linux + +package epoll + +import ( + "net" + "os" + "strconv" + "strings" + "testing" + + "golang.org/x/sys/unix" +) + +// celeris#885 item 54. parseAddr read the host with net.ParseIP, which rejects +// a zone, so a zoned IPv6 bind address ("[fe80::1234%dz0]:0"), which net.Listen +// and so the std engine bind, made both native engines fail to start on the +// same Config.Addr. The host's zone is now read as net.Listen reads it: the +// index of the interface of that name, else the zone as a number, else 0 (the +// bind then fails for a link-local address, as std's does). An IPv4-mapped host +// is bound as IPv4 and its zone ignored, as std does. + +// zoneID885 is std's rule for a zone (net's ipv6ZoneCache.index, go1.27): the +// index of the interface of that name, else the decimal digits the name starts +// with (dtoi: "12abc" is 12, a number of 0xFFFFFF or more saturates at +// 0xFFFFFF), else 0. +func zoneID885(zone string) uint32 { + if zone == "" { + return 0 + } + if ifi, err := net.InterfaceByName(zone); err == nil { + return uint32(ifi.Index) + } + n := 0 + for i := 0; i < len(zone) && '0' <= zone[i] && zone[i] <= '9'; i++ { + n = n*10 + int(zone[i]-'0') + if n >= 0xFFFFFF { + return 0xFFFFFF + } + } + return uint32(n) +} + +func TestParseAddrReadsAZonedIPv6Host885(t *testing.T) { + lo, err := net.InterfaceByName("lo") + if err != nil { + t.Skipf("no lo interface: %v", err) + } + v6 := func(port int, ip string, zone uint32) unix.Sockaddr { + sa := &unix.SockaddrInet6{Port: port, ZoneId: zone} + copy(sa.Addr[:], net.ParseIP(ip).To16()) + return sa + } + for _, tc := range []struct { + addr string + want unix.Sockaddr // nil: an error + }{ + {"[fe80::1%lo]:80", v6(80, "fe80::1", uint32(lo.Index))}, + {"[fe80::1234%lo]:0", v6(0, "fe80::1234", uint32(lo.Index))}, + {"[fe80::1%" + strconv.Itoa(lo.Index) + "]:8080", v6(8080, "fe80::1", uint32(lo.Index))}, + {"[fe80::1%7]:80", v6(80, "fe80::1", 7)}, + // An interface that does not exist and is not a number is scope 0, as + // std reads it: the kernel then refuses the bind of a link-local + // address, and a global one binds. + {"[fe80::1%nosuchif0]:80", v6(80, "fe80::1", 0)}, + // A malformed zone is read as std reads it: the digits it starts with. + {"[fe80::1%12abc]:80", v6(80, "fe80::1", 12)}, + {"[fe80::1%99999999999]:80", v6(80, "fe80::1", 0xFFFFFF)}, + {"[fe80::1%abc12]:80", v6(80, "fe80::1", 0)}, + {"[2001:db8::1%lo]:443", v6(443, "2001:db8::1", uint32(lo.Index))}, + {"[::%lo]:80", v6(80, "::", uint32(lo.Index))}, + // Everything after the first % is the zone, so this one names no interface. + {"[fe80::1%lo%lo]:80", v6(80, "fe80::1", 0)}, + // A mapped IPv4 address is IPv4 for std, and a zone means nothing to it. + {"[::ffff:1.2.3.4%lo]:80", &unix.SockaddrInet4{Port: 80, Addr: [4]byte{1, 2, 3, 4}}}, + // Not an address (std refuses them too): no zone on IPv4, an empty zone, a + // zone and no host. + {"1.2.3.4%lo:80", nil}, + {"[1.2.3.4%lo]:80", nil}, + {"[fe80::1%]:80", nil}, + {"[%lo]:80", nil}, + } { + got, err := parseAddr(tc.addr) + if tc.want == nil { + if err == nil { + t.Errorf("parseAddr(%q) = %v, want an error", tc.addr, got) + } + continue + } + if err != nil { + t.Errorf("parseAddr(%q): %v, want %v", tc.addr, err, tc.want) + continue + } + if !sockaddrEqual(got, tc.want) { + t.Errorf("parseAddr(%q) = %v, want %v", tc.addr, got, tc.want) + } + // std's reading of the same literal: the same IP, port and zone. + std, serr := net.ResolveTCPAddr("tcp", tc.addr) + if serr != nil { + t.Errorf("std refuses %q, which the table says is an address: %v", tc.addr, serr) + continue + } + if sa6, ok := got.(*unix.SockaddrInet6); ok { + if !net.IP(sa6.Addr[:]).Equal(std.IP) || sa6.Port != std.Port || sa6.ZoneId != zoneID885(std.Zone) { + t.Errorf("parseAddr(%q) = %v:%d scope %d, std reads %v:%d zone %q (scope %d)", + tc.addr, net.IP(sa6.Addr[:]), sa6.Port, sa6.ZoneId, std.IP, std.Port, std.Zone, zoneID885(std.Zone)) + } + } + } +} + +// zonedTarget885 returns a link-local address of this host with its zone, as +// "fe80::1234%dz0": the first up, non-loopback interface that has one. Without +// one the test skips, and fails when CELERIS_REQUIRE_ZONED_PEER=1 (a lane's +// container gets a dummy link, "ip link add dz0 type dummy; ip -6 addr add +// fe80::1234/64 dev dz0 nodad", with CAP_NET_ADMIN and +// net.ipv6.conf.all.disable_ipv6=0). +func zonedTarget885(t *testing.T) string { + t.Helper() + ifs, err := net.Interfaces() + if err != nil { + t.Fatalf("net.Interfaces: %v", err) + } + for _, ifi := range ifs { + if ifi.Flags&net.FlagUp == 0 || ifi.Flags&net.FlagLoopback != 0 { + continue + } + addrs, _ := ifi.Addrs() + for _, a := range addrs { + if ipn, ok := a.(*net.IPNet); ok && ipn.IP.To4() == nil && ipn.IP.IsLinkLocalUnicast() { + return ipn.IP.String() + "%" + ifi.Name + } + } + } + const env = "CELERIS_REQUIRE_ZONED_PEER" + if os.Getenv(env) == "1" { + t.Fatalf("no up, non-loopback interface has an fe80:: address, and %s=1 forbids skipping", env) + } + t.Skip("no up, non-loopback interface has an fe80:: address") + return "" +} + +// TestListenOnAZonedLinkLocalServes885: Listen on a zoned link-local address +// binds it, serves a request on it, and reports the address as net.Listen does +// ("[fe80::1234%dz0]:port"). +func TestListenOnAZonedLinkLocalServes885(t *testing.T) { + host := zonedTarget885(t) + std, err := net.Listen("tcp", net.JoinHostPort(host, "0")) + if err != nil { + t.Fatalf("std oracle: net.Listen(%q): %v", host, err) + } + _ = std.Close() + r := start895(t, net.JoinHostPort(host, "0"), 2, false) + addr := r.e.Addr().String() + if !strings.HasPrefix(addr, "["+host+"]:") { + t.Errorf("Addr() = %q, want the zoned address [%s]:port, as net.Listen reports it", addr, host) + } + get895(t, addr) +} diff --git a/internal/engine/epoll/pause_wake_886_linux_test.go b/internal/engine/epoll/pause_wake_886_linux_test.go new file mode 100644 index 000000000..cfa7837a4 --- /dev/null +++ b/internal/engine/epoll/pause_wake_886_linux_test.go @@ -0,0 +1,100 @@ +//go:build linux + +package epoll + +import ( + "net" + "strconv" + "testing" + "time" +) + +// celeris#886 item 24 (and the flap remainder of item 11). BeginPauseAccept +// stored the flag and woke nobody: a loop that is idle and still listening saw +// the pause only when its current wait ended, so its listener closed late. The +// resume side got the wake eventfd in #1013; this is the pause side. The test +// is #1013's lag test run the other way: the pause is DisableDeferAccept's (its +// listener closes at once, there is no linger), and the lag is the time from +// BeginPauseAccept until the kernel's table shows no listener of the engine. + +// pauseLag886 begins a pause and returns how long it took until no listen +// socket is left on port (the kernel's table, census895), or gaveup if that +// did not happen. It then resumes and waits for want listeners, so that the +// next cycle starts from a listening engine. +func pauseLag886(t *testing.T, e *Engine, port, want int, gaveup time.Duration) time.Duration { + t.Helper() + if n := census895(t).listening[port]; n < want { + t.Fatalf("celeris886 PREMISE: %d listen sockets on port %d before the pause, want %d", n, port, want) + } + t0 := time.Now() + e.BeginPauseAccept() + lag := gaveup + for time.Since(t0) < gaveup { + if census895(t).listening[port] == 0 { + lag = time.Since(t0) + break + } + time.Sleep(200 * time.Microsecond) + } + if err := e.ResumeAccept(); err != nil { + t.Fatalf("ResumeAccept: %v", err) + } + for dl := time.Now().Add(3 * time.Second); census895(t).listening[port] < want; time.Sleep(time.Millisecond) { + if time.Now().After(dl) { + t.Fatalf("celeris886: %d loops listening on port %d 3s after the resume, want %d", census895(t).listening[port], port, want) + } + } + return lag +} + +// TestBeginPauseAcceptReachesAnIdleLoop886: a pause must close every loop's +// listener within lagBound, whatever the loop is waiting for. The default arm +// is the guard that the 25 ms timerfd keeps the lag short; the arm without a +// header deadline has no timerfd and sits in epoll_wait for up to 500 ms, which +// is the one the eventfd signal is for. +func TestBeginPauseAcceptReachesAnIdleLoop886(t *testing.T) { + const ( + cycles = 8 + lagBound = 100 * time.Millisecond + ) + for _, tc := range []struct { + name string + rht time.Duration + conns int + }{ + {"default-read-header-timeout/live-connections", 0, 16}, + {"no-read-header-timeout/live-connections", -1, 16}, + {"default-read-header-timeout/no-connections", 0, 0}, + {"no-read-header-timeout/no-connections", -1, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + r := start886(t, 2, tc.rht) + r.e.mu.Lock() + want := len(r.e.loops) + r.e.mu.Unlock() + port := portOf895(t, r.e.Addr()) + addr := net.JoinHostPort("127.0.0.1", strconv.Itoa(port)) + live886(t, addr, tc.conns) + + // Settle first: the wait grows with the empty iterations the loop has + // had in a row (about 200 ms to reach its longest on epoll, 0.5 s on + // io_uring), and the lag is worst at the longest. + time.Sleep(1200 * time.Millisecond) + var lags []time.Duration + for range cycles { + // Idle for a while first: the loop is then deep in its wait, not + // just out of one, which is the worst case for the lag. + time.Sleep(300 * time.Millisecond) + lags = append(lags, pauseLag886(t, r.e, port, want, 3*time.Second)) + } + t.Logf("celeris886 PAUSE-LAG [%s] (%d loops, %d live connections, BeginPauseAccept to no listener): %v, bound %v", tc.name, want, tc.conns, lags, lagBound) + for i, lag := range lags { + if lag > lagBound { + t.Errorf("celeris886 PAUSE-LAG: cycle %d took %v to have no loop listening, want within %v: "+ + "an idle loop did not see the pause until its wait ended", i, lag, lagBound) + } + } + get895(t, addr) + }) + } +} diff --git a/internal/engine/epoll/pin_linux.go b/internal/engine/epoll/pin_linux.go index c868f39b1..933e735b4 100644 --- a/internal/engine/epoll/pin_linux.go +++ b/internal/engine/epoll/pin_linux.go @@ -4,6 +4,7 @@ package epoll import ( "log/slog" + "slices" "github.com/goceleris/celeris/internal/platform" ) @@ -17,6 +18,47 @@ var pinThreadToCPU = platform.PinToCPU // plan. var planWorkerCPUs = platform.PlanWorkerCPUsFor +// detectNUMA reads the host's NUMA topology, for the MaxEvents split in Listen. +// A variable so a test can stand in a topology. +var detectNUMA = platform.DetectNUMA + +// splitMaxEvents is the MaxEvents a loop gets: the configured figure divided by +// the number of NUMA nodes the process's allowed CPU set meets, never under 64 +// (celeris#885 item 31). It divided by the host's node count, so a process kept +// to one node (numactl --cpunodebind, a single-numa-node pod, a cpuset of one +// socket), whose every loop is on that node, still got MaxEvents / NumNodes. +// allowed is the plan's allowed set (platform.CPUPlan.Allowed). Where the nodes +// it meets cannot be told (no allowed set, no CPU lists for the nodes, or none +// of its CPUs on a listed node) the host's count is used, as before. +func splitMaxEvents(maxEvents int, topo platform.NUMATopology, allowed []int) int { + nodes := topo.NumNodes + if n := nodesMet(topo, allowed); n > 0 { + nodes = n + } + if nodes > 1 { + maxEvents /= nodes + if maxEvents < 64 { + maxEvents = 64 + } + } + return maxEvents +} + +// nodesMet is the number of nodes of topo that hold a CPU of allowed; 0 when +// that cannot be told. +func nodesMet(topo platform.NUMATopology, allowed []int) int { + if len(allowed) == 0 || len(topo.NodeCPUs) == 0 { + return 0 + } + n := 0 + for _, cpus := range topo.NodeCPUs { + if slices.ContainsFunc(cpus, func(c int) bool { return slices.Contains(allowed, c) }) { + n++ + } + } + return n +} + // annotateNICIRQs adds the report of the CPUs that service the interrupts of // the NIC the listener is reachable through to a plan that is already decided // (celeris#1036). It only reports: the loops are the plan's. A variable so a diff --git a/internal/engine/internal/bindiag/addr_format_test.go b/internal/engine/internal/bindiag/addr_format_test.go index 3c97dff31..e73dbed9d 100644 --- a/internal/engine/internal/bindiag/addr_format_test.go +++ b/internal/engine/internal/bindiag/addr_format_test.go @@ -224,3 +224,34 @@ func BenchmarkSockaddrString925(b *testing.B) { }) } } + +// TestZoneIDIsStdsRule (celeris#885 item 54): the scope id of a zone is read as +// net reads it for a listen address: the interface of that name, else the +// decimal digits the zone starts with (net's dtoi: "12abc" is 12, and a number +// of 0xFFFFFF or more saturates), else 0. +func TestZoneIDIsStdsRule(t *testing.T) { + lo, err := net.InterfaceByName("lo") + if err != nil { + t.Skipf("no lo interface: %v", err) + } + for _, tc := range []struct { + zone string + want uint32 + }{ + {"", 0}, + {"lo", uint32(lo.Index)}, + {"7", 7}, + {"12abc", 12}, + {"007", 7}, + {"abc", 0}, + {"abc12", 0}, + {"-5", 0}, + {"16777214", 16777214}, + {"16777215", 0xFFFFFF}, + {"99999999999", 0xFFFFFF}, + } { + if got := ZoneID(tc.zone); got != tc.want { + t.Errorf("ZoneID(%q) = %d, want %d", tc.zone, got, tc.want) + } + } +} diff --git a/internal/engine/internal/bindiag/bindiag.go b/internal/engine/internal/bindiag/bindiag.go index 5f097b01f..18c35abc4 100644 --- a/internal/engine/internal/bindiag/bindiag.go +++ b/internal/engine/internal/bindiag/bindiag.go @@ -166,6 +166,30 @@ func SockaddrString(fd int, sa unix.Sockaddr) string { return "" } +// ZoneID is the scope id of an IPv6 zone as net.Listen reads it (net's +// ipv6ZoneCache.index, go1.27): the index of the interface of that name, else +// the decimal digits the zone starts with (net's dtoi: "12abc" is 12, and a +// number of 0xFFFFFF or more saturates at 0xFFFFFF), else 0. A zone that names +// nothing is therefore not an error here, as it is not for std: the kernel +// refuses the bind of a link-local address with no scope, and a global one does +// not look at it (celeris#885 item 54). One netlink dump, at listen time. +func ZoneID(zone string) uint32 { + if zone == "" { + return 0 + } + if ifi, err := net.InterfaceByName(zone); err == nil { + return uint32(ifi.Index) + } + n := 0 + for i := 0; i < len(zone) && '0' <= zone[i] && zone[i] <= '9'; i++ { + n = n*10 + int(zone[i]-'0') + if n >= 0xFFFFFF { + return 0xFFFFFF + } + } + return uint32(n) +} + // BoundAddr is the bound address of the socket fd as a *net.TCPAddr, or nil // when Getsockname fails. The zone is set only for a nonzero scope id: an // unscoped "[::]:port" listener printed as "[::%0]:port" before celeris#925. diff --git a/internal/engine/iouring/abort_seam_844.go b/internal/engine/iouring/abort_seam_844.go index 53f3eea8c..150503801 100644 --- a/internal/engine/iouring/abort_seam_844.go +++ b/internal/engine/iouring/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 diff --git a/internal/engine/iouring/adopt_request_count_886_linux_test.go b/internal/engine/iouring/adopt_request_count_886_linux_test.go index da1305537..df6d4cab3 100644 --- a/internal/engine/iouring/adopt_request_count_886_linux_test.go +++ b/internal/engine/iouring/adopt_request_count_886_linux_test.go @@ -4,6 +4,7 @@ package iouring 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 TestIouringAdoptRequestCountSync886(t *testing.T) { e, _ := start908(t, false, false) adoptCount886(t, e, carryGood908) @@ -103,20 +117,22 @@ func TestIouringAdoptRequestCountEmptyCarry886(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 TestIouringAdoptRequestCountUpgradeCarry886(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 TestIouringAdoptRequestCountUpgradeCarryAsync886(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) } diff --git a/internal/engine/iouring/detach_rollback_889_linux_test.go b/internal/engine/iouring/detach_rollback_889_linux_test.go index 6f73c482c..fd7af5443 100644 --- a/internal/engine/iouring/detach_rollback_889_linux_test.go +++ b/internal/engine/iouring/detach_rollback_889_linux_test.go @@ -58,10 +58,14 @@ 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" ) @@ -69,6 +73,10 @@ const ( 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 }() @@ -108,7 +116,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" @@ -120,6 +128,12 @@ func TestOnDetachPanicLeavesTheConnAsIfNeverRan889(t *testing.T) { if r.w.detachedCount != 0 || r.cs.detachCounted { t.Errorf("detachedCount=%d counted=%v, want 0 false", r.w.detachedCount, r.cs.detachCounted) } + // The exported gauge goes back with the count (#889 item 48, the + // check TestFinalGaugeRolledBack of the final review of #1014 + // made). + if g := r.w.detachedConns.Load(); g != 0 { + t.Errorf("the detachedConns gauge is %d after the rollback, want 0", g) + } if r.cs.asyncDetachPending { t.Error("asyncDetachPending is still set: the drain would count a conn that never detached") } diff --git a/internal/engine/iouring/engine.go b/internal/engine/iouring/engine.go index 1ec8963c8..6554dee96 100644 --- a/internal/engine/iouring/engine.go +++ b/internal/engine/iouring/engine.go @@ -722,10 +722,12 @@ func (e *Engine) Type() engine.EngineType { // nothing yet is promoted by the kernel during the linger and served like // any other, instead of being reset by the close. // -// It does not promise a prompt start: a worker sees the flag at its next -// ring wakeup, which on an idle plain-HTTP/1 worker can be up to 100 ms -// away. Correctness does not depend on that, because each listener's -// deadline is taken at its own clear. +// It signals each worker's wake eventfd, whose POLL_ADD the ring keeps armed +// in every mode, so a worker idle in its ring wait sees the flag at once: it +// used to see it at its next ring wakeup, up to 100 ms away on an idle +// plain-HTTP/1 worker (measured, celeris#886 item 24). A worker that is +// running just reads the counter at its next wakeup. Correctness does not +// depend on that, because each listener's deadline is taken at its own clear. // // It reads net.ipv4.tcp_synack_retries once, here. When that reads 0, the // pausing listeners also get TCP_SYNCNT=1, without which the kernel would @@ -735,6 +737,13 @@ func (e *Engine) Type() engine.EngineType { func (e *Engine) BeginPauseAccept() { e.pause.Begin(e.cfg.Logger, "io_uring") e.acceptPaused.Store(true) + // After the store, as ResumeAccept signals after its own: a worker woken + // here reads the flag set. A leaf lock and one write(2) per worker. + e.mu.Lock() + for _, w := range e.workers { + w.wakeFD.Signal() + } + e.mu.Unlock() } // PauseAccept stops accepting new connections. It is BeginPauseAccept diff --git a/internal/engine/iouring/h1verdict.go b/internal/engine/iouring/h1verdict.go index 1382b6412..ab676eb8c 100644 --- a/internal/engine/iouring/h1verdict.go +++ b/internal/engine/iouring/h1verdict.go @@ -74,6 +74,22 @@ func (w *Worker) finishH1Upgrade(cs *connState, fd int, link bool) bool { // internal/engine/epoll's endH1Verdict does the same: one lock spanning the // flush and OnError. The lock is released before the close, which takes it // again. +// +// One difference: OnError is called only inside the `if mu != nil` below, so +// for a conn that has a detachMu; epoll calls it whenever one is registered. +// A sync conn with a callback and no detachMu therefore loses it here and gets +// it there (celeris#886 item 9). The recv path (handleRecv) and the transplant +// replay (endCarriedReplay) both end here, so on this engine the rule is the +// same for both. 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 here by +// TestIouringRecvPathOnErrorSyncNoDetachMuIsDropped886 and +// TestIouringAdoptCarriedOnErrorSyncNoDetachMuIsDropped886, and on epoll's side +// by TestEpollRecvPathOnErrorSync908 and TestEpollAdoptCarriedOnErrorSync908. +// Which rule is right is an open decision. func (w *Worker) endH1Verdict(cs *connState, fd int, err error) { if mu := cs.detachMu; mu != nil { mu.Lock() diff --git a/internal/engine/iouring/onerror_nodetachmu_886_linux_test.go b/internal/engine/iouring/onerror_nodetachmu_886_linux_test.go new file mode 100644 index 000000000..699f55a7b --- /dev/null +++ b/internal/engine/iouring/onerror_nodetachmu_886_linux_test.go @@ -0,0 +1,60 @@ +//go:build linux + +package iouring + +import ( + "bufio" + "net" + "testing" + "time" +) + +// celeris#886 item 9. OnError delivery on a conn with no detachMu differs by +// engine: epoll calls H1State.OnError on a processing error whenever one is +// registered (endH1Verdict), io_uring only under detachMu, so only for a conn +// that has one. A sync conn gets a detachMu at Detach; a handler that +// registers a callback (Context.SetWSErrorHandler, exported) and does not +// Detach, or whose Detach faulted and was rolled back (the rollback puts +// detachMu back to nil and leaves the callback), is such a conn. epoll's side +// is pinned by TestEpollRecvPathOnErrorSync908 and +// TestEpollAdoptCarriedOnErrorSync908; these are the other side. They pin what +// io_uring does today, and are the ones to flip if the rule is unified. + +func TestIouringRecvPathOnErrorSyncNoDetachMuIsDropped886(t *testing.T) { + e, h := start908(t, false, false) + client, err := net.DialTimeout("tcp", e.Addr().String(), 3*time.Second) + if err != nil { + t.Fatalf("dial: %v", err) + } + t.Cleanup(func() { _ = client.Close() }) + if _, err := client.Write([]byte("GET /onerr HTTP/1.1\r\nHost: x\r\n\r\n" + carryBad908)); err != nil { + t.Fatalf("write: %v", err) + } + br := bufio.NewReader(client) + for i, want := range []int{200, 400} { + code, _, err := readBody543(client, br, 3*time.Second) + if err != nil || code != want { + t.Fatalf("response %d: got %d, err %v, want %d", i+1, code, err, want) + } + } + // The engine calls OnError (if it does) before it closes the conn, so once + // the close is seen the callback has had its chance. + if !hasEOF908(client, br, 3*time.Second) { + t.Fatal("the conn stayed open after the 400") + } + select { + case err := <-h.errs: + t.Errorf("io_uring delivered %v to the OnError of a sync conn with no detachMu: the rule changed (epoll does deliver); update item 886.9's comments and this pin", err) + default: + } +} + +func TestIouringAdoptCarriedOnErrorSyncNoDetachMuIsDropped886(t *testing.T) { + e, h := start908(t, false, false) + runCarryReplies908(t, e.AdoptConn, "GET /onerr HTTP/1.1\r\nHost: x\r\n\r\n"+carryBad908, []step908{{200, "/onerr"}, {400, ""}}) + select { + case err := <-h.errs: + t.Errorf("io_uring delivered %v to the OnError of a sync conn with no detachMu on the replay: the rule changed (epoll does deliver); update item 886.9's comments and this pin", err) + default: + } +} diff --git a/internal/engine/iouring/parse_addr_mapped_engcfg_test.go b/internal/engine/iouring/parse_addr_mapped_engcfg_test.go index 90c523ee3..c21e3db92 100644 --- a/internal/engine/iouring/parse_addr_mapped_engcfg_test.go +++ b/internal/engine/iouring/parse_addr_mapped_engcfg_test.go @@ -47,9 +47,8 @@ func TestParseAddrReadsAnAddressAsNetParseIPDoes(t *testing.T) { {"0127.0.0.1:80", nil}, {"300.1.1.1:80", nil}, {"1.2.3.4.5:80", nil}, - // KNOWN GAP, not a goal: std binds a zoned address (net.ParseIP, which - // this reads, rejects the zone); tracked as checklist #885 item 54. - {"[fe80::1%eth0]:80", nil}, + // A zoned address is read by TestParseAddrReadsAZonedIPv6Host885 + // (checklist #885 item 54). {"localhost:80", nil}, } { got, err := parseAddr(tc.addr) diff --git a/internal/engine/iouring/parse_addr_zone_885_linux_test.go b/internal/engine/iouring/parse_addr_zone_885_linux_test.go new file mode 100644 index 000000000..772392cc5 --- /dev/null +++ b/internal/engine/iouring/parse_addr_zone_885_linux_test.go @@ -0,0 +1,159 @@ +//go:build linux + +package iouring + +import ( + "net" + "os" + "strconv" + "strings" + "testing" + + "golang.org/x/sys/unix" +) + +// celeris#885 item 54. parseAddr read the host with net.ParseIP, which rejects +// a zone, so a zoned IPv6 bind address ("[fe80::1234%dz0]:0"), which net.Listen +// and so the std engine bind, made both native engines fail to start on the +// same Config.Addr. The host's zone is now read as net.Listen reads it: the +// index of the interface of that name, else the zone as a number, else 0 (the +// bind then fails for a link-local address, as std's does). An IPv4-mapped host +// is bound as IPv4 and its zone ignored, as std does. + +// zoneID885 is std's rule for a zone (net's ipv6ZoneCache.index, go1.27): the +// index of the interface of that name, else the decimal digits the name starts +// with (dtoi: "12abc" is 12, a number of 0xFFFFFF or more saturates at +// 0xFFFFFF), else 0. +func zoneID885(zone string) uint32 { + if zone == "" { + return 0 + } + if ifi, err := net.InterfaceByName(zone); err == nil { + return uint32(ifi.Index) + } + n := 0 + for i := 0; i < len(zone) && '0' <= zone[i] && zone[i] <= '9'; i++ { + n = n*10 + int(zone[i]-'0') + if n >= 0xFFFFFF { + return 0xFFFFFF + } + } + return uint32(n) +} + +func TestParseAddrReadsAZonedIPv6Host885(t *testing.T) { + lo, err := net.InterfaceByName("lo") + if err != nil { + t.Skipf("no lo interface: %v", err) + } + v6 := func(port int, ip string, zone uint32) unix.Sockaddr { + sa := &unix.SockaddrInet6{Port: port, ZoneId: zone} + copy(sa.Addr[:], net.ParseIP(ip).To16()) + return sa + } + for _, tc := range []struct { + addr string + want unix.Sockaddr // nil: an error + }{ + {"[fe80::1%lo]:80", v6(80, "fe80::1", uint32(lo.Index))}, + {"[fe80::1234%lo]:0", v6(0, "fe80::1234", uint32(lo.Index))}, + {"[fe80::1%" + strconv.Itoa(lo.Index) + "]:8080", v6(8080, "fe80::1", uint32(lo.Index))}, + {"[fe80::1%7]:80", v6(80, "fe80::1", 7)}, + // An interface that does not exist and is not a number is scope 0, as + // std reads it: the kernel then refuses the bind of a link-local + // address, and a global one binds. + {"[fe80::1%nosuchif0]:80", v6(80, "fe80::1", 0)}, + // A malformed zone is read as std reads it: the digits it starts with. + {"[fe80::1%12abc]:80", v6(80, "fe80::1", 12)}, + {"[fe80::1%99999999999]:80", v6(80, "fe80::1", 0xFFFFFF)}, + {"[fe80::1%abc12]:80", v6(80, "fe80::1", 0)}, + {"[2001:db8::1%lo]:443", v6(443, "2001:db8::1", uint32(lo.Index))}, + {"[::%lo]:80", v6(80, "::", uint32(lo.Index))}, + // Everything after the first % is the zone, so this one names no interface. + {"[fe80::1%lo%lo]:80", v6(80, "fe80::1", 0)}, + // A mapped IPv4 address is IPv4 for std, and a zone means nothing to it. + {"[::ffff:1.2.3.4%lo]:80", &unix.SockaddrInet4{Port: 80, Addr: [4]byte{1, 2, 3, 4}}}, + // Not an address (std refuses them too): no zone on IPv4, an empty zone, a + // zone and no host. + {"1.2.3.4%lo:80", nil}, + {"[1.2.3.4%lo]:80", nil}, + {"[fe80::1%]:80", nil}, + {"[%lo]:80", nil}, + } { + got, err := parseAddr(tc.addr) + if tc.want == nil { + if err == nil { + t.Errorf("parseAddr(%q) = %v, want an error", tc.addr, got) + } + continue + } + if err != nil { + t.Errorf("parseAddr(%q): %v, want %v", tc.addr, err, tc.want) + continue + } + if !sockaddrEqual(got, tc.want) { + t.Errorf("parseAddr(%q) = %v, want %v", tc.addr, got, tc.want) + } + // std's reading of the same literal: the same IP, port and zone. + std, serr := net.ResolveTCPAddr("tcp", tc.addr) + if serr != nil { + t.Errorf("std refuses %q, which the table says is an address: %v", tc.addr, serr) + continue + } + if sa6, ok := got.(*unix.SockaddrInet6); ok { + if !net.IP(sa6.Addr[:]).Equal(std.IP) || sa6.Port != std.Port || sa6.ZoneId != zoneID885(std.Zone) { + t.Errorf("parseAddr(%q) = %v:%d scope %d, std reads %v:%d zone %q (scope %d)", + tc.addr, net.IP(sa6.Addr[:]), sa6.Port, sa6.ZoneId, std.IP, std.Port, std.Zone, zoneID885(std.Zone)) + } + } + } +} + +// zonedTarget885 returns a link-local address of this host with its zone, as +// "fe80::1234%dz0": the first up, non-loopback interface that has one. Without +// one the test skips, and fails when CELERIS_REQUIRE_ZONED_PEER=1 (a lane's +// container gets a dummy link, "ip link add dz0 type dummy; ip -6 addr add +// fe80::1234/64 dev dz0 nodad", with CAP_NET_ADMIN and +// net.ipv6.conf.all.disable_ipv6=0). +func zonedTarget885(t *testing.T) string { + t.Helper() + ifs, err := net.Interfaces() + if err != nil { + t.Fatalf("net.Interfaces: %v", err) + } + for _, ifi := range ifs { + if ifi.Flags&net.FlagUp == 0 || ifi.Flags&net.FlagLoopback != 0 { + continue + } + addrs, _ := ifi.Addrs() + for _, a := range addrs { + if ipn, ok := a.(*net.IPNet); ok && ipn.IP.To4() == nil && ipn.IP.IsLinkLocalUnicast() { + return ipn.IP.String() + "%" + ifi.Name + } + } + } + const env = "CELERIS_REQUIRE_ZONED_PEER" + if os.Getenv(env) == "1" { + t.Fatalf("no up, non-loopback interface has an fe80:: address, and %s=1 forbids skipping", env) + } + t.Skip("no up, non-loopback interface has an fe80:: address") + return "" +} + +// TestListenOnAZonedLinkLocalServes885: Listen on a zoned link-local address +// binds it, serves a request on it, and reports the address as net.Listen does +// ("[fe80::1234%dz0]:port"). +func TestListenOnAZonedLinkLocalServes885(t *testing.T) { + host := zonedTarget885(t) + std, err := net.Listen("tcp", net.JoinHostPort(host, "0")) + if err != nil { + t.Fatalf("std oracle: net.Listen(%q): %v", host, err) + } + _ = std.Close() + r := start895(t, net.JoinHostPort(host, "0"), 2, 1, false) + addr := r.e.Addr().String() + if !strings.HasPrefix(addr, "["+host+"]:") { + t.Errorf("Addr() = %q, want the zoned address [%s]:port, as net.Listen reports it", addr, host) + } + get895(t, addr) +} diff --git a/internal/engine/iouring/pause_wake_886_linux_test.go b/internal/engine/iouring/pause_wake_886_linux_test.go new file mode 100644 index 000000000..4745bddb2 --- /dev/null +++ b/internal/engine/iouring/pause_wake_886_linux_test.go @@ -0,0 +1,200 @@ +//go:build linux + +package iouring + +import ( + "context" + "io" + "log/slog" + "net" + "strconv" + "testing" + "time" + + "golang.org/x/sys/unix" + + "github.com/goceleris/celeris/internal/engine" + "github.com/goceleris/celeris/internal/resource" + "github.com/goceleris/celeris/internal/wakefd" +) + +// startPause886 is start895 on a loopback port 0 with DisableDeferAccept (a +// pause closes at once) and a ReadHeaderTimeout: a negative one disables the +// header deadline, and with it the 25 ms cap on the worker's ring wait, which +// leaves the 100 ms of Worker.baseTimeout. +func startPause886(t *testing.T, workers int, readHeaderTimeout time.Duration) *run895 { + t.Helper() + needWorkers895(t, 1) + var run *run895 + _, _, err := retryRingENOMEM(func() error { + e, err := New(resource.Config{ + Addr: "127.0.0.1:0", + Engine: engine.IOUring, + Protocol: engine.HTTP1, + Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + DisableDeferAccept: true, + ReadHeaderTimeout: readHeaderTimeout, + Resources: resource.Resources{Workers: workers}, + }, respondingHandler{}) + if err != nil { + skipOrFail656(t, "New: %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- e.Listen(ctx) }() + for dl := time.Now().Add(5 * time.Second); time.Now().Before(dl); { + if e.Addr() != nil { + run = &run895{e: e, cancel: cancel, done: done} + t.Cleanup(func() { cancel(); <-done }) + return nil + } + select { + case lerr := <-done: + cancel() + return lerr + case <-time.After(time.Millisecond): + } + } + cancel() + <-done + t.Fatalf("engine never published an address") + return nil + }) + if err != nil { + if ioUringUnavailable639(err) { + skipUnlessIOUringUnusable(t, err) + } + t.Fatalf("engine start: %v", err) + } + return run +} + +// celeris#886 item 24 (and the flap remainder of item 11). BeginPauseAccept +// stored the flag and woke nobody: a worker that is idle and still listening +// saw the pause only when its ring wait ended (Worker.baseTimeout: 100 ms, 50 ms +// with detached conns, 25 ms with a ReadHeaderTimeout), so its listener closed +// late. The resume side got the wake eventfd in #1013; this is the pause side. +// The test is #1013's lag test run the other way: the pause is +// DisableDeferAccept's (its listener closes at once, there is no linger), and +// the lag is the time from BeginPauseAccept until the kernel's table shows no +// listener of the engine. + +// pauseLag886 begins a pause and returns how long it took until no listen +// socket is left on port (the kernel's table, census895), or gaveup if that +// did not happen. It then resumes and waits for want listeners, so that the +// next cycle starts from a listening engine. +func pauseLag886(t *testing.T, e *Engine, port, want int, gaveup time.Duration) time.Duration { + t.Helper() + if n := census895(t).listening[port]; n < want { + t.Fatalf("celeris886 PREMISE: %d listen sockets on port %d before the pause, want %d", n, port, want) + } + t0 := time.Now() + e.BeginPauseAccept() + lag := gaveup + for time.Since(t0) < gaveup { + if census895(t).listening[port] == 0 { + lag = time.Since(t0) + break + } + time.Sleep(200 * time.Microsecond) + } + if err := e.ResumeAccept(); err != nil { + t.Fatalf("ResumeAccept: %v", err) + } + for dl := time.Now().Add(3 * time.Second); census895(t).listening[port] < want; time.Sleep(time.Millisecond) { + if time.Now().After(dl) { + t.Fatalf("celeris886: %d workers listening on port %d 3s after the resume, want %d", census895(t).listening[port], port, want) + } + } + return lag +} + +// TestBeginPauseAcceptReachesAnIdleWorker886: a pause must close every +// worker's listener within lagBound, whatever the worker is waiting for. Two +// workers are asked for: with memlock raised both start, and a fix that +// signalled only one of them fails; at the runner's 8 MiB one starts (see the +// PREMISE check). +func TestBeginPauseAcceptReachesAnIdleWorker886(t *testing.T) { + const cycles = 8 + for _, tc := range []struct { + name string + rht time.Duration + conns int + lagBound time.Duration + }{ + // The default arms are the guard that the 25 ms cap of a ReadHeaderTimeout + // keeps the lag short; the others have the 100 ms wait the signal is for. + // The bound is loose on purpose (CI runs this under -race on two vCPUs): + // the lag is a few ms with the signal, and up to 99 ms without it + // (measured), a tail too rare to fail on reliably. The deterministic test + // below is the one that fails before the change. + {"default-read-header-timeout/live-connections", 0, 16, 100 * time.Millisecond}, + {"no-read-header-timeout/live-connections", -1, 16, 100 * time.Millisecond}, + {"default-read-header-timeout/no-connections", 0, 0, 100 * time.Millisecond}, + {"no-read-header-timeout/no-connections", -1, 0, 100 * time.Millisecond}, + } { + t.Run(tc.name, func(t *testing.T) { + lagBound := tc.lagBound + r := startPause886(t, 2, tc.rht) + r.e.mu.Lock() + want := len(r.e.workers) + r.e.mu.Unlock() + if maxW := MaxWorkersForMemlock(); want < 2 && (maxW < 0 || maxW >= 2) { + t.Fatalf("celeris886 PREMISE: %d io_uring worker(s) started with memlock funding %d (-1: unlimited), want 2", want, maxW) + } + port := portOf895(t, r.e.Addr()) + addr := net.JoinHostPort("127.0.0.1", strconv.Itoa(port)) + live886(t, addr, tc.conns) + + // Settle first: the wait grows with the empty iterations the worker has + // had in a row (about 200 ms to reach its longest on epoll, 0.5 s on + // io_uring), and the lag is worst at the longest. + time.Sleep(1200 * time.Millisecond) + var lags []time.Duration + for range cycles { + // Idle for a while first: the worker is then deep in its ring + // wait, not just out of one, the worst case for the lag. + time.Sleep(300 * time.Millisecond) + lags = append(lags, pauseLag886(t, r.e, port, want, 3*time.Second)) + } + t.Logf("celeris886 PAUSE-LAG [%s] (%d workers, %d live connections, BeginPauseAccept to no listener): %v, bound %v", tc.name, want, tc.conns, lags, lagBound) + for i, lag := range lags { + if lag > lagBound { + t.Errorf("celeris886 PAUSE-LAG: cycle %d took %v to have no worker listening, want within %v: "+ + "an idle worker did not see the pause until its wait ended", i, lag, lagBound) + } + } + get895(t, addr) + }) + } +} + +// TestBeginPauseAcceptSignalsEveryWorkerWakeFD886 is the deterministic half: +// the pause must write the wake eventfd of every worker, which is what ends +// the ring wait of a worker that has not parked. The end-to-end lag above +// depends on where in its wait each worker is, so it cannot fail reliably +// before the change; the eventfd counter can. +func TestBeginPauseAcceptSignalsEveryWorkerWakeFD886(t *testing.T) { + e := &Engine{} + var efds []int + for range 3 { + efd, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC) + if err != nil { + t.Fatalf("eventfd: %v", err) + } + t.Cleanup(func() { _ = unix.Close(efd) }) + efds = append(efds, efd) + e.workers = append(e.workers, &Worker{wakeFD: wakefd.New(efd)}) + } + e.BeginPauseAccept() + if !e.acceptPaused.Load() { + t.Fatal("BeginPauseAccept did not set the pause") + } + for i, efd := range efds { + var buf [8]byte + if n, err := unix.Read(efd, buf[:]); err != nil || n != 8 || buf == [8]byte{} { + t.Errorf("celeris886 worker %d: its wake eventfd was not signalled by BeginPauseAccept (read n=%d err=%v): "+ + "an idle worker sees the pause only when its ring wait ends", i, n, err) + } + } +} diff --git a/internal/engine/iouring/transplant.go b/internal/engine/iouring/transplant.go index d0bf9d5f7..80160110f 100644 --- a/internal/engine/iouring/transplant.go +++ b/internal/engine/iouring/transplant.go @@ -214,12 +214,15 @@ func (w *Worker) attachAdoptedFD(newFD int, carry engine.Carryover) { } else { perr := conn.ProcessH1(cs.ctx, carry.Buffered, cs.h1State, w.handler, cs.writeFn) // The replay is the recv event of this conn's first bytes and - // counts one, whatever the verdict (celeris#886). handleRecv - // 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. - w.reqBatch++ + // counts one, whatever the verdict but one (celeris#886). + // handleRecv 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). + if !errors.Is(perr, conn.ErrUpgradeH2C) { + w.reqBatch++ + } if perr != nil { if !w.endCarriedReplay(cs, newFD, perr) { return // hijacked (handed off) or closed on error — nothing more to arm @@ -277,11 +280,13 @@ func (w *Worker) replayCarriedAsync(cs *connState, fd int, data []byte) bool { return false } // Counted for every outcome but the promotion above, whose - // promoteConnToAsync counts it: the replay is the recv event of this - // conn's first bytes (celeris#886). Unlike handleRecv it also counts a - // carry that ends in an h2c upgrade, which that path returns before - // counting. - w.reqBatch++ + // promoteConnToAsync counts it, and an h2c upgrade: the replay is the recv + // event of this conn's first bytes (celeris#886). Like handleRecv, which + // returns on the upgrade verdict before it counts, it counts a carry that + // ends in an h2c upgrade as none (item 23). + if !errors.Is(perr, conn.ErrUpgradeH2C) { + w.reqBatch++ + } if perr != nil { // Nothing below may touch cs.h1State: an h2c upgrade has replaced it. return w.endCarriedReplay(cs, fd, perr) diff --git a/internal/engine/iouring/worker.go b/internal/engine/iouring/worker.go index 89a98a60e..23dfa8da7 100644 --- a/internal/engine/iouring/worker.go +++ b/internal/engine/iouring/worker.go @@ -9,6 +9,7 @@ import ( "io" "log/slog" "net" + "net/netip" "os" "runtime" "runtime/debug" @@ -3427,6 +3428,52 @@ func (w *Worker) initProtocol(cs *connState) { prevRaw, prevPause, prevResume = h1.RawWriteFn, h1.PauseRecv, h1.ResumeRecv } counted, finished := false, false + // unlockDetachMu: this goroutine holds detachMu (the promoted async path) and + // releases it on the way out; a conn that is not on that path is rolled back + // instead. Decided here, from the snapshot, so that the rollback below is + // registered before the first mutation and an edit that adds one between + // the snapshot and the install is inside it (#889 item 48). It is the gate + // the hook used after it had allocated detachMu: in sync mode w.async is false + // either way, and in async mode detachMu was allocated at acquireConnState, + // so prevMu is not nil. + unlockDetachMu := w.async && cs.asyncPromoted.Load() && prevMu != nil && !cs.asyncDetachUnlocked + if !unlockDetachMu { + // A conn that is not on the promoted async path goes on after a panic + // out of this hook: Context.Detach has not marked the Context detached, + // the router answers 500, and the next request is served on the same + // conn. Put back everything the hook changes from here on, so that it is + // exactly as if Detach had not run: a guarded writeFn left behind + // locks detachMu again on the first write of the H2C branch and of + // serveAsync, which hold it across theirs (the loop thread, or the + // dispatch goroutine, deadlocks on itself), a count left behind is + // made again by the next Detach, and closeConn undoes one. The + // promoted path closes the conn instead (the deferred hook below). + // Assignments only, under no lock: this goroutine owns the conn's + // state here, as it did when the hook installed it (#889 item 34). + defer func() { + if finished { + return + } + cs.writeFn = prevWrite + if h1 != nil { + h1.UpdateWriteFn(prevWrite) + h1.SetWriteErrQuery(cs) + h1.RawWriteFn, h1.PauseRecv, h1.ResumeRecv = prevRaw, prevPause, prevResume + } + cs.detachMu = prevMu + cs.asyncDetachPending = prevPending + if counted { + // Sync mode only. The next Detach counts again, and closeConn undoes the count + // of a conn with detachCounted set, so the sync path relies on this leaving it + // false (#889 item 48). + w.detachedCount-- + cs.detachCounted = false + if w.detachedConns != nil { + w.detachedConns.Add(-1) + } + } + }() + } mu := cs.detachMu if mu == nil { mu = &sync.Mutex{} @@ -3451,6 +3498,9 @@ func (w *Worker) initProtocol(cs *connState) { } else { cs.asyncDetachPending = true } + if f := detachPostCountHook.Load(); f != nil { + (*f)() // test seam, #889 item 48 (between the snapshot and the rollback) + } orig := cs.writeFn wake := w.wakeFD guarded := func(data []byte) { @@ -3575,46 +3625,13 @@ func (w *Worker) initProtocol(cs *connState) { // CloseH1 instead of recycling the Context/stream this WS/SSE // middleware goroutine is still using — the use-after-recycle that // nil-derefs WSRawWriteFn under a peer RST mid-upgrade. - unlockDetachMu := w.async && cs.asyncPromoted.Load() && cs.detachMu != nil && !cs.asyncDetachUnlocked + // unlockDetachMu was decided at the top of the hook, before the first mutation, so that the rollback is + // registered before any of them. // completed: this closure ran to its end. Nothing is owed when this // goroutine holds no lock here. completed := !unlockDetachMu // flagged: asyncDetachUnlocked has been published by this closure. flagged := false - if !unlockDetachMu { - // A conn that is not on the promoted async path goes on after a panic - // out of this hook: Context.Detach has not marked the Context detached, - // the router answers 500, and the next request is served on the same - // conn. Put back everything installed above and below, so that it is - // exactly as if Detach had not run: a guarded writeFn left behind - // locks detachMu again on the first write of the H2C branch and of - // serveAsync, which hold it across theirs (the loop thread, or the - // dispatch goroutine, deadlocks on itself), a count left behind is - // made again by the next Detach, and closeConn undoes one. The - // promoted path closes the conn instead (the deferred hook below). - // Assignments only, under no lock: this goroutine owns the conn's - // state here, as it did when the hook installed it (#889 item 34). - defer func() { - if finished { - return - } - cs.writeFn = prevWrite - if h1 != nil { - h1.UpdateWriteFn(prevWrite) - h1.SetWriteErrQuery(cs) - h1.RawWriteFn, h1.PauseRecv, h1.ResumeRecv = prevRaw, prevPause, prevResume - } - cs.detachMu = prevMu - cs.asyncDetachPending = prevPending - if counted { - w.detachedCount-- - cs.detachCounted = false - if w.detachedConns != nil { - w.detachedConns.Add(-1) - } - } - }() - } if unlockDetachMu { // Registered before guarded is installed (celeris#844): from the moment // cs.writeFn is guarded, a write through the response adapter locks @@ -7803,7 +7820,22 @@ const ( // no new stream on it, as net/http's graceful shutdown does, and a stream it // opens anyway is refused. The wait is bounded by the budget the last // Engine.Shutdown handed over, as epoll's is, and never ends before -// h2PoolDrainFloor. Worker thread. +// h2PoolDrainFloor. +// +// A GOAWAY the write hook refuses (the conn's backlog is over the HTTP/2 cap, +// 64 MiB, so its peer has stopped reading) is lost and sets writeRefused; this +// pass does not look at it (celeris#885 item 14, read from the code, not +// measured). The conn is not left open for it: the next drainH2Writes with a +// frame queued closes the conn (closeConn sends what was staged first), a recv +// of the conn that finds writeRefused ends it as a request error does +// (handleRecv), and a conn with none of those is not busy here, so the pass +// settles and shutdown closes it. What the lost GOAWAY costs is the client's +// graceful notice, from a peer that was not reading; what a conn that stays +// busy (OutboundPending, response DATA waiting for a WINDOW_UPDATE that will not +// come) costs is the wait up to the drain's bound, the same as for any conn +// whose peer is not taking its response. +// +// Worker thread. func (w *Worker) h2PoolSettled() bool { if len(w.h2Conns) == 0 { return true @@ -8471,6 +8503,21 @@ func parseAddr(addr string) (unix.Sockaddr, error) { // own rules: a byte over 255 ("300.1.1.1" bound 44.1.1.1), a leading zero // ("0127.0.0.1": octal in some parsers), a fifth byte, and the shorthands // inet_aton reads ("10", "127.1": "1.2.3" bound 1.2.3.0). + // + // A zoned IPv6 host ("fe80::1%eth0") is what net.Listen binds and net.ParseIP + // rejects: it is read with netip, which keeps the zone, and the zone becomes + // the socket address's scope id as net.Listen makes it (bindiag.ZoneID). An + // IPv4-mapped host is IPv4 whatever its zone, as for std (celeris#885 + // item 54). + var zone string + if strings.IndexByte(host, '%') >= 0 { + za, err := netip.ParseAddr(host) + if err != nil || !za.Is6() { + return nil, fmt.Errorf("invalid addr: %s", addr) + } + zone = za.Zone() + host = za.WithZone("").String() + } ip := net.ParseIP(host) if ip == nil { return nil, fmt.Errorf("invalid addr: %s", addr) @@ -8480,7 +8527,7 @@ func parseAddr(addr string) (unix.Sockaddr, error) { copy(sa.Addr[:], ip4) return sa, nil } - sa := &unix.SockaddrInet6{Port: port} + sa := &unix.SockaddrInet6{Port: port, ZoneId: bindiag.ZoneID(zone)} copy(sa.Addr[:], ip.To16()) return sa, nil } diff --git a/internal/platform/cpuplan_rxfree_linux.go b/internal/platform/cpuplan_rxfree_linux.go index a7ce13ea5..73ccd1ff0 100644 --- a/internal/platform/cpuplan_rxfree_linux.go +++ b/internal/platform/cpuplan_rxfree_linux.go @@ -145,6 +145,23 @@ const ( // port is freed with the rest: the evidence of msr1 is that CPU 8, which // services enp49s0's TX completion, can lose its loop while CPU 1 keeps one. // +// When it does not engage. The rule reads the counters since boot, once, at the +// first Listen, before any workload: the packet shares of step 1 and 7 are +// those of the host's whole uptime, not of the load the server is about to +// carry. On msr1 under kernel 7.0.0-38 the two ports of the bond (802.3ad, +// layer3+4 hash, receive hashing off on both) each take about half of a cell's +// packets (enp49s0: median 50.0% with a range of 39.2% to 61.8% over 64 cells +// measured), so no window, recent or since boot, reliably shows the one port +// over rxPortPercent that step 1 needs, and the rule took the two CPUs only +// where the counters had drifted to one port over 60% through a long uptime. A +// host whose counters do not show one port over 60% with a million packets (a +// freshly booted one, under rxMinPackets, or a network that splits evenly like +// msr1's) does not engage the rule, and its plan is the one without it: one +// loop on every big CPU (celeris#1049). That is the fallback working as +// intended, not a fault, and the rule is left as it is. A run that compares the +// plan lines of two builds on such a host should expect the rule to have acted +// on neither. +// // Any read that fails, any format it does not know, a busy vector whose CPU is // not a single one: no CPU is freed and Why says why. It reads nothing when the // floor leaves no CPU to free. diff --git a/internal/platform/nicrx_addrs_885_linux_test.go b/internal/platform/nicrx_addrs_885_linux_test.go new file mode 100644 index 000000000..4eb1e32d8 --- /dev/null +++ b/internal/platform/nicrx_addrs_885_linux_test.go @@ -0,0 +1,110 @@ +//go:build linux + +package platform + +import ( + "net" + "net/netip" + "reflect" + "syscall" + "testing" +) + +// celeris#885 item 53. osIfaceAddrs built the interface table of the NIC +// interrupt detection (PR #1062) with net.Interfaces and then +// (*Interface).Addrs for each of up to 512 interfaces, and every Addrs call is +// a full RTM_GETADDR dump of all addresses of the host, down and loopback +// interfaces included. A hostNetwork node with hundreds of veths paid hundreds +// of dumps, at start and again at the adaptive engine's standby Listen. The +// table is now one RTM_GETADDR dump, mapped by interface index. + +// refIfaceAddrs885 is osIfaceAddrs as it was (one Addrs call per interface), +// the oracle for the table. +func refIfaceAddrs885() ([]ifaceAddr, error) { + ifs, err := net.Interfaces() + if err != nil { + return nil, err + } + var out []ifaceAddr + for _, ifc := range ifs { + if len(out) >= 512 { + break + } + a := ifaceAddr{ + Name: ifc.Name, + Up: ifc.Flags&net.FlagUp != 0, + Loopback: ifc.Flags&net.FlagLoopback != 0, + } + addrs, err := ifc.Addrs() + if err != nil { + continue + } + for _, ad := range addrs { + var ip net.IP + switch v := ad.(type) { + case *net.IPNet: + ip = v.IP + case *net.IPAddr: + ip = v.IP + } + if p, ok := netip.AddrFromSlice(ip); ok { + a.Addrs = append(a.Addrs, p.Unmap()) + } + } + out = append(out, a) + } + return out, nil +} + +// TestOSIfaceAddrsIsOneAddressDump885: the table equals the one the +// per-interface calls give, and costs one RTM_GETADDR dump however many +// interfaces the host has. +func TestOSIfaceAddrsIsOneAddressDump885(t *testing.T) { + want, err := refIfaceAddrs885() + if err != nil { + t.Skipf("net.Interfaces: %v", err) + } + var dumps int + old := netlinkRIB + t.Cleanup(func() { netlinkRIB = old }) + netlinkRIB = func(proto, family int) ([]byte, error) { + if proto == syscall.RTM_GETADDR { + dumps++ + } + return old(proto, family) + } + got, err := osIfaceAddrs() + if err != nil { + t.Fatalf("osIfaceAddrs: %v", err) + } + if dumps != 1 { + t.Errorf("osIfaceAddrs made %d RTM_GETADDR dumps through netlinkRIB for %d interfaces, want 1 (net.Interface.Addrs, one dump per interface, does not use it)", dumps, len(got)) + } + if !reflect.DeepEqual(got, want) { + t.Errorf("table differs from the per-interface one:\n got %+v\n want %+v", got, want) + } +} + +// BenchmarkOSIfaceAddrs885 is the table's cost on this host: the figure moves +// with the interface count (a lane's container gets a few hundred dummy links). +func BenchmarkOSIfaceAddrs885(b *testing.B) { + for b.Loop() { + if _, err := osIfaceAddrs(); err != nil { + b.Fatal(err) + } + } + ifs, _ := net.Interfaces() + b.ReportMetric(float64(len(ifs)), "ifaces") +} + +// BenchmarkRefIfaceAddrs885 is the same for the per-interface calls this +// replaces. +func BenchmarkRefIfaceAddrs885(b *testing.B) { + for b.Loop() { + if _, err := refIfaceAddrs885(); err != nil { + b.Fatal(err) + } + } + ifs, _ := net.Interfaces() + b.ReportMetric(float64(len(ifs)), "ifaces") +} diff --git a/internal/platform/nicrx_linux.go b/internal/platform/nicrx_linux.go index 2a4da1383..957d498f9 100644 --- a/internal/platform/nicrx_linux.go +++ b/internal/platform/nicrx_linux.go @@ -3,6 +3,7 @@ package platform import ( + "encoding/binary" "errors" "io" "io/fs" @@ -13,6 +14,7 @@ import ( "slices" "strconv" "strings" + "syscall" ) // Reading the NICs behind a listener (celeris#1036 item 1). The rx-port rule @@ -122,13 +124,25 @@ func (osNICFS) exists(name string) bool { return err == nil } -// osIfaceAddrs lists the interfaces and their addresses with the net package -// (one netlink dump per interface; no cgo and no ip binary). +// netlinkRIB is the netlink dump osIfaceAddrs reads addresses with. A variable +// so a test can count the dumps (celeris#885 item 53). +var netlinkRIB = syscall.NetlinkRIB + +// osIfaceAddrs lists the interfaces and their addresses: net.Interfaces for the +// interfaces (one RTM_GETLINK dump) and one RTM_GETADDR dump for all their +// addresses, mapped by interface index. It was net.Interfaces and then +// (*Interface).Addrs for each of up to 512 interfaces, and every Addrs call is +// a full RTM_GETADDR dump of the host's addresses, so a hostNetwork node with +// hundreds of veths paid hundreds of dumps at start, and again at the adaptive +// engine's standby Listen while the server is serving (celeris#885 item 53). +// No cgo and no ip binary. If the dump cannot be read the table is made the old +// way. func osIfaceAddrs() ([]ifaceAddr, error) { ifs, err := net.Interfaces() if err != nil { return nil, err } + byIndex, ok := addrsByIndex() var out []ifaceAddr for _, ifc := range ifs { if len(out) >= 512 { @@ -139,20 +153,24 @@ func osIfaceAddrs() ([]ifaceAddr, error) { Up: ifc.Flags&net.FlagUp != 0, Loopback: ifc.Flags&net.FlagLoopback != 0, } - addrs, err := ifc.Addrs() - if err != nil { - continue - } - for _, ad := range addrs { - var ip net.IP - switch v := ad.(type) { - case *net.IPNet: - ip = v.IP - case *net.IPAddr: - ip = v.IP + if ok { + a.Addrs = byIndex[ifc.Index] + } else { + addrs, err := ifc.Addrs() + if err != nil { + continue } - if p, ok := netip.AddrFromSlice(ip); ok { - a.Addrs = append(a.Addrs, p.Unmap()) + for _, ad := range addrs { + var ip net.IP + switch v := ad.(type) { + case *net.IPNet: + ip = v.IP + case *net.IPAddr: + ip = v.IP + } + if p, ok := netip.AddrFromSlice(ip); ok { + a.Addrs = append(a.Addrs, p.Unmap()) + } } } out = append(out, a) @@ -160,6 +178,92 @@ func osIfaceAddrs() ([]ifaceAddr, error) { return out, nil } +// addrsByIndex is the addresses of every interface of the host from one +// RTM_GETADDR dump, by interface index, each in the order the kernel sent them +// (the order (*net.Interface).Addrs gives). ok is false if the dump cannot be +// read or parsed. +func addrsByIndex() (m map[int][]netip.Addr, ok bool) { + // syscall.ParseNetlinkRouteAttr, which parseAddrRIB calls and net calls on + // the same bytes, slices past the message for a final attribute that is not + // padded to 4. A kernel dump always pads, but a table that is promised to fall + // back must not panic at Listen for one that does not (review of #1127). + defer func() { + if recover() != nil { + m, ok = nil, false + } + }() + rib, err := netlinkRIB(syscall.RTM_GETADDR, syscall.AF_UNSPEC) + if err != nil { + return nil, false + } + return parseAddrRIB(rib) +} + +// parseAddrRIB maps the RTM_NEWADDR messages of a netlink dump to addresses by +// interface index. It reads an address as net does (interface_linux.go, +// newAddr): the IFA_LOCAL attribute where the message has one (a point-to-point +// link, whose IFA_ADDRESS is the peer), else IFA_ADDRESS; IPv4 mapped IPv6 +// forms are unmapped like the table does. A dump that was interrupted by a change +// of the addresses (NLM_F_DUMP_INTR) is read as it is, as net reads it: the table +// is an annotation of the NIC report, and may then miss or repeat an address. +func parseAddrRIB(rib []byte) (map[int][]netip.Addr, bool) { + msgs, err := syscall.ParseNetlinkMessage(rib) + if err != nil { + return nil, false + } + out := make(map[int][]netip.Addr) +loop: + for i := range msgs { + m := &msgs[i] + switch m.Header.Type { + case syscall.NLMSG_DONE: + break loop + case syscall.RTM_NEWADDR: + // struct ifaddrmsg { u8 family, prefixlen, flags, scope; u32 index }. + if len(m.Data) < syscall.SizeofIfAddrmsg { + return nil, false + } + family := m.Data[0] + index := int(binary.NativeEndian.Uint32(m.Data[4:8])) + attrs, err := syscall.ParseNetlinkRouteAttr(m) + if err != nil { + return nil, false + } + var addr, local []byte + for _, at := range attrs { + switch at.Attr.Type { + case syscall.IFA_ADDRESS: + if addr == nil { + addr = at.Value + } + case syscall.IFA_LOCAL: + if local == nil { + local = at.Value + } + } + } + raw := addr + if local != nil { + raw = local + } + want := 0 + switch family { + case syscall.AF_INET: + want = 4 + case syscall.AF_INET6: + want = 16 + } + if want == 0 || len(raw) < want { + continue + } + if p, ok := netip.AddrFromSlice(raw[:want]); ok { + out[index] = append(out[index], p.Unmap()) + } + } + } + return out, true +} + // satAdd is a+b, held at the largest uint64. func satAdd(a, b uint64) uint64 { if s := a + b; s >= a { diff --git a/internal/platform/nicrx_ribparse_885_linux_test.go b/internal/platform/nicrx_ribparse_885_linux_test.go new file mode 100644 index 000000000..35beee742 --- /dev/null +++ b/internal/platform/nicrx_ribparse_885_linux_test.go @@ -0,0 +1,201 @@ +//go:build linux + +package platform + +import ( + "encoding/binary" + "net/netip" + "reflect" + "slices" + "syscall" + "testing" +) + +// The parser of the RTM_GETADDR dump (celeris#885 item 53), against dumps built +// byte by byte: the host's own dump (TestOSIfaceAddrsIsOneAddressDump885) has +// no point-to-point link, no mapped address and no truncated message. + +// attr885 is one rtattr: len, type, payload, padded to 4 bytes. +func attr885(typ uint16, payload []byte) []byte { + b := make([]byte, 4, 4+len(payload)+3) + binary.NativeEndian.PutUint16(b[0:], uint16(4+len(payload))) + binary.NativeEndian.PutUint16(b[2:], typ) + b = append(b, payload...) + for len(b)%4 != 0 { + b = append(b, 0) + } + return b +} + +// msg885 is one netlink message of a dump: a header and data. +func msg885(typ uint16, data []byte) []byte { + n := syscall.NLMSG_HDRLEN + len(data) + b := make([]byte, syscall.NLMSG_HDRLEN, (n+3)&^3) + binary.NativeEndian.PutUint32(b[0:], uint32(n)) + binary.NativeEndian.PutUint16(b[4:], typ) + b = append(b, data...) + for len(b)%4 != 0 { + b = append(b, 0) + } + return b +} + +// newaddr885 is an RTM_NEWADDR message for the interface index. +func newaddr885(family byte, index uint32, attrs ...[]byte) []byte { + d := make([]byte, syscall.SizeofIfAddrmsg) + d[0] = family + d[1] = 24 + binary.NativeEndian.PutUint32(d[4:], index) + for _, a := range attrs { + d = append(d, a...) + } + return msg885(syscall.RTM_NEWADDR, d) +} + +func ip885(s string) []byte { + a := netip.MustParseAddr(s) + return a.AsSlice() +} + +func done885() []byte { return msg885(syscall.NLMSG_DONE, make([]byte, 4)) } + +func cat885(parts ...[]byte) []byte { + var out []byte + for _, p := range parts { + out = append(out, p...) + } + return out +} + +// unpadded885 is an RTM_NEWADDR whose last attribute carries n payload bytes +// and is not padded to 4, with the message length counting exactly its bytes. +func unpadded885(n int) []byte { + d := make([]byte, syscall.SizeofIfAddrmsg) + d[0] = syscall.AF_INET + binary.NativeEndian.PutUint32(d[4:], 2) + d = append(d, attr885(syscall.IFA_ADDRESS, ip885("10.0.0.1"))...) + last := make([]byte, 4+n) + binary.NativeEndian.PutUint16(last[0:], uint16(4+n)) + binary.NativeEndian.PutUint16(last[2:], syscall.IFA_LABEL) + d = append(d, last...) + return msg885(syscall.RTM_NEWADDR, d) +} + +// viaDump885 reads rib the way Listen does: as the dump netlinkRIB returned. +func viaDump885(rib []byte) (map[int][]netip.Addr, bool) { + old := netlinkRIB + netlinkRIB = func(int, int) ([]byte, error) { return rib, nil } + defer func() { netlinkRIB = old }() + return addrsByIndex() +} + +// TestOSIfaceAddrsFallsBackOnAnUnparseableDump885: a dump the parser cannot +// read gives the table the per-interface calls give, and does not panic. +func TestOSIfaceAddrsFallsBackOnAnUnparseableDump885(t *testing.T) { + want, err := refIfaceAddrs885() + if err != nil { + t.Skipf("net.Interfaces: %v", err) + } + old := netlinkRIB + t.Cleanup(func() { netlinkRIB = old }) + netlinkRIB = func(int, int) ([]byte, error) { return unpadded885(5), nil } + got, err := osIfaceAddrs() + if err != nil { + t.Fatalf("osIfaceAddrs: %v", err) + } + if !reflect.DeepEqual(got, want) { + t.Errorf("fallback table differs from the per-interface one:\n got %+v\n want %+v", got, want) + } +} + +func TestParseAddrRIB885(t *testing.T) { + a := netip.MustParseAddr + for _, tc := range []struct { + name string + rib []byte + want map[int][]netip.Addr + ok bool + }{ + {"ipv4", cat885(newaddr885(syscall.AF_INET, 2, attr885(syscall.IFA_ADDRESS, ip885("10.0.0.1")), attr885(syscall.IFA_LOCAL, ip885("10.0.0.1"))), done885()), + map[int][]netip.Addr{2: {a("10.0.0.1")}}, true}, + // A point-to-point link: IFA_ADDRESS is the peer, IFA_LOCAL is ours, in + // either order. + {"p2p-address-first", cat885(newaddr885(syscall.AF_INET, 5, attr885(syscall.IFA_ADDRESS, ip885("10.0.0.9")), attr885(syscall.IFA_LOCAL, ip885("10.0.0.1"))), done885()), + map[int][]netip.Addr{5: {a("10.0.0.1")}}, true}, + {"p2p-local-first", cat885(newaddr885(syscall.AF_INET, 5, attr885(syscall.IFA_LOCAL, ip885("10.0.0.1")), attr885(syscall.IFA_ADDRESS, ip885("10.0.0.9"))), done885()), + map[int][]netip.Addr{5: {a("10.0.0.1")}}, true}, + // Only IFA_ADDRESS (every IPv6 address, an unnumbered IPv4 link). + {"address-only-v6", cat885(newaddr885(syscall.AF_INET6, 3, attr885(syscall.IFA_ADDRESS, ip885("2001:db8::1"))), done885()), + map[int][]netip.Addr{3: {a("2001:db8::1")}}, true}, + // Order is the kernel's, per interface, across interleaved interfaces. + {"interleaved", cat885( + newaddr885(syscall.AF_INET, 2, attr885(syscall.IFA_ADDRESS, ip885("10.0.0.1"))), + newaddr885(syscall.AF_INET, 3, attr885(syscall.IFA_ADDRESS, ip885("10.1.0.1"))), + newaddr885(syscall.AF_INET6, 2, attr885(syscall.IFA_ADDRESS, ip885("fe80::1"))), + newaddr885(syscall.AF_INET, 2, attr885(syscall.IFA_ADDRESS, ip885("10.0.0.2"))), done885()), + map[int][]netip.Addr{2: {a("10.0.0.1"), a("fe80::1"), a("10.0.0.2")}, 3: {a("10.1.0.1")}}, true}, + // An IPv4-mapped IPv6 address is unmapped, as the table did. + {"mapped", cat885(newaddr885(syscall.AF_INET6, 4, attr885(syscall.IFA_ADDRESS, ip885("::ffff:1.2.3.4"))), done885()), + map[int][]netip.Addr{4: {a("1.2.3.4")}}, true}, + // Not an IP family, a payload too short for its family, no address + // attribute: nothing for that message, and no failure. + {"other-family", cat885(newaddr885(syscall.AF_PACKET, 2, attr885(syscall.IFA_ADDRESS, ip885("10.0.0.1"))), done885()), map[int][]netip.Addr{}, true}, + {"short-payload", cat885(newaddr885(syscall.AF_INET6, 2, attr885(syscall.IFA_ADDRESS, ip885("10.0.0.1"))), done885()), map[int][]netip.Addr{}, true}, + {"no-address-attribute", cat885(newaddr885(syscall.AF_INET, 2, attr885(syscall.IFA_LABEL, []byte("eth0\x00"))), done885()), map[int][]netip.Addr{}, true}, + // Messages of another type are skipped; the dump ends at NLMSG_DONE. + {"foreign-type-and-after-done", cat885(msg885(syscall.RTM_NEWLINK, make([]byte, 16)), + newaddr885(syscall.AF_INET, 2, attr885(syscall.IFA_ADDRESS, ip885("10.0.0.1"))), done885(), + newaddr885(syscall.AF_INET, 2, attr885(syscall.IFA_ADDRESS, ip885("10.0.0.2")))), + map[int][]netip.Addr{2: {a("10.0.0.1")}}, true}, + {"empty", nil, map[int][]netip.Addr{}, true}, + // A message that cannot be read makes the whole table fall back. + {"short-ifaddrmsg", cat885(msg885(syscall.RTM_NEWADDR, make([]byte, 4)), done885()), nil, false}, + {"length-past-the-buffer", func() []byte { + m := newaddr885(syscall.AF_INET, 2, attr885(syscall.IFA_ADDRESS, ip885("10.0.0.1"))) + binary.NativeEndian.PutUint32(m[0:], 4096) + return m + }(), nil, false}, + // A final attribute whose length is not a multiple of 4 and that has no + // padding: the std function that walks the attributes slices past the end + // of the message and panics. A kernel dump pads every attribute, so this + // is not a dump the kernel makes, but the table is built at Listen and + // must fall back, not panic (review of #1127, minor 1). The first row is + // the fuzz crasher of that review. + {"unpadded-final-attribute-fuzz-crasher", []byte("1\x00\x00\x00\x14\x00000000000000000000\b\x00000000\b\x00000000\t\x000000000000"), nil, false}, + {"unpadded-final-attribute-1", unpadded885(1), nil, false}, + {"unpadded-final-attribute-2", unpadded885(2), nil, false}, + {"unpadded-final-attribute-3", unpadded885(3), nil, false}, + {"unpadded-final-attribute-5", unpadded885(5), nil, false}, + {"unpadded-final-attribute-6", unpadded885(6), nil, false}, + {"unpadded-final-attribute-7", unpadded885(7), nil, false}, + } { + t.Run(tc.name, func(t *testing.T) { + got, ok := viaDump885(tc.rib) + if ok != tc.ok { + t.Fatalf("ok = %v, want %v", ok, tc.ok) + } + if ok && !reflect.DeepEqual(got, tc.want) { + t.Errorf("parseAddrRIB = %v, want %v", got, tc.want) + } + }) + } +} + +// TestHostDumpParses885: the host's own RTM_GETADDR dump is read and parsed (the +// table is not made the old way, one Addrs call per interface, which the +// equivalence test would also accept), and holds loopback's 127.0.0.1. +func TestHostDumpParses885(t *testing.T) { + byIndex, ok := addrsByIndex() + if !ok { + t.Fatal("the host's address dump could not be read or parsed: osIfaceAddrs falls back to one dump per interface") + } + found := false + for _, addrs := range byIndex { + if slices.Contains(addrs, netip.MustParseAddr("127.0.0.1")) { + found = true + } + } + if !found { + t.Errorf("127.0.0.1 is in no interface's addresses: %v", byIndex) + } +} diff --git a/test/server/large_response_linux_test.go b/test/server/large_response_linux_test.go index 39a52181a..1302c6101 100644 --- a/test/server/large_response_linux_test.go +++ b/test/server/large_response_linux_test.go @@ -15,10 +15,12 @@ import ( "path/filepath" "strconv" "strings" + "sync" "testing" "time" "github.com/goceleris/celeris" + "github.com/goceleris/celeris/internal/conn" ) // Tests for celeris#761. On epoll and io_uring the per-connection write @@ -412,20 +414,34 @@ func TestBackloggedPeerIsClosed(t *testing.T) { } defer func() { _ = raw.Close() }() _ = raw.(*net.TCPConn).SetReadBuffer(64 << 10) + refused := watchBacklogRefusal890(t, raw.LocalAddr().String()) if _, err := io.WriteString(raw, strings.Repeat("GET /big HTTP/1.1\r\nHost: x\r\n\r\n", requests)); err != nil { t.Fatal(err) } - // The server stages what it will before the client reads. Two - // conditions are visible from here: the server has read every request - // (receive queue empty) and its kernel send queue has stopped growing. - // Neither is exact: the stop point is overBacklogH1 (pendingBytes > - // maxPendingBytes), which is unexported, and the kernel send queue - // plateaus before the engine stops staging, so a short fixed grace - // follows (celeris#890 item 22, open question: it needs an engine-side - // observable; was a 300 ms sleep). + // The server stages what it will before the client reads, and the + // native engines then refuse the request that finds the unsent + // responses over the 4 MiB cap (maxPendingBytes, maxSendQueueBytes: + // unexported). That refusal is the server-side event to wait for + // (conn.BacklogRefusedHook, celeris#890 item 22; it was a fixed 100 + // ms after the kernel send queue had plateaued): once it has fired + // the engine closes the connection when what is staged has gone out, + // whatever the client does next, so the client may start reading. + // The hook is keyed on the exact peer address string (the client's + // LocalAddr): if the server ever formats the peer differently (a + // v4-mapped peer on a dual-stack listener), this wait times out + // after 10 s and fails loudly; it cannot pass falsely. + // std has no cap to refuse at: its handler blocks on the socket. waitServerReadAll761(t, raw, addr) - waitServerSendQueued761(raw, addr) - time.Sleep(backlogGrace761) + if e.eng != celeris.Std { + select { + case <-refused: + case <-time.After(10 * time.Second): + t.Fatalf("%s/%s: no request was refused within 10s of the server reading all %d requests: "+ + "the engine did not apply its back-pressure cap to a client that read nothing, "+ + "or conn.BacklogRefusedHook is not called where a request is refused", + e.name, sh.name, requests) + } + } br := bufio.NewReaderSize(idleConn761{raw}, 64<<10) whole := 0 for whole < requests { @@ -661,10 +677,33 @@ func hexPort761(field string) int { return int(p) } -// backlogGrace761 is the fixed wait that follows the server-side conditions in -// TestBackloggedPeerIsClosed, for the engine to finish staging after the -// kernel send queue has plateaued. -const backlogGrace761 = 100 * time.Millisecond +// backlogWatch890 maps the peer address of a connection a test watches to the +// channel its refusal is reported on. One hook serves every test; it is +// installed on first use and tells only the watched connection's test. +var ( + backlogWatch890 sync.Map // remote address string -> chan struct{} (buffer 1) + backlogHookOnce890 sync.Once + backlogHookSeen890 = func(remote string) { + if v, ok := backlogWatch890.Load(remote); ok { + select { + case v.(chan struct{}) <- struct{}{}: + default: + } + } + } +) + +// watchBacklogRefusal890 returns a channel that receives when the engine +// refuses a request on the connection whose client end is clientAddr (the +// server sees it as that remote address). +func watchBacklogRefusal890(t *testing.T, clientAddr string) <-chan struct{} { + t.Helper() + backlogHookOnce890.Do(func() { conn.BacklogRefusedHook.Store(&backlogHookSeen890) }) + ch := make(chan struct{}, 1) + backlogWatch890.Store(clientAddr, ch) + t.Cleanup(func() { backlogWatch890.Delete(clientAddr) }) + return ch +} // waitServerReadAll761 waits until the server has read everything sent on // raw so far: its receive queue is empty. Used where a test sends a request diff --git a/test/server/zoned_listen_885_linux_test.go b/test/server/zoned_listen_885_linux_test.go new file mode 100644 index 000000000..f19cd4b75 --- /dev/null +++ b/test/server/zoned_listen_885_linux_test.go @@ -0,0 +1,103 @@ +//go:build linux + +package celeris_test + +import ( + "bufio" + "context" + "io" + "net" + "net/http" + "strconv" + "strings" + "testing" + "time" + + "github.com/goceleris/celeris" +) + +// TestZonedListenAddressServesOnEveryEngine885 (checklist #885 item 54): a +// Config.Addr with an IPv6 zone ("[fe80::1234%dz0]:port"), which net.Listen and +// so the std engine bind, starts the native engines too, through the exported +// API (celeris.New, Start), not only through the engine packages: parseAddr of +// epoll and io_uring used to refuse the zone ("invalid addr"). Adaptive starts +// both of them on the address, with its port hold beside them. The server must +// answer a request sent to the zoned address and report that address, zone +// included, as std does. +// +// It skips without an up, non-loopback interface that has an fe80:: address, +// and fails under CELERIS_REQUIRE_ZONED_PEER=1 (see zonedTarget890). +func TestZonedListenAddressServesOnEveryEngine885(t *testing.T) { + host := zonedTarget890(t) + arms := append(keptArms(t), keptArm{"adaptive", celeris.Adaptive, false}) + for _, arm := range arms { + t.Run(arm.name, func(t *testing.T) { + // A free port on the zoned address: std binds it, then lets it go. + probe, err := net.Listen("tcp", net.JoinHostPort(host, "0")) + if err != nil { + t.Fatalf("std cannot listen on %s: %v", host, err) + } + port := probe.Addr().(*net.TCPAddr).Port + _ = probe.Close() + addr := net.JoinHostPort(host, strconv.Itoa(port)) + + s := celeris.New(celeris.Config{Engine: arm.engine, AsyncHandlers: arm.async, Addr: addr}) + s.GET("/", func(c *celeris.Context) error { return c.String(http.StatusOK, "zoned") }) + done := make(chan error, 1) + go func() { done <- s.Start() }() + returned := false // Start has returned and its result was read below + t.Cleanup(func() { + if returned { + return + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _ = s.Shutdown(ctx) + select { + case <-done: + case <-time.After(15 * time.Second): + t.Errorf("Start did not return within 15s of Shutdown") + } + }) + deadline := time.Now().Add(15 * time.Second) + for s.Addr() == nil { + select { + case err := <-done: + returned = true + t.Fatalf("Start on %s returned before the server was ready: %v", addr, err) + case <-time.After(5 * time.Millisecond): + } + if time.Now().After(deadline) { + t.Fatalf("no address published for %s within 15s", addr) + } + } + // The native engines name the zone of the address they bound + // (bindiag.BoundAddr). std's own Addr() is not asserted: it is the + // oracle here, and on a hosted runner it prints the address without + // the zone (CI run 40425220210), which is not this change's to fix. + if got := s.Addr().String(); got != addr && arm.engine != celeris.Std { + t.Errorf("Addr() = %q, want %q (std's form, the zone included)", got, addr) + } else if got != addr { + t.Logf("std reports %q for %q", got, addr) + } + c, err := net.DialTimeout("tcp", addr, 3*time.Second) + if err != nil { + t.Fatalf("dial %s: %v", addr, err) + } + defer func() { _ = c.Close() }() + _ = c.SetDeadline(time.Now().Add(10 * time.Second)) + if _, err := io.WriteString(c, "GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"); err != nil { + t.Fatalf("write: %v", err) + } + resp, err := http.ReadResponse(bufio.NewReader(c), nil) + if err != nil { + t.Fatalf("read: %v", err) + } + body, _ := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if resp.StatusCode != http.StatusOK || !strings.Contains(string(body), "zoned") { + t.Errorf("answered %d %q, want 200 zoned", resp.StatusCode, body) + } + }) + } +}