warn on uselessly low MTU

.github/actions/code-sign/action.yml

@@ -25,9 +25,9 @@ inputs:
     required: false
     default: "code-signer"
   key-prefix:
-    description: "S3 key prefix to write under; defaults to code-signing/<owner>/<repo> of the calling repo"
+    description: "S3 key prefix the caller is authorized to write under"
     required: false
-    default: ""
+    default: "code-signing/slackhq/nebula"
 
 runs:
   using: composite
@@ -57,9 +57,6 @@ runs:
         KEY_PREFIX: ${{ inputs.key-prefix }}
       run: |
         set -eu
-        # Default the prefix to this repo so the S3 key attributes the sign correctly.
-        # nebula-nightly runs this same action but writes under its own repo's prefix.
-        KEY_PREFIX="${KEY_PREFIX:-code-signing/$GITHUB_REPOSITORY}"
         RUN="${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
 
         find "$SIGN_PATH" -name '*.exe' -print | while read -r path

control_lifecycle_test.go

@@ -11,7 +11,6 @@ import (
 
 	"github.com/gaissmai/bart"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/test"
 	"github.com/slackhq/nebula/udp"
@@ -31,9 +30,9 @@ func newFakeDevice() *fakeDevice {
 
 // Read blocks until Close like a real tun with no traffic, then reports EOF
 // the same way a closed device does
-func (d *fakeDevice) Read() ([]tio.Packet, error) {
+func (d *fakeDevice) Read(p []byte) (int, error) {
 	<-d.closedCh
-	return nil, io.EOF
+	return 0, io.EOF
 }
 
 func (d *fakeDevice) Write(p []byte) (int, error) { return len(p), nil }
@@ -50,8 +49,10 @@ func (d *fakeDevice) Activate() error                       { return nil }
 func (d *fakeDevice) Networks() []netip.Prefix              { return nil }
 func (d *fakeDevice) Name() string                          { return "fake" }
 func (d *fakeDevice) RoutesFor(netip.Addr) routing.Gateways { return nil }
-
+func (d *fakeDevice) SupportsMultiqueue() bool              { return false }
-func (d *fakeDevice) Queues(int) ([]tio.Queue, error) { return []tio.Queue{d}, nil }
+func (d *fakeDevice) NewMultiQueueReader() (io.ReadWriteCloser, error) {
+	return nil, errors.New("unsupported")
+}
 
 // newReadyControl hand-builds the minimum Control that Main would have
 // produced right before Start, including the construction token NewInterface
@@ -77,6 +78,7 @@ func newReadyControl(t *testing.T) (*Control, *fakeDevice, *fakeConn) {
 		inside:     dev,
 		outside:    conn,
 		writers:    []udp.Conn{conn},
+		readers:    make([]io.ReadWriteCloser, 1),
 		routines:   1,
 		hostMap:    newHostMap(l),
 		lightHouse: lh,
@@ -153,14 +155,7 @@ type multiqueueDevice struct {
 	*fakeDevice
 }
 
-// Queues claims multiqueue support but fails to open the second queue,
+func (d *multiqueueDevice) SupportsMultiqueue() bool { return true }
-// exercising the activation error path.
-func (d *multiqueueDevice) Queues(n int) ([]tio.Queue, error) {
-	if n > 1 {
-		return nil, errors.New("second queue failed to open")
-	}
-	return d.fakeDevice.Queues(n)
-}
 
 func TestControl_StartMultiqueueFailureReleases(t *testing.T) {
 	dev := &multiqueueDevice{fakeDevice: newFakeDevice()}
@@ -171,6 +166,7 @@ func TestControl_StartMultiqueueFailureReleases(t *testing.T) {
 		inside:   dev,
 		outside:  conn,
 		writers:  []udp.Conn{conn},
+		readers:  make([]io.ReadWriteCloser, 2),
 		routines: 2,
 		l:        test.NewLogger(),
 	}

examples/config.yml

@@ -110,6 +110,15 @@ lighthouse:
     #- "1.1.1.1:4242"
     #- "1.2.3.4:0" # port will be replaced with the real listening port
 
+  # Locally discovered addresses are checked against the MTU of the link they were found on. If the link cannot fit
+  # a full-size packet from the nebula tun device (`tun.mtu` plus encapsulation overhead, which is larger for relayed
+  # traffic) without fragmenting, a warning is logged.
+  # When omit_low_mtu_addrs is true, addresses whose links cannot fit normal nebula traffic are dropped from
+  # lighthouse reports entirely.
+  # Addresses that can fit normal nebula traffic but not relayed traffic are always still advertised.
+  # This does not apply to addresses listed in advertise_addrs.
+  #omit_low_mtu_addrs: false
+
   # EXPERIMENTAL: This option may change or disappear in the future.
   # This setting allows us to "guess" what the remote might be for a host
   # while we wait for the lighthouse response.
@@ -254,20 +263,6 @@ 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
-  # 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

firewall.go

@@ -44,8 +44,8 @@ type Firewall struct {
 	InRules  *FirewallTable
 	OutRules *FirewallTable
 
-	InboundSendReject  bool
+	InSendReject  bool
-	OutboundSendReject bool
+	OutSendReject bool
 
 	//TODO: we should have many more options for TCP, an option for ICMP, and mimic the kernel a bit better
 	// https://www.kernel.org/doc/Documentation/networking/nf_conntrack-sysctl.txt
@@ -216,23 +216,23 @@ func NewFirewallFromConfig(l *slog.Logger, cs *CertState, c *config.C) (*Firewal
 	inboundAction := c.GetString("firewall.inbound_action", "drop")
 	switch inboundAction {
 	case "reject":
-		fw.InboundSendReject = true
+		fw.InSendReject = true
 	case "drop":
-		fw.InboundSendReject = false
+		fw.InSendReject = false
 	default:
 		l.Warn("invalid firewall.inbound_action, defaulting to `drop`", "action", inboundAction)
-		fw.InboundSendReject = false
+		fw.InSendReject = false
 	}
 
 	outboundAction := c.GetString("firewall.outbound_action", "drop")
 	switch outboundAction {
 	case "reject":
-		fw.OutboundSendReject = true
+		fw.OutSendReject = true
 	case "drop":
-		fw.OutboundSendReject = false
+		fw.OutSendReject = false
 	default:
 		l.Warn("invalid firewall.outbound_action, defaulting to `drop`", "action", outboundAction)
-		fw.OutboundSendReject = false
+		fw.OutSendReject = false
 	}
 
 	err := AddFirewallRulesFromConfig(l, false, c, fw)

hostmap.go

@@ -869,9 +869,17 @@ func (i *HostInfo) logger(l *slog.Logger) *slog.Logger {
 
 // Utility functions
 
-func localAddrs(l *slog.Logger, allowList *LocalAllowList) []netip.Addr {
+// localAddr is a locally discovered address candidate for lighthouse
+// advertisement, along with details about the link it was found on.
+type localAddr struct {
+	addr    netip.Addr
+	ifName  string
+	linkMTU int // MTU reported for the link, or <= 0 if unknown
+}
+
+func localAddrs(l *slog.Logger, allowList *LocalAllowList) []localAddr {
 	//FIXME: This function is pretty garbage
-	var finalAddrs []netip.Addr
+	var finalAddrs []localAddr
 	ifaces, _ := net.Interfaces()
 	for _, i := range ifaces {
 		allow := allowList.AllowName(i.Name)
@@ -916,7 +924,7 @@ func localAddrs(l *slog.Logger, allowList *LocalAllowList) []netip.Addr {
 					continue
 				}
 
-				finalAddrs = append(finalAddrs, addr)
+				finalAddrs = append(finalAddrs, localAddr{addr: addr, ifName: i.Name, linkMTU: i.MTU})
 			}
 		}
 	}

inside.go

@@ -37,7 +37,7 @@ func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *firewall.Packet
 		// routes packets from the Nebula addr to the Nebula addr through the Nebula
 		// TUN device.
 		if immediatelyForwardToSelf {
-			_, err := f.queues[q].Write(packet)
+			_, err := f.readers[q].Write(packet)
 			if err != nil {
 				f.l.Error("Failed to forward to tun", "error", err)
 			}
@@ -87,7 +87,7 @@ func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *firewall.Packet
 }
 
 func (f *Interface) rejectInside(packet []byte, out []byte, q int) {
-	if !f.firewall.OutboundSendReject {
+	if !f.firewall.InSendReject {
 		return
 	}
 
@@ -96,14 +96,14 @@ func (f *Interface) rejectInside(packet []byte, out []byte, q int) {
 		return
 	}
 
-	_, err := f.queues[q].Write(out)
+	_, err := f.readers[q].Write(out)
 	if err != nil {
 		f.l.Error("Failed to write to tun", "error", err)
 	}
 }
 
 func (f *Interface) rejectOutside(packet []byte, ci *ConnectionState, hostinfo *HostInfo, nb, out []byte, q int) {
-	if !f.firewall.InboundSendReject {
+	if !f.firewall.OutSendReject {
 		return
 	}
 

interface.go

@@ -4,9 +4,9 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"log/slog"
 	"net/netip"
-	"runtime"
 	"slices"
 	"sync"
 	"sync/atomic"
@@ -20,9 +20,7 @@ import (
 	"github.com/slackhq/nebula/firewall"
 	"github.com/slackhq/nebula/header"
 	"github.com/slackhq/nebula/overlay"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/udp"
-	"github.com/slackhq/nebula/util"
 )
 
 const mtu = 9001
@@ -51,19 +49,7 @@ type InterfaceConfig struct {
 	reQueryWait     time.Duration
 
 	ConntrackCacheTimeout time.Duration
-
+	l                     *slog.Logger
-	// 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. 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 UDP sends on one XPS-selected NIC TX ring so per-flow
-	// packets stay ordered on the wire.
-	PinThreads bool
-
-	l *slog.Logger
 }
 
 type Interface struct {
@@ -87,16 +73,7 @@ type Interface struct {
 	routines              int
 	disconnectInvalid     atomic.Bool
 	closed                atomic.Bool
-	// cpuAffinity, when non-empty, names the CPUs each TUN reader goroutine
+	relayManager          *relayManager
-	// should pin to. Queue i pins to cpuAffinity[i % len(cpuAffinity)].
-	// 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
-	relayManager *relayManager
 
 	tryPromoteEvery atomic.Uint32
 	reQueryEvery    atomic.Uint32
@@ -113,7 +90,7 @@ type Interface struct {
 
 	ctx     context.Context
 	writers []udp.Conn
-	queues  []tio.Queue
+	readers []io.ReadWriteCloser
 	wg      sync.WaitGroup
 
 	// fatalErr holds the first unexpected reader error that caused shutdown.
@@ -212,6 +189,7 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) {
 		routines:              c.routines,
 		version:               c.version,
 		writers:               make([]udp.Conn, c.routines),
+		readers:               make([]io.ReadWriteCloser, c.routines),
 		myVpnNetworks:         cs.myVpnNetworks,
 		myVpnNetworksTable:    cs.myVpnNetworksTable,
 		myVpnAddrs:            cs.myVpnAddrs,
@@ -220,8 +198,6 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) {
 		relayManager:          c.relayManager,
 		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,
@@ -264,27 +240,27 @@ func (f *Interface) activate() error {
 		"boringcrypto", boringEnabled(),
 	)
 
-	if f.routines > 1 && !f.outside.SupportsMultipleReaders() {
+	if f.routines > 1 {
-		f.routines = 1
+		if !f.inside.SupportsMultiqueue() || !f.outside.SupportsMultipleReaders() {
-		f.l.Warn("multiple udp readers are not supported on this platform, falling back to a single routine")
+			f.routines = 1
+			f.l.Warn("routines is not supported on this platform, falling back to a single routine")
+		}
 	}
 
-	// Prepare the tun queues. A device that can't open that many hands back
-	// fewer (a single queue on platforms without multiqueue support) and we
-	// size the reader routines to what we actually got.
-	queues, err := f.inside.Queues(f.routines)
-	if err != nil {
-		return err
-	}
-	if len(queues) < f.routines {
-		f.l.Warn("tun multiqueue is not supported on this platform, falling back to fewer routines",
-			"requested", f.routines, "opened", len(queues))
-		f.routines = len(queues)
-	}
-	f.queues = queues
-
 	metrics.GetOrRegisterGauge("routines", nil).Update(int64(f.routines))
 
+	// Prepare n tun queues
+	var reader io.ReadWriteCloser = f.inside
+	for i := 0; i < f.routines; i++ {
+		if i > 0 {
+			reader, err = f.inside.NewMultiQueueReader()
+			if err != nil {
+				return err
+			}
+		}
+		f.readers[i] = reader
+	}
+
 	// On error the caller owns the cleanup, Control.Start cancels the service context
 	// before releasing our resources so a waiter never observes a live context
 	if err = f.inside.Activate(); err != nil {
@@ -305,7 +281,7 @@ func (f *Interface) run() {
 	// Launch n queues to read packets from tun dev
 	for i := 0; i < f.routines; i++ {
 		f.wg.Go(func() {
-			f.listenIn(f.queues[i], i)
+			f.listenIn(f.readers[i], i)
 		})
 	}
 
@@ -360,29 +336,8 @@ func (f *Interface) listenOut(i int) {
 	f.l.Debug("underlay reader is done", "reader", i)
 }
 
-func (f *Interface) listenIn(queue tio.Queue, i int) {
+func (f *Interface) listenIn(reader io.ReadWriteCloser, i int) {
-	// Pinning this thread (and goroutine) to a single CPU keeps every UDP send from this goroutine going through
+	packet := make([]byte, mtu)
-	// the same TX ring on the nic (XPS selects the ring by CPU), 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)
-		}
-	}
-
 	out := make([]byte, mtu)
 	fwPacket := &firewall.Packet{}
 	nb := make([]byte, 12, 12)
@@ -390,7 +345,7 @@ func (f *Interface) listenIn(queue tio.Queue, i int) {
 	conntrackCache := firewall.NewConntrackCacheTicker(f.ctx, f.l, f.conntrackCacheTimeout)
 
 	for {
-		pkts, err := queue.Read()
+		n, err := reader.Read(packet)
 		if err != nil {
 			// Same shutdown noise handling as listenOut
 			if !f.closed.Load() && f.ctx.Err() == nil {
@@ -400,11 +355,7 @@ func (f *Interface) listenIn(queue tio.Queue, i int) {
 			break
 		}
 
-		for _, pkt := range pkts {
+		f.consumeInsidePacket(packet[:n], fwPacket, nb, out, i, conntrackCache.Get())
-			// borrowed: pkt.Bytes is owned by the queue and only valid until
-			// the next Read; consumeInsidePacket reads it synchronously.
-			f.consumeInsidePacket(pkt.Bytes, fwPacket, nb, out, i, conntrackCache.Get())
-		}
 	}
 
 	f.l.Debug("overlay reader is done", "reader", i)

lighthouse.go

@@ -19,8 +19,11 @@ import (
 	"github.com/slackhq/nebula/config"
 	"github.com/slackhq/nebula/header"
 	"github.com/slackhq/nebula/logging"
+	"github.com/slackhq/nebula/overlay"
 	"github.com/slackhq/nebula/udp"
 	"github.com/slackhq/nebula/util"
+	"golang.org/x/net/ipv4"
+	"golang.org/x/net/ipv6"
 )
 
 var ErrHostNotKnown = errors.New("host not known")
@@ -65,6 +68,18 @@ type LightHouse struct {
 
 	advertiseAddrs atomic.Pointer[[]netip.AddrPort]
 
+	// tunMTU mirrors tun.mtu so locally discovered addrs can be checked for
+	// links too small to carry a full-size nebula packet without fragmenting.
+	tunMTU atomic.Int64
+	// omitLowMTUAddrs drops such addrs from lighthouse updates (and demotes
+	// the associated warnings to debug logs) instead of advertising them.
+	omitLowMTUAddrs atomic.Bool
+	// mtuWarned tracks the last classification logged per local addr so a
+	// warning is only emitted when the classification changes, not on every
+	// periodic update.
+	mtuWarnLock sync.Mutex
+	mtuWarned   map[mtuWarnKey]linkMTUTier
+
 	// Addr's of relays that can be used by peers to access me
 	relaysForMe atomic.Pointer[[]netip.Addr]
 
@@ -105,6 +120,7 @@ func NewLightHouseFromConfig(ctx context.Context, l *slog.Logger, c *config.C, c
 		punchy:             p,
 		updateTrigger:      make(chan struct{}, 1),
 		queryChan:          make(chan netip.Addr, c.GetUint32("handshakes.query_buffer", 64)),
+		mtuWarned:          make(map[mtuWarnKey]linkMTUTier),
 		l:                  l,
 	}
 	lighthouses := make([]netip.Addr, 0)
@@ -216,6 +232,23 @@ func (lh *LightHouse) reload(c *config.C, initial bool) error {
 		}
 	}
 
+	if initial || c.HasChanged("tun.mtu") || c.HasChanged("lighthouse.omit_low_mtu_addrs") {
+		lh.tunMTU.Store(int64(c.GetInt("tun.mtu", overlay.DefaultMTU)))
+		lh.omitLowMTUAddrs.Store(c.GetBool("lighthouse.omit_low_mtu_addrs", false))
+
+		// Re-log any addrs whose links are still too small under the new values
+		lh.mtuWarnLock.Lock()
+		clear(lh.mtuWarned)
+		lh.mtuWarnLock.Unlock()
+
+		if !initial {
+			lh.l.Info("tun.mtu and/or lighthouse.omit_low_mtu_addrs has changed",
+				"tunMTU", lh.tunMTU.Load(),
+				"omitLowMTUAddrs", lh.omitLowMTUAddrs.Load(),
+			)
+		}
+	}
+
 	if initial || c.HasChanged("lighthouse.interval") {
 		lh.interval.Store(int64(c.GetInt("lighthouse.interval", 10)))
 
@@ -905,6 +938,108 @@ func (lh *LightHouse) TriggerUpdate() {
 	}
 }
 
+// linkMTUTier classifies how well a local addr's link MTU can carry
+// full-size nebula packets built from a tun packet of tun.mtu bytes.
+type linkMTUTier uint8
+
+// mtuWarnKey identifies a local addr for MTU warning dedup purposes. The
+// interface name is included because the same addr can exist on multiple
+// links with different MTUs.
+type mtuWarnKey struct {
+	ifName string
+	addr   netip.Addr
+}
+
+const (
+	// The link can carry both normal and relayed nebula traffic
+	linkMTUOk linkMTUTier = iota
+	// The link can carry normal nebula traffic, but relayed traffic (which
+	// adds a second layer of encapsulation) will not fit
+	linkMTUTooSmallForRelay
+	// Even normal nebula traffic will not fit
+	linkMTUTooSmall
+)
+
+const (
+	// Both AES-256-GCM and ChaCha20-Poly1305 append a 16 byte AEAD tag
+	cipherTagLen = 16
+	udpHeaderLen = 8
+)
+
+// requiredLinkMTU returns the minimum underlay link MTU that can carry a
+// full-size tun packet to an addr of the given family without fragmentation,
+// both directly and via a relay (which wraps the packet in a second nebula
+// header and AEAD tag).
+func requiredLinkMTU(tunMTU int, is4 bool) (direct, relayed int) {
+	ipHeaderLen := ipv6.HeaderLen
+	if is4 {
+		ipHeaderLen = ipv4.HeaderLen
+	}
+
+	direct = tunMTU + header.Len + cipherTagLen + udpHeaderLen + ipHeaderLen
+	relayed = direct + header.Len + cipherTagLen
+	return direct, relayed
+}
+
+// checkLocalLinkMTU classifies e's link MTU, logs when the classification
+// changes, and reports whether e should be advertised to lighthouses.
+func (lh *LightHouse) checkLocalLinkMTU(e localAddr) bool {
+	tunMTU := int(lh.tunMTU.Load())
+	omit := lh.omitLowMTUAddrs.Load()
+
+	tier := linkMTUOk
+	direct, relayed := requiredLinkMTU(tunMTU, e.addr.Is4())
+	if e.linkMTU > 0 { // links with an unknown MTU are advertised as-is
+		if e.linkMTU < direct {
+			tier = linkMTUTooSmall
+		} else if e.linkMTU < relayed {
+			tier = linkMTUTooSmallForRelay
+		}
+	}
+	advertise := tier != linkMTUTooSmall || !omit
+
+	key := mtuWarnKey{ifName: e.ifName, addr: e.addr}
+	lh.mtuWarnLock.Lock()
+	changed := lh.mtuWarned[key] != tier
+	if changed {
+		if tier == linkMTUOk {
+			delete(lh.mtuWarned, key)
+		} else {
+			lh.mtuWarned[key] = tier
+		}
+	}
+	lh.mtuWarnLock.Unlock()
+
+	if !changed || tier == linkMTUOk {
+		return advertise
+	}
+
+	level := slog.LevelWarn
+	if omit {
+		level = slog.LevelDebug
+	}
+
+	if lh.l.Enabled(context.Background(), level) {
+		msg := "Link MTU too small for nebula traffic, expect fragmentation or drops"
+		if !advertise {
+			msg = "Omitting addr with too-small link MTU from lighthouse report"
+		} else if tier == linkMTUTooSmallForRelay {
+			msg = "Link MTU too small for relayed nebula traffic"
+		}
+
+		lh.l.Log(context.Background(), level, msg,
+			"localAddr", e.addr,
+			"interface", e.ifName,
+			"linkMTU", e.linkMTU,
+			"requiredMTU", direct,
+			"requiredRelayMTU", relayed,
+			"tunMTU", tunMTU,
+		)
+	}
+
+	return advertise
+}
+
 func (lh *LightHouse) SendUpdate() {
 	var v4 []*V4AddrPort
 	var v6 []*V6AddrPort
@@ -919,15 +1054,19 @@ func (lh *LightHouse) SendUpdate() {
 
 	lal := lh.GetLocalAllowList()
 	for _, e := range localAddrs(lh.l, lal) {
-		if lh.myVpnNetworksTable.Contains(e) {
+		if lh.myVpnNetworksTable.Contains(e.addr) {
+			continue
+		}
+
+		if !lh.checkLocalLinkMTU(e) {
 			continue
 		}
 
 		// Only add addrs that aren't my VPN/tun networks
-		if e.Is4() {
+		if e.addr.Is4() {
-			v4 = append(v4, netAddrToProtoV4AddrPort(e, uint16(lh.nebulaPort)))
+			v4 = append(v4, netAddrToProtoV4AddrPort(e.addr, uint16(lh.nebulaPort)))
 		} else {
-			v6 = append(v6, netAddrToProtoV6AddrPort(e, uint16(lh.nebulaPort)))
+			v6 = append(v6, netAddrToProtoV6AddrPort(e.addr, uint16(lh.nebulaPort)))
 		}
 	}
 

lighthouse_test.go

@@ -738,3 +738,86 @@ func TestLighthouse_DeletesWork(t *testing.T) {
 	out = lh.Query(testHost)
 	assert.Nil(t, out)
 }
+
+func Test_requiredLinkMTU(t *testing.T) {
+	// tun packet + nebula header (16) + AEAD tag (16) + udp (8) + ip header
+	direct, relayed := requiredLinkMTU(1300, true)
+	assert.Equal(t, 1360, direct)
+	assert.Equal(t, 1392, relayed)
+
+	direct, relayed = requiredLinkMTU(1300, false)
+	assert.Equal(t, 1380, direct)
+	assert.Equal(t, 1412, relayed)
+}
+
+func Test_checkLocalLinkMTU(t *testing.T) {
+	lh := &LightHouse{l: test.NewLogger(), mtuWarned: make(map[mtuWarnKey]linkMTUTier)}
+	lh.tunMTU.Store(1300)
+
+	v4 := netip.MustParseAddr("192.168.1.2")
+	v6 := netip.MustParseAddr("fd00::2")
+	mkAddr := func(a netip.Addr, mtu int) localAddr {
+		return localAddr{addr: a, ifName: "test0", linkMTU: mtu}
+	}
+
+	// Plenty of room, no state recorded
+	assert.True(t, lh.checkLocalLinkMTU(mkAddr(v4, 1500)))
+	assert.Empty(t, lh.mtuWarned)
+
+	// Unknown link MTU is not classified
+	assert.True(t, lh.checkLocalLinkMTU(mkAddr(v4, 0)))
+	assert.Empty(t, lh.mtuWarned)
+
+	// Too small for even normal traffic, still advertised by default
+	assert.True(t, lh.checkLocalLinkMTU(mkAddr(v4, 1359)))
+	assert.Equal(t, linkMTUTooSmall, lh.mtuWarned[mtuWarnKey{ifName: "test0", addr: v4}])
+
+	// Fits normal traffic but not relayed traffic
+	assert.True(t, lh.checkLocalLinkMTU(mkAddr(v4, 1360)))
+	assert.Equal(t, linkMTUTooSmallForRelay, lh.mtuWarned[mtuWarnKey{ifName: "test0", addr: v4}])
+
+	// Exactly enough for relayed traffic clears the state
+	assert.True(t, lh.checkLocalLinkMTU(mkAddr(v4, 1392)))
+	assert.Empty(t, lh.mtuWarned)
+
+	// v6 addrs need 20 more bytes of headroom
+	assert.True(t, lh.checkLocalLinkMTU(mkAddr(v6, 1380)))
+	assert.Equal(t, linkMTUTooSmallForRelay, lh.mtuWarned[mtuWarnKey{ifName: "test0", addr: v6}])
+	assert.True(t, lh.checkLocalLinkMTU(mkAddr(v6, 1379)))
+	assert.Equal(t, linkMTUTooSmall, lh.mtuWarned[mtuWarnKey{ifName: "test0", addr: v6}])
+	assert.True(t, lh.checkLocalLinkMTU(mkAddr(v6, 1412)))
+	assert.Empty(t, lh.mtuWarned)
+
+	// With omit enabled, only addrs that can't fit normal traffic are dropped
+	lh.omitLowMTUAddrs.Store(true)
+	assert.False(t, lh.checkLocalLinkMTU(mkAddr(v4, 1359)))
+	assert.Equal(t, linkMTUTooSmall, lh.mtuWarned[mtuWarnKey{ifName: "test0", addr: v4}])
+	assert.True(t, lh.checkLocalLinkMTU(mkAddr(v4, 1360)))
+	assert.Equal(t, linkMTUTooSmallForRelay, lh.mtuWarned[mtuWarnKey{ifName: "test0", addr: v4}])
+	assert.False(t, lh.checkLocalLinkMTU(mkAddr(v4, 1359)))
+}
+
+func Test_lighthouseMTUConfig(t *testing.T) {
+	l := test.NewLogger()
+	myVpnNet := netip.MustParsePrefix("10.128.0.1/16")
+	nt := new(bart.Lite)
+	nt.Insert(myVpnNet)
+	cs := &CertState{
+		myVpnNetworks:      []netip.Prefix{myVpnNet},
+		myVpnNetworksTable: nt,
+	}
+
+	c := config.NewC(l)
+	lh, err := NewLightHouseFromConfig(t.Context(), l, c, cs, nil, nil)
+	require.NoError(t, err)
+	assert.Equal(t, int64(1300), lh.tunMTU.Load())
+	assert.False(t, lh.omitLowMTUAddrs.Load())
+
+	c = config.NewC(l)
+	c.Settings["tun"] = map[string]any{"mtu": 8000}
+	c.Settings["lighthouse"] = map[string]any{"omit_low_mtu_addrs": true}
+	lh, err = NewLightHouseFromConfig(t.Context(), l, c, cs, nil, nil)
+	require.NoError(t, err)
+	assert.Equal(t, int64(8000), lh.tunMTU.Load())
+	assert.True(t, lh.omitLowMTUAddrs.Load())
+}

main.go

@@ -7,7 +7,6 @@ import (
 	"net"
 	"net/netip"
 	"runtime/debug"
-	"slices"
 	"strings"
 	"time"
 
@@ -232,8 +231,6 @@ 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),
 		l:                     l,
 	}
 
@@ -285,70 +282,6 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev
 	}, nil
 }
 
-// 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 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 {
-		return nil
-	}
-	rv, ok := raw.([]any)
-	if !ok {
-		l.Warn("tun.cpu_affinity must be a list of integers; ignoring", "value", raw)
-		return nil
-	}
-	// 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
-		switch v := e.(type) {
-		case int:
-			cpu = v
-		case int64:
-			cpu = int(v)
-		case float64:
-			cpu = int(v)
-		default:
-			l.Warn("tun.cpu_affinity entry not an integer; ignoring affinity",
-				"index", i, "value", e)
-			return nil
-		}
-		if cpu < 0 {
-			l.Warn("tun.cpu_affinity entry out of range; ignoring affinity",
-				"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)
-	}
-	if len(cpus) != routines {
-		l.Warn("tun.cpu_affinity length doesn't match routines; queues will modulo-cycle through the list",
-			"affinity_len", len(cpus), "routines", routines)
-	}
-	return cpus
-}
-
 func moduleVersion() string {
 	info, ok := debug.ReadBuildInfo()
 	if !ok {

main_test.go

@@ -1,51 +0,0 @@
-package nebula
-
-import (
-	"testing"
-
-	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/test"
-	"github.com/slackhq/nebula/util"
-	"github.com/stretchr/testify/assert"
-)
-
-func TestParseCpuAffinity(t *testing.T) {
-	l := test.NewLogger()
-
-	// newConfig returns a config.C with tun.cpu_affinity set to v. A nil v
-	// leaves the key unset.
-	newConfig := func(v any) *config.C {
-		c := config.NewC(l)
-		if v != nil {
-			c.Settings["tun"] = map[string]any{"cpu_affinity": v}
-		}
-		return c
-	}
-
-	// unset -> nil (listenIn falls back to spreading across the allowed set)
-	assert.Nil(t, parseCpuAffinity(newConfig(nil), l, 1))
-
-	// Pick a CPU we're actually allowed to run on so a valid list survives
-	// validation regardless of the host's affinity mask.
-	allowed, _ := util.AllowedCPUs()
-	validCPU := 0
-	if len(allowed) > 0 {
-		validCPU = allowed[0]
-	}
-
-	// valid list -> parsed through unchanged
-	assert.Equal(t, []int{validCPU, validCPU}, parseCpuAffinity(newConfig([]any{validCPU, validCPU}), l, 2))
-
-	// a negative entry is out of range on every platform -> disables the override
-	assert.Nil(t, parseCpuAffinity(newConfig([]any{validCPU, -1}), l, 2))
-
-	// a non-integer entry -> disables the override
-	assert.Nil(t, parseCpuAffinity(newConfig([]any{validCPU, "not-a-cpu"}), l, 2))
-
-	// 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
-	// any representable CPU id so it can never be in the mask.
-	if len(allowed) > 0 {
-		assert.Nil(t, parseCpuAffinity(newConfig([]any{1 << 20}), l, 1))
-	}
-}

outside.go

@@ -542,7 +542,7 @@ func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, out []byte, p
 		return
 	}
 
-	_, err = f.queues[q].Write(out)
+	_, err = f.readers[q].Write(out)
 	if err != nil {
 		f.l.Error("Failed to write to tun", "error", err)
 	}

overlay/device.go

@@ -4,25 +4,15 @@ import (
 	"io"
 	"net/netip"
 
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 )
 
-// defaultBatchBufSize is the per-Queue scratch size for Read. 65535 covers
-// any single IP packet.
-const defaultBatchBufSize = 65535
-
 type Device interface {
-	io.Closer
+	io.ReadWriteCloser
 	Activate() error
 	Networks() []netip.Prefix
 	Name() string
 	RoutesFor(netip.Addr) routing.Gateways
-	// Queues returns the device's packet queues, opening additional ones as
+	SupportsMultiqueue() bool
-	// needed until there are n. Platforms without multiqueue support return
+	NewMultiQueueReader() (io.ReadWriteCloser, error)
-	// their single queue regardless of n, so callers must size reader loops
-	// to len(result), not n; implementations never return more than n. An
-	// error means a queue that should have opened could not; the caller owns
-	// cleanup via Close. Called once, during interface activation.
-	Queues(n int) ([]tio.Queue, error)
 }

overlay/overlaytest/noop.go

@@ -3,9 +3,10 @@
 package overlaytest
 
 import (
+	"errors"
+	"io"
 	"net/netip"
 
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 )
 
@@ -30,16 +31,20 @@ func (NoopTun) Name() string {
 	return "noop"
 }
 
-func (NoopTun) Read() ([]tio.Packet, error) {
+func (NoopTun) Read([]byte) (int, error) {
-	return nil, nil
+	return 0, nil
 }
 
 func (NoopTun) Write([]byte) (int, error) {
 	return 0, nil
 }
 
-func (NoopTun) Queues(int) ([]tio.Queue, error) {
+func (NoopTun) SupportsMultiqueue() bool {
-	return []tio.Queue{NoopTun{}}, nil
+	return false
+}
+
+func (NoopTun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
+	return nil, errors.New("unsupported")
 }
 
 func (NoopTun) Close() error {

overlay/tio/blockon_linux.go

@@ -1,45 +0,0 @@
-//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.
-//
-// 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
-}

overlay/tio/queueset_poll_linux.go

@@ -1,90 +0,0 @@
-//go:build linux && !android
-// +build linux,!android
-
-package tio
-
-import (
-	"encoding/binary"
-	"errors"
-	"fmt"
-	"sync/atomic"
-
-	"golang.org/x/sys/unix"
-)
-
-type pollQueueSet struct {
-	pq []*Poll
-	// pqi is exactly the same as pq, but stored as the interface type
-	pqi        []Queue
-	shutdownFd int
-	closed     atomic.Bool
-}
-
-func NewPollQueueSet() (QueueSet, error) {
-	shutdownFd, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create eventfd: %w", err)
-	}
-
-	out := &pollQueueSet{
-		pq:         []*Poll{},
-		pqi:        []Queue{},
-		shutdownFd: shutdownFd,
-	}
-
-	return out, nil
-}
-
-func (c *pollQueueSet) Queues() []Queue {
-	return c.pqi
-}
-
-func (c *pollQueueSet) Add(fd int) error {
-	x, err := newPoll(fd, c.shutdownFd)
-	if err != nil {
-		return err
-	}
-	c.pq = append(c.pq, x)
-	c.pqi = append(c.pqi, x)
-
-	return nil
-}
-
-func (c *pollQueueSet) wakeForShutdown() error {
-	var buf [8]byte
-	binary.NativeEndian.PutUint64(buf[:], 1)
-	_, err := unix.Write(int(c.shutdownFd), buf[:])
-	return err
-}
-
-func (c *pollQueueSet) Close() error {
-	if c.closed.Swap(true) {
-		return nil
-	}
-
-	errs := []error{}
-
-	// Wake any reader blocked in poll so it observes POLLIN on the shutdown
-	// eventfd and returns os.ErrClosed.
-	if err := c.wakeForShutdown(); err != nil {
-		errs = append(errs, err)
-	}
-
-	// Close the per-queue tun fds; this also unblocks any in-flight reads.
-	// The per-queue Close deliberately leaves shutdownFd alone - it belongs
-	// to this container.
-	for _, x := range c.pq {
-		if err := x.Close(); err != nil {
-			errs = append(errs, err)
-		}
-	}
-
-	// Close the shutdown eventfd last: every reader's pollfd set references
-	// it, so it must outlive the wake + per-queue teardown above.
-	if err := unix.Close(c.shutdownFd); err != nil {
-		errs = append(errs, err)
-	}
-	c.shutdownFd = -1
-
-	return errors.Join(errs...)
-}

overlay/tio/single.go

@@ -1,50 +0,0 @@
-package tio
-
-import "io"
-
-// singleQueue adapts a legacy one-datagram-per-Read source into a Queue.
-// Read fills a private scratch buffer and returns exactly one Packet whose
-// Bytes borrow from that buffer, valid only until the next Read, per the
-// Queue contract. Single-reader like every Queue; Write is exactly as safe
-// for concurrent use as the underlying source's Write.
-type singleQueue struct {
-	rw     io.ReadWriter
-	closer io.Closer // nil: Close is a no-op (the source is shared and owned elsewhere)
-	buf    []byte
-	ret    [1]Packet
-}
-
-// NewSingleQueue wraps a one-datagram-per-Read ReadWriteCloser (a legacy tun
-// device) into a Queue. bufSize is the per-queue read scratch size and must
-// be at least the largest datagram the source can return. Close closes rwc.
-func NewSingleQueue(rwc io.ReadWriteCloser, bufSize int) Queue {
-	return &singleQueue{rw: rwc, closer: rwc, buf: make([]byte, bufSize)}
-}
-
-// NewSingleQueueNoClose is NewSingleQueue for a source owned by someone else,
-// e.g. several queues sharing one device. Close on the returned Queue is a
-// no-op so one queue can't tear the shared source out from under its
-// siblings; the owner remains responsible for closing the source itself.
-func NewSingleQueueNoClose(rw io.ReadWriter, bufSize int) Queue {
-	return &singleQueue{rw: rw, buf: make([]byte, bufSize)}
-}
-
-func (q *singleQueue) Read() ([]Packet, error) {
-	n, err := q.rw.Read(q.buf)
-	if err != nil {
-		return nil, err
-	}
-	q.ret[0] = Packet{Bytes: q.buf[:n]}
-	return q.ret[:], nil
-}
-
-func (q *singleQueue) Write(p []byte) (int, error) {
-	return q.rw.Write(p)
-}
-
-func (q *singleQueue) Close() error {
-	if q.closer == nil {
-		return nil
-	}
-	return q.closer.Close()
-}

overlay/tio/tio.go

@@ -1,52 +0,0 @@
-package tio
-
-import (
-	"io"
-)
-
-// QueueSet holds one or many Queue objects and helps close them in an orderly way.
-type QueueSet interface {
-	io.Closer
-	Queues() []Queue
-
-	// Add takes a tun fd, adds it to the set, and prepares it for use as a Queue.
-	Add(fd int) error
-}
-
-// Queue is a readable/writable packet queue. Concurrency contract: a single
-// read goroutine drives Read; plain Write is safe for concurrent callers.
-type Queue interface {
-	io.Closer
-
-	// Read returns one or more packets. The returned Packet.Bytes slices
-	// are borrowed from the Queue's internal buffer and are only valid
-	// until the next Read or Close on this Queue - callers must encrypt
-	// or copy each slice before the next call. 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. Safe for concurrent use.
-	Write(p []byte) (int, error)
-}
-
-// Packet is the unit Queue.Read returns. Bytes points into the queue's
-// internal buffer and is only valid until the next Read or Close on the
-// queue that produced it.
-type Packet struct {
-	Bytes []byte
-}
-
-// Clone returns a Packet whose Bytes is a freshly allocated copy of p.Bytes,
-// safe to retain past the next Read or Close on the originating Queue.
-// Use this only when a caller genuinely needs to outlive the borrowed-slice
-// contract — the hot path reads should continue to consume the borrow
-// synchronously to avoid the allocation.
-func (p Packet) Clone() Packet {
-	if p.Bytes == nil {
-		return p
-	}
-	cp := make([]byte, len(p.Bytes))
-	copy(cp, p.Bytes)
-	return Packet{Bytes: cp}
-}

overlay/tio/tio_poll_linux.go

@@ -1,116 +0,0 @@
-//go:build linux && !android
-// +build linux,!android
-
-package tio
-
-import (
-	"fmt"
-	"os"
-	"sync/atomic"
-
-	"golang.org/x/sys/unix"
-)
-
-// Maximum size we accept for a single read from a TUN. 65535 covers any
-// single IP packet.
-const tunReadBufSize = 65535
-
-type Poll struct {
-	fd         int
-	shutdownFd int
-	closed     atomic.Bool
-
-	readBuf  []byte
-	batchRet [1]Packet
-}
-
-// newPoll wraps an existing tun fd. On failure it does NOT close fd: the
-// caller owns fd and is the sole closer (see pollQueueSet.Add callers in
-// overlay/tun_linux.go, which unix.Close on Add error). This keeps closes
-// at exactly one on every path.
-func newPoll(fd int, shutdownFd int) (*Poll, error) {
-	if err := unix.SetNonblock(fd, true); err != nil {
-		return nil, fmt.Errorf("failed to set Poll device as nonblocking: %w", err)
-	}
-
-	out := &Poll{
-		fd:         fd,
-		shutdownFd: shutdownFd,
-		readBuf:    make([]byte, tunReadBufSize),
-	}
-	return out, nil
-}
-
-// 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 {
-	return blockOn(int32(t.fd), int32(t.shutdownFd), unix.POLLIN)
-}
-
-func (t *Poll) blockOnWrite() error {
-	return blockOn(int32(t.fd), int32(t.shutdownFd), unix.POLLOUT)
-}
-
-func (t *Poll) Read() ([]Packet, error) {
-	n, err := t.readOne(t.readBuf)
-	if err != nil {
-		return nil, err
-	}
-	t.batchRet[0] = Packet{Bytes: t.readBuf[:n]}
-	return t.batchRet[:], nil
-}
-
-func (t *Poll) readOne(to []byte) (int, error) {
-	for {
-		n, errno := unix.Read(t.fd, to)
-		if errno == nil {
-			return n, nil
-		}
-		switch errno {
-		case unix.EAGAIN:
-			if err := t.blockOnRead(); err != nil {
-				return 0, err
-			}
-		case unix.EINTR:
-			// retry
-		case unix.EBADF:
-			return 0, os.ErrClosed
-		default:
-			return 0, errno
-		}
-	}
-}
-
-// Write is safe for concurrent use
-func (t *Poll) Write(from []byte) (int, error) {
-	for {
-		n, errno := unix.Write(t.fd, from)
-		if errno == nil {
-			return n, nil
-		}
-		switch errno {
-		case unix.EAGAIN:
-			if err := t.blockOnWrite(); err != nil {
-				return 0, err
-			}
-		case unix.EINTR:
-			// retry
-		case unix.EBADF:
-			return 0, os.ErrClosed
-		default:
-			return 0, errno
-		}
-	}
-}
-
-func (t *Poll) Close() error {
-	if t.closed.Swap(true) {
-		return nil
-	}
-	//shutdownFd is owned by the container, so we should not close it
-	// Close the underlying fd but do NOT null t.fd: a reader may still be
-	// loading it in readOne, and mutating the field would race that load.
-	// It gets EBADF -> os.ErrClosed (or wakes via the shutdown eventfd's
-	// ppoll first). closed.Swap already guarantees we only close once.
-	return unix.Close(t.fd)
-}

overlay/tio/tun_file_linux_test.go

@@ -1,208 +0,0 @@
-//go:build linux && !android && !e2e_testing
-// +build linux,!android,!e2e_testing
-
-package tio
-
-import (
-	"errors"
-	"os"
-	"sync"
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/require"
-	"golang.org/x/sys/unix"
-)
-
-// newReadPipe returns a read fd. The matching write fd is registered for cleanup.
-// The caller takes ownership of the read fd (pass it into a QueueSet).
-func newReadPipe(t *testing.T) int {
-	t.Helper()
-	var fds [2]int
-	if err := unix.Pipe2(fds[:], unix.O_CLOEXEC); err != nil {
-		t.Fatalf("pipe2: %v", err)
-	}
-	t.Cleanup(func() { _ = unix.Close(fds[1]) })
-	return fds[0]
-}
-
-func TestPoll_WakeForShutdown_WakesFriends(t *testing.T) {
-	pipe1 := newReadPipe(t)
-	pipe2 := newReadPipe(t)
-	parent, err := NewPollQueueSet()
-	require.NoError(t, err)
-	require.NoError(t, parent.Add(pipe1))
-	require.NoError(t, parent.Add(pipe2))
-	t.Cleanup(func() {
-		_ = unix.Close(pipe1)
-		_ = unix.Close(pipe2)
-	})
-
-	readers := parent.Queues()
-	errs := make([]error, len(readers))
-	var wg sync.WaitGroup
-	for i, r := range readers {
-		wg.Add(1)
-		go func(i int, r Queue) {
-			defer wg.Done()
-			_, errs[i] = r.Read()
-		}(i, r)
-	}
-
-	time.Sleep(50 * time.Millisecond)
-
-	if err := parent.Close(); err != nil {
-		t.Fatalf("Close: %v", err)
-	}
-
-	done := make(chan struct{})
-	go func() { wg.Wait(); close(done) }()
-	select {
-	case <-done:
-	case <-time.After(2 * time.Second):
-		t.Fatal("readers did not wake")
-	}
-
-	for i, err := range errs {
-		if !errors.Is(err, os.ErrClosed) {
-			t.Errorf("reader %d: expected os.ErrClosed, got %v", i, err)
-		}
-	}
-}
-
-// 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,
-// the poll path would double-close on Add error. We force the failure with
-// an O_PATH descriptor: fcntl(F_SETFL) — which SetNonblock performs — is not
-// permitted on O_PATH fds and fails with EBADF, while the fd itself stays
-// open so we can observe that newPoll left it alone.
-func TestPoll_NewPoll_DoesNotCloseFdOnFailure(t *testing.T) {
-	fd, err := unix.Open("/", unix.O_PATH|unix.O_CLOEXEC, 0)
-	require.NoError(t, err)
-	t.Cleanup(func() { _ = unix.Close(fd) })
-
-	p, err := newPoll(fd, 1)
-	require.Error(t, err, "SetNonblock on an O_PATH fd should fail")
-	require.Nil(t, p)
-
-	// If newPoll had closed fd, F_GETFD would report it closed. It staying
-	// open proves newPoll left the fd for the caller to close exactly once.
-	require.True(t, fdOpen(t, fd), "newPoll must not close fd on failure; caller is the sole closer")
-}
-
-func TestPoll_Close_Idempotent(t *testing.T) {
-	tf, err := newPoll(newReadPipe(t), 1)
-	require.NoError(t, err)
-	if err := tf.Close(); err != nil {
-		t.Fatalf("first Close: %v", err)
-	}
-	if err := tf.Close(); err != nil {
-		t.Fatalf("second Close should be a no-op, got %v", err)
-	}
-}
-
-// fdOpen reports whether fd currently refers to an open file description.
-// A closed (or never-allocated) fd makes F_GETFD fail with EBADF.
-func fdOpen(t *testing.T, fd int) bool {
-	t.Helper()
-	_, err := unix.FcntlInt(uintptr(fd), unix.F_GETFD, 0)
-	if err == nil {
-		return true
-	}
-	if errors.Is(err, unix.EBADF) {
-		return false
-	}
-	t.Fatalf("unexpected fcntl(F_GETFD) error on fd %d: %v", fd, err)
-	return false
-}
-
-// TestPollQueueSet_Close_ClosesShutdownFd is the regression test for the
-// leaked shutdown eventfd: the container that owns shutdownFd must close it in
-// Close, and a second Close must be a safe no-op.
-func TestPollQueueSet_Close_ClosesShutdownFd(t *testing.T) {
-	qs, err := NewPollQueueSet()
-	require.NoError(t, err)
-	c, ok := qs.(*pollQueueSet)
-	require.True(t, ok)
-	require.NoError(t, qs.Add(newReadPipe(t)))
-
-	shutdownFd := c.shutdownFd
-	require.True(t, fdOpen(t, shutdownFd), "shutdown eventfd should be open before Close")
-
-	require.NoError(t, qs.Close())
-	require.False(t, fdOpen(t, shutdownFd), "shutdown eventfd should be closed after Close")
-
-	// Second Close must not touch fds (shutdownFd is now -1) and must return nil.
-	require.NoError(t, qs.Close())
-}

overlay/tun_android.go

@@ -13,7 +13,6 @@ import (
 
 	"github.com/gaissmai/bart"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/util"
 )
@@ -64,7 +63,7 @@ func (t *tun) RoutesFor(ip netip.Addr) routing.Gateways {
 	return r
 }
 
-func (t *tun) Activate() error {
+func (t tun) Activate() error {
 	return nil
 }
 
@@ -97,6 +96,10 @@ func (t *tun) Name() string {
 	return "android"
 }
 
-func (t *tun) Queues(int) ([]tio.Queue, error) {
+func (t *tun) SupportsMultiqueue() bool {
-	return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil
+	return false
+}
+
+func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
+	return nil, fmt.Errorf("TODO: multiqueue not implemented for android")
 }

overlay/tun_darwin.go

@@ -6,6 +6,7 @@ package overlay
 import (
 	"errors"
 	"fmt"
+	"io"
 	"log/slog"
 	"net/netip"
 	"os"
@@ -15,7 +16,6 @@ import (
 
 	"github.com/gaissmai/bart"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/util"
 	netroute "golang.org/x/net/route"
@@ -606,6 +606,10 @@ func (t *tun) Name() string {
 	return t.Device
 }
 
-func (t *tun) Queues(int) ([]tio.Queue, error) {
+func (t *tun) SupportsMultiqueue() bool {
-	return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil
+	return false
+}
+
+func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
+	return nil, fmt.Errorf("TODO: multiqueue not implemented for darwin")
 }

overlay/tun_disabled.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/rcrowley/go-metrics"
 	"github.com/slackhq/nebula/iputil"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 )
 
@@ -24,23 +23,6 @@ type disabledTun struct {
 	l  *slog.Logger
 }
 
-// Read hands the next queued packet to a reader, copying it into b. Reads
-// from concurrent queues are safe: the channel receive serializes them and
-// each queue copies into its own private scratch buffer.
-func (t *disabledTun) Read(b []byte) (int, error) {
-	r, ok := <-t.read
-	if !ok {
-		return 0, io.EOF
-	}
-
-	t.tx.Inc(1)
-	if t.l.Enabled(context.Background(), slog.LevelDebug) {
-		t.l.Debug("Write payload", "raw", prettyPacket(r))
-	}
-
-	return copy(b, r), nil
-}
-
 func newDisabledTun(vpnNetworks []netip.Prefix, queueLen int, metricsEnabled bool, l *slog.Logger) *disabledTun {
 	tun := &disabledTun{
 		vpnNetworks: vpnNetworks,
@@ -75,6 +57,24 @@ func (*disabledTun) Name() string {
 	return "disabled"
 }
 
+func (t *disabledTun) Read(b []byte) (int, error) {
+	r, ok := <-t.read
+	if !ok {
+		return 0, io.EOF
+	}
+
+	if len(r) > len(b) {
+		return 0, fmt.Errorf("packet larger than mtu: %d > %d bytes", len(r), len(b))
+	}
+
+	t.tx.Inc(1)
+	if t.l.Enabled(context.Background(), slog.LevelDebug) {
+		t.l.Debug("Write payload", "raw", prettyPacket(r))
+	}
+
+	return copy(b, r), nil
+}
+
 func (t *disabledTun) handleICMPEchoRequest(b []byte) bool {
 	out := make([]byte, len(b))
 	out = iputil.CreateICMPEchoResponse(b, out)
@@ -106,14 +106,12 @@ func (t *disabledTun) Write(b []byte) (int, error) {
 	return len(b), nil
 }
 
-func (t *disabledTun) Queues(n int) ([]tio.Queue, error) {
+func (t *disabledTun) SupportsMultiqueue() bool {
-	out := make([]tio.Queue, n)
+	return true
-	for i := range out {
+}
-		// NoClose: the shared channel and metrics are owned by the
+
-		// disabledTun; Close on the device tears them down once for everybody.
+func (t *disabledTun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
-		out[i] = tio.NewSingleQueueNoClose(t, defaultBatchBufSize)
+	return t, nil
-	}
-	return out, nil
 }
 
 func (t *disabledTun) Close() error {

overlay/tun_file_linux_test.go

@@ -0,0 +1,120 @@
+//go:build linux && !android && !e2e_testing
+// +build linux,!android,!e2e_testing
+
+package overlay
+
+import (
+	"errors"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"golang.org/x/sys/unix"
+)
+
+// newReadPipe returns a read fd. The matching write fd is registered for cleanup.
+// The caller takes ownership of the read fd (pass it to newTunFd / newFriend).
+func newReadPipe(t *testing.T) int {
+	t.Helper()
+	var fds [2]int
+	if err := unix.Pipe2(fds[:], unix.O_CLOEXEC); err != nil {
+		t.Fatalf("pipe2: %v", err)
+	}
+	t.Cleanup(func() { _ = unix.Close(fds[1]) })
+	return fds[0]
+}
+
+func TestTunFile_WakeForShutdown_UnblocksRead(t *testing.T) {
+	tf, err := newTunFd(newReadPipe(t))
+	if err != nil {
+		t.Fatalf("newTunFd: %v", err)
+	}
+	t.Cleanup(func() { _ = tf.Close() })
+
+	done := make(chan error, 1)
+	go func() {
+		_, err := tf.Read(make([]byte, 64))
+		done <- err
+	}()
+
+	// Verify Read is actually blocked in poll.
+	select {
+	case err := <-done:
+		t.Fatalf("Read returned before shutdown signal: %v", err)
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	if err := tf.wakeForShutdown(); err != nil {
+		t.Fatalf("wakeForShutdown: %v", err)
+	}
+
+	select {
+	case err := <-done:
+		if !errors.Is(err, os.ErrClosed) {
+			t.Fatalf("expected os.ErrClosed, got %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("Read did not wake on shutdown")
+	}
+}
+
+func TestTunFile_WakeForShutdown_WakesFriends(t *testing.T) {
+	parent, err := newTunFd(newReadPipe(t))
+	if err != nil {
+		t.Fatalf("newTunFd: %v", err)
+	}
+	friend, err := parent.newFriend(newReadPipe(t))
+	if err != nil {
+		_ = parent.Close()
+		t.Fatalf("newFriend: %v", err)
+	}
+	t.Cleanup(func() {
+		_ = friend.Close()
+		_ = parent.Close()
+	})
+
+	readers := []*tunFile{parent, friend}
+	errs := make([]error, len(readers))
+	var wg sync.WaitGroup
+	for i, r := range readers {
+		wg.Add(1)
+		go func(i int, r *tunFile) {
+			defer wg.Done()
+			_, errs[i] = r.Read(make([]byte, 64))
+		}(i, r)
+	}
+
+	time.Sleep(50 * time.Millisecond)
+
+	if err := parent.wakeForShutdown(); err != nil {
+		t.Fatalf("wakeForShutdown: %v", err)
+	}
+
+	done := make(chan struct{})
+	go func() { wg.Wait(); close(done) }()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("readers did not wake")
+	}
+
+	for i, err := range errs {
+		if !errors.Is(err, os.ErrClosed) {
+			t.Errorf("reader %d: expected os.ErrClosed, got %v", i, err)
+		}
+	}
+}
+
+func TestTunFile_Close_Idempotent(t *testing.T) {
+	tf, err := newTunFd(newReadPipe(t))
+	if err != nil {
+		t.Fatalf("newTunFd: %v", err)
+	}
+	if err := tf.Close(); err != nil {
+		t.Fatalf("first Close: %v", err)
+	}
+	if err := tf.Close(); err != nil {
+		t.Fatalf("second Close should be a no-op, got %v", err)
+	}
+}

overlay/tun_freebsd.go

@@ -7,6 +7,7 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"io"
 	"io/fs"
 	"log/slog"
 	"net/netip"
@@ -19,7 +20,7 @@ import (
 	"github.com/gaissmai/bart"
 
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
+
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/util"
 	netroute "golang.org/x/net/route"
@@ -560,8 +561,12 @@ func (t *tun) Name() string {
 	return t.Device
 }
 
-func (t *tun) Queues(int) ([]tio.Queue, error) {
+func (t *tun) SupportsMultiqueue() bool {
-	return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil
+	return false
+}
+
+func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
+	return nil, fmt.Errorf("TODO: multiqueue not implemented for freebsd")
 }
 
 func (t *tun) addRoutes(logErrors bool) error {

overlay/tun_ios.go

@@ -16,7 +16,6 @@ import (
 
 	"github.com/gaissmai/bart"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/util"
 	"golang.org/x/sys/unix"
@@ -160,6 +159,10 @@ func (t *tun) Name() string {
 	return "iOS"
 }
 
-func (t *tun) Queues(int) ([]tio.Queue, error) {
+func (t *tun) SupportsMultiqueue() bool {
-	return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil
+	return false
+}
+
+func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
+	return nil, fmt.Errorf("TODO: multiqueue not implemented for ios")
 }

overlay/tun_linux.go

@@ -4,7 +4,9 @@
 package overlay
 
 import (
+	"encoding/binary"
 	"fmt"
+	"io"
 	"log/slog"
 	"net"
 	"net/netip"
@@ -17,15 +19,180 @@ import (
 
 	"github.com/gaissmai/bart"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/util"
 	"github.com/vishvananda/netlink"
 	"golang.org/x/sys/unix"
 )
 
+// tunFile wraps a TUN file descriptor with poll-based reads. The FD provided will be changed to non-blocking.
+// A shared eventfd allows Close to wake all readers blocked in poll.
+type tunFile struct {
+	fd         int
+	shutdownFd int
+	lastOne    bool
+	readPoll   [2]unix.PollFd
+	writePoll  [2]unix.PollFd
+	closed     bool
+}
+
+// newFriend makes a tunFile for a MultiQueueReader that copies the shutdown eventfd from the parent tun
+func (r *tunFile) newFriend(fd int) (*tunFile, error) {
+	if err := unix.SetNonblock(fd, true); err != nil {
+		return nil, fmt.Errorf("failed to set tun fd non-blocking: %w", err)
+	}
+	return &tunFile{
+		fd:         fd,
+		shutdownFd: r.shutdownFd,
+		readPoll: [2]unix.PollFd{
+			{Fd: int32(fd), Events: unix.POLLIN},
+			{Fd: int32(r.shutdownFd), Events: unix.POLLIN},
+		},
+		writePoll: [2]unix.PollFd{
+			{Fd: int32(fd), Events: unix.POLLOUT},
+			{Fd: int32(r.shutdownFd), Events: unix.POLLIN},
+		},
+	}, nil
+}
+
+func newTunFd(fd int) (*tunFile, error) {
+	if err := unix.SetNonblock(fd, true); err != nil {
+		return nil, fmt.Errorf("failed to set tun fd non-blocking: %w", err)
+	}
+
+	shutdownFd, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create eventfd: %w", err)
+	}
+
+	out := &tunFile{
+		fd:         fd,
+		shutdownFd: shutdownFd,
+		lastOne:    true,
+		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},
+		},
+	}
+
+	return out, nil
+}
+
+func (r *tunFile) 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
+}
+
+func (r *tunFile) 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!
+	tunEvents := r.writePoll[0].Revents
+	shutdownEvents := r.writePoll[1].Revents
+	r.writePoll[0].Revents = 0
+	r.writePoll[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
+}
+
+func (r *tunFile) Read(buf []byte) (int, error) {
+	for {
+		if n, err := unix.Read(r.fd, buf); err == nil {
+			return n, nil
+		} else if err == unix.EAGAIN {
+			if err = r.blockOnRead(); err != nil {
+				return 0, err
+			}
+			continue
+		} else if err == unix.EINTR {
+			continue
+		} else if err == unix.EBADF {
+			return 0, os.ErrClosed
+		} else {
+			return 0, err
+		}
+	}
+}
+
+func (r *tunFile) Write(buf []byte) (int, error) {
+	for {
+		if n, err := unix.Write(r.fd, buf); err == nil {
+			return n, nil
+		} else if err == unix.EAGAIN {
+			if err = r.blockOnWrite(); err != nil {
+				return 0, err
+			}
+			continue
+		} else if err == unix.EINTR {
+			continue
+		} else if err == unix.EBADF {
+			return 0, os.ErrClosed
+		} else {
+			return 0, err
+		}
+	}
+}
+
+func (r *tunFile) wakeForShutdown() error {
+	var buf [8]byte
+	binary.NativeEndian.PutUint64(buf[:], 1)
+	_, err := unix.Write(int(r.readPoll[1].Fd), buf[:])
+	return err
+}
+
+func (r *tunFile) Close() error {
+	if r.closed { // avoid closing more than once. Technically a fd could get re-used, which would be a problem
+		return nil
+	}
+	r.closed = true
+	if r.lastOne {
+		_ = unix.Close(r.shutdownFd)
+	}
+	return unix.Close(r.fd)
+}
+
 type tun struct {
-	readers     tio.QueueSet
+	*tunFile
+	readers     []*tunFile
 	closeLock   sync.Mutex
 	Device      string
 	vpnNetworks []netip.Prefix
@@ -82,57 +249,44 @@ func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip
 	return t, nil
 }
 
-// openTunDev opens /dev/net/tun, creating the device node first if it's
-// missing (docker containers occasionally omit it).
-func openTunDev() (int, error) {
-	fd, err := unix.Open("/dev/net/tun", os.O_RDWR, 0)
-	if err == nil {
-		return fd, nil
-	}
-	if !os.IsNotExist(err) {
-		return -1, err
-	}
-	if err = os.MkdirAll("/dev/net", 0755); err != nil {
-		return -1, fmt.Errorf("/dev/net/tun doesn't exist, failed to mkdir -p /dev/net: %w", err)
-	}
-	if err = unix.Mknod("/dev/net/tun", unix.S_IFCHR|0600, int(unix.Mkdev(10, 200))); err != nil {
-		return -1, fmt.Errorf("failed to create /dev/net/tun: %w", err)
-	}
-	fd, err = unix.Open("/dev/net/tun", os.O_RDWR, 0)
-	if err != nil {
-		return -1, fmt.Errorf("created /dev/net/tun, but still failed: %w", err)
-	}
-	return fd, nil
-}
-
-// tunSetIff runs TUNSETIFF with the given flags and returns the kernel-chosen
-// device name on success.
-func tunSetIff(fd int, name string, flags uint16) (string, error) {
-	var req ifReq
-	req.Flags = flags
-	copy(req.Name[:], name)
-	if err := ioctl(uintptr(fd), uintptr(unix.TUNSETIFF), uintptr(unsafe.Pointer(&req))); err != nil {
-		return "", err
-	}
-	return strings.Trim(string(req.Name[:]), "\x00"), nil
-}
-
 func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, multiqueue bool) (*tun, error) {
-	baseFlags := uint16(unix.IFF_TUN | unix.IFF_NO_PI)
+	fd, err := unix.Open("/dev/net/tun", os.O_RDWR, 0)
+	if err != nil {
+		// If /dev/net/tun doesn't exist, try to create it (will happen in docker)
+		if os.IsNotExist(err) {
+			err = os.MkdirAll("/dev/net", 0755)
+			if err != nil {
+				return nil, fmt.Errorf("/dev/net/tun doesn't exist, failed to mkdir -p /dev/net: %w", err)
+			}
+			err = unix.Mknod("/dev/net/tun", unix.S_IFCHR|0600, int(unix.Mkdev(10, 200)))
+			if err != nil {
+				return nil, fmt.Errorf("failed to create /dev/net/tun: %w", err)
+			}
+
+			fd, err = unix.Open("/dev/net/tun", os.O_RDWR, 0)
+			if err != nil {
+				return nil, fmt.Errorf("created /dev/net/tun, but still failed: %w", err)
+			}
+		} else {
+			return nil, err
+		}
+	}
+
+	var req ifReq
+	req.Flags = uint16(unix.IFF_TUN | unix.IFF_NO_PI)
 	if multiqueue {
-		baseFlags |= unix.IFF_MULTI_QUEUE
+		req.Flags |= unix.IFF_MULTI_QUEUE
 	}
 	nameStr := c.GetString("tun.dev", "")
-
+	copy(req.Name[:], nameStr)
-	fd, err := openTunDev()
+	if err = ioctl(uintptr(fd), uintptr(unix.TUNSETIFF), uintptr(unsafe.Pointer(&req))); err != nil {
-	if err != nil {
-		return nil, err
-	}
-	name, err := tunSetIff(fd, nameStr, baseFlags)
-	if err != nil {
 		_ = unix.Close(fd)
-		return nil, &NameError{Name: nameStr, Underlying: err}
+		return nil, &NameError{
+			Name:       nameStr,
+			Underlying: err,
+		}
 	}
+	name := strings.Trim(string(req.Name[:]), "\x00")
 
 	t, err := newTunGeneric(c, l, fd, vpnNetworks)
 	if err != nil {
@@ -144,22 +298,16 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, multiqueue
 	return t, nil
 }
 
-// newTunGeneric does all the stuff common to different tun initialization
+// newTunGeneric does all the stuff common to different tun initialization paths. It will close your files on error.
-// paths. It will close your files on error.
 func newTunGeneric(c *config.C, l *slog.Logger, fd int, vpnNetworks []netip.Prefix) (*tun, error) {
-	qs, err := tio.NewPollQueueSet()
+	tfd, err := newTunFd(fd)
 	if err != nil {
 		_ = unix.Close(fd)
 		return nil, err
 	}
-	err = qs.Add(fd)
-	if err != nil {
-		_ = unix.Close(fd)
-		return nil, err
-	}
-
 	t := &tun{
-		readers:                   qs,
+		tunFile:                   tfd,
+		readers:                   []*tunFile{tfd},
 		closeLock:                 sync.Mutex{},
 		vpnNetworks:               vpnNetworks,
 		TXQueueLen:                c.GetInt("tun.tx_queue", 500),
@@ -258,41 +406,36 @@ func (t *tun) reload(c *config.C, initial bool) error {
 	return nil
 }
 
-// Queues opens additional kernel multiqueue fds until the device has n
+func (t *tun) SupportsMultiqueue() bool {
-// queues, then returns them all. The first queue was opened by newTun.
+	return true
-func (t *tun) Queues(n int) ([]tio.Queue, error) {
-	for len(t.readers.Queues()) < n {
-		if err := t.addQueue(); err != nil {
-			return nil, err
-		}
-	}
-	return t.readers.Queues(), nil
 }
 
-// addQueue opens one more IFF_MULTI_QUEUE fd on the device and adds it to
+func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
-// the queue set.
-func (t *tun) addQueue() error {
 	t.closeLock.Lock()
 	defer t.closeLock.Unlock()
 
 	fd, err := unix.Open("/dev/net/tun", os.O_RDWR, 0)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	flags := uint16(unix.IFF_TUN | unix.IFF_NO_PI | unix.IFF_MULTI_QUEUE)
+	var req ifReq
-	if _, err = tunSetIff(fd, t.Device, flags); err != nil {
+	req.Flags = uint16(unix.IFF_TUN | unix.IFF_NO_PI | unix.IFF_MULTI_QUEUE)
+	copy(req.Name[:], t.Device)
+	if err = ioctl(uintptr(fd), uintptr(unix.TUNSETIFF), uintptr(unsafe.Pointer(&req))); err != nil {
 		_ = unix.Close(fd)
-		return err
+		return nil, err
 	}
 
-	err = t.readers.Add(fd)
+	out, err := t.tunFile.newFriend(fd)
 	if err != nil {
 		_ = unix.Close(fd)
-		return err
+		return nil, err
 	}
 
-	return nil
+	t.readers = append(t.readers, out)
+
+	return out, nil
 }
 
 func (t *tun) RoutesFor(ip netip.Addr) routing.Gateways {
@@ -735,10 +878,32 @@ func (t *tun) Close() error {
 		t.routeChan = nil
 	}
 
+	// Signal all readers blocked in poll to wake up and exit
+	_ = t.tunFile.wakeForShutdown()
+
 	if t.ioctlFd > 0 {
 		_ = unix.Close(int(t.ioctlFd))
 		t.ioctlFd = 0
 	}
 
-	return t.readers.Close()
+	for i := range t.readers {
+		if i == 0 {
+			continue //we want to close the zeroth reader last
+		}
+		err := t.readers[i].Close()
+		if err != nil {
+			t.l.Error("error closing tun reader", "reader", i, "error", err)
+		} else {
+			t.l.Info("closed tun reader", "reader", i)
+		}
+	}
+
+	//this is t.readers[0] too
+	err := t.tunFile.Close()
+	if err != nil {
+		t.l.Error("error closing tun reader", "reader", 0, "error", err)
+	} else {
+		t.l.Info("closed tun reader", "reader", 0)
+	}
+	return err
 }

overlay/tun_netbsd.go

@@ -6,6 +6,7 @@ package overlay
 import (
 	"errors"
 	"fmt"
+	"io"
 	"log/slog"
 	"net/netip"
 	"os"
@@ -16,7 +17,6 @@ import (
 
 	"github.com/gaissmai/bart"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/util"
 	netroute "golang.org/x/net/route"
@@ -390,8 +390,12 @@ func (t *tun) Name() string {
 	return t.Device
 }
 
-func (t *tun) Queues(int) ([]tio.Queue, error) {
+func (t *tun) SupportsMultiqueue() bool {
-	return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil
+	return false
+}
+
+func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
+	return nil, fmt.Errorf("TODO: multiqueue not implemented for netbsd")
 }
 
 func (t *tun) addRoutes(logErrors bool) error {

overlay/tun_openbsd.go

@@ -6,6 +6,7 @@ package overlay
 import (
 	"errors"
 	"fmt"
+	"io"
 	"log/slog"
 	"net/netip"
 	"os"
@@ -16,7 +17,6 @@ import (
 
 	"github.com/gaissmai/bart"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/util"
 	netroute "golang.org/x/net/route"
@@ -138,8 +138,8 @@ func tunWritev(fd int, iovecs []unix.Iovec) (n int, err error)
 //go:noescape
 func tunReadv(fd int, iovecs []unix.Iovec) (n int, err error)
 
-// Read pulls one IP packet off the tun device, scattering the 4 byte protocol header away from
+// Read pulls one IP packet off the tun device, scattering the 4 byte protocol header away from the
-// the packet so the payload lands directly in to.
+// packet so the payload lands directly in to.
 func (t *tun) Read(to []byte) (int, error) {
 	var head [4]byte
 
@@ -369,8 +369,12 @@ func (t *tun) Name() string {
 	return t.Device
 }
 
-func (t *tun) Queues(int) ([]tio.Queue, error) {
+func (t *tun) SupportsMultiqueue() bool {
-	return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil
+	return false
+}
+
+func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
+	return nil, fmt.Errorf("TODO: multiqueue not implemented for openbsd")
 }
 
 func (t *tun) addRoutes(logErrors bool) error {

overlay/tun_tester.go

@@ -14,7 +14,6 @@ import (
 
 	"github.com/gaissmai/bart"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/udp"
 )
@@ -178,6 +177,10 @@ func (t *TestTun) Read(b []byte) (int, error) {
 	return n, nil
 }
 
-func (t *TestTun) Queues(int) ([]tio.Queue, error) {
+func (t *TestTun) SupportsMultiqueue() bool {
-	return []tio.Queue{tio.NewSingleQueue(t, udp.MTU)}, nil
+	return false
+}
+
+func (t *TestTun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
+	return nil, fmt.Errorf("TODO: multiqueue not implemented")
 }

overlay/tun_windows.go

@@ -6,6 +6,7 @@ package overlay
 import (
 	"crypto"
 	"fmt"
+	"io"
 	"log/slog"
 	"net/netip"
 	"os"
@@ -17,7 +18,6 @@ import (
 
 	"github.com/gaissmai/bart"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/util"
 	"github.com/slackhq/nebula/wintun"
@@ -47,10 +47,6 @@ type winTun struct {
 	tun *wintun.NativeTun
 }
 
-func (t *winTun) Read(b []byte) (int, error) {
-	return t.tun.Read(b, 0)
-}
-
 func newTunFromFd(_ *config.C, _ *slog.Logger, _ int, _ []netip.Prefix) (Device, error) {
 	return nil, fmt.Errorf("newTunFromFd not supported in Windows")
 }
@@ -259,12 +255,20 @@ func (t *winTun) Name() string {
 	return t.Device
 }
 
+func (t *winTun) Read(b []byte) (int, error) {
+	return t.tun.Read(b, 0)
+}
+
 func (t *winTun) Write(b []byte) (int, error) {
 	return t.tun.Write(b, 0)
 }
 
-func (t *winTun) Queues(int) ([]tio.Queue, error) {
+func (t *winTun) SupportsMultiqueue() bool {
-	return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil
+	return false
+}
+
+func (t *winTun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
+	return nil, fmt.Errorf("TODO: multiqueue not implemented for windows")
 }
 
 func (t *winTun) Close() error {

overlay/user.go

@@ -6,7 +6,6 @@ import (
 	"net/netip"
 
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 )
 
@@ -47,16 +46,12 @@ func (d *UserDevice) RoutesFor(ip netip.Addr) routing.Gateways {
 	return routing.Gateways{routing.NewGateway(ip, 1)}
 }
 
-func (d *UserDevice) Queues(n int) ([]tio.Queue, error) {
+func (d *UserDevice) SupportsMultiqueue() bool {
-	out := make([]tio.Queue, n)
+	return true
-	for i := range out {
+}
-		// All queues share the underlying pipes (the io.Pipe serializes
+
-		// concurrent callers) but each owns a private scratch buffer so
+func (d *UserDevice) NewMultiQueueReader() (io.ReadWriteCloser, error) {
-		// concurrent Reads across queues never alias. NoClose: the pipes are
+	return d, nil
-		// owned by the UserDevice and torn down once by UserDevice.Close.
-		out[i] = tio.NewSingleQueueNoClose(d, defaultBatchBufSize)
-	}
-	return out, nil
 }
 
 func (d *UserDevice) Pipe() (*io.PipeReader, *io.PipeWriter) {
@@ -66,11 +61,9 @@ func (d *UserDevice) Pipe() (*io.PipeReader, *io.PipeWriter) {
 func (d *UserDevice) Read(p []byte) (n int, err error) {
 	return d.outboundReader.Read(p)
 }
-
 func (d *UserDevice) Write(p []byte) (n int, err error) {
 	return d.inboundWriter.Write(p)
 }
-
 func (d *UserDevice) Close() error {
 	d.inboundWriter.Close()
 	d.outboundWriter.Close()

overlay/user_test.go

@@ -1,163 +0,0 @@
-package overlay
-
-import (
-	"fmt"
-	"net/netip"
-	"sync"
-	"testing"
-
-	"github.com/slackhq/nebula/overlay/tio"
-)
-
-// newTestUserDevice returns the concrete *UserDevice so tests can reach Pipe()
-// and the internal queue plumbing.
-func newTestUserDevice(t *testing.T) *UserDevice {
-	t.Helper()
-	dev, err := NewUserDevice([]netip.Prefix{netip.MustParsePrefix("10.0.0.1/24")})
-	if err != nil {
-		t.Fatalf("NewUserDevice: %v", err)
-	}
-	ud, ok := dev.(*UserDevice)
-	if !ok {
-		t.Fatalf("NewUserDevice returned %T, want *UserDevice", dev)
-	}
-	return ud
-}
-
-// TestUserDeviceReadersDistinctBuffers ensures each Queue is actually different
-func TestUserDeviceReadersDistinctBuffers(t *testing.T) {
-	d := newTestUserDevice(t)
-
-	readers, err := d.Queues(2)
-	if err != nil {
-		t.Fatalf("Queues: %v", err)
-	}
-	if len(readers) != 2 {
-		t.Fatalf("Queues(2) returned %d queues, want 2", len(readers))
-	}
-
-	// Distinct queue objects.
-	if readers[0] == readers[1] {
-		t.Fatal("Queues(2) returned the same queue object twice")
-	}
-
-	// Drive one packet through each queue and confirm the borrowed bytes from
-	// the first read are NOT clobbered by the second read. With a shared
-	// buffer, reading pkt1 into q1 would corrupt q0's still-borrowed slice.
-	_, ow := d.Pipe()
-
-	pkt0 := []byte("packet-zero-aaaaaaaa")
-	pkt1 := []byte("packet-one-bbbbbbbbb")
-
-	// The pipe is unbuffered, so writes block until a reader consumes them.
-	// Serialize: write pkt0 (read on q0), then write pkt1 (read on q1).
-	go func() {
-		if _, err := ow.Write(pkt0); err != nil {
-			t.Errorf("write pkt0: %v", err)
-		}
-		if _, err := ow.Write(pkt1); err != nil {
-			t.Errorf("write pkt1: %v", err)
-		}
-	}()
-
-	got0, err := readers[0].Read()
-	if err != nil {
-		t.Fatalf("q0.Read: %v", err)
-	}
-	if len(got0) != 1 || string(got0[0].Bytes) != string(pkt0) {
-		t.Fatalf("q0 first read = %q, want %q", firstBytes(got0), pkt0)
-	}
-	// Hold onto q0's borrowed slice across q1's read.
-	borrowed := got0[0].Bytes
-
-	got1, err := readers[1].Read()
-	if err != nil {
-		t.Fatalf("q1.Read: %v", err)
-	}
-	if len(got1) != 1 || string(got1[0].Bytes) != string(pkt1) {
-		t.Fatalf("q1 read = %q, want %q", firstBytes(got1), pkt1)
-	}
-
-	// q0's borrowed bytes must still hold pkt0 - a shared buffer would now
-	// show pkt1's contents.
-	if string(borrowed) != string(pkt0) {
-		t.Fatalf("q0 borrowed bytes were clobbered by q1's read: got %q, want %q", borrowed, pkt0)
-	}
-}
-
-// TestUserDeviceReadersConcurrentRace exercises two queues reading distinct
-// packets concurrently. Run it under `go test -race`: with the old
-// shared-buffer implementation the concurrent Reads raced on readBuf/batchRet
-// and corrupted each other's returned slices.
-func TestUserDeviceReadersConcurrentRace(t *testing.T) {
-	d := newTestUserDevice(t)
-	readers, err := d.Queues(2)
-	if err != nil {
-		t.Fatalf("Queues: %v", err)
-	}
-	_, ow := d.Pipe()
-
-	const iterations = 200
-
-	errs := make(chan error, 3)
-
-	// Each reader parks in Read on the shared outboundReader; io.Pipe hands
-	// each write to whichever reader is currently waiting. We only care that
-	// concurrent Reads into distinct buffers are race-free, so any parked
-	// reader may serve any write.
-	var wg sync.WaitGroup
-	run := func(idx int) {
-		defer wg.Done()
-		for i := 0; i < iterations; i++ {
-			pkts, err := readers[idx].Read()
-			if err != nil {
-				errs <- err
-				return
-			}
-			if len(pkts) != 1 {
-				errs <- fmt.Errorf("reader %d: got %d packets, want 1", idx, len(pkts))
-				return
-			}
-			// Touch every byte of the borrowed slice while the other reader
-			// may be mid-Read; a shared buffer would race here.
-			total := 0
-			for _, c := range pkts[0].Bytes {
-				total += int(c)
-			}
-			_ = total
-		}
-	}
-
-	wg.Add(2)
-	go run(0)
-	go run(1)
-
-	// Feed 2*iterations packets. io.Pipe copies each write straight into the
-	// waiting reader's private buffer, so reusing buf between writes is safe.
-	go func() {
-		buf := make([]byte, 32)
-		for i := 0; i < 2*iterations; i++ {
-			for j := range buf {
-				buf[j] = byte(i + j)
-			}
-			if _, err := ow.Write(buf); err != nil {
-				errs <- err
-				return
-			}
-		}
-	}()
-
-	wg.Wait()
-	select {
-	case err := <-errs:
-		t.Fatalf("concurrent reader failed: %v", err)
-	default:
-	}
-}
-
-func firstBytes(p []tio.Packet) []byte {
-	if len(p) == 0 {
-		return nil
-	}
-	return p[0].Bytes
-}

util/cpupin_linux.go

@@ -1,43 +0,0 @@
-//go:build linux && !android && !e2e_testing
-
-package util
-
-import (
-	"runtime"
-
-	"golang.org/x/sys/unix"
-)
-
-// PinThreadToCPU restricts the calling OS thread to the given CPU via
-// sched_setaffinity(2). Combined with runtime.LockOSThread on the
-// goroutine, this prevents the kernel from migrating us across CPUs and
-// in turn keeps every UDP send from this goroutine going through the
-// same XPS-selected TX ring, eliminating the wire-side reorder that
-// otherwise fragments one nebula flow across multiple rings.
-func PinThreadToCPU(cpu int) error {
-	runtime.LockOSThread()
-	var set unix.CPUSet
-	set.Zero()
-	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
-}

util/cpupin_other.go

@@ -1,18 +0,0 @@
-//go:build !linux || android || e2e_testing
-
-package util
-
-// PinThreadToCPU is a no-op outside Linux: only Linux exposes a stable
-// per-thread CPU affinity API and only Linux has XPS-driven TX ring
-// selection in the first place. On every other platform there's nothing
-// to fix here.
-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
-}