diff --git a/control_lifecycle_test.go b/control_lifecycle_test.go index 0b5d106d..0aab85cc 100644 --- a/control_lifecycle_test.go +++ b/control_lifecycle_test.go @@ -11,6 +11,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/test" "github.com/slackhq/nebula/udp" @@ -30,9 +31,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(p []byte) (int, error) { +func (d *fakeDevice) Read() ([]tio.Packet, error) { <-d.closedCh - return 0, io.EOF + return nil, io.EOF } func (d *fakeDevice) Write(p []byte) (int, error) { return len(p), nil } @@ -49,10 +50,8 @@ 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) NewMultiQueueReader() (io.ReadWriteCloser, error) { - return nil, errors.New("unsupported") -} + +func (d *fakeDevice) Queues(int) ([]tio.Queue, error) { return []tio.Queue{d}, nil } // newReadyControl hand-builds the minimum Control that Main would have // produced right before Start, including the construction token NewInterface @@ -78,7 +77,6 @@ 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, @@ -155,7 +153,14 @@ type multiqueueDevice struct { *fakeDevice } -func (d *multiqueueDevice) SupportsMultiqueue() bool { return true } +// Queues claims multiqueue support but fails to open the second queue, +// 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()} @@ -166,7 +171,6 @@ func TestControl_StartMultiqueueFailureReleases(t *testing.T) { inside: dev, outside: conn, writers: []udp.Conn{conn}, - readers: make([]io.ReadWriteCloser, 2), routines: 2, l: test.NewLogger(), } diff --git a/inside.go b/inside.go index 163a6034..be44310a 100644 --- a/inside.go +++ b/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.readers[q].Write(packet) + _, err := f.queues[q].Write(packet) if err != nil { f.l.Error("Failed to forward to tun", "error", err) } @@ -96,7 +96,7 @@ func (f *Interface) rejectInside(packet []byte, out []byte, q int) { return } - _, err := f.readers[q].Write(out) + _, err := f.queues[q].Write(out) if err != nil { f.l.Error("Failed to write to tun", "error", err) } diff --git a/interface.go b/interface.go index c44f38b3..07384376 100644 --- a/interface.go +++ b/interface.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "io" "log/slog" "net/netip" "slices" @@ -20,6 +19,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" ) @@ -90,7 +90,7 @@ type Interface struct { ctx context.Context writers []udp.Conn - readers []io.ReadWriteCloser + queues []tio.Queue wg sync.WaitGroup // fatalErr holds the first unexpected reader error that caused shutdown. @@ -189,7 +189,6 @@ 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, @@ -240,27 +239,27 @@ func (f *Interface) activate() error { "boringcrypto", boringEnabled(), ) - if f.routines > 1 { - if !f.inside.SupportsMultiqueue() || !f.outside.SupportsMultipleReaders() { - f.routines = 1 - f.l.Warn("routines is not supported on this platform, falling back to a single routine") - } + if f.routines > 1 && !f.outside.SupportsMultipleReaders() { + f.routines = 1 + f.l.Warn("multiple udp readers are 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 { @@ -281,7 +280,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.readers[i], i) + f.listenIn(f.queues[i], i) }) } @@ -336,8 +335,7 @@ func (f *Interface) listenOut(i int) { f.l.Debug("underlay reader is done", "reader", i) } -func (f *Interface) listenIn(reader io.ReadWriteCloser, i int) { - packet := make([]byte, mtu) +func (f *Interface) listenIn(queue tio.Queue, i int) { out := make([]byte, mtu) fwPacket := &firewall.Packet{} nb := make([]byte, 12, 12) @@ -345,7 +343,7 @@ func (f *Interface) listenIn(reader io.ReadWriteCloser, i int) { conntrackCache := firewall.NewConntrackCacheTicker(f.ctx, f.l, f.conntrackCacheTimeout) for { - n, err := reader.Read(packet) + pkts, err := queue.Read() if err != nil { // Same shutdown noise handling as listenOut if !f.closed.Load() && f.ctx.Err() == nil { @@ -355,7 +353,11 @@ func (f *Interface) listenIn(reader io.ReadWriteCloser, i int) { break } - f.consumeInsidePacket(packet[:n], fwPacket, nb, out, i, conntrackCache.Get()) + for _, pkt := range pkts { + // 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) diff --git a/outside.go b/outside.go index 4464acdf..177d4598 100644 --- a/outside.go +++ b/outside.go @@ -542,7 +542,7 @@ func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, out []byte, p return } - _, err = f.readers[q].Write(out) + _, err = f.queues[q].Write(out) if err != nil { f.l.Error("Failed to write to tun", "error", err) } diff --git a/overlay/device.go b/overlay/device.go index b6077aba..742f7510 100644 --- a/overlay/device.go +++ b/overlay/device.go @@ -4,15 +4,25 @@ 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.ReadWriteCloser + io.Closer Activate() error Networks() []netip.Prefix Name() string RoutesFor(netip.Addr) routing.Gateways - SupportsMultiqueue() bool - NewMultiQueueReader() (io.ReadWriteCloser, error) + // 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) } diff --git a/overlay/overlaytest/noop.go b/overlay/overlaytest/noop.go index 956da7dd..0268c9ec 100644 --- a/overlay/overlaytest/noop.go +++ b/overlay/overlaytest/noop.go @@ -3,10 +3,9 @@ package overlaytest import ( - "errors" - "io" "net/netip" + "github.com/slackhq/nebula/overlay/tio" "github.com/slackhq/nebula/routing" ) @@ -31,20 +30,16 @@ func (NoopTun) Name() string { return "noop" } -func (NoopTun) Read([]byte) (int, error) { - return 0, nil +func (NoopTun) Read() ([]tio.Packet, error) { + return nil, nil } func (NoopTun) Write([]byte) (int, error) { return 0, nil } -func (NoopTun) SupportsMultiqueue() bool { - return false -} - -func (NoopTun) NewMultiQueueReader() (io.ReadWriteCloser, error) { - return nil, errors.New("unsupported") +func (NoopTun) Queues(int) ([]tio.Queue, error) { + return []tio.Queue{NoopTun{}}, nil } func (NoopTun) Close() error { diff --git a/overlay/tio/blockon_linux.go b/overlay/tio/blockon_linux.go new file mode 100644 index 00000000..84be1a2c --- /dev/null +++ b/overlay/tio/blockon_linux.go @@ -0,0 +1,45 @@ +//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 +} diff --git a/overlay/tio/queueset_poll_linux.go b/overlay/tio/queueset_poll_linux.go new file mode 100644 index 00000000..da97c09e --- /dev/null +++ b/overlay/tio/queueset_poll_linux.go @@ -0,0 +1,90 @@ +//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...) +} diff --git a/overlay/tio/single.go b/overlay/tio/single.go new file mode 100644 index 00000000..7a5be8bc --- /dev/null +++ b/overlay/tio/single.go @@ -0,0 +1,50 @@ +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() +} diff --git a/overlay/tio/tio.go b/overlay/tio/tio.go new file mode 100644 index 00000000..20d1a337 --- /dev/null +++ b/overlay/tio/tio.go @@ -0,0 +1,52 @@ +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} +} diff --git a/overlay/tio/tio_poll_linux.go b/overlay/tio/tio_poll_linux.go new file mode 100644 index 00000000..de2ae425 --- /dev/null +++ b/overlay/tio/tio_poll_linux.go @@ -0,0 +1,116 @@ +//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) +} diff --git a/overlay/tio/tun_file_linux_test.go b/overlay/tio/tun_file_linux_test.go new file mode 100644 index 00000000..85bc2363 --- /dev/null +++ b/overlay/tio/tun_file_linux_test.go @@ -0,0 +1,208 @@ +//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()) +} diff --git a/overlay/tun_android.go b/overlay/tun_android.go index e4080b41..f7ab417a 100644 --- a/overlay/tun_android.go +++ b/overlay/tun_android.go @@ -13,6 +13,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" ) @@ -63,7 +64,7 @@ func (t *tun) RoutesFor(ip netip.Addr) routing.Gateways { return r } -func (t tun) Activate() error { +func (t *tun) Activate() error { return nil } @@ -96,10 +97,6 @@ func (t *tun) Name() string { return "android" } -func (t *tun) SupportsMultiqueue() bool { - return false -} - -func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) { - return nil, fmt.Errorf("TODO: multiqueue not implemented for android") +func (t *tun) Queues(int) ([]tio.Queue, error) { + return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil } diff --git a/overlay/tun_darwin.go b/overlay/tun_darwin.go index d30148b9..ee77c06a 100644 --- a/overlay/tun_darwin.go +++ b/overlay/tun_darwin.go @@ -6,7 +6,6 @@ package overlay import ( "errors" "fmt" - "io" "log/slog" "net/netip" "os" @@ -16,6 +15,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" @@ -606,10 +606,6 @@ func (t *tun) Name() string { return t.Device } -func (t *tun) SupportsMultiqueue() bool { - return false -} - -func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) { - return nil, fmt.Errorf("TODO: multiqueue not implemented for darwin") +func (t *tun) Queues(int) ([]tio.Queue, error) { + return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil } diff --git a/overlay/tun_disabled.go b/overlay/tun_disabled.go index f47880dd..82204ad6 100644 --- a/overlay/tun_disabled.go +++ b/overlay/tun_disabled.go @@ -10,6 +10,7 @@ import ( "github.com/rcrowley/go-metrics" "github.com/slackhq/nebula/iputil" + "github.com/slackhq/nebula/overlay/tio" "github.com/slackhq/nebula/routing" ) @@ -23,6 +24,23 @@ 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, @@ -57,24 +75,6 @@ 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,12 +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() (io.ReadWriteCloser, error) { - return t, nil +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, nil } func (t *disabledTun) Close() error { diff --git a/overlay/tun_file_linux_test.go b/overlay/tun_file_linux_test.go deleted file mode 100644 index 5ab87e05..00000000 --- a/overlay/tun_file_linux_test.go +++ /dev/null @@ -1,120 +0,0 @@ -//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) - } -} diff --git a/overlay/tun_freebsd.go b/overlay/tun_freebsd.go index 79f55697..e6479001 100644 --- a/overlay/tun_freebsd.go +++ b/overlay/tun_freebsd.go @@ -7,7 +7,6 @@ import ( "bytes" "errors" "fmt" - "io" "io/fs" "log/slog" "net/netip" @@ -20,7 +19,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" @@ -561,12 +560,8 @@ func (t *tun) Name() string { return t.Device } -func (t *tun) SupportsMultiqueue() bool { - return false -} - -func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) { - return nil, 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 { diff --git a/overlay/tun_ios.go b/overlay/tun_ios.go index 27bf558b..56603b02 100644 --- a/overlay/tun_ios.go +++ b/overlay/tun_ios.go @@ -16,6 +16,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" "golang.org/x/sys/unix" @@ -159,10 +160,6 @@ func (t *tun) Name() string { return "iOS" } -func (t *tun) SupportsMultiqueue() bool { - return false -} - -func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) { - return nil, fmt.Errorf("TODO: multiqueue not implemented for ios") +func (t *tun) Queues(int) ([]tio.Queue, error) { + return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil } diff --git a/overlay/tun_linux.go b/overlay/tun_linux.go index c6cfb686..c6f308a5 100644 --- a/overlay/tun_linux.go +++ b/overlay/tun_linux.go @@ -4,9 +4,7 @@ package overlay import ( - "encoding/binary" "fmt" - "io" "log/slog" "net" "net/netip" @@ -19,180 +17,15 @@ 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 { - *tunFile - readers []*tunFile + readers tio.QueueSet closeLock sync.Mutex Device string vpnNetworks []netip.Prefix @@ -249,44 +82,57 @@ func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip return t, nil } -func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, multiqueue bool) (*tun, error) { +// 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 { - // 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 - } + 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 = uint16(unix.IFF_TUN | unix.IFF_NO_PI) + 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) if multiqueue { - req.Flags |= unix.IFF_MULTI_QUEUE + baseFlags |= unix.IFF_MULTI_QUEUE } nameStr := c.GetString("tun.dev", "") - copy(req.Name[:], nameStr) - if err = ioctl(uintptr(fd), uintptr(unix.TUNSETIFF), uintptr(unsafe.Pointer(&req))); err != nil { - _ = unix.Close(fd) - return nil, &NameError{ - Name: nameStr, - Underlying: err, - } + + fd, err := openTunDev() + 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} } - name := strings.Trim(string(req.Name[:]), "\x00") t, err := newTunGeneric(c, l, fd, vpnNetworks) if err != nil { @@ -298,16 +144,22 @@ 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 paths. It will close your files on error. +// newTunGeneric does all the stuff common to different tun initialization +// paths. It will close your files on error. func newTunGeneric(c *config.C, l *slog.Logger, fd int, vpnNetworks []netip.Prefix) (*tun, error) { - tfd, err := newTunFd(fd) + qs, err := tio.NewPollQueueSet() if err != nil { _ = unix.Close(fd) return nil, err } + err = qs.Add(fd) + if err != nil { + _ = unix.Close(fd) + return nil, err + } + t := &tun{ - tunFile: tfd, - readers: []*tunFile{tfd}, + readers: qs, closeLock: sync.Mutex{}, vpnNetworks: vpnNetworks, TXQueueLen: c.GetInt("tun.tx_queue", 500), @@ -406,36 +258,41 @@ 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. +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() (io.ReadWriteCloser, 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() fd, err := unix.Open("/dev/net/tun", os.O_RDWR, 0) if err != nil { - return nil, err + return err } - var req ifReq - 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 { + flags := uint16(unix.IFF_TUN | unix.IFF_NO_PI | unix.IFF_MULTI_QUEUE) + if _, err = tunSetIff(fd, t.Device, flags); err != nil { _ = unix.Close(fd) - return nil, err + return err } - out, err := t.tunFile.newFriend(fd) + err = t.readers.Add(fd) if err != nil { _ = unix.Close(fd) - return nil, err + return err } - t.readers = append(t.readers, out) - - return out, nil + return nil } func (t *tun) RoutesFor(ip netip.Addr) routing.Gateways { @@ -878,32 +735,10 @@ 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 } - 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 + return t.readers.Close() } diff --git a/overlay/tun_netbsd.go b/overlay/tun_netbsd.go index c971bb6e..97691543 100644 --- a/overlay/tun_netbsd.go +++ b/overlay/tun_netbsd.go @@ -6,7 +6,6 @@ package overlay import ( "errors" "fmt" - "io" "log/slog" "net/netip" "os" @@ -17,6 +16,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" @@ -390,12 +390,8 @@ func (t *tun) Name() string { return t.Device } -func (t *tun) SupportsMultiqueue() bool { - return false -} - -func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) { - return nil, 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 { diff --git a/overlay/tun_openbsd.go b/overlay/tun_openbsd.go index 41224777..23816b0d 100644 --- a/overlay/tun_openbsd.go +++ b/overlay/tun_openbsd.go @@ -6,7 +6,6 @@ package overlay import ( "errors" "fmt" - "io" "log/slog" "net/netip" "os" @@ -17,6 +16,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" @@ -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 the -// packet so the payload lands directly in to. +// Read pulls one IP packet off the tun device, scattering the 4 byte protocol header away from +// the packet so the payload lands directly in to. func (t *tun) Read(to []byte) (int, error) { var head [4]byte @@ -369,12 +369,8 @@ func (t *tun) Name() string { return t.Device } -func (t *tun) SupportsMultiqueue() bool { - return false -} - -func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) { - return nil, 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 { diff --git a/overlay/tun_tester.go b/overlay/tun_tester.go index 8acd83f0..4b2685e0 100644 --- a/overlay/tun_tester.go +++ b/overlay/tun_tester.go @@ -14,6 +14,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/udp" ) @@ -177,10 +178,6 @@ func (t *TestTun) Read(b []byte) (int, error) { return n, nil } -func (t *TestTun) SupportsMultiqueue() bool { - return false -} - -func (t *TestTun) NewMultiQueueReader() (io.ReadWriteCloser, error) { - return nil, fmt.Errorf("TODO: multiqueue not implemented") +func (t *TestTun) Queues(int) ([]tio.Queue, error) { + return []tio.Queue{tio.NewSingleQueue(t, udp.MTU)}, nil } diff --git a/overlay/tun_windows.go b/overlay/tun_windows.go index cf01615f..6be85ffc 100644 --- a/overlay/tun_windows.go +++ b/overlay/tun_windows.go @@ -6,7 +6,6 @@ package overlay import ( "crypto" "fmt" - "io" "log/slog" "net/netip" "os" @@ -18,6 +17,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" "github.com/slackhq/nebula/wintun" @@ -47,6 +47,10 @@ 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") } @@ -255,20 +259,12 @@ 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) SupportsMultiqueue() bool { - return false -} - -func (t *winTun) NewMultiQueueReader() (io.ReadWriteCloser, error) { - return nil, fmt.Errorf("TODO: multiqueue not implemented for windows") +func (t *winTun) Queues(int) ([]tio.Queue, error) { + return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil } func (t *winTun) Close() error { diff --git a/overlay/user.go b/overlay/user.go index e5f27f37..2d775bde 100644 --- a/overlay/user.go +++ b/overlay/user.go @@ -6,6 +6,7 @@ import ( "net/netip" "github.com/slackhq/nebula/config" + "github.com/slackhq/nebula/overlay/tio" "github.com/slackhq/nebula/routing" ) @@ -46,12 +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() (io.ReadWriteCloser, error) { - return d, nil +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, nil } func (d *UserDevice) Pipe() (*io.PipeReader, *io.PipeWriter) { @@ -61,9 +66,11 @@ 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() diff --git a/overlay/user_test.go b/overlay/user_test.go new file mode 100644 index 00000000..9e0e9c9c --- /dev/null +++ b/overlay/user_test.go @@ -0,0 +1,163 @@ +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 +}