From e04473f30b26938112ee0138902e1357240524a1 Mon Sep 17 00:00:00 2001 From: JackDoan Date: Mon, 3 Aug 2026 10:51:37 -0500 Subject: [PATCH] crazy core pinning junk --- cpupick/cpupick.go | 187 +++++++++++++++++++++++++++++++++++++ cpupick/cpupick_test.go | 171 +++++++++++++++++++++++++++++++++ cpupick/perf_linux.go | 154 ++++++++++++++++++++++++++++++ cpupick/perf_linux_test.go | 163 ++++++++++++++++++++++++++++++++ cpupick/perf_other.go | 10 ++ cpupick/topo_linux.go | 118 +++++++++++++++++++++++ cpupick/topo_linux_test.go | 111 ++++++++++++++++++++++ cpupick/topo_other.go | 10 ++ main.go | 21 ++++- 9 files changed, 943 insertions(+), 2 deletions(-) create mode 100644 cpupick/cpupick.go create mode 100644 cpupick/cpupick_test.go create mode 100644 cpupick/perf_linux.go create mode 100644 cpupick/perf_linux_test.go create mode 100644 cpupick/perf_other.go create mode 100644 cpupick/topo_linux.go create mode 100644 cpupick/topo_linux_test.go create mode 100644 cpupick/topo_other.go diff --git a/cpupick/cpupick.go b/cpupick/cpupick.go new file mode 100644 index 00000000..49f791e8 --- /dev/null +++ b/cpupick/cpupick.go @@ -0,0 +1,187 @@ +// Package cpupick chooses which CPUs the tun reader threads pin to when the +// operator has not chosen for us (tun.cpu_affinity). The stock spread — +// allowed[i] for routine i — has two failure modes this package exists to fix: +// +// - every co-located nebula starts its spread at allowed[0], so N instances +// on one box stack their readers onto the same cores, and allowed[0] is +// usually CPU 0, the core housekeeping and default IRQ affinity already +// favor; +// - on heterogeneous CPUs (ARM big.LITTLE, Intel P/E hybrids, AMD compact +// cores) low IDs are not necessarily fast cores, and pinning an encrypt +// thread to an efficiency core caps that queue's throughput. +// +// Default instead returns a preference-ordered pin list: the allowed set +// filtered to performance cores (when the platform distinguishes them and +// enough remain for every routine), confined to a single NUMA node and spread +// across distinct physical cores when the topology permits, CPU 0's physical +// core demoted to last resort, and the order rotated by a stable per-instance +// key so co-located instances spread instead of stacking. +package cpupick + +import ( + "log/slog" + + "github.com/slackhq/nebula/util" +) + +// topology is the slice of machine layout arrange consults: the NUMA node +// and the physical core behind each candidate CPU, plus which core CPU 0 +// lives on (zeroCore, -1 when unknown — tracked separately because CPU 0's +// SMT sibling deserves demotion even when CPU 0 itself isn't a candidate). +// Probed from sysfs on Linux; flatTopology stands in when the platform can't +// say, which turns every topology rule into a no-op rather than a wrong +// answer. +type topology struct { + nodeOf map[int]int + coreOf map[int]int + zeroCore int +} + +// flatTopology places every CPU on node 0 and on a physical core of its own. +func flatTopology(cpus []int) topology { + t := topology{ + nodeOf: make(map[int]int, len(cpus)), + coreOf: make(map[int]int, len(cpus)), + zeroCore: -1, + } + for i, c := range cpus { + t.nodeOf[c] = 0 + t.coreOf[c] = i + if c == 0 { + t.zeroCore = i + } + } + return t +} + +// Default computes the pin order for `routines` tun readers. key is any +// stable per-instance value; the bound UDP port is ideal — distinct across +// co-located instances, stable across restarts so benchmark runs stay +// comparable. Returns nil when there is nothing useful to say (no affinity +// support on this platform, lookup failure); callers keep their existing +// fallback spread. +func Default(routines int, key uint64, l *slog.Logger) []int { + allowed, err := util.AllowedCPUs() + if err != nil || len(allowed) == 0 { + return nil + } + perf, signal := perfCPUs(allowed) + cands := pickCandidates(allowed, perf, routines) + if len(cands) == 0 { + return nil + } + if len(perf) < routines { + signal = "" + } + cpus := arrange(cands, readTopology(cands), routines, splitmix64(key)) + if l != nil { + l.Info("chose default pin CPUs for tun readers", + "cpus", cpus[:min(routines, len(cpus))], + "perfSignal", signal) + } + return cpus +} + +// pickCandidates applies the enough-for-everyone guard: a perf filter that +// leaves fewer candidates than routines is discarded — giving every reader +// its own (possibly slow) core beats stacking two readers on a fast one. +func pickCandidates(allowed, perf []int, routines int) []int { + if len(perf) < routines { + return allowed + } + return perf +} + +// arrange turns the candidate set into the final pin order: +// +// 1. NUMA: when at least one node holds enough candidates for every +// routine, confine to one such node, chosen by the instance hash. The +// readers share hostmap and cipher state, so splitting one instance +// across nodes taxes every packet — and co-located instances that hash +// to different nodes stop competing entirely. When no node is big +// enough, span nodes rather than stack readers. +// 2. Rotate the preferred candidates by the hash so instances spread. +// 3. SMT: emit one thread per physical core before any of their siblings — +// two encrypt threads on one core split its execution units. Siblings +// still follow for the routines > cores case. +// 4. CPU 0's whole physical core goes last: housekeeping and default IRQ +// noise on CPU 0 bleeds into its SMT sibling too. Within that tail the +// sibling precedes CPU 0 itself, which only catches the bleed-through. +// +// The rotation happens before the SMT pass so each instance's one-per-core +// walk also starts at a different core, and CPU 0's core is excluded from +// the rotation so no hash value can put it back at the front. +func arrange(cands []int, topo topology, routines int, h uint64) []int { + byNode := map[int][]int{} + var nodes []int + for _, c := range cands { + n := topo.nodeOf[c] + if _, ok := byNode[n]; !ok { + nodes = append(nodes, n) + } + byNode[n] = append(byNode[n], c) + } + var eligible []int + for _, n := range nodes { + if len(byNode[n]) >= routines { + eligible = append(eligible, n) + } + } + if len(eligible) > 0 { + cands = byNode[eligible[int(h%uint64(len(eligible)))]] + } + + // Split off CPU 0's core: its siblings tail the list, CPU 0 tails them. + preferred := make([]int, 0, len(cands)) + var zeroTail []int + hasZero := false + for _, c := range cands { + switch { + case c == 0: + hasZero = true + case topo.zeroCore >= 0 && topo.coreOf[c] == topo.zeroCore: + zeroTail = append(zeroTail, c) + default: + preferred = append(preferred, c) + } + } + if hasZero { + zeroTail = append(zeroTail, 0) + } + if len(preferred) == 0 { + return zeroTail // CPU 0's core is all we have + } + + // The node pick consumed the low hash bits; rotate by the high ones so + // the two choices stay independent. + off := int((h >> 32) % uint64(len(preferred))) + rot := make([]int, 0, len(preferred)) + rot = append(rot, preferred[off:]...) + rot = append(rot, preferred[:off]...) + + seenCore := make(map[int]bool, len(rot)) + out := make([]int, 0, len(cands)) + var siblings []int + for _, c := range rot { + g := topo.coreOf[c] + if seenCore[g] { + siblings = append(siblings, c) + continue + } + seenCore[g] = true + out = append(out, c) + } + out = append(out, siblings...) + out = append(out, zeroTail...) + return out +} + +// splitmix64 decorrelates instance keys before the selection modulos: ports +// on one box often share spacing (4242/4243, or round steps like +1000) that +// raw key%len arithmetic would fold onto the same offset. +func splitmix64(x uint64) uint64 { + x += 0x9e3779b97f4a7c15 + x = (x ^ (x >> 30)) * 0xbf58476d1ce4e5b9 + x = (x ^ (x >> 27)) * 0x94d049bb133111eb + return x ^ (x >> 31) +} diff --git a/cpupick/cpupick_test.go b/cpupick/cpupick_test.go new file mode 100644 index 00000000..bd81c1c9 --- /dev/null +++ b/cpupick/cpupick_test.go @@ -0,0 +1,171 @@ +package cpupick + +import ( + "slices" + "testing" +) + +// pairTopo builds a topology where consecutive candidate pairs are SMT +// siblings: (cpus[0],cpus[1]) share a core, (cpus[2],cpus[3]) the next, ... +// All CPUs land on node 0. +func pairTopo(cpus []int) topology { + t := topology{ + nodeOf: make(map[int]int, len(cpus)), + coreOf: make(map[int]int, len(cpus)), + zeroCore: -1, + } + for i, c := range cpus { + t.nodeOf[c] = 0 + t.coreOf[c] = i / 2 + if c == 0 { + t.zeroCore = i / 2 + } + } + return t +} + +func TestArrangeDemotesZeroForEveryKey(t *testing.T) { + candidates := []int{0, 1, 2, 3, 4, 5, 6, 7} + for key := range uint64(64) { + got := arrange(candidates, flatTopology(candidates), 4, splitmix64(key)) + if len(got) != len(candidates) { + t.Fatalf("key %d: len=%d want %d", key, len(got), len(candidates)) + } + if got[0] == 0 { + t.Errorf("key %d: CPU 0 at the front: %v", key, got) + } + if got[len(got)-1] != 0 { + t.Errorf("key %d: CPU 0 not demoted to last: %v", key, got) + } + sorted := slices.Clone(got) + slices.Sort(sorted) + if !slices.Equal(sorted, candidates) { + t.Errorf("key %d: not a permutation: %v", key, got) + } + } +} + +func TestArrangeDemotesZeroSiblings(t *testing.T) { + // Pairs (0,1),(2,3),(4,5),(6,7): CPU 0's core — 0 and its sibling 1 — + // must tail the list, sibling ahead of 0 itself. + candidates := []int{0, 1, 2, 3, 4, 5, 6, 7} + for key := range uint64(64) { + got := arrange(candidates, pairTopo(candidates), 2, splitmix64(key)) + n := len(got) + if got[n-1] != 0 || got[n-2] != 1 { + t.Fatalf("key %d: tail = %v, want [... 1 0]", key, got) + } + } +} + +func TestArrangeZeroSiblingWithoutZero(t *testing.T) { + // CPU 0 excluded (cpuset) but its sibling 1 remains: the sibling still + // tails the list when the topology knows which core CPU 0 lives on. + candidates := []int{1, 2, 3, 4, 5} + topo := pairTopo([]int{0, 1, 2, 3, 4, 5}) + got := arrange(candidates, topo, 2, splitmix64(7)) + if got[len(got)-1] != 1 { + t.Errorf("CPU 0's sibling not demoted: %v", got) + } +} + +func TestArrangeRotatesByKey(t *testing.T) { + candidates := []int{1, 2, 3, 4, 5, 6, 7, 8} + seen := map[int]bool{} + for key := range uint64(64) { + seen[arrange(candidates, flatTopology(candidates), 4, splitmix64(key))[0]] = true + } + // 64 hashed keys over 8 slots must hit more than one starting CPU, or + // co-located instances would all stack again. + if len(seen) < 2 { + t.Errorf("rotation never varied across keys: %v", seen) + } +} + +func TestArrangeStableForSameKey(t *testing.T) { + candidates := []int{0, 2, 4, 6} + topo := flatTopology(candidates) + a := arrange(candidates, topo, 2, splitmix64(4242)) + b := arrange(candidates, topo, 2, splitmix64(4242)) + if !slices.Equal(a, b) { + t.Errorf("same key ordered differently: %v vs %v", a, b) + } +} + +func TestArrangeZeroOnly(t *testing.T) { + if got := arrange([]int{0}, flatTopology([]int{0}), 1, splitmix64(7)); !slices.Equal(got, []int{0}) { + t.Errorf("sole CPU 0 must survive: %v", got) + } +} + +func TestArrangeSMTSiblingsLast(t *testing.T) { + // Pairs (1,2),(3,4),(5,6),(7,8): the first four picks must cover four + // distinct physical cores before any sibling repeats. + candidates := []int{1, 2, 3, 4, 5, 6, 7, 8} + topo := pairTopo(candidates) + for key := range uint64(16) { + got := arrange(candidates, topo, 4, splitmix64(key)) + seen := map[int]bool{} + for _, c := range got[:4] { + g := topo.coreOf[c] + if seen[g] { + t.Fatalf("key %d: sibling before all cores covered: %v", key, got) + } + seen[g] = true + } + } +} + +func TestArrangeNUMAConfinesToOneNode(t *testing.T) { + // Two nodes of four; both fit routines=3, so the result must sit + // entirely inside one of them, and the hash must pick both across keys. + candidates := []int{1, 2, 3, 4, 10, 11, 12, 13} + topo := flatTopology(candidates) + for _, c := range []int{10, 11, 12, 13} { + topo.nodeOf[c] = 1 + } + nodesSeen := map[int]bool{} + for key := range uint64(32) { + got := arrange(candidates, topo, 3, splitmix64(key)) + if len(got) != 4 { + t.Fatalf("key %d: not confined to one node: %v", key, got) + } + n := topo.nodeOf[got[0]] + for _, c := range got { + if topo.nodeOf[c] != n { + t.Fatalf("key %d: spans nodes: %v", key, got) + } + } + nodesSeen[n] = true + } + if len(nodesSeen) != 2 { + t.Errorf("hash never spread instances across nodes: %v", nodesSeen) + } +} + +func TestArrangeNUMASpansWhenNoNodeFits(t *testing.T) { + candidates := []int{1, 2, 3, 4, 10, 11, 12, 13} + topo := flatTopology(candidates) + for _, c := range []int{10, 11, 12, 13} { + topo.nodeOf[c] = 1 + } + got := arrange(candidates, topo, 6, splitmix64(1)) + if len(got) != len(candidates) { + t.Errorf("undersized nodes must span, got %v", got) + } +} + +func TestPickCandidates(t *testing.T) { + allowed := []int{0, 1, 2, 3, 4, 5, 6, 7} + perf := []int{4, 5} + + // Enough perf cores for every routine: only they are used. + if got := pickCandidates(allowed, perf, 2); !slices.Equal(got, perf) { + t.Errorf("perf filter not applied: %v", got) + } + // Perf filter too small for the routine count: discarded, everyone + // gets their own core from the full allowed set. + if got := pickCandidates(allowed, perf, 4); !slices.Equal(got, allowed) { + t.Errorf("undersized perf filter not discarded: %v", got) + } +} diff --git a/cpupick/perf_linux.go b/cpupick/perf_linux.go new file mode 100644 index 00000000..f5bf76b5 --- /dev/null +++ b/cpupick/perf_linux.go @@ -0,0 +1,154 @@ +//go:build linux + +package cpupick + +import ( + "fmt" + "os" + "path/filepath" + "strconv" + "strings" +) + +// capacityKeepPct is the cpu_capacity admission threshold, relative to the +// fastest allowed core. LITTLE cores are normalized to ~250-400 of the big +// core's 1024 while mid cores sit at ~75%+, so half of max separates little +// from the rest without splitting prime from mid on three-tier parts. +const capacityKeepPct = 50 + +// freqKeepPct is the cpuinfo_max_freq admission threshold. Favored-core +// turbo skew is 2-4% and ARM mid-vs-prime ~12%, while E-cores, LITTLE +// cores, and AMD compact cores all sit >= 20% below their siblings' max. +const freqKeepPct = 85 + +// perfCPUs partitions allowed into the subset that are "performance" cores, +// consulting (in order of authority): +// +// 1. cpu_capacity — arch_topology's normalized per-CPU capacity, exposed on +// arm/arm64/riscv; the scheduler's own view of big vs LITTLE. +// 2. /sys/devices/cpu_core/cpus — the Intel hybrid P-core PMU mask, present +// only on P/E parts (x86 has no cpu_capacity) and naming P cores outright. +// 3. cpuinfo_max_freq — the cross-vendor fallback; catches AMD compact +// cores, which neither of the above covers. +// +// Returns allowed unchanged (signal "") when nothing distinguishes the +// cores: homogeneous parts, VMs without cpufreq, sysfs unavailable. +func perfCPUs(allowed []int) ([]int, string) { + return perfCPUsFrom("/sys/devices/system/cpu", "/sys/devices/cpu_core/cpus", allowed) +} + +func perfCPUsFrom(cpuDir, intelCoreMask string, allowed []int) ([]int, string) { + if cpus, ok := byPerCPUValue(cpuDir, "cpu_capacity", allowed, capacityKeepPct); ok { + return cpus, "cpu_capacity" + } + if cpus, ok := byIntelCoreMask(intelCoreMask, allowed); ok { + return cpus, "intel_core_pmu" + } + if cpus, ok := byPerCPUValue(cpuDir, "cpufreq/cpuinfo_max_freq", allowed, freqKeepPct); ok { + return cpus, "max_freq" + } + return allowed, "" +} + +// byPerCPUValue keeps the allowed CPUs whose per-CPU sysfs value is at least +// keepPct percent of the maximum across allowed. Inconclusive (ok=false) +// when any CPU is missing the file or when every value is equal. +func byPerCPUValue(cpuDir, file string, allowed []int, keepPct int) ([]int, bool) { + vals := make([]int, len(allowed)) + minV, maxV := 0, 0 + for i, cpu := range allowed { + v, err := readIntFile(filepath.Join(cpuDir, fmt.Sprintf("cpu%d", cpu), file)) + if err != nil { + return nil, false + } + vals[i] = v + if i == 0 || v < minV { + minV = v + } + if v > maxV { + maxV = v + } + } + if minV == maxV { + return nil, false // homogeneous by this signal; try the next one + } + keep := make([]int, 0, len(allowed)) + for i, cpu := range allowed { + if vals[i]*100 >= maxV*keepPct { + keep = append(keep, cpu) + } + } + return keep, true +} + +// byIntelCoreMask keeps the allowed CPUs named by the hybrid P-core PMU +// mask. Inconclusive when the file is absent (non-hybrid x86, other arches) +// or no allowed CPU is in the mask (the process was deliberately confined +// to E-cores; nothing useful to prefer within that). +func byIntelCoreMask(maskPath string, allowed []int) ([]int, bool) { + b, err := os.ReadFile(maskPath) + if err != nil { + return nil, false + } + set, err := parseCPUList(strings.TrimSpace(string(b))) + if err != nil || len(set) == 0 { + return nil, false + } + pcore := make(map[int]bool, len(set)) + for _, c := range set { + pcore[c] = true + } + keep := make([]int, 0, len(allowed)) + for _, cpu := range allowed { + if pcore[cpu] { + keep = append(keep, cpu) + } + } + if len(keep) == 0 { + return nil, false + } + return keep, true +} + +// parseCPUList decodes the kernel's cpulist format ("0-7,16-23", "3") into +// individual CPU IDs. Empty input yields an empty list. +func parseCPUList(s string) ([]int, error) { + if s == "" { + return nil, nil + } + var out []int + for part := range strings.SplitSeq(s, ",") { + part = strings.TrimSpace(part) + if part == "" { + continue + } + lo, hi, isRange := strings.Cut(part, "-") + a, err := strconv.Atoi(lo) + if err != nil { + return nil, fmt.Errorf("bad cpulist entry %q: %w", part, err) + } + if !isRange { + out = append(out, a) + continue + } + b, err := strconv.Atoi(hi) + if err != nil { + return nil, fmt.Errorf("bad cpulist entry %q: %w", part, err) + } + if b < a || b-a > 8192 { + return nil, fmt.Errorf("bad cpulist range %q", part) + } + for v := a; v <= b; v++ { + out = append(out, v) + } + } + return out, nil +} + +func readIntFile(path string) (int, error) { + b, err := os.ReadFile(path) + if err != nil { + return 0, err + } + return strconv.Atoi(strings.TrimSpace(string(b))) +} diff --git a/cpupick/perf_linux_test.go b/cpupick/perf_linux_test.go new file mode 100644 index 00000000..00aee422 --- /dev/null +++ b/cpupick/perf_linux_test.go @@ -0,0 +1,163 @@ +//go:build linux + +package cpupick + +import ( + "fmt" + "os" + "path/filepath" + "slices" + "testing" +) + +// fakeSysfs builds a cpuDir tree with the given per-CPU file values. +// A nil map for a file means "file absent on every CPU". +func fakeSysfs(t *testing.T, capacity, maxFreq map[int]int) string { + t.Helper() + dir := t.TempDir() + write := func(cpu int, rel string, v int) { + p := filepath.Join(dir, fmt.Sprintf("cpu%d", cpu), rel) + if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(p, fmt.Appendf(nil, "%d\n", v), 0o644); err != nil { + t.Fatal(err) + } + } + for cpu, v := range capacity { + write(cpu, "cpu_capacity", v) + } + for cpu, v := range maxFreq { + write(cpu, "cpufreq/cpuinfo_max_freq", v) + } + return dir +} + +func writeCoreMask(t *testing.T, mask string) string { + t.Helper() + p := filepath.Join(t.TempDir(), "cpus") + if err := os.WriteFile(p, []byte(mask+"\n"), 0o644); err != nil { + t.Fatal(err) + } + return p +} + +func TestPerfCPUsBigLittleCapacity(t *testing.T) { + // 4 big (1024) + 4 LITTLE (~290): capacity is authoritative on ARM. + dir := fakeSysfs(t, map[int]int{ + 0: 1024, 1: 1024, 2: 1024, 3: 1024, + 4: 290, 5: 290, 6: 290, 7: 290, + }, nil) + got, signal := perfCPUsFrom(dir, filepath.Join(dir, "nope"), []int{0, 1, 2, 3, 4, 5, 6, 7}) + if signal != "cpu_capacity" { + t.Fatalf("signal = %q", signal) + } + if !slices.Equal(got, []int{0, 1, 2, 3}) { + t.Errorf("got %v", got) + } +} + +func TestPerfCPUsThreeTierKeepsMid(t *testing.T) { + // prime (1024) + mid (~780) + little (~280): 50% keeps prime+mid. + dir := fakeSysfs(t, map[int]int{ + 0: 280, 1: 280, 2: 280, 3: 280, + 4: 780, 5: 780, 6: 780, + 7: 1024, + }, nil) + got, _ := perfCPUsFrom(dir, filepath.Join(dir, "nope"), []int{0, 1, 2, 3, 4, 5, 6, 7}) + if !slices.Equal(got, []int{4, 5, 6, 7}) { + t.Errorf("got %v", got) + } +} + +func TestPerfCPUsIntelHybridMask(t *testing.T) { + // No cpu_capacity on x86; the P-core PMU mask decides. + dir := fakeSysfs(t, nil, nil) + mask := writeCoreMask(t, "0-7") + got, signal := perfCPUsFrom(dir, mask, []int{0, 1, 2, 3, 8, 9, 10, 11}) + if signal != "intel_core_pmu" { + t.Fatalf("signal = %q", signal) + } + if !slices.Equal(got, []int{0, 1, 2, 3}) { + t.Errorf("got %v", got) + } +} + +func TestPerfCPUsIntelMaskDisjointFallsThrough(t *testing.T) { + // Confined to E-cores only: the mask can't help, and equal freqs below + // mean nothing else distinguishes them either -> allowed unchanged. + dir := fakeSysfs(t, nil, map[int]int{8: 4300000, 9: 4300000}) + mask := writeCoreMask(t, "0-7") + got, signal := perfCPUsFrom(dir, mask, []int{8, 9}) + if signal != "" || !slices.Equal(got, []int{8, 9}) { + t.Errorf("got %v signal %q", got, signal) + } +} + +func TestPerfCPUsMaxFreqCompactCores(t *testing.T) { + // AMD-style compact cores: no capacity, no Intel mask; 3.3 vs 5.7 GHz. + dir := fakeSysfs(t, nil, map[int]int{ + 0: 5700000, 1: 5700000, 2: 3300000, 3: 3300000, + }) + got, signal := perfCPUsFrom(dir, filepath.Join(dir, "nope"), []int{0, 1, 2, 3}) + if signal != "max_freq" { + t.Fatalf("signal = %q", signal) + } + if !slices.Equal(got, []int{0, 1}) { + t.Errorf("got %v", got) + } +} + +func TestPerfCPUsFavoredCoreSkewKept(t *testing.T) { + // Turbo Boost Max favored cores run a few percent hot; they must not + // shrink the candidate set to one or two cores. + dir := fakeSysfs(t, nil, map[int]int{ + 0: 5800000, 1: 5700000, 2: 5700000, 3: 5600000, + }) + got, _ := perfCPUsFrom(dir, filepath.Join(dir, "nope"), []int{0, 1, 2, 3}) + if !slices.Equal(got, []int{0, 1, 2, 3}) { + t.Errorf("favored-core skew filtered CPUs: %v", got) + } +} + +func TestPerfCPUsHomogeneousInconclusive(t *testing.T) { + dir := fakeSysfs(t, nil, map[int]int{0: 3000000, 1: 3000000}) + got, signal := perfCPUsFrom(dir, filepath.Join(dir, "nope"), []int{0, 1}) + if signal != "" || !slices.Equal(got, []int{0, 1}) { + t.Errorf("got %v signal %q", got, signal) + } +} + +func TestPerfCPUsNoSysfs(t *testing.T) { + dir := t.TempDir() + got, signal := perfCPUsFrom(dir, filepath.Join(dir, "nope"), []int{0, 1, 2}) + if signal != "" || !slices.Equal(got, []int{0, 1, 2}) { + t.Errorf("got %v signal %q", got, signal) + } +} + +func TestParseCPUList(t *testing.T) { + cases := []struct { + in string + want []int + wantErr bool + }{ + {"0-3", []int{0, 1, 2, 3}, false}, + {"0-1,16-17", []int{0, 1, 16, 17}, false}, + {"5", []int{5}, false}, + {"", nil, false}, + {"3-1", nil, true}, + {"a-b", nil, true}, + {"1,x", nil, true}, + } + for _, c := range cases { + got, err := parseCPUList(c.in) + if (err != nil) != c.wantErr { + t.Errorf("%q: err=%v wantErr=%v", c.in, err, c.wantErr) + continue + } + if !c.wantErr && !slices.Equal(got, c.want) { + t.Errorf("%q: got %v want %v", c.in, got, c.want) + } + } +} diff --git a/cpupick/perf_other.go b/cpupick/perf_other.go new file mode 100644 index 00000000..9be94583 --- /dev/null +++ b/cpupick/perf_other.go @@ -0,0 +1,10 @@ +//go:build !linux + +package cpupick + +// perfCPUs is Linux-only sysfs walking; elsewhere report "no distinction". +// Default already returns nil off-Linux (util.AllowedCPUs has no answer +// there), so this exists to keep the package compiling everywhere. +func perfCPUs(allowed []int) ([]int, string) { + return allowed, "" +} diff --git a/cpupick/topo_linux.go b/cpupick/topo_linux.go new file mode 100644 index 00000000..f72ed950 --- /dev/null +++ b/cpupick/topo_linux.go @@ -0,0 +1,118 @@ +//go:build linux + +package cpupick + +import ( + "fmt" + "os" + "path/filepath" + "strconv" + "strings" +) + +// readTopology probes the NUMA node and physical-core layout of cpus from +// sysfs. Anything sysfs won't say degrades toward flatTopology: an unknown +// node becomes node 0, an unknown core becomes a core of its own — either +// way the corresponding arrange rule becomes a no-op instead of a wrong +// answer. +func readTopology(cpus []int) topology { + return readTopologyFrom("/sys/devices/system/node", "/sys/devices/system/cpu", cpus) +} + +func readTopologyFrom(nodeDir, cpuDir string, cpus []int) topology { + coreOf, zeroCore := coreGroups(cpuDir, cpus) + return topology{ + nodeOf: numaNodes(nodeDir, cpus), + coreOf: coreOf, + zeroCore: zeroCore, + } +} + +// numaNodes maps each cpu to its NUMA node via +// /sys/devices/system/node/nodeN/cpulist. CPUs no node claims (or no node +// dirs at all: VMs, non-NUMA kernels) land on node 0. +func numaNodes(nodeDir string, cpus []int) map[int]int { + out := make(map[int]int, len(cpus)) + for _, c := range cpus { + out[c] = 0 + } + entries, err := os.ReadDir(nodeDir) + if err != nil { + return out + } + want := make(map[int]bool, len(cpus)) + for _, c := range cpus { + want[c] = true + } + for _, e := range entries { + id, ok := strings.CutPrefix(e.Name(), "node") + if !ok { + continue + } + n, err := strconv.Atoi(id) + if err != nil { + continue // has_cpu, possible, ... share the prefix + } + b, err := os.ReadFile(filepath.Join(nodeDir, e.Name(), "cpulist")) + if err != nil { + continue + } + list, err := parseCPUList(strings.TrimSpace(string(b))) + if err != nil { + continue + } + for _, c := range list { + if want[c] { + out[c] = n + } + } + } + return out +} + +// coreGroups maps each cpu to a dense physical-core id derived from its +// (physical_package_id, core_id) pair — core_id alone repeats across +// sockets. CPUs whose topology files are unreadable get a core of their own. +// The second return is the group id of the core CPU 0 lives on, or -1 when +// that can't be determined; CPU 0's own files are consulted even when 0 is +// not a candidate, so its SMT siblings are recognized under cpusets that +// exclude CPU 0 itself. +func coreGroups(cpuDir string, cpus []int) (map[int]int, int) { + type pkgCore struct{ pkg, core int } + pairOf := func(cpu int) (pkgCore, bool) { + topoDir := filepath.Join(cpuDir, fmt.Sprintf("cpu%d", cpu), "topology") + pkg, err1 := readIntFile(filepath.Join(topoDir, "physical_package_id")) + core, err2 := readIntFile(filepath.Join(topoDir, "core_id")) + if err1 != nil || err2 != nil { + return pkgCore{}, false + } + return pkgCore{pkg, core}, true + } + + ids := map[pkgCore]int{} + out := make(map[int]int, len(cpus)) + next := 0 + for _, cpu := range cpus { + k, ok := pairOf(cpu) + if !ok { + out[cpu] = next + next++ + continue + } + id, ok := ids[k] + if !ok { + id = next + next++ + ids[k] = id + } + out[cpu] = id + } + + zeroCore := -1 + if k, ok := pairOf(0); ok { + if id, ok := ids[k]; ok { + zeroCore = id + } + } + return out, zeroCore +} diff --git a/cpupick/topo_linux_test.go b/cpupick/topo_linux_test.go new file mode 100644 index 00000000..2b914328 --- /dev/null +++ b/cpupick/topo_linux_test.go @@ -0,0 +1,111 @@ +//go:build linux + +package cpupick + +import ( + "fmt" + "os" + "path/filepath" + "testing" +) + +// fakeTopoSysfs builds nodeDir/cpuDir trees. nodes maps node id -> cpulist +// string; cores maps cpu -> (package, core) pair. +func fakeTopoSysfs(t *testing.T, nodes map[int]string, cores map[int][2]int) (string, string) { + t.Helper() + base := t.TempDir() + nodeDir := filepath.Join(base, "node") + cpuDir := filepath.Join(base, "cpu") + for n, list := range nodes { + d := filepath.Join(nodeDir, fmt.Sprintf("node%d", n)) + if err := os.MkdirAll(d, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(d, "cpulist"), []byte(list+"\n"), 0o644); err != nil { + t.Fatal(err) + } + } + for cpu, pc := range cores { + d := filepath.Join(cpuDir, fmt.Sprintf("cpu%d", cpu), "topology") + if err := os.MkdirAll(d, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(d, "physical_package_id"), fmt.Appendf(nil, "%d\n", pc[0]), 0o644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(d, "core_id"), fmt.Appendf(nil, "%d\n", pc[1]), 0o644); err != nil { + t.Fatal(err) + } + } + return nodeDir, cpuDir +} + +func TestReadTopology(t *testing.T) { + // Two nodes; SMT pairs (0,4),(1,5) on node 0 and (2,6),(3,7) on node 1. + // core_id repeats across packages on purpose: the pair must disambiguate. + nodeDir, cpuDir := fakeTopoSysfs(t, + map[int]string{0: "0-1,4-5", 1: "2-3,6-7"}, + map[int][2]int{ + 0: {0, 0}, 4: {0, 0}, 1: {0, 1}, 5: {0, 1}, + 2: {1, 0}, 6: {1, 0}, 3: {1, 1}, 7: {1, 1}, + }) + cpus := []int{0, 1, 2, 3, 4, 5, 6, 7} + topo := readTopologyFrom(nodeDir, cpuDir, cpus) + + for _, c := range []int{0, 1, 4, 5} { + if topo.nodeOf[c] != 0 { + t.Errorf("cpu %d on node %d, want 0", c, topo.nodeOf[c]) + } + } + for _, c := range []int{2, 3, 6, 7} { + if topo.nodeOf[c] != 1 { + t.Errorf("cpu %d on node %d, want 1", c, topo.nodeOf[c]) + } + } + pairs := [][2]int{{0, 4}, {1, 5}, {2, 6}, {3, 7}} + for _, p := range pairs { + if topo.coreOf[p[0]] != topo.coreOf[p[1]] { + t.Errorf("siblings %v not grouped: %d vs %d", p, topo.coreOf[p[0]], topo.coreOf[p[1]]) + } + } + if topo.coreOf[0] == topo.coreOf[2] { + t.Error("cross-package cores with equal core_id must not merge") + } + if topo.zeroCore != topo.coreOf[0] { + t.Errorf("zeroCore = %d, want %d", topo.zeroCore, topo.coreOf[0]) + } +} + +func TestReadTopologyZeroCoreWithoutZeroCandidate(t *testing.T) { + // CPU 0 is not a candidate (cpuset excludes it) but its sibling 4 is: + // zeroCore must still identify their shared core. + nodeDir, cpuDir := fakeTopoSysfs(t, + map[int]string{0: "0-7"}, + map[int][2]int{0: {0, 0}, 4: {0, 0}, 1: {0, 1}, 5: {0, 1}}) + topo := readTopologyFrom(nodeDir, cpuDir, []int{1, 4, 5}) + if topo.zeroCore < 0 || topo.coreOf[4] != topo.zeroCore { + t.Errorf("zeroCore = %d, coreOf[4] = %d; sibling of CPU 0 not identified", topo.zeroCore, topo.coreOf[4]) + } + if topo.coreOf[1] == topo.zeroCore { + t.Error("cpu 1 wrongly grouped with CPU 0's core") + } +} + +func TestReadTopologyMissingSysfs(t *testing.T) { + base := t.TempDir() + cpus := []int{0, 1, 2} + topo := readTopologyFrom(filepath.Join(base, "nope"), filepath.Join(base, "also-nope"), cpus) + seen := map[int]bool{} + for _, c := range cpus { + if topo.nodeOf[c] != 0 { + t.Errorf("cpu %d node = %d, want 0", c, topo.nodeOf[c]) + } + if seen[topo.coreOf[c]] { + t.Errorf("cpu %d shares a fallback core group", c) + } + seen[topo.coreOf[c]] = true + } + if topo.zeroCore != -1 { + t.Errorf("zeroCore = %d, want -1 when unknown", topo.zeroCore) + } +} diff --git a/cpupick/topo_other.go b/cpupick/topo_other.go new file mode 100644 index 00000000..8003ce4a --- /dev/null +++ b/cpupick/topo_other.go @@ -0,0 +1,10 @@ +//go:build !linux + +package cpupick + +// readTopology has no sysfs to consult off Linux; the flat stand-in makes +// arrange's NUMA and SMT rules no-ops. Default is already nil off Linux +// (util.AllowedCPUs has no answer there) — this keeps the package compiling. +func readTopology(cpus []int) topology { + return flatTopology(cpus) +} diff --git a/main.go b/main.go index db38a1a5..4fb47ed6 100644 --- a/main.go +++ b/main.go @@ -6,12 +6,14 @@ import ( "log/slog" "net" "net/netip" + "os" "runtime/debug" "slices" "strings" "time" "github.com/slackhq/nebula/config" + "github.com/slackhq/nebula/cpupick" "github.com/slackhq/nebula/overlay" "github.com/slackhq/nebula/sshd" "github.com/slackhq/nebula/udp" @@ -220,6 +222,21 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev l.Warn("Failed to start DNS responder", "error", err) } + pinThreads := c.GetBool("tun.pin_threads", true) + cpuAffinity := parseCpuAffinity(c, l, routines) + if pinThreads && routines > 1 && len(cpuAffinity) == 0 && !configTest { + // The operator didn't choose pin CPUs, so pick a default set that + // prefers performance cores and doesn't stack co-located instances + // onto allowed[0]. The bound UDP port keys the per-instance spread: + // distinct across instances sharing a box, stable across restarts. + // A nil result keeps listenIn's stock allowed[i] fallback. + key := uint64(os.Getpid()) + if ap, err := udpConns[0].LocalAddr(); err == nil && ap.Port() != 0 { + key = uint64(ap.Port()) + } + cpuAffinity = cpupick.Default(routines, key, l) + } + ifConfig := &InterfaceConfig{ HostMap: hostMap, Inside: tun, @@ -241,8 +258,8 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev relayManager: NewRelayManager(ctx, l, hostMap, c), punchy: punchy, ConntrackCacheTimeout: conntrackCacheTimeout, - CpuAffinity: parseCpuAffinity(c, l, routines), - PinThreads: c.GetBool("tun.pin_threads", true), + CpuAffinity: cpuAffinity, + PinThreads: pinThreads, l: l, }