From ca476f55799f25d3b35c5bb02fab165444e5f54e Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 01:06:10 +0100 Subject: [PATCH 1/7] engines: tests for the cross-engine small fixes, red on main (v1.6.0, Refs #885 #886 #889 #890) Failing-first half of the lane. Every behaviour change below has a test here that fails on baa43d99; the fixes follow in the next commit. 886.23 has its own test and fix commits at the end, so that it can be dropped on its own. - 889.48: OnDetach panic injected right after the count (detachPostCountHook, a no-op seam called between the rollback's snapshot and the place the rollback is registered); io_uring rollback test also asserts the detachedConns gauge. - 890.22: TestBackloggedPeerIsClosed waits for the engine's refusal (conn.BacklogRefusedHook, defined here, called in the next commit) instead of sleeping. - 885.54: parseAddr of a zoned IPv6 host, Listen on a zoned link-local address on both engines, and a Server start on one for every engine including adaptive. - 886.24: BeginPauseAccept must signal the io_uring workers' wake eventfd; pause-to-closed lag measurements for both engines. - 885.31: the MaxEvents split follows the nodes the allowed set meets (splitMaxEvents, detectNUMA seam; the split still uses the host count here). - 885.53: osIfaceAddrs is one RTM_GETADDR dump (netlinkRIB seam). --- internal/conn/backlog_seam_890.go | 17 ++ internal/engine/epoll/abort_seam_844.go | 7 + .../epoll/detach_rollback_889_linux_test.go | 16 +- internal/engine/epoll/engine.go | 9 +- internal/engine/epoll/loop.go | 3 + .../epoll/maxevents_nodes_885_linux_test.go | 90 ++++++++ .../epoll/parse_addr_mapped_engcfg_test.go | 5 +- .../epoll/parse_addr_zone_885_linux_test.go | 150 +++++++++++++ .../engine/epoll/pause_wake_886_linux_test.go | 100 +++++++++ internal/engine/epoll/pin_linux.go | 17 ++ internal/engine/iouring/abort_seam_844.go | 7 + .../iouring/detach_rollback_889_linux_test.go | 22 +- .../iouring/parse_addr_mapped_engcfg_test.go | 5 +- .../iouring/parse_addr_zone_885_linux_test.go | 150 +++++++++++++ .../iouring/pause_wake_886_linux_test.go | 200 ++++++++++++++++++ internal/engine/iouring/worker.go | 3 + .../platform/nicrx_addrs_885_linux_test.go | 110 ++++++++++ internal/platform/nicrx_linux.go | 5 + test/server/large_response_linux_test.go | 63 ++++-- test/server/zoned_listen_885_linux_test.go | 103 +++++++++ 20 files changed, 1047 insertions(+), 35 deletions(-) create mode 100644 internal/conn/backlog_seam_890.go create mode 100644 internal/engine/epoll/maxevents_nodes_885_linux_test.go create mode 100644 internal/engine/epoll/parse_addr_zone_885_linux_test.go create mode 100644 internal/engine/epoll/pause_wake_886_linux_test.go create mode 100644 internal/engine/iouring/parse_addr_zone_885_linux_test.go create mode 100644 internal/engine/iouring/pause_wake_886_linux_test.go create mode 100644 internal/platform/nicrx_addrs_885_linux_test.go create mode 100644 test/server/zoned_listen_885_linux_test.go diff --git a/internal/conn/backlog_seam_890.go b/internal/conn/backlog_seam_890.go new file mode 100644 index 000000000..50c7c7c80 --- /dev/null +++ b/internal/conn/backlog_seam_890.go @@ -0,0 +1,17 @@ +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)] 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/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..010f8a75e 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) diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go index 6bc92dfab..29fe73dae 100644 --- a/internal/engine/epoll/loop.go +++ b/internal/engine/epoll/loop.go @@ -2868,6 +2868,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() 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..2e0503038 --- /dev/null +++ b/internal/engine/epoll/parse_addr_zone_885_linux_test.go @@ -0,0 +1,150 @@ +//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.zoneToInt): the interface's index, +// else the number, else 0. +func zoneID885(zone string) uint32 { + if zone == "" { + return 0 + } + if ifi, err := net.InterfaceByName(zone); err == nil { + return uint32(ifi.Index) + } + n, err := strconv.ParseUint(zone, 10, 32) + if err != nil { + return 0 + } + 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)}, + {"[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..55fadc539 100644 --- a/internal/engine/epoll/pin_linux.go +++ b/internal/engine/epoll/pin_linux.go @@ -17,6 +17,23 @@ 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 on a host with topo: the +// configured figure divided by the number of NUMA nodes, never under 64. +func splitMaxEvents(maxEvents int, topo platform.NUMATopology, allowed []int) int { + _ = allowed + if topo.NumNodes > 1 { + maxEvents /= topo.NumNodes + if maxEvents < 64 { + maxEvents = 64 + } + } + return maxEvents +} + // 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/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/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/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..4e6fa6521 --- /dev/null +++ b/internal/engine/iouring/parse_addr_zone_885_linux_test.go @@ -0,0 +1,150 @@ +//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.zoneToInt): the interface's index, +// else the number, else 0. +func zoneID885(zone string) uint32 { + if zone == "" { + return 0 + } + if ifi, err := net.InterfaceByName(zone); err == nil { + return uint32(ifi.Index) + } + n, err := strconv.ParseUint(zone, 10, 32) + if err != nil { + return 0 + } + 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)}, + {"[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/worker.go b/internal/engine/iouring/worker.go index 22cb9a1b1..6395a78cc 100644 --- a/internal/engine/iouring/worker.go +++ b/internal/engine/iouring/worker.go @@ -3421,6 +3421,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) { 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..b8dfe2d3c 100644 --- a/internal/platform/nicrx_linux.go +++ b/internal/platform/nicrx_linux.go @@ -13,6 +13,7 @@ import ( "slices" "strconv" "strings" + "syscall" ) // Reading the NICs behind a listener (celeris#1036 item 1). The rx-port rule @@ -122,6 +123,10 @@ func (osNICFS) exists(name string) bool { return err == nil } +// 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 with the net package // (one netlink dump per interface; no cgo and no ip binary). func osIfaceAddrs() ([]ifaceAddr, error) { diff --git a/test/server/large_response_linux_test.go b/test/server/large_response_linux_test.go index 39a52181a..4a8ca87e5 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,30 @@ 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. + // 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 +673,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) + } + }) + } +} From 2fe0f01c0442f0ea1329599994b25ee265d00e7d Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 01:06:10 +0100 Subject: [PATCH 2/7] engines: cross-engine small fixes (v1.6.0, Refs #885 #886 #889 #890) The fixes for the tests of the previous commit. - 889.48: OnDetach's rollback is registered right after the snapshot, before detachMu is allocated and the count is made, and the promoted-path gate is decided there from the snapshot's detachMu; the sync path's reliance on the rollback leaving detachCounted false is commented. Both engines. - 890.22: conn.BacklogRefusedHook is called where a request is refused for the back-pressure cap, and TestBackloggedPeerIsClosed waits for it instead of sleeping 100 ms. - 885.54: parseAddr reads a zoned IPv6 host with netip and sets the scope id as net.Listen does (bindiag.ZoneID). Both engines. - 886.24: io_uring's BeginPauseAccept signals every worker's wake eventfd after it stores the flag. epoll's is not changed: its listening loop waits at most 4 ms (measured 0.8 to 5.9 ms). The 2-worker pause test joins the CI step that runs the multi-worker listener tests by name (ci.yml). - 885.31: epoll divides MaxEvents by the NUMA nodes the plan's allowed set meets, not the host's node count. - 885.53: osIfaceAddrs reads all addresses from one RTM_GETADDR dump, mapped by interface index, instead of one dump per interface (311 interfaces: 21.6 ms to 0.94 ms, 313 netlink requests to 3). - 885.14: h2PoolSettled's doc says what becomes of a refused GOAWAY. --- .github/workflows/ci.yml | 8 +- internal/conn/backlog_seam_890.go | 12 ++ internal/conn/h1.go | 1 + internal/conn/h1_backlog_bench_890_test.go | 57 +++++++ internal/engine/epoll/engine.go | 5 +- internal/engine/epoll/loop.go | 113 ++++++++++---- internal/engine/epoll/pin_linux.go | 35 ++++- internal/engine/internal/bindiag/bindiag.go | 20 +++ internal/engine/iouring/engine.go | 17 +- internal/engine/iouring/worker.go | 118 +++++++++----- internal/platform/nicrx_linux.go | 118 ++++++++++++-- .../platform/nicrx_ribparse_885_linux_test.go | 147 ++++++++++++++++++ 12 files changed, 552 insertions(+), 99 deletions(-) create mode 100644 internal/conn/h1_backlog_bench_890_test.go create mode 100644 internal/platform/nicrx_ribparse_885_linux_test.go 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 index 50c7c7c80..a81b1a533 100644 --- a/internal/conn/backlog_seam_890.go +++ b/internal/conn/backlog_seam_890.go @@ -15,3 +15,15 @@ import "sync/atomic" // 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/engine.go b/internal/engine/epoll/engine.go index 010f8a75e..4ff50e0ee 100644 --- a/internal/engine/epoll/engine.go +++ b/internal/engine/epoll/engine.go @@ -337,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/loop.go b/internal/engine/epoll/loop.go index 29fe73dae..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 @@ -2980,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 @@ -5335,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 @@ -5605,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) @@ -5614,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/pin_linux.go b/internal/engine/epoll/pin_linux.go index 55fadc539..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" ) @@ -21,12 +22,21 @@ var planWorkerCPUs = platform.PlanWorkerCPUsFor // A variable so a test can stand in a topology. var detectNUMA = platform.DetectNUMA -// splitMaxEvents is the MaxEvents a loop gets on a host with topo: the -// configured figure divided by the number of NUMA nodes, never under 64. +// 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 { - _ = allowed - if topo.NumNodes > 1 { - maxEvents /= topo.NumNodes + nodes := topo.NumNodes + if n := nodesMet(topo, allowed); n > 0 { + nodes = n + } + if nodes > 1 { + maxEvents /= nodes if maxEvents < 64 { maxEvents = 64 } @@ -34,6 +44,21 @@ func splitMaxEvents(maxEvents int, topo platform.NUMATopology, allowed []int) in 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/bindiag.go b/internal/engine/internal/bindiag/bindiag.go index 5f097b01f..caf79b9f3 100644 --- a/internal/engine/internal/bindiag/bindiag.go +++ b/internal/engine/internal/bindiag/bindiag.go @@ -166,6 +166,26 @@ func SockaddrString(fd int, sa unix.Sockaddr) string { return "" } +// ZoneID is the scope id of an IPv6 zone as net.Listen reads it (net.zoneToInt): +// the index of the interface of that name, else the zone as a number, 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, err := strconv.ParseUint(zone, 10, 32) + if err != nil { + return 0 + } + 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/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/worker.go b/internal/engine/iouring/worker.go index 6395a78cc..20467b672 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" @@ -3397,6 +3398,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{} @@ -3548,46 +3595,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 @@ -7773,7 +7787,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 @@ -8400,6 +8429,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) @@ -8409,7 +8453,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/nicrx_linux.go b/internal/platform/nicrx_linux.go index b8dfe2d3c..3a4469e66 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" @@ -127,13 +128,21 @@ func (osNICFS) exists(name string) bool { // so a test can count the dumps (celeris#885 item 53). var netlinkRIB = syscall.NetlinkRIB -// osIfaceAddrs lists the interfaces and their addresses with the net package -// (one netlink dump per interface; no cgo and no ip binary). +// 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 { @@ -144,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) @@ -165,6 +178,81 @@ 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() (map[int][]netip.Addr, bool) { + 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. +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..f4d413bb9 --- /dev/null +++ b/internal/platform/nicrx_ribparse_885_linux_test.go @@ -0,0 +1,147 @@ +//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 +} + +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}, + } { + t.Run(tc.name, func(t *testing.T) { + got, ok := parseAddrRIB(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) + } +} From 87f09614d8e3e0c12d67a6cc1171709ae3d36bf0 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 01:06:10 +0100 Subject: [PATCH 3/7] engines: comments for the items that need no code (v1.6.0, Refs #885 #886) No behaviour change. - 886.9: endH1Verdict of both engines states the OnError difference on a conn with no detachMu (epoll delivers, io_uring does not), how a sync conn gets into that state, and that the rule is an open decision. The io_uring side is pinned by two tests (the epoll side already was). - U-12: the rx-port rule's doc says when it does not engage (since-boot counters at the first Listen; an evenly split network) and that it is left as it is. Godoc only; nothing in README. --- internal/engine/epoll/h1verdict.go | 17 +++++- internal/engine/iouring/h1verdict.go | 16 +++++ .../onerror_nodetachmu_886_linux_test.go | 60 +++++++++++++++++++ internal/platform/cpuplan_rxfree_linux.go | 17 ++++++ 4 files changed, 109 insertions(+), 1 deletion(-) create mode 100644 internal/engine/iouring/onerror_nodetachmu_886_linux_test.go 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/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/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. From 2e6194a257cf60eaf83b6230ea0e807fd788a4df Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 01:06:10 +0100 Subject: [PATCH 4/7] engines: 886.23 test, red on main: an h2c upgrade carry counts no request (v1.6.0, Refs #886) The four h2c-upgrade-carry replay tests expect 0 requests and read the 101 off the client first (expectCount886 returns at once on 0 and only holds 400 ms, so without the 101 it could pass before the replay ran). They fail on baa43d99 (the replay counts 1). Kept apart from the rest of the lane on purpose: the fix touches the two reqBatch++ lines of epoll/adopt.go that #1122 rewrites (recvCounted), so this pair is the one to drop or rebase. --- .../adopt_request_count_886_linux_test.go | 40 +++++++++++++------ .../adopt_request_count_886_linux_test.go | 40 +++++++++++++------ 2 files changed, 56 insertions(+), 24 deletions(-) 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/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) } From e39fd57891906ac9aeeea5136a7e71916cc6b847 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 01:06:10 +0100 Subject: [PATCH 5/7] engines: an h2c upgrade carry counts no request on the transplant replay (v1.6.0, Refs #886) 886.23: the recv path (drainRead, handleRecv) returns on the upgrade verdict before it counts, so a conn that upgraded on a recv counted its upgrade request as none, while the replay of a carry that is an upgrade request counted one. The replay now counts none, at the four sites (epoll sync and async, io_uring sync and async). The other rule needs drainRead/handleRecv. h2 conns count no request on either engine, so BytesPerReq moves by one request per upgraded conn that was replayed. On epoll the condition sits beside #1122's: `!recvCounted && !errors.Is(perr, conn.ErrUpgradeH2C)`. A reclaim of an upgrade carry keeps the count drainRead made when it read the bytes (it counts before it transplants), so that one case still counts 1; it needs drainRead. --- internal/engine/epoll/adopt.go | 27 ++++++++++++++------------- internal/engine/iouring/transplant.go | 27 ++++++++++++++++----------- 2 files changed, 30 insertions(+), 24 deletions(-) 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/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) From b1f1246cb0fb1edd809aaa1ab79273243c4b05ab Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 02:33:40 +0100 Subject: [PATCH 6/7] engines: tests for the review of #1127, red: unpadded dump attribute, malformed zone (v1.6.0, Refs #885) - 885.53: a dump whose last attribute is not padded to 4 makes the std attribute walker panic (slice bounds out of range [12:9]); rows through addrsByIndex for the review's fuzz crasher and six hand-built ones, and TestOSIfaceAddrsFallsBackOnAnUnparseableDump885. The binary panics before the fix. - 885.54: ZoneID rows for a malformed zone ("12abc" is 12, a number of 0xFFFFFF or more saturates), in bindiag and in both engines' parseAddr table, whose oracle now follows net's rule. --- .../epoll/parse_addr_zone_885_linux_test.go | 19 +++++-- .../internal/bindiag/addr_format_test.go | 31 ++++++++++ .../iouring/parse_addr_zone_885_linux_test.go | 19 +++++-- .../platform/nicrx_ribparse_885_linux_test.go | 56 ++++++++++++++++++- 4 files changed, 114 insertions(+), 11 deletions(-) diff --git a/internal/engine/epoll/parse_addr_zone_885_linux_test.go b/internal/engine/epoll/parse_addr_zone_885_linux_test.go index 2e0503038..ed09c31b6 100644 --- a/internal/engine/epoll/parse_addr_zone_885_linux_test.go +++ b/internal/engine/epoll/parse_addr_zone_885_linux_test.go @@ -20,8 +20,10 @@ import ( // 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.zoneToInt): the interface's index, -// else the number, else 0. +// 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 @@ -29,9 +31,12 @@ func zoneID885(zone string) uint32 { if ifi, err := net.InterfaceByName(zone); err == nil { return uint32(ifi.Index) } - n, err := strconv.ParseUint(zone, 10, 32) - if err != nil { - return 0 + 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) } @@ -58,6 +63,10 @@ func TestParseAddrReadsAZonedIPv6Host885(t *testing.T) { // 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. 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/iouring/parse_addr_zone_885_linux_test.go b/internal/engine/iouring/parse_addr_zone_885_linux_test.go index 4e6fa6521..772392cc5 100644 --- a/internal/engine/iouring/parse_addr_zone_885_linux_test.go +++ b/internal/engine/iouring/parse_addr_zone_885_linux_test.go @@ -20,8 +20,10 @@ import ( // 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.zoneToInt): the interface's index, -// else the number, else 0. +// 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 @@ -29,9 +31,12 @@ func zoneID885(zone string) uint32 { if ifi, err := net.InterfaceByName(zone); err == nil { return uint32(ifi.Index) } - n, err := strconv.ParseUint(zone, 10, 32) - if err != nil { - return 0 + 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) } @@ -58,6 +63,10 @@ func TestParseAddrReadsAZonedIPv6Host885(t *testing.T) { // 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. diff --git a/internal/platform/nicrx_ribparse_885_linux_test.go b/internal/platform/nicrx_ribparse_885_linux_test.go index f4d413bb9..35beee742 100644 --- a/internal/platform/nicrx_ribparse_885_linux_test.go +++ b/internal/platform/nicrx_ribparse_885_linux_test.go @@ -67,6 +67,47 @@ func cat885(parts ...[]byte) []byte { 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 { @@ -114,9 +155,22 @@ func TestParseAddrRIB885(t *testing.T) { 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 := parseAddrRIB(tc.rib) + got, ok := viaDump885(tc.rib) if ok != tc.ok { t.Fatalf("ok = %v, want %v", ok, tc.ok) } From 26b2c539b78415dbb481eab22a524e1c9c74f8a7 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 02:35:25 +0100 Subject: [PATCH 7/7] engines: the NIC table falls back instead of panicking; ZoneID reads a malformed zone as std does (v1.6.0, Refs #885) Fixes for the tests of the previous commit (review of #1127). - 885.53: addrsByIndex recovers from the std attribute walker's panic on an unpadded final attribute and reports not-ok, so osIfaceAddrs builds the table the old way, as its doc says. A comment says an interrupted dump (NLM_F_DUMP_INTR) is read as net reads it. - 885.54: bindiag.ZoneID reads the digits a zone starts with and saturates at 0xFFFFFF, as net's dtoi does, instead of returning 0 for "12abc" and for a number past 32 bits. The smaller change of the two options (matching std rather than narrowing the godoc). - the backlog test's comment says a peer-address mismatch fails after 10 s and cannot pass falsely. --- internal/engine/internal/bindiag/bindiag.go | 22 ++++++++++++--------- internal/platform/nicrx_linux.go | 15 ++++++++++++-- test/server/large_response_linux_test.go | 4 ++++ 3 files changed, 30 insertions(+), 11 deletions(-) diff --git a/internal/engine/internal/bindiag/bindiag.go b/internal/engine/internal/bindiag/bindiag.go index caf79b9f3..18c35abc4 100644 --- a/internal/engine/internal/bindiag/bindiag.go +++ b/internal/engine/internal/bindiag/bindiag.go @@ -166,12 +166,13 @@ func SockaddrString(fd int, sa unix.Sockaddr) string { return "" } -// ZoneID is the scope id of an IPv6 zone as net.Listen reads it (net.zoneToInt): -// the index of the interface of that name, else the zone as a number, 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. +// 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 @@ -179,9 +180,12 @@ func ZoneID(zone string) uint32 { if ifi, err := net.InterfaceByName(zone); err == nil { return uint32(ifi.Index) } - n, err := strconv.ParseUint(zone, 10, 32) - if err != nil { - return 0 + 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) } diff --git a/internal/platform/nicrx_linux.go b/internal/platform/nicrx_linux.go index 3a4469e66..957d498f9 100644 --- a/internal/platform/nicrx_linux.go +++ b/internal/platform/nicrx_linux.go @@ -182,7 +182,16 @@ func osIfaceAddrs() ([]ifaceAddr, error) { // 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() (map[int][]netip.Addr, bool) { +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 @@ -194,7 +203,9 @@ func addrsByIndex() (map[int][]netip.Addr, bool) { // 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. +// 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 { diff --git a/test/server/large_response_linux_test.go b/test/server/large_response_linux_test.go index 4a8ca87e5..1302c6101 100644 --- a/test/server/large_response_linux_test.go +++ b/test/server/large_response_linux_test.go @@ -426,6 +426,10 @@ func TestBackloggedPeerIsClosed(t *testing.T) { // 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) if e.eng != celeris.Std {