cpupick: Make tun.avoid_cpus a pin selection rule instead of a process-wide mask

This commit is contained in:
JackDoan committed 2026-10-15 12:13:21 -05:00
1 parent 36579bae03
commit df1c4d9364
7 files changed
+127 -58

No files matched your search

+2 -2
View File
@@ -15,8 +15,8 @@ import (
// Avoid removes cpus from the affinity of every thread of the process, for use at startup. New threads inherit // Avoid removes cpus from the affinity of every thread of the process, for use at startup. New threads inherit
// their parent's mask, so it repeats until a pass finds nothing left to change; a thread that exits meanwhile is // their parent's mask, so it repeats until a pass finds nothing left to change; a thread that exits meanwhile is
// skipped. Every mask is checked before any is changed. With apply false it only validates, for configuration tests. // skipped. Every mask is checked before any is changed. With apply false it only validates, for configuration tests.
// This affects the entire process, including other users of an embedded Nebula. Run it before Default, which then // This affects the entire process, including other users of an embedded Nebula. Not wired up yet: tun.avoid_cpus
// picks from what remains. // only steers Default for now.
func Avoid(cpus []int, apply bool) error { func Avoid(cpus []int, apply bool) error {
const capacity = int(unsafe.Sizeof(unix.CPUSet{})) * 8 const capacity = int(unsafe.Sizeof(unix.CPUSet{})) * 8
for _, cpu := range cpus { for _, cpu := range cpus {
+16 -9
View File
@@ -17,13 +17,14 @@
// core demoted to last resort, and the order rotated by a stable per-instance // core demoted to last resort, and the order rotated by a stable per-instance
// key so co-located instances spread instead of stacking. // key so co-located instances spread instead of stacking.
// //
// Avoid narrows the process-wide allowed set before any of that happens // CPUs the operator asked to keep readers off (tun.avoid_cpus, typically the
// (tun.avoid_cpus), so Default and tun.cpu_affinity validation only ever see // underlay NIC's receive IRQ CPUs) are removed from the allowed set before
// the CPUs left over. // any of that, so no rule can bring one back.
package cpupick package cpupick
import ( import (
"log/slog" "log/slog"
"slices"
"github.com/slackhq/nebula/util" "github.com/slackhq/nebula/util"
) )
@@ -61,12 +62,17 @@ func flatTopology(cpus []int) topology {
// Default computes the pin order for `routines` tun readers. key is any // Default computes the pin order for `routines` tun readers. key is any
// stable per-instance value; the bound UDP port is ideal — distinct across // stable per-instance value; the bound UDP port is ideal — distinct across
// co-located instances, stable across restarts so benchmark runs stay // co-located instances, stable across restarts so benchmark runs stay
// comparable. Returns nil when there is nothing useful to say (no affinity // comparable. avoid lists CPUs never to return, matched literally. Returns
// support on this platform, lookup failure); callers keep their existing // nil when there is nothing useful to say (no affinity support on this
// fallback spread. // platform, lookup failure, avoid covering every allowed CPU); callers keep
func Default(routines int, key uint64, l *slog.Logger) []int { // their existing fallback spread.
func Default(routines int, key uint64, avoid []int, l *slog.Logger) []int {
allowed, err := util.AllowedCPUs() allowed, err := util.AllowedCPUs()
if err != nil || len(allowed) == 0 { if err != nil {
return nil
}
allowed = slices.DeleteFunc(allowed, func(c int) bool { return slices.Contains(avoid, c) })
if len(allowed) == 0 {
return nil return nil
} }
perf, signal := perfCPUs(allowed) perf, signal := perfCPUs(allowed)
@@ -81,7 +87,8 @@ func Default(routines int, key uint64, l *slog.Logger) []int {
if l != nil { if l != nil {
l.Info("chose default pin CPUs for tun readers", l.Info("chose default pin CPUs for tun readers",
"cpus", cpus[:min(routines, len(cpus))], "cpus", cpus[:min(routines, len(cpus))],
"perfSignal", signal) "perfSignal", signal,
"avoided", avoid)
} }
return cpus return cpus
} }
+35
View File
@@ -0,0 +1,35 @@
//go:build linux && !android && !e2e_testing
package cpupick
import (
"slices"
"testing"
"github.com/slackhq/nebula/util"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestDefaultSkipsAvoided(t *testing.T) {
allowed, err := util.AllowedCPUs()
require.NoError(t, err)
if len(allowed) < 2 {
t.Skip("requires at least two allowed CPUs")
}
avoid := allowed[:len(allowed)/2]
for key := range uint64(16) {
got := Default(len(allowed), key, avoid, nil)
require.NotEmpty(t, got)
for _, cpu := range got {
assert.NotContains(t, avoid, cpu, "key %d", key)
assert.Contains(t, allowed, cpu, "key %d", key)
}
}
assert.Nil(t, Default(2, 0, allowed, nil), "avoiding everything leaves nothing to suggest")
// Without avoid every allowed CPU stays in play
got := Default(len(allowed), 0, nil, nil)
slices.Sort(got)
assert.Equal(t, allowed, got)
}
+7 -6
View File
@@ -295,12 +295,13 @@ tun:
# - 2 # - 2
# - 4 # - 4
# Linux only. avoid_cpus removes these CPU IDs from the affinity of every thread of the process at startup, # Linux only. avoid_cpus lists CPU IDs the tun readers are never pinned to, exactly as listed (SMT siblings are not
# embedding applications included; later threads inherit the remaining set. Applied before cpu_affinity and the # added for you). The default pin selection skips them, even with a single routine, and a cpu_affinity entry naming
# default pin selection, independent of pin_threads. Invalid IDs, or IDs that would leave no CPU, fail # one is rejected like any other invalid entry. Other threads are not restricted, and the setting does nothing with
# startup. Not reloadable; ignored on other platforms. To keep Nebula off the underlay NIC's receive IRQ CPUs, find # pin_threads: false. Invalid IDs, or IDs covering every allowed CPU, fail startup. Not reloadable. To keep Nebula
# them in /proc/interrupts and /proc/irq/<IRQ>/effective_affinity_list, add their SMT siblings from # off the underlay NIC's receive IRQ CPUs, find them in /proc/interrupts and /proc/irq/<IRQ>/effective_affinity_list,
# /sys/devices/system/cpu/cpu<N>/topology/thread_siblings_list, and revisit the list when IRQ placement changes. # add their SMT siblings from /sys/devices/system/cpu/cpu<N>/topology/thread_siblings_list if you want those kept
# clear too, and revisit the list when IRQ placement changes.
#avoid_cpus: [] #avoid_cpus: []
# Route based MTU overrides, you have known vpn ip paths that can support larger MTUs you can increase/decrease them here # Route based MTU overrides, you have known vpn ip paths that can support larger MTUs you can increase/decrease them here
+1 -1
View File
@@ -437,7 +437,7 @@ func (f *Interface) pinThisThread(i int) {
var cpu int var cpu int
if n := len(f.cpuAffinity); n > 0 { if n := len(f.cpuAffinity); n > 0 {
// Explicit tun.cpu_affinity list wins; parseCpuAffinity already // Explicit tun.cpu_affinity list wins; parseCpuAffinity already
// validated the entries against the allowed CPU set. // validated the entries against the allowed CPU set and tun.avoid_cpus.
cpu = f.cpuAffinity[i%n] cpu = f.cpuAffinity[i%n]
} else if allowed, err := util.AllowedCPUs(); err == nil && len(allowed) > 0 { } else if allowed, err := util.AllowedCPUs(); err == nil && len(allowed) > 0 {
// Default: spread queues across the CPUs we're actually allowed to // Default: spread queues across the CPUs we're actually allowed to
+38 -31
View File
@@ -2,7 +2,6 @@ package nebula
import ( import (
"context" "context"
"errors"
"fmt" "fmt"
"log/slog" "log/slog"
"net" "net"
@@ -32,10 +31,6 @@ const maxRoutines = 1 << 16
const _ = noiseutil.RejectHeadroom - 4*maxRoutines const _ = noiseutil.RejectHeadroom - 4*maxRoutines
func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, deviceFactory overlay.DeviceFactory) (retcon *Control, reterr error) { func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, deviceFactory overlay.DeviceFactory) (retcon *Control, reterr error) {
// Apply before starting services or selecting per-reader pin CPUs.
if err := configureAvoidCPUs(c, l, configTest); err != nil {
return nil, err
}
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
// The goroutines started below stop only when this context does, and only a caller holding the // The goroutines started below stop only when this context does, and only a caller holding the
// Control can arrange that. Cancel whenever we are not handing one back, which covers an error // Control can arrange that. Cancel whenever we are not handing one back, which covers an error
@@ -248,11 +243,24 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev
} }
pinThreads := c.GetBool("tun.pin_threads", true) pinThreads := c.GetBool("tun.pin_threads", true)
cpuAffinity := parseCpuAffinity(c, l, routines) avoidCPUs, err := parseAvoidCPUs(c)
if pinThreads && routines > 1 && len(cpuAffinity) == 0 && !configTest { if err != nil {
return nil, err
}
if len(avoidCPUs) > 0 && !pinThreads {
l.Warn("tun.avoid_cpus has no effect with tun.pin_threads disabled")
}
// Keeping every thread of the process off avoidCPUs, not just the pinned readers, is parked for now:
//if err := cpupick.Avoid(avoidCPUs, !configTest); err != nil && !errors.Is(err, errors.ErrUnsupported) {
// return nil, fmt.Errorf("tun.avoid_cpus: %w", err)
//}
cpuAffinity := parseCpuAffinity(c, l, routines, avoidCPUs)
if pinThreads && (routines > 1 || len(avoidCPUs) > 0) && len(cpuAffinity) == 0 && !configTest {
// The operator didn't choose pin CPUs, so pick a default set that // The operator didn't choose pin CPUs, so pick a default set that
// prefers performance cores and doesn't stack co-located instances // prefers performance cores, skips tun.avoid_cpus and doesn't stack
// onto allowed[0]. // co-located instances onto allowed[0]. A single reader would
// otherwise just take allowed[0], so it only comes here to dodge
// tun.avoid_cpus.
// key is used to seed the spreading of routines->cores. // key is used to seed the spreading of routines->cores.
// use PID if you want to ensure many different Nebulas in VMs or containers land on different cores // use PID if you want to ensure many different Nebulas in VMs or containers land on different cores
@@ -275,7 +283,7 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev
l.Warn("tun.pin_threads_key is invalid, using PID") l.Warn("tun.pin_threads_key is invalid, using PID")
} }
cpuAffinity = cpupick.Default(routines, key, l) cpuAffinity = cpupick.Default(routines, key, avoidCPUs, l)
} }
ifConfig := &InterfaceConfig{ ifConfig := &InterfaceConfig{
@@ -368,8 +376,10 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev
// mask (util.AllowedCPUs) rather than 0..NumCPU-1: under a cgroup cpuset or // mask (util.AllowedCPUs) rather than 0..NumCPU-1: under a cgroup cpuset or
// taskset the runnable IDs are frequently not that contiguous range, and // 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 // pinning to an unrunnable ID always fails. If the allowed set can't be
// determined we fall back to a plain non-negative check. // determined we fall back to a plain non-negative check. An entry listed in
func parseCpuAffinity(c *config.C, l *slog.Logger, routines int) []int { // avoid (tun.avoid_cpus) contradicts the operator's own config and is treated
// the same as any other invalid entry.
func parseCpuAffinity(c *config.C, l *slog.Logger, routines int, avoid []int) []int {
raw := c.Get("tun.cpu_affinity") raw := c.Get("tun.cpu_affinity")
if raw == nil { if raw == nil {
return nil return nil
@@ -412,6 +422,11 @@ func parseCpuAffinity(c *config.C, l *slog.Logger, routines int) []int {
"index", i, "cpu", cpu, "allowed", allowed) "index", i, "cpu", cpu, "allowed", allowed)
return nil return nil
} }
if slices.Contains(avoid, cpu) {
l.Warn("tun.cpu_affinity entry is listed in tun.avoid_cpus; ignoring affinity",
"index", i, "cpu", cpu)
return nil
}
cpus = append(cpus, cpu) cpus = append(cpus, cpu)
} }
if len(cpus) != routines { if len(cpus) != routines {
@@ -421,39 +436,31 @@ func parseCpuAffinity(c *config.C, l *slog.Logger, routines int) []int {
return cpus return cpus
} }
// configureAvoidCPUs applies tun.avoid_cpus, a list of CPU IDs to keep every thread of the process off. Unlike // parseAvoidCPUs reads tun.avoid_cpus, a list of CPU IDs the tun readers must not be pinned to. IDs outside the
// tun.cpu_affinity, a bad entry fails startup rather than being ignored. Configuration tests only validate it. // allowed set are accepted, since one config may serve differently shaped hosts. Unlike tun.cpu_affinity, a bad
func configureAvoidCPUs(c *config.C, l *slog.Logger, configTest bool) error { // entry, or a list covering every allowed CPU, fails startup rather than being ignored.
func parseAvoidCPUs(c *config.C) ([]int, error) {
raw := c.Get("tun.avoid_cpus") raw := c.Get("tun.avoid_cpus")
if raw == nil { if raw == nil {
return nil return nil, nil
} }
values, ok := raw.([]any) values, ok := raw.([]any)
if !ok { if !ok {
return fmt.Errorf("tun.avoid_cpus must be a list of non-negative integers") return nil, fmt.Errorf("tun.avoid_cpus must be a list of non-negative integers")
} }
cpus := make([]int, 0, len(values)) cpus := make([]int, 0, len(values))
for _, value := range values { for _, value := range values {
cpu, ok := value.(int) cpu, ok := value.(int)
if !ok || cpu < 0 { if !ok || cpu < 0 {
return fmt.Errorf("tun.avoid_cpus must contain non-negative integers, got %v", value) return nil, fmt.Errorf("tun.avoid_cpus must contain non-negative integers, got %v", value)
} }
cpus = append(cpus, cpu) cpus = append(cpus, cpu)
} }
if len(cpus) == 0 { if allowed, err := util.AllowedCPUs(); err == nil && len(allowed) > 0 &&
return nil !slices.ContainsFunc(allowed, func(cpu int) bool { return !slices.Contains(cpus, cpu) }) {
return nil, fmt.Errorf("tun.avoid_cpus %v leaves none of the allowed CPUs %v", cpus, allowed)
} }
if err := cpupick.Avoid(cpus, !configTest); err != nil { return cpus, nil
if errors.Is(err, errors.ErrUnsupported) {
l.Warn("tun.avoid_cpus is only supported on Linux; ignoring")
return nil
}
return fmt.Errorf("tun.avoid_cpus: %w", err)
}
if !configTest {
l.Info("Excluded CPUs from process threads", "cpus", cpus)
}
return nil
} }
func moduleVersion() string { func moduleVersion() string {
+28 -9
View File
@@ -32,7 +32,7 @@ func TestParseCpuAffinity(t *testing.T) {
} }
// unset -> nil (listenIn falls back to spreading across the allowed set) // unset -> nil (listenIn falls back to spreading across the allowed set)
assert.Nil(t, parseCpuAffinity(newConfig(nil), l, 1)) assert.Nil(t, parseCpuAffinity(newConfig(nil), l, 1, nil))
// Pick a CPU we're actually allowed to run on so a valid list survives // Pick a CPU we're actually allowed to run on so a valid list survives
// validation regardless of the host's affinity mask. // validation regardless of the host's affinity mask.
@@ -43,41 +43,60 @@ func TestParseCpuAffinity(t *testing.T) {
} }
// valid list -> parsed through unchanged // valid list -> parsed through unchanged
assert.Equal(t, []int{validCPU, validCPU}, parseCpuAffinity(newConfig([]any{validCPU, validCPU}), l, 2)) assert.Equal(t, []int{validCPU, validCPU}, parseCpuAffinity(newConfig([]any{validCPU, validCPU}), l, 2, nil))
// a negative entry is out of range on every platform -> disables the override // a negative entry is out of range on every platform -> disables the override
assert.Nil(t, parseCpuAffinity(newConfig([]any{validCPU, -1}), l, 2)) assert.Nil(t, parseCpuAffinity(newConfig([]any{validCPU, -1}), l, 2, nil))
// a non-integer entry -> disables the override // a non-integer entry -> disables the override
assert.Nil(t, parseCpuAffinity(newConfig([]any{validCPU, "not-a-cpu"}), l, 2)) assert.Nil(t, parseCpuAffinity(newConfig([]any{validCPU, "not-a-cpu"}), l, 2, nil))
// a CPU id outside the allowed set -> disables the override. Only assertable // a CPU id outside the allowed set -> disables the override. Only assertable
// where we can enumerate the allowed set (e.g. linux); 1<<20 is far beyond // where we can enumerate the allowed set (e.g. linux); 1<<20 is far beyond
// any representable CPU id so it can never be in the mask. // any representable CPU id so it can never be in the mask.
if len(allowed) > 0 { if len(allowed) > 0 {
assert.Nil(t, parseCpuAffinity(newConfig([]any{1 << 20}), l, 1)) assert.Nil(t, parseCpuAffinity(newConfig([]any{1 << 20}), l, 1, nil))
} }
// an entry also listed in tun.avoid_cpus -> disables the override
assert.Nil(t, parseCpuAffinity(newConfig([]any{validCPU}), l, 1, []int{validCPU}))
} }
func TestConfigureAvoidCPUs(t *testing.T) { func TestParseAvoidCPUs(t *testing.T) {
l := test.NewLogger() l := test.NewLogger()
for _, tc := range []struct { for _, tc := range []struct {
value any value any
want []int
valid bool valid bool
}{ }{
{nil, true}, {[]any{}, true}, {"auto", false}, {"1,2", false}, {[]any{-1}, false}, {[]any{1.5}, false}, {nil, nil, true}, {[]any{}, []int{}, true}, {[]any{1, 1 << 20}, []int{1, 1 << 20}, true},
{[]any{"2"}, false}, {"auto", nil, false}, {"1,2", nil, false}, {[]any{-1}, nil, false}, {[]any{1.5}, nil, false},
{[]any{"2"}, nil, false},
} { } {
c := config.NewC(l) c := config.NewC(l)
if tc.value != nil { if tc.value != nil {
c.Settings["tun"] = map[string]any{"avoid_cpus": tc.value} c.Settings["tun"] = map[string]any{"avoid_cpus": tc.value}
} }
if err := configureAvoidCPUs(c, l, true); tc.valid { got, err := parseAvoidCPUs(c)
if tc.valid {
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, tc.want, got)
} else { } else {
require.Error(t, err, "%v", tc.value) require.Error(t, err, "%v", tc.value)
} }
} }
// avoiding every allowed CPU leaves nowhere to pin; only checkable where the allowed set is known
if allowed, _ := util.AllowedCPUs(); len(allowed) > 0 {
all := make([]any, len(allowed))
for i, cpu := range allowed {
all[i] = cpu
}
c := config.NewC(l)
c.Settings["tun"] = map[string]any{"avoid_cpus": all}
_, err := parseAvoidCPUs(c)
require.Error(t, err)
}
} }
// TestMain_ConfigTestReleasesItsGoroutines pins the rule that Main only leaves goroutines running // TestMain_ConfigTestReleasesItsGoroutines pins the rule that Main only leaves goroutines running