personal_interest_mirrors/nebula
1
0
Fork 0
mirror of https://github.com/slackhq/nebula.git synced 2026-08-15 09:36:58 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

simplify making new Queues

Browse Source
This commit is contained in:
JackDoan
2026-07-14 10:56:21 -05:00
parent 05443523bd
commit 720990ddcd
18 changed files with 105 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files
overlay/device.go
+7 -3
View File
@@ -18,7 +18,11 @@ type Device interface {
Networks() []netip.Prefix
Name() string
RoutesFor(netip.Addr) routing.Gateways
SupportsMultiqueue() bool
NewMultiQueueReader() error
Readers() []tio.Queue
// Queues returns the device's packet queues, opening additional ones as
// needed until there are n. Platforms without multiqueue support return
// 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
+2 -11
View File
@@ -3,7 +3,6 @@
package overlaytest
import (
"errors"
"net/netip"
"github.com/slackhq/nebula/overlay/tio"
@@ -39,16 +38,8 @@ func (NoopTun) Write([]byte) (int, error) {
return 0, nil
}
func (NoopTun) SupportsMultiqueue() bool {
return false
}
func (NoopTun) NewMultiQueueReader() error {
return errors.New("unsupported")
}
func (NoopTun) Readers() []tio.Queue {
return []tio.Queue{NoopTun{}}
func (NoopTun) Queues(int) ([]tio.Queue, error) {
return []tio.Queue{NoopTun{}}, nil
}
func (NoopTun) Close() error {
overlay/tun_android.go
+2 -10
View File
@@ -97,14 +97,6 @@ func (t *tun) Name() string {
return "android"
}
func (t *tun) SupportsMultiqueue() bool {
return false
}
func (t *tun) NewMultiQueueReader() error {
return fmt.Errorf("TODO: multiqueue not implemented for android")
}
func (t *tun) Readers() []tio.Queue {
return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}
func (t *tun) Queues(int) ([]tio.Queue, error) {
return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil
}
overlay/tun_darwin.go
+2 -10
View File
@@ -606,14 +606,6 @@ func (t *tun) Name() string {
return t.Device
}
func (t *tun) SupportsMultiqueue() bool {
return false
}
func (t *tun) NewMultiQueueReader() error {
return fmt.Errorf("TODO: multiqueue not implemented for darwin")
}
func (t *tun) Readers() []tio.Queue {
return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}
func (t *tun) Queues(int) ([]tio.Queue, error) {
return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil
}
overlay/tun_disabled.go
+7 -18
View File
@@ -19,10 +19,9 @@ type disabledTun struct {
vpnNetworks []netip.Prefix
// Track these metrics since we don't have the tun device to do it for us
tx metrics.Counter
rx metrics.Counter
l *slog.Logger
numReaders int
tx metrics.Counter
rx metrics.Counter
l *slog.Logger
}
// Read hands the next queued packet to a reader, copying it into b. Reads
@@ -47,7 +46,6 @@ func newDisabledTun(vpnNetworks []netip.Prefix, queueLen int, metricsEnabled boo
vpnNetworks: vpnNetworks,
read: make(chan []byte, queueLen),
l: l,
numReaders: 1,
}
if metricsEnabled {
@@ -108,23 +106,14 @@ func (t *disabledTun) Write(b []byte) (int, error) {
return len(b), nil
}
func (t *disabledTun) SupportsMultiqueue() bool {
return true
}
func (t *disabledTun) NewMultiQueueReader() error {
t.numReaders++
return nil
}
func (t *disabledTun) Readers() []tio.Queue {
out := make([]tio.Queue, t.numReaders)
for i := range t.numReaders {
func (t *disabledTun) Queues(n int) ([]tio.Queue, error) {
out := make([]tio.Queue, n)
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.
out[i] = tio.NewSingleQueueNoClose(t, defaultBatchBufSize)
}
return out
return out, nil
}
func (t *disabledTun) Close() error {
overlay/tun_freebsd.go
+2 -10
View File
@@ -560,12 +560,8 @@ func (t *tun) Name() string {
return t.Device
}
func (t *tun) SupportsMultiqueue() bool {
return false
}
func (t *tun) NewMultiQueueReader() error {
return fmt.Errorf("TODO: multiqueue not implemented for freebsd")
func (t *tun) Queues(int) ([]tio.Queue, error) {
return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil
}
func (t *tun) addRoutes(logErrors bool) error {
@@ -592,10 +588,6 @@ func (t *tun) addRoutes(logErrors bool) error {
return nil
}
func (t *tun) Readers() []tio.Queue {
return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}
}
func (t *tun) removeRoutes(routes []Route) error {
for _, r := range routes {
if !r.Install {
overlay/tun_ios.go
+2 -10
View File
@@ -160,14 +160,6 @@ func (t *tun) Name() string {
return "iOS"
}
func (t *tun) SupportsMultiqueue() bool {
return false
}
func (t *tun) NewMultiQueueReader() error {
return fmt.Errorf("TODO: multiqueue not implemented for ios")
}
func (t *tun) Readers() []tio.Queue {
return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}
func (t *tun) Queues(int) ([]tio.Queue, error) {
return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil
}
overlay/tun_linux.go
+15 -9
View File
@@ -39,7 +39,7 @@ type tun struct {
// the kernel: usoOffloadFlags when USO was accepted, tsoOffloadFlags on
// the TSO-only fallback, or 0 when vnetHdr is off. TUNSETOFFLOAD is
// device-wide (drivers/net/tun.c set_offload updates tun->set_features
// for the whole netdev), so NewMultiQueueReader must replay this exact
// for the whole netdev), so addQueue must replay this exact
// mask on every added queue — issuing a narrower mask there would
// silently downgrade offloads (e.g. disable USO) for all queues while
// they still advertise the stale capability.
@@ -174,7 +174,7 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, multiqueue
}
vnetHdr := true
// offloadFlags is the exact TUN_F_* mask the kernel accepted. We remember
// it (rather than a plain bool) so NewMultiQueueReader can replay the
// it (rather than a plain bool) so addQueue can replay the
// identical device-wide mask on added queues instead of downgrading them.
var offloadFlags uint
name, err := tunSetIff(fd, nameStr, baseFlags|unix.IFF_VNET_HDR)
@@ -349,11 +349,21 @@ func (t *tun) reload(c *config.C, initial bool) error {
return nil
}
func (t *tun) SupportsMultiqueue() bool {
return true
// Queues opens additional kernel multiqueue fds until the device has n
// queues, then returns them all. The first queue was opened by newTun; each
// extra fd replays the negotiated offload state (see addQueue).
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
}
func (t *tun) NewMultiQueueReader() error {
// addQueue opens one more IFF_MULTI_QUEUE fd on the device and adds it to
// the queue set.
func (t *tun) addQueue() error {
t.closeLock.Lock()
defer t.closeLock.Unlock()
@@ -837,10 +847,6 @@ func (t *tun) updateRoutes(r netlink.RouteUpdate) {
t.routeTree.Store(newTree)
}
func (t *tun) Readers() []tio.Queue {
return t.readers.Queues()
}
func (t *tun) Close() error {
t.closeLock.Lock()
defer t.closeLock.Unlock()
overlay/tun_linux_test.go
+8 -9
View File
@@ -41,7 +41,7 @@ func TestTunAdvMSS(t *testing.T) {
func TestOffloadUSOEnabled(t *testing.T) {
// usoOffloadFlags must be a strict superset of tsoOffloadFlags. Otherwise
// the TSO-only fallback (and the historic hardcoded-mask bug in
// NewMultiQueueReader) would not actually be a downgrade.
// addQueue) would not actually be a downgrade.
if usoOffloadFlags&tsoOffloadFlags != tsoOffloadFlags {
t.Fatalf("usoOffloadFlags (%#x) is not a superset of tsoOffloadFlags (%#x)", usoOffloadFlags, tsoOffloadFlags)
}
@@ -67,20 +67,19 @@ func TestOffloadUSOEnabled(t *testing.T) {
}
}
// TestNewMultiQueueReaderReplaysNegotiatedMask guards the device-wide
// TUNSETOFFLOAD downgrade bug: NewMultiQueueReader must issue the exact mask
// newTun negotiated (t.offloadFlags), not a hardcoded TSO-only mask. Because
// TUNSETOFFLOAD is per-netdev, a narrower mask on an added queue silently
// disables USO for every queue on a USO-capable kernel while the queues keep
// advertising it.
// TestAddQueueReplaysNegotiatedMask guards the device-wide TUNSETOFFLOAD
// downgrade bug: addQueue must issue the exact mask newTun negotiated
// (t.offloadFlags), not a hardcoded TSO-only mask. Because TUNSETOFFLOAD is
// per-netdev, a narrower mask on an added queue silently disables USO for
// every queue on a USO-capable kernel while the queues keep advertising it.
//
// A full multi-queue exercise needs /dev/net/tun and CAP_NET_ADMIN, which are
// not available in CI/sandbox, so this asserts on the struct field that the
// TUNSETOFFLOAD argument is read from.
func TestNewMultiQueueReaderReplaysNegotiatedMask(t *testing.T) {
func TestAddQueueReplaysNegotiatedMask(t *testing.T) {
t.Run("uso-negotiated", func(t *testing.T) {
tn := &tun{vnetHdr: true, offloadFlags: usoOffloadFlags}
// The ioctl argument in NewMultiQueueReader is uintptr(t.offloadFlags);
// The ioctl argument in addQueue is uintptr(t.offloadFlags);
// it must equal the negotiated USO mask, and must NOT be the TSO-only
// mask (the original bug).
if tn.offloadFlags != usoOffloadFlags {
overlay/tun_netbsd.go
+2 -10
View File
@@ -68,10 +68,6 @@ type tun struct {
fd int
}
func (t *tun) Readers() []tio.Queue {
return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}
}
var deviceNameRE = regexp.MustCompile(`^tun[0-9]+$`)
func newTunFromFd(_ *config.C, _ *slog.Logger, _ int, _ []netip.Prefix) (*tun, error) {
@@ -394,12 +390,8 @@ func (t *tun) Name() string {
return t.Device
}
func (t *tun) SupportsMultiqueue() bool {
return false
}
func (t *tun) NewMultiQueueReader() error {
return fmt.Errorf("TODO: multiqueue not implemented for netbsd")
func (t *tun) Queues(int) ([]tio.Queue, error) {
return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil
}
func (t *tun) addRoutes(logErrors bool) error {
overlay/tun_openbsd.go
+2 -10
View File
@@ -369,12 +369,8 @@ func (t *tun) Name() string {
return t.Device
}
func (t *tun) SupportsMultiqueue() bool {
return false
}
func (t *tun) NewMultiQueueReader() error {
return fmt.Errorf("TODO: multiqueue not implemented for openbsd")
func (t *tun) Queues(int) ([]tio.Queue, error) {
return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil
}
func (t *tun) addRoutes(logErrors bool) error {
@@ -425,10 +421,6 @@ func (t *tun) deviceBytes() (o [16]byte) {
return
}
func (t *tun) Readers() []tio.Queue {
return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}
}
func addRoute(prefix netip.Prefix, gateways []netip.Prefix) error {
sock, err := unix.Socket(unix.AF_ROUTE, unix.SOCK_RAW, unix.AF_UNSPEC)
if err != nil {
overlay/tun_tester.go
+2 -10
View File
@@ -178,14 +178,6 @@ func (t *TestTun) Read(b []byte) (int, error) {
return n, nil
}
func (t *TestTun) Readers() []tio.Queue {
return []tio.Queue{tio.NewSingleQueue(t, udp.MTU)}
}
func (t *TestTun) SupportsMultiqueue() bool {
return false
}
func (t *TestTun) NewMultiQueueReader() error {
return fmt.Errorf("TODO: multiqueue not implemented")
func (t *TestTun) Queues(int) ([]tio.Queue, error) {
return []tio.Queue{tio.NewSingleQueue(t, udp.MTU)}, nil
}
overlay/tun_windows.go
+2 -10
View File
@@ -263,16 +263,8 @@ func (t *winTun) Write(b []byte) (int, error) {
return t.tun.Write(b, 0)
}
func (t *winTun) SupportsMultiqueue() bool {
return false
}
func (t *winTun) NewMultiQueueReader() error {
return fmt.Errorf("TODO: multiqueue not implemented for windows")
}
func (t *winTun) Readers() []tio.Queue {
return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}
func (t *winTun) Queues(int) ([]tio.Queue, error) {
return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil
}
func (t *winTun) Close() error {
overlay/user.go
+4 -15
View File
@@ -24,13 +24,11 @@ func NewUserDevice(vpnNetworks []netip.Prefix) (Device, error) {
outboundWriter: ow,
inboundReader: ir,
inboundWriter: iw,
numReaders: 1,
}, nil
}
type UserDevice struct {
vpnNetworks []netip.Prefix
numReaders int
outboundReader *io.PipeReader
outboundWriter *io.PipeWriter
@@ -49,25 +47,16 @@ func (d *UserDevice) RoutesFor(ip netip.Addr) routing.Gateways {
return routing.Gateways{routing.NewGateway(ip, 1)}
}
func (d *UserDevice) SupportsMultiqueue() bool {
return true
}
func (d *UserDevice) NewMultiQueueReader() error {
d.numReaders++
return nil
}
func (d *UserDevice) Readers() []tio.Queue {
out := make([]tio.Queue, d.numReaders)
for i := range d.numReaders {
func (d *UserDevice) Queues(n int) ([]tio.Queue, error) {
out := make([]tio.Queue, n)
for i := range out {
// All queues share the underlying pipes (the io.Pipe serializes
// concurrent callers) but each owns a private scratch buffer so
// concurrent Reads across queues never alias. NoClose: the pipes are
// owned by the UserDevice and torn down once by UserDevice.Close.
out[i] = tio.NewSingleQueueNoClose(d, defaultBatchBufSize)
}
return out
return out, nil
}
func (d *UserDevice) Pipe() (*io.PipeReader, *io.PipeWriter) {
overlay/user_test.go
+9 -17
View File
@@ -24,29 +24,21 @@ func newTestUserDevice(t *testing.T) *UserDevice {
return ud
}
// TestUserDeviceReadersDistinctBuffers is the regression test for the
// multiqueue packet-corruption bug: Readers() used to hand the same
// *UserDevice (and therefore the same read scratch buffer) to every queue, so
// one reader's borrowed Packet.Bytes was overwritten by another reader's
// concurrent Read. Readers() must now return numReaders DISTINCT queue
// objects, each with its own backing buffer — verified behaviorally below by
// holding one queue's borrowed slice across the other queue's Read.
// TestUserDeviceReadersDistinctBuffers ensures each Queue is actually different
func TestUserDeviceReadersDistinctBuffers(t *testing.T) {
d := newTestUserDevice(t)
// One extra reader => two queues total.
if err := d.NewMultiQueueReader(); err != nil {
t.Fatalf("NewMultiQueueReader: %v", err)
readers, err := d.Queues(2)
if err != nil {
t.Fatalf("Queues: %v", err)
}
readers := d.Readers()
if len(readers) != 2 {
t.Fatalf("Readers() returned %d queues, want 2", len(readers))
t.Fatalf("Queues(2) returned %d queues, want 2", len(readers))
}
// Distinct queue objects.
if readers[0] == readers[1] {
t.Fatal("Readers() returned the same queue object twice")
t.Fatal("Queues(2) returned the same queue object twice")
}
// Drive one packet through each queue and confirm the borrowed bytes from
@@ -99,10 +91,10 @@ func TestUserDeviceReadersDistinctBuffers(t *testing.T) {
// and corrupted each other's returned slices.
func TestUserDeviceReadersConcurrentRace(t *testing.T) {
d := newTestUserDevice(t)
if err := d.NewMultiQueueReader(); err != nil {
t.Fatalf("NewMultiQueueReader: %v", err)
readers, err := d.Queues(2)
if err != nil {
t.Fatalf("Queues: %v", err)
}
readers := d.Readers()
_, ow := d.Pipe()
const iterations = 200