diff --git a/Makefile b/Makefile index 24c71459..d84bc6fa 100644 --- a/Makefile +++ b/Makefile @@ -161,6 +161,10 @@ bin-pkcs11: BUILD_ARGS += -tags pkcs11 bin-pkcs11: CGO_ENABLED = 1 bin-pkcs11: bin +# Build with the pprof debug server (serves on :6060). See startPprofServer. +debug: BUILD_ARGS += -tags debug +debug: bin + bin: go build $(BUILD_ARGS) -ldflags "$(LDFLAGS)" -o ./nebula${NEBULA_CMD_SUFFIX} ${NEBULA_CMD_PATH} go build $(BUILD_ARGS) -ldflags "$(LDFLAGS)" -o ./nebula-cert${NEBULA_CMD_SUFFIX} ./cmd/nebula-cert @@ -280,5 +284,5 @@ smoke-vagrant/%: bin-docker build/%/nebula cd .github/workflows/smoke/ && ./smoke-vagrant.sh $* .FORCE: -.PHONY: all all-linux all-freebsd all-openbsd all-netbsd all-darwin all-windows all-cross-linux all-cross-linux-arm all-cross-linux-mips all-cross-linux-other all-cross-darwin all-cross-windows bench bench-cpu bench-cpu-long bin build-test-mobile e2e e2ev e2evv e2evvv e2evvvv proto release service smoke-docker smoke-docker-race test test-cov-html smoke-vagrant/% +.PHONY: all all-linux all-freebsd all-openbsd all-netbsd all-darwin all-windows all-cross-linux all-cross-linux-arm all-cross-linux-mips all-cross-linux-other all-cross-darwin all-cross-windows bench bench-cpu bench-cpu-long bin debug build-test-mobile e2e e2ev e2evv e2evvv e2evvvv proto release service smoke-docker smoke-docker-race test test-cov-html smoke-vagrant/% .DEFAULT_GOAL := bin diff --git a/examples/config.yml b/examples/config.yml index 4f7fd1e7..4c64f7ab 100644 --- a/examples/config.yml +++ b/examples/config.yml @@ -254,6 +254,20 @@ tun: # Default MTU for every packet, safe setting is (and the default) 1300 for internet based traffic mtu: 1300 + # Linux only. pin_threads pins each tun reader/encrypt OS thread to a single CPU. This keeps every goroutine's + # batched sends flowing through one XPS-selected NIC TX ring, so packets within a flow stay ordered on the wire + # instead of being sprayed across multiple TX rings and reordered. Not reloadable. + #pin_threads: true + + # Linux only. cpu_affinity overrides which CPUs the tun reader threads pin to: a list of CPU IDs, one per routine + # (see the top-level `routines` setting). Lists shorter than `routines` are modulo-cycled across the queues; extra + # entries are ignored. IDs must be within the process's allowed CPU set, so this respects taskset / cgroup cpusets; + # a non-integer or not-allowed entry disables the override and falls back to spreading queues across the allowed + # CPUs. Only meaningful while pin_threads is true. Not reloadable. + #cpu_affinity: + # - 2 + # - 4 + # Route based MTU overrides, you have known vpn ip paths that can support larger MTUs you can increase/decrease them here routes: #- mtu: 8800 @@ -390,6 +404,16 @@ logging: # This setting is reloadable #inactivity_timeout: 10m + # ecn (default true) propagates ECN (Explicit Congestion Notification) across the tunnel per RFC 6040: the inner + # packet's ECN codepoint is copied onto the outer carrier header on encapsulation, and an outer CE ("congestion + # experienced") mark is folded back into the inner header on decapsulation. On linux it additionally stamps + # RTAX_FEATURE_ECN on the routes nebula installs, so the kernel actively negotiates ECN for connections to mesh + # prefixes. Disable this only when an underlay middlebox mangles or clears ECN bits unpredictably. + # This setting is reloadable, BUT flipping it at runtime only updates the datapath (the inner<->outer copy/combine). + # The RTAX_FEATURE_ECN flag on already-installed routes is NOT revisited on reload, so nebula must be restarted for + # the route half of this setting to take effect. + #ecn: true + # Nebula security group configuration firewall: # Action to take when a packet is not allowed by the firewall rules. diff --git a/interface.go b/interface.go index 8b7d96ee..6625ea3f 100644 --- a/interface.go +++ b/interface.go @@ -56,8 +56,13 @@ type InterfaceConfig struct { // CpuAffinity, when non-empty, names the CPUs each TUN reader goroutine // should pin to. Queue i pins to CpuAffinity[i % len(CpuAffinity)] — // shorter lists than `routines` cycle. Empty list keeps the default - // pin-to-(i % NumCPU) behavior. + // pin-to-(i % NumCPU) behavior. Only consulted when PinThreads is true. CpuAffinity []int + // PinThreads controls whether each TUN reader OS thread is pinned to a + // single CPU (via tun.pin_threads, default true). Pinning keeps each + // goroutine's sendmmsg on one XPS-selected NIC TX ring so per-flow + // packets stay ordered on the wire. + PinThreads bool l *slog.Logger } @@ -85,8 +90,13 @@ type Interface struct { closed atomic.Bool // cpuAffinity, when non-empty, names the CPUs each TUN reader goroutine // should pin to. Queue i pins to cpuAffinity[i % len(cpuAffinity)]. - // Empty falls back to the default pin-to-(i % NumCPU) behavior. + // Empty falls back to the default pin-to-(allowed CPU) behavior. + // Only consulted when pinThreads is true. cpuAffinity []int + // pinThreads controls whether listenIn pins each TUN reader OS thread to + // a CPU at all (tun.pin_threads, default true). When false, threads are + // left free to migrate as on stock nebula. + pinThreads bool // ecnEnabled gates RFC 6040 underlay ECN propagation. When true, // inside.go copies the inner ECN onto the outer carrier on encap and // decryptToTun folds outer CE into the inner header on decap. Toggle @@ -223,6 +233,7 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) { connectionManager: c.connectionManager, conntrackCacheTimeout: c.ConntrackCacheTimeout, cpuAffinity: c.CpuAffinity, + pinThreads: c.PinThreads, metricHandshakes: metrics.GetOrRegisterHistogram("handshakes", nil, metrics.NewExpDecaySample(1028, 0.015)), messageMetrics: c.MessageMetrics, @@ -383,13 +394,24 @@ func (f *Interface) listenOut(i int) { func (f *Interface) listenIn(reader tio.Queue, i int) { // Pinning this thread (and goroutine) to a single CPU keeps every sendmmsg from this goroutine going through the - // same TX ring on the nic, so the wire sees per-flow order. - cpu := i % runtime.NumCPU() - if n := len(f.cpuAffinity); n > 0 { - cpu = f.cpuAffinity[i%n] - } - if err := util.PinThreadToCPU(cpu); err != nil { - f.l.Warn("failed to pin tun reader to CPU", "queue", i, "cpu", cpu, "err", err) + // same TX ring on the nic, so the wire sees per-flow order. Skip entirely when tun.pin_threads is false. + if f.pinThreads { + var cpu int + if n := len(f.cpuAffinity); n > 0 { + // Explicit tun.cpu_affinity list wins; parseCpuAffinity already + // validated the entries against the allowed CPU set. + cpu = f.cpuAffinity[i%n] + } else if allowed, err := util.AllowedCPUs(); err == nil && len(allowed) > 0 { + // Default: spread queues across the CPUs we're actually allowed to + // run on. Under a cpuset/taskset mask these aren't 0..NumCPU-1, so + // i % NumCPU would pick unrunnable IDs and every pin would fail. + cpu = allowed[i%len(allowed)] + } else { + cpu = i % runtime.NumCPU() + } + if err := util.PinThreadToCPU(cpu); err != nil { + f.l.Warn("failed to pin tun reader to CPU", "queue", i, "cpu", cpu, "err", err) + } } rejectBuf := make([]byte, mtu) @@ -568,9 +590,12 @@ func (f *Interface) reloadEcn(c *config.C) { initial := c.InitialLoad() if initial || c.HasChanged("tunnels.ecn") { v := c.GetBool("tunnels.ecn", true) - f.ecnEnabled.Store(v) + changed := f.ecnEnabled.Swap(v) != v if !initial { f.l.Info("tunnels.ecn changed", "enabled", v) + if changed { + f.l.Warn("tunnels.ecn datapath toggled, but route-level ECN negotiation (RTAX_FEATURE_ECN) retains its previous state until nebula is restarted", "enabled", v) + } } } } diff --git a/main.go b/main.go index 4646c565..5439b773 100644 --- a/main.go +++ b/main.go @@ -5,11 +5,9 @@ import ( "fmt" "log/slog" "net" - "net/http" - _ "net/http/pprof" "net/netip" - "runtime" "runtime/debug" + "slices" "strings" "time" @@ -36,17 +34,8 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev buildVersion = moduleVersion() } - //todo no merge - pprofServer := &http.Server{Addr: ":6060", Handler: nil} - go func() { - pprofServer.ListenAndServe() - }() - - // Shut down the server when context is cancelled - go func() { - <-ctx.Done() - pprofServer.Shutdown(context.Background()) - }() + // Debug builds (-tags debug) serve pprof on :6060; a no-op otherwise. + startPprofServer(ctx, l) // Print the config if in test, the exit comes later if configTest { @@ -247,6 +236,7 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev punchy: punchy, ConntrackCacheTimeout: conntrackCacheTimeout, CpuAffinity: parseCpuAffinity(c, l, routines), + PinThreads: c.GetBool("tun.pin_threads", true), l: l, } @@ -301,11 +291,16 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev // parseCpuAffinity reads `tun.cpu_affinity` from the config — a list of // integer CPU IDs, one per TUN reader goroutine. Empty / unset returns nil -// (listenIn falls back to its default `i % NumCPU` pinning). Length -// mismatch with `routines` is a warning, not an error: shorter lists are -// modulo-cycled across queues, longer lists' tail is ignored. Invalid -// entries (non-integer, out of range) are also a warning and disable the -// override entirely so we don't silently pin to the wrong CPU. +// (listenIn falls back to spreading queues across the allowed CPU set). +// Length mismatch with `routines` is a warning, not an error: shorter lists +// are modulo-cycled across queues, longer lists' tail is ignored. Invalid +// entries (non-integer, or a CPU ID we're not allowed to run on) are also a +// warning and disable the override entirely so we don't silently pin to the +// wrong CPU. Entries are validated against the process's current affinity +// mask (util.AllowedCPUs) rather than 0..NumCPU-1: under a cgroup cpuset or +// taskset the runnable IDs are frequently not that contiguous range, and +// pinning to an unrunnable ID always fails. If the allowed set can't be +// determined we fall back to a plain non-negative check. func parseCpuAffinity(c *config.C, l *slog.Logger, routines int) []int { raw := c.Get("tun.cpu_affinity") if raw == nil { @@ -316,7 +311,14 @@ func parseCpuAffinity(c *config.C, l *slog.Logger, routines int) []int { l.Warn("tun.cpu_affinity must be a list of integers; ignoring", "value", raw) return nil } - nCPU := runtime.NumCPU() + // allowed is the set of CPU IDs we're actually permitted to run on. A nil + // slice (unsupported platform or lookup error) means "can't tell", so we + // only apply the weaker non-negative check in that case. + allowed, err := util.AllowedCPUs() + if err != nil { + l.Warn("could not determine allowed CPUs; validating tun.cpu_affinity against non-negative only", "error", err) + allowed = nil + } cpus := make([]int, 0, len(rv)) for i, e := range rv { var cpu int @@ -332,9 +334,14 @@ func parseCpuAffinity(c *config.C, l *slog.Logger, routines int) []int { "index", i, "value", e) return nil } - if cpu < 0 || cpu >= nCPU { + if cpu < 0 { l.Warn("tun.cpu_affinity entry out of range; ignoring affinity", - "index", i, "cpu", cpu, "num_cpu", nCPU) + "index", i, "cpu", cpu) + return nil + } + if len(allowed) > 0 && !slices.Contains(allowed, cpu) { + l.Warn("tun.cpu_affinity entry not in allowed CPU set; ignoring affinity", + "index", i, "cpu", cpu, "allowed", allowed) return nil } cpus = append(cpus, cpu) diff --git a/overlay/batch/coalesce_core.go b/overlay/batch/coalesce_core.go index d46addc3..80100a81 100644 --- a/overlay/batch/coalesce_core.go +++ b/overlay/batch/coalesce_core.go @@ -93,8 +93,11 @@ func parseIPPrologue(pkt []byte, wantProto byte) (parsedIP, bool) { // ipHeadersMatch compares the IP portion of two packet header prefixes for // byte-for-byte equality on every field that must be identical across -// coalesced segments. Size/IPID/IPCsum and the 2-bit IP-level ECN field are -// masked out — the appendPayload step merges CE into the seed. +// coalesced segments. Size/IPID/IPCsum are masked out. The full DSCP/ECN +// byte (IPv4 ToS / IPv6 traffic class) is compared, matching Linux kernel +// GRO: segments with differing ECN codepoints must not coalesce, otherwise +// ORing e.g. ECT(0) with ECT(1) would fabricate a false CE (congestion) +// mark or mark a Not-ECT flow as ECN-capable. // // The transport (L4) portion of the header is checked separately by the // per-protocol matcher. @@ -102,11 +105,11 @@ func ipHeadersMatch(a, b []byte, isV6 bool) bool { if isV6 { // IPv6: byte 0 = version/TC[7:4], byte 1 = TC[3:0]/flow[19:16], // bytes [2:4] = flow[15:0], [6:8] = next_hdr/hop, [8:40] = src+dst. - // ECN lives in TC[1:0] = byte 1 mask 0x30. Skip [4:6] payload_len. + // Compare byte 1 fully so ECN (TC[1:0]) must match. Skip [4:6] payload_len. if a[0] != b[0] { return false } - if a[1]&^0x30 != b[1]&^0x30 { + if a[1] != b[1] { return false } if !bytes.Equal(a[2:4], b[2:4]) { @@ -119,11 +122,12 @@ func ipHeadersMatch(a, b []byte, isV6 bool) bool { } // IPv4: byte 0 = version/IHL, byte 1 = DSCP(6)|ECN(2), // [6:10] flags/fragoff/TTL/proto, [12:20] src+dst. + // Compare byte 1 fully so ECN must match. // Skip [2:4] total len, [4:6] id, [10:12] csum. if a[0] != b[0] { return false } - if a[1]&^0x03 != b[1]&^0x03 { + if a[1] != b[1] { return false } if !bytes.Equal(a[6:10], b[6:10]) { @@ -135,19 +139,6 @@ func ipHeadersMatch(a, b []byte, isV6 bool) bool { return true } -// mergeECNIntoSeed ORs the 2-bit IP-level ECN field of pkt's IP header -// onto the seed's IP header, so a CE mark on any coalesced segment -// propagates to the final superpacket. (CE is 0b11; ORing yields CE if -// any segment carried it.) Used by both TCP and UDP coalescers, so the -// invariant lives in one place. -func mergeECNIntoSeed(seedHdr, pktHdr []byte, isV6 bool) { - if isV6 { - seedHdr[1] |= pktHdr[1] & 0x30 - } else { - seedHdr[1] |= pktHdr[1] & 0x03 - } -} - // Arena is an injectable byte-slab that hands out non-overlapping borrowed // slices via Reserve and releases them in bulk via Reset. Coalescers take // an *Arena at construction so the caller controls the slab lifetime and diff --git a/overlay/batch/tcp_coalesce.go b/overlay/batch/tcp_coalesce.go index c50667e0..f7630c12 100644 --- a/overlay/batch/tcp_coalesce.go +++ b/overlay/batch/tcp_coalesce.go @@ -365,9 +365,6 @@ func (c *TCPCoalescer) appendPayload(s *coalesceSlot, pkt []byte, info parsedTCP // last segment. Without this the sender's push signal is dropped. s.hdrBuf[s.ipHdrLen+13] |= tcpFlagPsh } - // Merge IP-level CE marks into the seed: headersMatch ignores ECN, so - // this is the one place the signal is preserved. - mergeECNIntoSeed(s.hdrBuf[:s.ipHdrLen], pkt[:s.ipHdrLen], s.isV6) if info.payLen < s.gsoSize || info.flags&tcpFlagPsh != 0 { s.psh = true } @@ -424,8 +421,9 @@ func (c *TCPCoalescer) flushSlot(s *coalesceSlot) error { // headersMatch compares two IP+TCP header prefixes for byte-for-byte // equality on every field that must be identical across coalesced -// segments. Size/IPID/IPCsum/seq/flags/tcpCsum are masked out, as is the -// 2-bit IP-level ECN field — appendPayload merges CE into the seed. +// segments. Size/IPID/IPCsum/seq/flags/tcpCsum are masked out. The IP-level +// ECN codepoint is compared (via ipHeadersMatch) so segments with differing +// ECN don't coalesce, matching kernel GRO. func headersMatch(a, b []byte, isV6 bool, ipHdrLen int) bool { if len(a) != len(b) { return false @@ -632,6 +630,9 @@ func flowKeyCompare(a, b flowKey) int { // ECE state must agree across both slots: PSH is a semantic delimiter // (preserving the sender's push boundary) and ECE state must be uniform // across a window (the same rule canAppend enforces for in-flow appends). +// The IP-level ECN codepoint must also match: this check calls headersMatch +// → ipHeadersMatch, which compares the full DSCP/ECN byte, so two slots with +// differing ECN marks stay separate superpackets, each keeping its own mark. // // Note: a slot sealed by reorder (canAppend returned false on seq // mismatch) keeps psh=false, so this restriction does not block the @@ -670,10 +671,9 @@ func canMergeSlots(prev, s *coalesceSlot) bool { } // mergeSlots folds src into dst in place: payIovs concatenated, counters -// and totals updated, PSH and IP-level CE bits OR'd into the seed header -// so neither the push signal nor a CE mark is lost. The seed header's -// seq, gsoSize, and fk are unchanged. Caller is responsible for releasing -// src (it's no longer in c.slots after this call). +// and totals updated, PSH OR'd into the seed header so the push signal is +// not lost. The seed header's seq, gsoSize, and fk are unchanged. Caller +// is responsible for releasing src (it's no longer in c.slots after this call). func mergeSlots(dst, src *coalesceSlot) { dst.payIovs = append(dst.payIovs, src.payIovs...) dst.numSeg += src.numSeg @@ -683,7 +683,6 @@ func mergeSlots(dst, src *coalesceSlot) { dst.psh = true dst.hdrBuf[dst.ipHdrLen+13] |= tcpFlagPsh } - mergeECNIntoSeed(dst.hdrBuf[:dst.ipHdrLen], src.hdrBuf[:src.ipHdrLen], dst.isV6) } // ipv4HdrChecksum computes the IPv4 header checksum over hdr (which must diff --git a/overlay/batch/tcp_coalesce_test.go b/overlay/batch/tcp_coalesce_test.go index 6da60861..941161fa 100644 --- a/overlay/batch/tcp_coalesce_test.go +++ b/overlay/batch/tcp_coalesce_test.go @@ -762,39 +762,88 @@ func TestCoalescerEceMismatchReseeds(t *testing.T) { } } -// TestCoalescerMergesCEMark confirms that an ECT(0) burst with a single -// CE-marked packet still coalesces, and the merged superpacket carries CE. -func TestCoalescerMergesCEMark(t *testing.T) { +// TestCoalescerDifferingECNReseeds confirms that segments with differing IP +// ECN codepoints do NOT coalesce: headersMatch compares the full ToS byte, +// matching kernel GRO. Two ECT(0) segments merge; a CE stamp mid-run seals +// the ECT(0) chain and starts a fresh superpacket that keeps CE; a trailing +// ECT(0) starts yet another. Each superpacket keeps its own codepoint — +// ORing the marks (the old buggy behavior) would have fabricated a false CE +// across the whole burst. +func TestCoalescerDifferingECNReseeds(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} c := NewTCPCoalescer(w, test.NewLogger(), NewArena(0)) pay := make([]byte, 1200) if err := c.Commit(buildTCPv4WithToS(ecnECT0, 1000, tcpAck, pay)); err != nil { t.Fatal(err) } - // Router along the path stamped CE on this one. - if err := c.Commit(buildTCPv4WithToS(ecnCE, 2200, tcpAck, pay)); err != nil { + if err := c.Commit(buildTCPv4WithToS(ecnECT0, 2200, tcpAck, pay)); err != nil { t.Fatal(err) } - if err := c.Commit(buildTCPv4WithToS(ecnECT0, 3400, tcpAck, pay)); err != nil { + // Router along the path stamped CE on this one. + if err := c.Commit(buildTCPv4WithToS(ecnCE, 3400, tcpAck, pay)); err != nil { + t.Fatal(err) + } + if err := c.Commit(buildTCPv4WithToS(ecnECT0, 4600, tcpAck, pay)); err != nil { t.Fatal(err) } if err := c.Flush(); err != nil { t.Fatal(err) } - if len(w.gsoWrites) != 1 { - t.Fatalf("want 1 merged gso write, got %d (plain=%d)", len(w.gsoWrites), len(w.writes)) + if len(w.gsoWrites) != 3 { + t.Fatalf("want 3 superpackets (ECN split), got %d (plain=%d)", len(w.gsoWrites), len(w.writes)) } - g := w.gsoWrites[0] - if len(g.pays) != 3 { - t.Errorf("pay count=%d want 3", len(g.pays)) + // gso[0]: the two ECT(0) segments merged; gso[1]: CE alone; gso[2]: + // trailing ECT(0) alone. Emitted in seq order. + type want struct { + pays int + ecn byte } - if got := g.hdr[1] & 0x03; got != ecnCE { - t.Errorf("seed ECN=0x%02x want CE 0x%02x", got, ecnCE) + wants := []want{{2, ecnECT0}, {1, ecnCE}, {1, ecnECT0}} + for i, wnt := range wants { + g := w.gsoWrites[i] + if len(g.pays) != wnt.pays { + t.Errorf("gso %d pay count=%d want %d", i, len(g.pays), wnt.pays) + } + if got := g.hdr[1] & 0x03; got != wnt.ecn { + t.Errorf("gso %d ECN=0x%02x want 0x%02x", i, got, wnt.ecn) + } } } -// TestCoalescerDscpMismatchReseeds confirms that the new ECN-mask in -// headersMatch did not also relax DSCP — different DSCP must still split. +// TestCoalescerECT0ThenECT1NoCE is the core regression for the ECN merge +// bug: ORing ECT(0)=0b10 with ECT(1)=0b01 fabricates CE=0b11. The two +// segments must land in separate superpackets, each preserving its own +// codepoint, and neither may end up CE-marked. +func TestCoalescerECT0ThenECT1NoCE(t *testing.T) { + w := &fakeTunWriter{gsoEnabled: true} + c := NewTCPCoalescer(w, test.NewLogger(), NewArena(0)) + pay := make([]byte, 1200) + if err := c.Commit(buildTCPv4WithToS(ecnECT0, 1000, tcpAck, pay)); err != nil { + t.Fatal(err) + } + if err := c.Commit(buildTCPv4WithToS(ecnECT1, 2200, tcpAck, pay)); err != nil { + t.Fatal(err) + } + if err := c.Flush(); err != nil { + t.Fatal(err) + } + if len(w.gsoWrites) != 2 { + t.Fatalf("want 2 separate superpackets (ECT0 vs ECT1), got %d", len(w.gsoWrites)) + } + wantECN := []byte{ecnECT0, ecnECT1} + for i, g := range w.gsoWrites { + if got := g.hdr[1] & 0x03; got != wantECN[i] { + t.Errorf("gso %d ECN=0x%02x want 0x%02x", i, got, wantECN[i]) + } + if got := g.hdr[1] & 0x03; got == ecnCE { + t.Errorf("gso %d fabricated CE from ECT merge", i) + } + } +} + +// TestCoalescerDscpMismatchReseeds confirms that a DSCP difference (same +// ECN) still splits — headersMatch compares the full ToS byte, so the upper +// six DSCP bits must match too. func TestCoalescerDscpMismatchReseeds(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} c := NewTCPCoalescer(w, test.NewLogger(), NewArena(0)) @@ -995,9 +1044,10 @@ func TestCoalescerSortKeepsPassthroughBarrier(t *testing.T) { } } -// TestCoalescerIPv6MergesCEMark is the IPv6 analogue of -// TestCoalescerMergesCEMark. ECN bits live in TC[1:0] = byte 1 mask 0x30. -func TestCoalescerIPv6MergesCEMark(t *testing.T) { +// TestCoalescerIPv6DifferingECNReseeds is the IPv6 analogue of +// TestCoalescerDifferingECNReseeds. ECN bits live in TC[1:0] = byte 1 mask +// 0x30, so ipHeadersMatch (comparing byte 1 fully) still splits them. +func TestCoalescerIPv6DifferingECNReseeds(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} c := NewTCPCoalescer(w, test.NewLogger(), NewArena(0)) pay := make([]byte, 1200) @@ -1005,20 +1055,36 @@ func TestCoalescerIPv6MergesCEMark(t *testing.T) { if err := c.Commit(buildTCPv6(ecnECT0, 1000, tcpAck, pay)); err != nil { t.Fatal(err) } - if err := c.Commit(buildTCPv6(ecnCE, 2200, tcpAck, pay)); err != nil { + if err := c.Commit(buildTCPv6(ecnECT0, 2200, tcpAck, pay)); err != nil { + t.Fatal(err) + } + if err := c.Commit(buildTCPv6(ecnCE, 3400, tcpAck, pay)); err != nil { + t.Fatal(err) + } + if err := c.Commit(buildTCPv6(ecnECT0, 4600, tcpAck, pay)); err != nil { t.Fatal(err) } if err := c.Flush(); err != nil { t.Fatal(err) } - if len(w.gsoWrites) != 1 { - t.Fatalf("want 1 merged gso write, got %d", len(w.gsoWrites)) + if len(w.gsoWrites) != 3 { + t.Fatalf("want 3 superpackets (ECN split), got %d", len(w.gsoWrites)) } - g := w.gsoWrites[0] // Byte 1 high nibble holds TC[3:0]; ECN is the low 2 bits of that nibble, // which appears in byte 1 mask 0x30 (>>4 to read the codepoint value). - if got := (g.hdr[1] >> 4) & 0x03; got != ecnCE { - t.Errorf("seed v6 ECN=0x%02x want CE 0x%02x", got, ecnCE) + type want struct { + pays int + ecn byte + } + wants := []want{{2, ecnECT0}, {1, ecnCE}, {1, ecnECT0}} + for i, wnt := range wants { + g := w.gsoWrites[i] + if len(g.pays) != wnt.pays { + t.Errorf("gso %d pay count=%d want %d", i, len(g.pays), wnt.pays) + } + if got := (g.hdr[1] >> 4) & 0x03; got != wnt.ecn { + t.Errorf("gso %d v6 ECN=0x%02x want 0x%02x", i, got, wnt.ecn) + } } } diff --git a/overlay/batch/udp_coalesce.go b/overlay/batch/udp_coalesce.go index d7ad2506..7bd7c89f 100644 --- a/overlay/batch/udp_coalesce.go +++ b/overlay/batch/udp_coalesce.go @@ -257,8 +257,6 @@ func (c *UDPCoalescer) appendPayload(s *udpSlot, pkt []byte, info parsedUDP) { s.payIovs = append(s.payIovs, pkt[info.hdrLen:info.hdrLen+info.payLen]) s.numSeg++ s.totalPay += info.payLen - // Merge IP-level CE marks into the seed (same trick TCP coalescer uses). - mergeECNIntoSeed(s.hdrBuf[:s.ipHdrLen], pkt[:s.ipHdrLen], s.isV6) if info.payLen < s.gsoSize { // Last-segment-can-be-shorter: this seals the chain. s.sealed = true @@ -329,8 +327,9 @@ func (c *UDPCoalescer) flushSlot(s *udpSlot) error { // udpHeadersMatch compares two IP+UDP header prefixes for byte-equality on // every field that must be identical across coalesced segments. Length -// fields and the ECN bits in IP TOS/TC are masked out — appendPayload -// merges CE into the seed; flushSlot rewrites lengths. +// fields are masked out (flushSlot rewrites them), but the IP-level ECN +// codepoint is compared (via ipHeadersMatch) so segments with differing ECN +// don't coalesce, matching kernel GRO. func udpHeadersMatch(a, b []byte, isV6 bool, ipHdrLen int) bool { if len(a) != len(b) { return false diff --git a/overlay/batch/udp_coalesce_test.go b/overlay/batch/udp_coalesce_test.go index 1e01760d..273c7d69 100644 --- a/overlay/batch/udp_coalesce_test.go +++ b/overlay/batch/udp_coalesce_test.go @@ -261,32 +261,37 @@ func TestUDPCoalescerCapsAtMaxSegs(t *testing.T) { } } -// CE marks on appended segments must be merged into the seed's IP TOS. -func TestUDPCoalescerMergesCEMark(t *testing.T) { +// Differing IP ECN codepoints must not coalesce: udpHeadersMatch compares +// the full ToS byte (matching kernel GRO). A CE-marked datagram mid-run +// seals the Not-ECT chain and seeds a fresh superpacket that keeps CE; the +// trailing Not-ECT datagram seeds another. +func TestUDPCoalescerDifferingECNReseeds(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} c := NewUDPCoalescer(w, NewArena(0)) pay := make([]byte, 800) - pkt0 := buildUDPv4(1000, 53, pay) // ECN=00 + pkt0 := buildUDPv4(1000, 53, pay) // ECN=00 (Not-ECT) pkt1 := buildUDPv4(1000, 53, pay) - pkt1[1] = 0x03 // CE - pkt2 := buildUDPv4(1000, 53, pay) - if err := c.Commit(pkt0); err != nil { - t.Fatal(err) - } - if err := c.Commit(pkt1); err != nil { - t.Fatal(err) - } - if err := c.Commit(pkt2); err != nil { - t.Fatal(err) + pkt1[1] = 0x03 // CE + pkt2 := buildUDPv4(1000, 53, pay) // ECN=00 again + for _, p := range [][]byte{pkt0, pkt1, pkt2} { + if err := c.Commit(p); err != nil { + t.Fatal(err) + } } if err := c.Flush(); err != nil { t.Fatal(err) } - if len(w.gsoWrites) != 1 { - t.Fatalf("want 1 merged gso write, got %d (plain=%d)", len(w.gsoWrites), len(w.writes)) + if len(w.gsoWrites) != 3 { + t.Fatalf("want 3 separate seeds (differing ECN), got %d (plain=%d)", len(w.gsoWrites), len(w.writes)) } - if w.gsoWrites[0].hdr[1]&0x03 != 0x03 { - t.Errorf("CE not merged into seed (tos=%#x)", w.gsoWrites[0].hdr[1]) + wantECN := []byte{0x00, 0x03, 0x00} + for i, g := range w.gsoWrites { + if len(g.pays) != 1 { + t.Errorf("gso %d pay count=%d want 1", i, len(g.pays)) + } + if got := g.hdr[1] & 0x03; got != wantECN[i] { + t.Errorf("gso %d ECN=%#x want %#x", i, got, wantECN[i]) + } } } @@ -326,7 +331,7 @@ func TestUDPCoalescerIPv6Coalesces(t *testing.T) { } } -// DSCP differences must reseed (headers don't match outside ECN). +// DSCP differences must reseed: udpHeadersMatch compares the full ToS byte. func TestUDPCoalescerDSCPMismatchReseeds(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} c := NewUDPCoalescer(w, NewArena(0)) diff --git a/overlay/tio/blockon_linux.go b/overlay/tio/blockon_linux.go new file mode 100644 index 00000000..9c87f5e9 --- /dev/null +++ b/overlay/tio/blockon_linux.go @@ -0,0 +1,50 @@ +//go:build linux && !android +// +build linux,!android + +package tio + +import ( + "os" + + "golang.org/x/sys/unix" +) + +// blockOn parks the calling goroutine until fd is ready (events is POLLIN for +// reads, POLLOUT for writes) or shutdownFd signals teardown. It builds the +// pollfd array on the stack every call, so concurrent callers on the same +// Queue never share Revents storage: the previous shared-array implementation +// was a genuine Go data race when two writers parked in poll(2) at once (the +// kernel writing Revents while another goroutine zeroed it). Level-triggered +// events kept it from deadlocking, but it was still a race. +// +// Poll(2) is looped over EINTR. err is checked before the Revents bits are +// trusted, since a failed poll may leave them bogus. Returns os.ErrClosed when +// shutdown was signaled (POLLIN on shutdownFd) or either fd reported a problem +// condition (POLLHUP|POLLNVAL|POLLERR). +func blockOn(fd, shutdownFd int32, events int16) error { + const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR + pfds := [2]unix.PollFd{ + {Fd: fd, Events: events}, + {Fd: shutdownFd, Events: unix.POLLIN}, + } + var err error + for { + _, err = unix.Poll(pfds[:], -1) + if err != unix.EINTR { + break + } + } + tunEvents := pfds[0].Revents + shutdownEvents := pfds[1].Revents + // Check err before trusting the potentially bogus bits we just got. + if err != nil { + return err + } + if shutdownEvents&(unix.POLLIN|problemFlags) != 0 { + return os.ErrClosed + } + if tunEvents&problemFlags != 0 { + return os.ErrClosed + } + return nil +} diff --git a/overlay/tio/tio.go b/overlay/tio/tio.go index 2d94c764..c680f452 100644 --- a/overlay/tio/tio.go +++ b/overlay/tio/tio.go @@ -26,8 +26,9 @@ type Capabilities struct { USO bool } -// Queue is a readable/writable Poll queue. One Queue is driven by a single -// read goroutine plus a single writer (see Write below). +// Queue is a readable/writable Poll queue. Concurrency contract: a single +// read goroutine drives Read; plain Write is safe for concurrent callers; +// WriteGSO (on Queues that implement GSOWriter) is single-writer per queue. type Queue interface { io.Closer @@ -37,11 +38,12 @@ type Queue interface { // or copy each slice before the next call. A Packet may carry a // GSO/USO superpacket (see GSOInfo); when GSO.IsSuperpacket() is // true the caller must segment Bytes before treating it as a single - // IP datagram. Not safe for concurrent Reads. + // IP datagram. Single-reader only: not safe for concurrent Reads (it + // reuses per-queue rx scratch each call). Read() ([]Packet, error) // Write emits a single packet on the plaintext (outside→inside) - // delivery path. Not safe for concurrent Writes. + // delivery path. Safe for concurrent use. Write(p []byte) (int, error) } diff --git a/overlay/tio/tio_gso_linux.go b/overlay/tio/tio_gso_linux.go index beb4952b..460db0ad 100644 --- a/overlay/tio/tio_gso_linux.go +++ b/overlay/tio/tio_gso_linux.go @@ -8,7 +8,6 @@ import ( "io" "log/slog" "os" - "sync" "sync/atomic" "syscall" "unsafe" @@ -62,17 +61,10 @@ var validVnetHdr = [virtio.Size]byte{unix.VIRTIO_NET_HDR_F_DATA_VALID} type Offload struct { fd int shutdownFd int - readPoll [2]unix.PollFd - writePoll [2]unix.PollFd - // writeLock serializes blockOnWrite's read+clear of writePoll[*].Revents. - // Any goroutine that calls Write may end up parked in poll(2); without - // the lock concurrent waiters could race the Revents reset and lose - // events. - writeLock sync.Mutex - closed atomic.Bool - rxBuf []byte // backing store for kernel-handed packets read this drain - rxOff int // cursor into rxBuf for the current Read drain - pending []Packet // packets returned from the most recent Read + closed atomic.Bool + rxBuf []byte // backing store for kernel-handed packets read this drain + rxOff int // cursor into rxBuf for the current Read drain + pending []Packet // packets returned from the most recent Read // readVnetScratch holds the 10-byte virtio_net_hdr split off the front of // every TUN read via readv(2). Decoupling the header from the packet body @@ -109,15 +101,6 @@ func newOffload(fd int, shutdownFd int, usoEnabled bool) (*Offload, error) { shutdownFd: shutdownFd, usoEnabled: usoEnabled, closed: atomic.Bool{}, - readPoll: [2]unix.PollFd{ - {Fd: int32(fd), Events: unix.POLLIN}, - {Fd: int32(shutdownFd), Events: unix.POLLIN}, - }, - writePoll: [2]unix.PollFd{ - {Fd: int32(fd), Events: unix.POLLOUT}, - {Fd: int32(shutdownFd), Events: unix.POLLIN}, - }, - writeLock: sync.Mutex{}, rxBuf: make([]byte, tunRxBufCap), gsoIovs: make([]unix.Iovec, 2, gsoMaxIovs), @@ -135,57 +118,11 @@ func newOffload(fd int, shutdownFd int, usoEnabled bool) (*Offload, error) { } func (r *Offload) blockOnRead() error { - const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR - var err error - for { - _, err = unix.Poll(r.readPoll[:], -1) - if err != unix.EINTR { - break - } - } - //always reset these! - tunEvents := r.readPoll[0].Revents - shutdownEvents := r.readPoll[1].Revents - r.readPoll[0].Revents = 0 - r.readPoll[1].Revents = 0 - //do the err check before trusting the potentially bogus bits we just got - if err != nil { - return err - } - if shutdownEvents&(unix.POLLIN|problemFlags) != 0 { - return os.ErrClosed - } else if tunEvents&problemFlags != 0 { - return os.ErrClosed - } - return nil + return blockOn(int32(r.fd), int32(r.shutdownFd), unix.POLLIN) } func (r *Offload) blockOnWrite() error { - const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR - var err error - for { - _, err = unix.Poll(r.writePoll[:], -1) - if err != unix.EINTR { - break - } - } - //always reset these! - r.writeLock.Lock() - tunEvents := r.writePoll[0].Revents - shutdownEvents := r.writePoll[1].Revents - r.writePoll[0].Revents = 0 - r.writePoll[1].Revents = 0 - r.writeLock.Unlock() - //do the err check before trusting the potentially bogus bits we just got - if err != nil { - return err - } - if shutdownEvents&(unix.POLLIN|problemFlags) != 0 { - return os.ErrClosed - } else if tunEvents&problemFlags != 0 { - return os.ErrClosed - } - return nil + return blockOn(int32(r.fd), int32(r.shutdownFd), unix.POLLOUT) } // readPacket issues a single readv(2) splitting the virtio_net_hdr off diff --git a/overlay/tio/tio_poll_linux.go b/overlay/tio/tio_poll_linux.go index 975eff1f..22ee1631 100644 --- a/overlay/tio/tio_poll_linux.go +++ b/overlay/tio/tio_poll_linux.go @@ -17,11 +17,9 @@ import ( const tunReadBufSize = 65535 type Poll struct { - fd int - - readPoll [2]unix.PollFd - writePoll [2]unix.PollFd - closed atomic.Bool + fd int + shutdownFd int + closed atomic.Bool readBuf []byte batchRet [1]Packet @@ -37,16 +35,9 @@ func newPoll(fd int, shutdownFd int) (*Poll, error) { } out := &Poll{ - fd: fd, - readBuf: make([]byte, tunReadBufSize), - readPoll: [2]unix.PollFd{ - {Fd: int32(fd), Events: unix.POLLIN}, - {Fd: int32(shutdownFd), Events: unix.POLLIN}, - }, - writePoll: [2]unix.PollFd{ - {Fd: int32(fd), Events: unix.POLLOUT}, - {Fd: int32(shutdownFd), Events: unix.POLLIN}, - }, + fd: fd, + shutdownFd: shutdownFd, + readBuf: make([]byte, tunReadBufSize), } return out, nil } @@ -54,53 +45,11 @@ func newPoll(fd int, shutdownFd int) (*Poll, error) { // blockOnRead waits until the Poll fd is readable or shutdown has been signaled. // Returns os.ErrClosed if Close was called. func (t *Poll) blockOnRead() error { - const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR - var err error - for { - _, err = unix.Poll(t.readPoll[:], -1) - if err != unix.EINTR { - break - } - } - tunEvents := t.readPoll[0].Revents - shutdownEvents := t.readPoll[1].Revents - t.readPoll[0].Revents = 0 - t.readPoll[1].Revents = 0 - if err != nil { - return err - } - if shutdownEvents&(unix.POLLIN|problemFlags) != 0 { - return os.ErrClosed - } - if tunEvents&problemFlags != 0 { - return os.ErrClosed - } - return nil + return blockOn(int32(t.fd), int32(t.shutdownFd), unix.POLLIN) } func (t *Poll) blockOnWrite() error { - const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR - var err error - for { - _, err = unix.Poll(t.writePoll[:], -1) - if err != unix.EINTR { - break - } - } - tunEvents := t.writePoll[0].Revents - shutdownEvents := t.writePoll[1].Revents - t.writePoll[0].Revents = 0 - t.writePoll[1].Revents = 0 - if err != nil { - return err - } - if shutdownEvents&(unix.POLLIN|problemFlags) != 0 { - return os.ErrClosed - } - if tunEvents&problemFlags != 0 { - return os.ErrClosed - } - return nil + return blockOn(int32(t.fd), int32(t.shutdownFd), unix.POLLOUT) } func (t *Poll) Read() ([]Packet, error) { @@ -133,7 +82,7 @@ func (t *Poll) readOne(to []byte) (int, error) { } } -// Write is only valid for single threaded use +// Write is safe for concurrent use func (t *Poll) Write(from []byte) (int, error) { for { n, errno := unix.Write(t.fd, from) diff --git a/overlay/tio/tun_file_linux_test.go b/overlay/tio/tun_file_linux_test.go index d72202ea..7d8b5c4e 100644 --- a/overlay/tio/tun_file_linux_test.go +++ b/overlay/tio/tun_file_linux_test.go @@ -70,6 +70,76 @@ func TestPoll_WakeForShutdown_WakesFriends(t *testing.T) { } } +// TestPoll_ConcurrentWrite_NoRace hammers a single Poll queue from two writer +// goroutines while a reader drains the other end of the pipe. The writers +// overflow the pipe buffer, so both repeatedly park in blockOnWrite at the same +// time — the exact scenario that raced on the old shared writePoll member +// array. Run under -race; a shared-array regression trips the detector here. +func TestPoll_ConcurrentWrite_NoRace(t *testing.T) { + var fds [2]int + require.NoError(t, unix.Pipe2(fds[:], unix.O_CLOEXEC)) + readFd, writeFd := fds[0], fds[1] + + shutdownFd, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC) + require.NoError(t, err) + t.Cleanup(func() { _ = unix.Close(shutdownFd) }) + + p, err := newPoll(writeFd, shutdownFd) + require.NoError(t, err) + + const writers = 2 + const perWriter = 4000 + payload := make([]byte, 100) + total := writers * perWriter * len(payload) + + // Reader: drain the read end (blocking) until every writer's bytes are + // consumed, so the writers keep making progress rather than wedging on a + // permanently full pipe. + readDone := make(chan struct{}) + go func() { + defer close(readDone) + buf := make([]byte, 4096) + got := 0 + for got < total { + n, rerr := unix.Read(readFd, buf) + got += n + if rerr != nil { + if rerr == unix.EINTR { + continue + } + return + } + if n == 0 { // EOF + return + } + } + }() + + var wg sync.WaitGroup + for w := 0; w < writers; w++ { + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < perWriter; i++ { + if _, werr := p.Write(payload); werr != nil { + t.Errorf("write: %v", werr) + return + } + } + }() + } + wg.Wait() + + select { + case <-readDone: + case <-time.After(10 * time.Second): + t.Fatal("reader did not drain") + } + + require.NoError(t, p.Close()) + _ = unix.Close(readFd) +} + // TestPoll_NewPoll_DoesNotCloseFdOnFailure pins the ownership rule: when // newPoll fails, it must leave fd open so the caller (pollQueueSet.Add's // callers in tun_linux.go) is the sole closer. If newPoll also closed fd, diff --git a/overlay/tio/virtio/segment_linux.go b/overlay/tio/virtio/segment_linux.go index a5666f6e..679eab45 100644 --- a/overlay/tio/virtio/segment_linux.go +++ b/overlay/tio/virtio/segment_linux.go @@ -143,7 +143,7 @@ func CorrectHdrLen(pkt []byte, hdr *Hdr) error { if hdr.HdrLen < hdr.CsumStart { return fmt.Errorf("virtioNetHdr.HdrLen (%d) < virtioNetHdr.CsumStart (%d)", hdr.HdrLen, hdr.CsumStart) } - cSumAt := int(hdr.CsumStart + hdr.CsumStart) + cSumAt := int(hdr.CsumStart + hdr.CsumOffset) if cSumAt+1 >= len(pkt) { return fmt.Errorf("end of checksum offset (%d) exceeds packet length (%d)", cSumAt+1, len(pkt)) } diff --git a/overlay/tio/virtio/segment_linux_test.go b/overlay/tio/virtio/segment_linux_test.go index 6ba5f2e6..9c6799d5 100644 --- a/overlay/tio/virtio/segment_linux_test.go +++ b/overlay/tio/virtio/segment_linux_test.go @@ -211,6 +211,55 @@ func TestSegmentTCPHeaderNotCorrupted(t *testing.T) { } } +// TestCorrectHdrLenChecksumBound guards the checksum-field bounds check in +// CorrectHdrLen. The checksum field sits at CsumStart+CsumOffset, so the check +// must be computed from CsumStart+CsumOffset — NOT CsumStart+CsumStart, a +// regression that doubled CsumStart and thus over-tightened the bound (since +// CsumOffset, 6 for UDP / 16 for TCP, is always < CsumStart >= 20). That bogus +// bound spuriously rejected valid small USO superpackets in decodeRead. +func TestCorrectHdrLenChecksumBound(t *testing.T) { + // A valid IPv4 USO superpacket: 20B IPv4 + 8B UDP + two 6-byte segments + // (payload 12) = 40 bytes total. CsumStart=20, CsumOffset=6, so the UDP + // checksum field lives at bytes 26..27, comfortably inside the 40-byte + // packet. The OLD formula computed cSumAt = CsumStart+CsumStart = 40 and + // rejected on cSumAt+1 (41) >= len(pkt) (40); the fix (CsumStart+CsumOffset + // = 26) accepts. This case FAILS against the CsumStart+CsumStart regression. + t.Run("valid-small-uso-accepted", func(t *testing.T) { + pkt, _, csumStart := buildUDPv4Super(12) // total len 40 + hdr := Hdr{ + Flags: unix.VIRTIO_NET_HDR_F_NEEDS_CSUM, + GSOType: unix.VIRTIO_NET_HDR_GSO_UDP_L4, + GSOSize: 6, // two 6-byte segments + CsumStart: csumStart, + CsumOffset: 6, + } + if err := CorrectHdrLen(pkt, &hdr); err != nil { + t.Fatalf("CorrectHdrLen rejected a valid 40-byte USO superpacket: %v", err) + } + if hdr.HdrLen != csumStart+udpHeaderLen { + t.Errorf("HdrLen = %d, want %d", hdr.HdrLen, csumStart+udpHeaderLen) + } + }) + + // A genuinely-too-short packet: CsumStart=20, CsumOffset=6 means the + // checksum field would end at byte 27, but the packet is only 25 bytes + // (CsumStart+CsumOffset+2 = 28 > 25). CorrectHdrLen must still reject it. + t.Run("too-short-rejected", func(t *testing.T) { + pkt := make([]byte, 25) + pkt[0] = 0x45 // IPv4, IHL 5 + hdr := Hdr{ + Flags: unix.VIRTIO_NET_HDR_F_NEEDS_CSUM, + GSOType: unix.VIRTIO_NET_HDR_GSO_UDP_L4, + GSOSize: 6, + CsumStart: 20, + CsumOffset: 6, + } + if err := CorrectHdrLen(pkt, &hdr); err == nil { + t.Fatalf("CorrectHdrLen accepted a too-short (25-byte) packet") + } + }) +} + // TestSegmentUDPHeaderNotCorrupted is the USO counterpart: SegmentUDP performs // the same header stamp and must be correct when gsoSize < headerLen. func TestSegmentUDPHeaderNotCorrupted(t *testing.T) { diff --git a/pprof_debug.go b/pprof_debug.go new file mode 100644 index 00000000..5e1470d4 --- /dev/null +++ b/pprof_debug.go @@ -0,0 +1,33 @@ +//go:build debug + +package nebula + +import ( + "context" + "errors" + "log/slog" + "net/http" + _ "net/http/pprof" // registers pprof handlers on http.DefaultServeMux +) + +// startPprofServer serves net/http/pprof on :6060 for the life of ctx. It is +// only compiled into debug builds (`-tags debug`, `make debug`), so a debug +// build announces itself with the Info line below. +func startPprofServer(ctx context.Context, l *slog.Logger) { + server := &http.Server{Addr: ":6060", Handler: nil} + l.Info("Starting pprof debug server (debug build)", "addr", server.Addr) + + go func() { + if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + l.Error("pprof debug server stopped", "error", err) + } + }() + + // Shut down the server when the context is cancelled. + go func() { + <-ctx.Done() + if err := server.Shutdown(context.Background()); err != nil { + l.Debug("Error shutting down pprof debug server", "error", err) + } + }() +} diff --git a/pprof_nodebug.go b/pprof_nodebug.go new file mode 100644 index 00000000..f74be0d1 --- /dev/null +++ b/pprof_nodebug.go @@ -0,0 +1,11 @@ +//go:build !debug + +package nebula + +import ( + "context" + "log/slog" +) + +// startPprofServer is a no-op unless built with `-tags debug` (see make debug). +func startPprofServer(_ context.Context, _ *slog.Logger) {} diff --git a/util/cpupin_linux.go b/util/cpupin_linux.go index a5acc01d..016e9bf8 100644 --- a/util/cpupin_linux.go +++ b/util/cpupin_linux.go @@ -21,3 +21,23 @@ func PinThreadToCPU(cpu int) error { set.Set(cpu) return unix.SchedSetaffinity(0, &set) } + +// AllowedCPUs returns the CPU IDs the calling process is currently allowed to +// run on, as reported by sched_getaffinity(2). Under a cgroup cpuset or a +// `taskset` mask the allowed IDs are frequently not the contiguous range +// 0..NumCPU-1 (e.g. pinned to CPUs 4-7: NumCPU reports 4 while the valid IDs +// are 4,5,6,7). Callers that need a real CPU to pin to must choose from this +// set rather than assuming i % NumCPU is runnable, or every pin fails. +func AllowedCPUs() ([]int, error) { + var set unix.CPUSet + if err := unix.SchedGetaffinity(0, &set); err != nil { + return nil, err + } + cpus := make([]int, 0, set.Count()) + for cpu := 0; cpu < len(set)*64; cpu++ { + if set.IsSet(cpu) { + cpus = append(cpus, cpu) + } + } + return cpus, nil +} diff --git a/util/cpupin_other.go b/util/cpupin_other.go index ed28d82f..50522b9a 100644 --- a/util/cpupin_other.go +++ b/util/cpupin_other.go @@ -9,3 +9,10 @@ package util func PinThreadToCPU(_ int) error { return nil } + +// AllowedCPUs has no meaningful answer off Linux (no sched_getaffinity), so it +// reports "unknown" by returning a nil slice and nil error. Callers treat an +// empty result as "fall back to the default CPU choice". +func AllowedCPUs() ([]int, error) { + return nil, nil +}