From 9e7c783eb300dce3e8b4495542156ae5e8790d45 Mon Sep 17 00:00:00 2001 From: JackDoan Date: Tue, 14 Jul 2026 10:40:57 -0500 Subject: [PATCH] checkpt --- overlay/tio/single.go | 50 +++++++++++++++++++++++++++++++++++++++++ overlay/tun_android.go | 33 +++++---------------------- overlay/tun_darwin.go | 19 +++------------- overlay/tun_disabled.go | 42 +++++++++++----------------------- overlay/tun_freebsd.go | 17 ++------------ overlay/tun_netbsd.go | 17 ++------------ overlay/tun_openbsd.go | 19 +++------------- overlay/tun_tester.go | 19 ++-------------- overlay/tun_windows.go | 15 +++---------- overlay/user.go | 50 ++++++++--------------------------------- overlay/user_test.go | 22 +++++------------- 11 files changed, 99 insertions(+), 204 deletions(-) create mode 100644 overlay/tio/single.go 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/tun_android.go b/overlay/tun_android.go index 804400ab..c1ca1220 100644 --- a/overlay/tun_android.go +++ b/overlay/tun_android.go @@ -19,32 +19,12 @@ import ( ) type tun struct { - rwc io.ReadWriteCloser + io.ReadWriteCloser fd int vpnNetworks []netip.Prefix Routes atomic.Pointer[[]Route] routeTree atomic.Pointer[bart.Table[routing.Gateways]] l *slog.Logger - - readBuf []byte - batchRet [1]tio.Packet -} - -func (t *tun) Read() ([]tio.Packet, error) { - n, err := t.rwc.Read(t.readBuf) - if err != nil { - return nil, err - } - t.batchRet[0] = tio.Packet{Bytes: t.readBuf[:n]} - return t.batchRet[:], nil -} - -func (t *tun) Write(p []byte) (int, error) { - return t.rwc.Write(p) -} - -func (t *tun) Close() error { - return t.rwc.Close() } func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip.Prefix) (*tun, error) { @@ -53,11 +33,10 @@ func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip file := os.NewFile(uintptr(deviceFd), "/dev/net/tun") t := &tun{ - rwc: file, - fd: deviceFd, - vpnNetworks: vpnNetworks, - l: l, - readBuf: make([]byte, defaultBatchBufSize), + ReadWriteCloser: file, + fd: deviceFd, + vpnNetworks: vpnNetworks, + l: l, } err := t.reload(c, true) @@ -127,5 +106,5 @@ func (t *tun) NewMultiQueueReader() error { } func (t *tun) Readers() []tio.Queue { - return []tio.Queue{t} + return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)} } diff --git a/overlay/tun_darwin.go b/overlay/tun_darwin.go index 39074cbb..b966330e 100644 --- a/overlay/tun_darwin.go +++ b/overlay/tun_darwin.go @@ -31,9 +31,6 @@ type tun struct { routeTree atomic.Pointer[bart.Table[routing.Gateways]] linkAddr *netroute.LinkAddr l *slog.Logger - - readBuf []byte - batchRet [1]tio.Packet } type ifReq struct { @@ -129,7 +126,6 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, _ bool) (*t vpnNetworks: vpnNetworks, DefaultMTU: c.GetInt("tun.mtu", DefaultMTU), l: l, - readBuf: make([]byte, defaultBatchBufSize), } err = t.reload(c, true) @@ -519,9 +515,9 @@ func tunWritev(fd int, iovecs []unix.Iovec) (n int, err error) //go:noescape func tunReadv(fd int, iovecs []unix.Iovec) (n int, err error) -// readOne pulls one IP packet off the utun device, scattering the 4 byte protocol header away from +// Read pulls one IP packet off the utun device, scattering the 4 byte protocol header away from // the packet so the payload lands directly in to. -func (t *tun) readOne(to []byte) (int, error) { +func (t *tun) Read(to []byte) (int, error) { var head [4]byte rc, err := t.f.SyscallConn() @@ -554,15 +550,6 @@ func (t *tun) readOne(to []byte) (int, error) { return n - 4, nil } -func (t *tun) Read() ([]tio.Packet, error) { - n, err := t.readOne(t.readBuf) - if err != nil { - return nil, err - } - t.batchRet[0] = tio.Packet{Bytes: t.readBuf[:n]} - return t.batchRet[:], nil -} - // Write pushes one IP packet onto the utun device. Only valid for single threaded use. func (t *tun) Write(from []byte) (int, error) { if len(from) == 0 { @@ -628,5 +615,5 @@ func (t *tun) NewMultiQueueReader() error { } func (t *tun) Readers() []tio.Queue { - return []tio.Queue{t} + return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)} } diff --git a/overlay/tun_disabled.go b/overlay/tun_disabled.go index ff86bc29..870b318e 100644 --- a/overlay/tun_disabled.go +++ b/overlay/tun_disabled.go @@ -25,39 +25,21 @@ type disabledTun struct { numReaders int } -// disabledQueue is one tio.Queue view onto a shared disabledTun. Each queue -// owns a private batchRet so concurrent Read calls from different reader -// goroutines do not race on the returned slice. -type disabledQueue struct { - parent *disabledTun - batchRet [1]tio.Packet -} - -func (q *disabledQueue) Read() ([]tio.Packet, error) { - r, ok := <-q.parent.read +// 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 nil, io.EOF + return 0, io.EOF } - q.parent.tx.Inc(1) - if q.parent.l.Enabled(context.Background(), slog.LevelDebug) { - q.parent.l.Debug("Write payload", "raw", prettyPacket(r)) + t.tx.Inc(1) + if t.l.Enabled(context.Background(), slog.LevelDebug) { + t.l.Debug("Write payload", "raw", prettyPacket(r)) } - q.batchRet[0] = tio.Packet{Bytes: r} - return q.batchRet[:], nil -} - -// Write on a queue forwards to the underlying disabledTun. All queues share -// one ICMP-handling/log path so this is a thin pass-through. -func (q *disabledQueue) Write(b []byte) (int, error) { - return q.parent.Write(b) -} - -// Close on a queue is a no-op. The shared channel and metrics are owned by -// the disabledTun; Close on the device tears them down once for everybody. -func (q *disabledQueue) Close() error { - return nil + return copy(b, r), nil } func newDisabledTun(vpnNetworks []netip.Prefix, queueLen int, metricsEnabled bool, l *slog.Logger) *disabledTun { @@ -138,7 +120,9 @@ func (t *disabledTun) NewMultiQueueReader() error { func (t *disabledTun) Readers() []tio.Queue { out := make([]tio.Queue, t.numReaders) for i := range t.numReaders { - out[i] = &disabledQueue{parent: t} + // 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 } diff --git a/overlay/tun_freebsd.go b/overlay/tun_freebsd.go index 21860fcf..914d70f8 100644 --- a/overlay/tun_freebsd.go +++ b/overlay/tun_freebsd.go @@ -102,9 +102,6 @@ type tun struct { readPoll [2]unix.PollFd writePoll [2]unix.PollFd closed atomic.Bool - - readBuf []byte - batchRet [1]tio.Packet } // blockOnRead waits until the tun fd is readable or shutdown has been signaled. @@ -159,16 +156,7 @@ func (t *tun) blockOnWrite() error { return nil } -func (t *tun) Read() ([]tio.Packet, error) { - n, err := t.readOne(t.readBuf) - if err != nil { - return nil, err - } - t.batchRet[0] = tio.Packet{Bytes: t.readBuf[:n]} - return t.batchRet[:], nil -} - -func (t *tun) readOne(to []byte) (int, error) { +func (t *tun) Read(to []byte) (int, error) { // first 4 bytes is protocol family, in network byte order var head [4]byte iovecs := [2]syscall.Iovec{ @@ -386,7 +374,6 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, _ bool) (*t MTU: c.GetInt("tun.mtu", DefaultMTU), l: l, fd: fd, - readBuf: make([]byte, defaultBatchBufSize), shutdownR: shutdownR, shutdownW: shutdownW, readPoll: [2]unix.PollFd{ @@ -606,7 +593,7 @@ func (t *tun) addRoutes(logErrors bool) error { } func (t *tun) Readers() []tio.Queue { - return []tio.Queue{t} + return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)} } func (t *tun) removeRoutes(routes []Route) error { diff --git a/overlay/tun_netbsd.go b/overlay/tun_netbsd.go index e8678959..9904a88a 100644 --- a/overlay/tun_netbsd.go +++ b/overlay/tun_netbsd.go @@ -66,22 +66,10 @@ type tun struct { l *slog.Logger f *os.File fd int - - readBuf []byte - batchRet [1]tio.Packet -} - -func (t *tun) Read() ([]tio.Packet, error) { - n, err := t.readOne(t.readBuf) - if err != nil { - return nil, err - } - t.batchRet[0] = tio.Packet{Bytes: t.readBuf[:n]} - return t.batchRet[:], nil } func (t *tun) Readers() []tio.Queue { - return []tio.Queue{t} + return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)} } var deviceNameRE = regexp.MustCompile(`^tun[0-9]+$`) @@ -118,7 +106,6 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, _ bool) (*t vpnNetworks: vpnNetworks, MTU: c.GetInt("tun.mtu", DefaultMTU), l: l, - readBuf: make([]byte, defaultBatchBufSize), } err = t.reload(c, true) @@ -158,7 +145,7 @@ func (t *tun) Close() error { return nil } -func (t *tun) readOne(to []byte) (int, error) { +func (t *tun) Read(to []byte) (int, error) { rc, err := t.f.SyscallConn() if err != nil { return 0, fmt.Errorf("failed to get syscall conn for tun: %w", err) diff --git a/overlay/tun_openbsd.go b/overlay/tun_openbsd.go index 28f20123..c4b1a302 100644 --- a/overlay/tun_openbsd.go +++ b/overlay/tun_openbsd.go @@ -57,18 +57,6 @@ type tun struct { l *slog.Logger f *os.File fd int - - readBuf []byte - batchRet [1]tio.Packet -} - -func (t *tun) Read() ([]tio.Packet, error) { - n, err := t.readOne(t.readBuf) - if err != nil { - return nil, err - } - t.batchRet[0] = tio.Packet{Bytes: t.readBuf[:n]} - return t.batchRet[:], nil } var deviceNameRE = regexp.MustCompile(`^tun[0-9]+$`) @@ -105,7 +93,6 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, _ bool) (*t vpnNetworks: vpnNetworks, MTU: c.GetInt("tun.mtu", DefaultMTU), l: l, - readBuf: make([]byte, defaultBatchBufSize), } err = t.reload(c, true) @@ -151,9 +138,9 @@ func tunWritev(fd int, iovecs []unix.Iovec) (n int, err error) //go:noescape func tunReadv(fd int, iovecs []unix.Iovec) (n int, err error) -// readOne 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 packet so the payload lands directly in to. -func (t *tun) readOne(to []byte) (int, error) { +func (t *tun) Read(to []byte) (int, error) { var head [4]byte rc, err := t.f.SyscallConn() @@ -439,7 +426,7 @@ func (t *tun) deviceBytes() (o [16]byte) { } func (t *tun) Readers() []tio.Queue { - return []tio.Queue{t} + return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)} } func addRoute(prefix netip.Prefix, gateways []netip.Prefix) error { diff --git a/overlay/tun_tester.go b/overlay/tun_tester.go index 898adc23..34911527 100644 --- a/overlay/tun_tester.go +++ b/overlay/tun_tester.go @@ -29,8 +29,6 @@ type TestTun struct { closed atomic.Bool rxPackets chan []byte // Packets to receive into nebula TxPackets chan []byte // Packets transmitted outside by nebula - - batchRet [1]tio.Packet } func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, _ bool) (*TestTun, error) { @@ -51,9 +49,6 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, _ bool) (*T l: l, rxPackets: make(chan []byte, 10), TxPackets: make(chan []byte, 10), - batchRet: [1]tio.Packet{ - tio.Packet{Bytes: make([]byte, udp.MTU)}, - }, }, nil } @@ -168,17 +163,7 @@ func (t *TestTun) Close() error { return nil } -func (t *TestTun) Read() ([]tio.Packet, error) { - t.batchRet[0].Bytes = t.batchRet[0].Bytes[:udp.MTU] - n, err := t.read(t.batchRet[0].Bytes) - if err != nil { - return nil, err - } - t.batchRet[0].Bytes = t.batchRet[0].Bytes[:n] - return t.batchRet[:], nil -} - -func (t *TestTun) read(b []byte) (int, error) { +func (t *TestTun) Read(b []byte) (int, error) { p, ok := <-t.rxPackets if !ok { return 0, os.ErrClosed @@ -194,7 +179,7 @@ func (t *TestTun) read(b []byte) (int, error) { } func (t *TestTun) Readers() []tio.Queue { - return []tio.Queue{t} + return []tio.Queue{tio.NewSingleQueue(t, udp.MTU)} } func (t *TestTun) SupportsMultiqueue() bool { diff --git a/overlay/tun_windows.go b/overlay/tun_windows.go index a5ee063c..a5ae036e 100644 --- a/overlay/tun_windows.go +++ b/overlay/tun_windows.go @@ -45,18 +45,10 @@ type winTun struct { l *slog.Logger tun *wintun.NativeTun - - readBuf []byte - batchRet [1]tio.Packet } -func (t *winTun) Read() ([]tio.Packet, error) { - n, err := t.tun.Read(t.readBuf, 0) - if err != nil { - return nil, err - } - t.batchRet[0] = tio.Packet{Bytes: t.readBuf[:n]} - return t.batchRet[:], nil +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) { @@ -81,7 +73,6 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, _ bool) (*w } t := &winTun{ - readBuf: make([]byte, defaultBatchBufSize), Device: deviceName, vpnNetworks: vpnNetworks, MTU: c.GetInt("tun.mtu", DefaultMTU), @@ -281,7 +272,7 @@ func (t *winTun) NewMultiQueueReader() error { } func (t *winTun) Readers() []tio.Queue { - return []tio.Queue{t} + return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)} } func (t *winTun) Close() error { diff --git a/overlay/user.go b/overlay/user.go index e4892414..7693f653 100644 --- a/overlay/user.go +++ b/overlay/user.go @@ -39,40 +39,6 @@ type UserDevice struct { inboundWriter *io.PipeWriter } -// userDeviceQueue is a single tio.Queue over a UserDevice's shared pipes. -// One is handed to each tun read goroutine by Readers(). All queues delegate -// reads to the same outboundReader and writes to the same inboundWriter (the -// io.Pipe serializes concurrent callers), but every queue owns a private -// readBuf/batchRet so the borrowed Packet.Bytes slice one goroutine returns is -// never clobbered by another goroutine's concurrent Read. -type userDeviceQueue struct { - outboundReader *io.PipeReader - inboundWriter *io.PipeWriter - - readBuf []byte - batchRet [1]tio.Packet -} - -func (q *userDeviceQueue) Read() ([]tio.Packet, error) { - n, err := q.outboundReader.Read(q.readBuf) - if err != nil { - return nil, err - } - q.batchRet[0] = tio.Packet{Bytes: q.readBuf[:n]} - return q.batchRet[:], nil -} - -func (q *userDeviceQueue) Write(p []byte) (int, error) { - return q.inboundWriter.Write(p) -} - -// Close is a no-op: the shared pipes are owned by the UserDevice and torn -// down by UserDevice.Close, so an individual queue must not close them out -// from under its siblings. -func (q *userDeviceQueue) Close() error { - return nil -} - func (d *UserDevice) Activate() error { return nil } @@ -95,13 +61,11 @@ func (d *UserDevice) NewMultiQueueReader() error { func (d *UserDevice) Readers() []tio.Queue { out := make([]tio.Queue, d.numReaders) for i := range d.numReaders { - // Each queue shares the underlying pipes but owns its own scratch - // buffer so concurrent Reads across queues never alias. - out[i] = &userDeviceQueue{ - outboundReader: d.outboundReader, - inboundWriter: d.inboundWriter, - readBuf: make([]byte, defaultBatchBufSize), - } + // 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 } @@ -110,6 +74,10 @@ func (d *UserDevice) Pipe() (*io.PipeReader, *io.PipeWriter) { return d.inboundReader, d.outboundWriter } +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) } diff --git a/overlay/user_test.go b/overlay/user_test.go index 2bf11ad4..f0904bb1 100644 --- a/overlay/user_test.go +++ b/overlay/user_test.go @@ -26,10 +26,11 @@ func newTestUserDevice(t *testing.T) *UserDevice { // TestUserDeviceReadersDistinctBuffers is the regression test for the // multiqueue packet-corruption bug: Readers() used to hand the same -// *UserDevice (and therefore the same readBuf/batchRet) 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. +// *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. func TestUserDeviceReadersDistinctBuffers(t *testing.T) { d := newTestUserDevice(t) @@ -43,21 +44,10 @@ func TestUserDeviceReadersDistinctBuffers(t *testing.T) { t.Fatalf("Readers() returned %d queues, want 2", len(readers)) } - q0 := readers[0].(*userDeviceQueue) - q1 := readers[1].(*userDeviceQueue) - // Distinct queue objects. - if q0 == q1 { + if readers[0] == readers[1] { t.Fatal("Readers() returned the same queue object twice") } - // Distinct backing buffers (the actual regression: shared readBuf). - if &q0.readBuf[0] == &q1.readBuf[0] { - t.Fatal("queues share the same readBuf backing array") - } - // Shared underlying pipes. - if q0.outboundReader != q1.outboundReader || q0.inboundWriter != q1.inboundWriter { - t.Fatal("queues do not share the underlying pipes") - } // 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