diff --git a/overlay/tio/single.go b/overlay/tio/single.go index 61960c70..a24c2242 100644 --- a/overlay/tio/single.go +++ b/overlay/tio/single.go @@ -15,13 +15,9 @@ type singleQueue struct { // 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. The Queue implements BatchWriter when rwc does. +// Close closes rwc. func NewSingleQueue(rwc io.ReadWriteCloser, bufSize int) Queue { - q := &singleQueue{rw: rwc, closer: rwc, buf: make([]byte, bufSize)} - if bw, ok := rwc.(BatchWriter); ok { - return &batchSingleQueue{singleQueue: q, bw: bw} - } - return q + return &singleQueue{rw: rwc, closer: rwc, buf: make([]byte, bufSize)} } // NewSingleQueueNoClose is NewSingleQueue for a source owned by someone else, @@ -51,14 +47,3 @@ func (q *singleQueue) Close() error { } return q.closer.Close() } - -// batchSingleQueue is a singleQueue whose source also implements BatchWriter. -// It is a separate type so a plain singleQueue never advertises WriteBatch. -type batchSingleQueue struct { - *singleQueue - bw BatchWriter -} - -func (q *batchSingleQueue) WriteBatch(pkts [][]byte) error { - return q.bw.WriteBatch(pkts) -} diff --git a/overlay/tio/tio_utun_darwin.go b/overlay/tio/tio_utun_darwin.go new file mode 100644 index 00000000..d3fe72dd --- /dev/null +++ b/overlay/tio/tio_utun_darwin.go @@ -0,0 +1,338 @@ +//go:build darwin && !ios + +package tio + +import ( + "fmt" + "log/slog" + "os" + "sync" + "sync/atomic" + "syscall" + "unsafe" + + "golang.org/x/sys/unix" +) + +// tunMaxPacket is the largest packet utun can return at any device MTU. +const tunMaxPacket = 65535 + +// tunReadBatch is the most packets one Utun.Read returns. +const tunReadBatch = 64 + +// tunReadArena is Utun's receive buffer. Draining stops once less than tunMaxPacket of it is left, +// so every readv has room for the largest packet utun can return. +const tunReadArena = 4 * tunMaxPacket + +// tunWriteBatch is how many packets one sendmsg_x call may carry. +const tunWriteBatch = 64 + +// Utun is a darwin utun control socket as a Queue. Every packet crosses the socket behind a 4 byte +// address family prefix, which Read strips and the writes add. +// +// The fd lives in the runtime poller, so Close wakes a Read parked on it; darwin needs no QueueSet +// shutdown fd. +type Utun struct { + f *os.File + l *slog.Logger + + // readBuf and readRet are Read's scratch, owned by the single reader. + readBuf []byte + readRet [tunReadBatch]Packet + + // noSendmsgX is set once sendmsg_x is refused; WriteBatch then writes one packet at a time. + noSendmsgX atomic.Bool + batchMu sync.Mutex + batch tunBatch +} + +// tunBatch is WriteBatch's scratch, guarded by Utun.batchMu. The kernel reads every entry in place. +type tunBatch struct { + heads [tunWriteBatch][4]byte + iovs [tunWriteBatch][2]unix.Iovec + hdrs [tunWriteBatch]msghdrX + pkts [tunWriteBatch][]byte +} + +// msghdrX mirrors xnu's struct msghdr_x (bsd/sys/socket_private.h). +// Keep in sync with msghdrX in udp/udp_darwin.go. +type msghdrX struct { + Name *byte + Namelen uint32 + Iov *unix.Iovec + Iovlen int32 + Control *byte + Controllen uint32 + Flags int32 + Datalen uint64 +} + +// NewUtun wraps a utun control socket fd and makes it non-blocking. +// On failure it does NOT close fd: the caller owns fd until NewUtun succeeds, after which Close closes it. +func NewUtun(fd int, l *slog.Logger) (*Utun, error) { + // Before os.NewFile, which only hands a non-blocking fd to the runtime poller. + if err := unix.SetNonblock(fd, true); err != nil { + return nil, fmt.Errorf("failed to set the tun fd to non-blocking mode: %w", err) + } + return &Utun{ + f: os.NewFile(uintptr(fd), "utun"), + l: l, + readBuf: make([]byte, tunReadArena), + }, nil +} + +// tunWritev and tunReadv are linkname'd to x/sys/unix's libc-routed writev/readv stubs so the +// calls go through libSystem's pinned trampoline. A raw syscall.Syscall(SYS_WRITEV/SYS_READV, ...) +// on darwin/arm64 emits an SVC #0x80 trap (see $GOROOT/src/syscall/asm_darwin_arm64.s), the path +// Apple keeps warning they will eventually disallow. We pull the low-level stubs instead of calling +// unix.Writev/unix.Readv because those take [][]byte and rebuild the []Iovec every call, which +// heap-allocates the header; linkname'ing the stubs lets us hand them our own stack-allocated +// iovecs. See golang/go#78049. + +//go:linkname tunWritev golang.org/x/sys/unix.writev +//go:noescape +func tunWritev(fd int, iovecs []unix.Iovec) (n int, err error) + +//go:linkname tunReadv golang.org/x/sys/unix.readv +//go:noescape +func tunReadv(fd int, iovecs []unix.Iovec) (n int, err error) + +// Read waits for the utun to become readable, then reads packets until it would block, up to +// tunReadBatch, so the caller encrypts and sends them as one batch rather than one per wakeup. Each +// packet is read with its AF prefix scattered away, so the payload lands directly in readBuf. +// An error after at least one packet is dropped in favor of returning those packets; a persistent +// one recurs on the next Read. +func (u *Utun) Read() ([]Packet, error) { + rc, err := u.f.SyscallConn() + if err != nil { + return nil, err + } + + var head [4]byte + n, off := 0, 0 + var callErr error + err = rc.Read(func(fd uintptr) bool { + for n < tunReadBatch && len(u.readBuf)-off >= tunMaxPacket { + iovecs := [2]unix.Iovec{ + {Base: &head[0], Len: 4}, + {Base: &u.readBuf[off], Len: uint64(len(u.readBuf) - off)}, + } + l, e := tunReadv(int(fd), iovecs[:]) + if e != nil { + if errno, ok := e.(syscall.Errno); ok && errno.Temporary() { + // Park on the poller only while there is nothing to hand back. + return n > 0 + } + callErr = e + return true + } + if l < 4 { + // A datagram too short to carry the AF prefix, or end of file; stop rather than spin on it. + return true + } + end := off + l - 4 + u.readRet[n] = Packet{Bytes: u.readBuf[off:end:end]} + n++ + off = end + } + return true + }) + if n > 0 { + return u.readRet[:n], nil + } + if err != nil { + return nil, err + } + return nil, callErr +} + +// Write pushes one IP packet onto the utun device. Safe for concurrent use: +// the AF prefix and iovecs are per-call stack state, and the fd write itself +// serializes on the runtime's fd mutex (see the Queue contract in tio.go). +func (u *Utun) Write(from []byte) (int, error) { + if len(from) == 0 { + return 0, syscall.EIO + } + + var head [4]byte + af, err := tunAF(from) + if err != nil { + return 0, err + } + head[3] = af + + // Grab rc as a local so the compiler can devirtualize the call and keep the closure on the stack. + rc, err := u.f.SyscallConn() + if err != nil { + return 0, err + } + + var n int + var callErr error + err = rc.Write(func(fd uintptr) bool { + iovecs := []unix.Iovec{ + {Base: &head[0], Len: 4}, + {Base: &from[0], Len: uint64(len(from))}, + } + n, callErr = tunWritev(int(fd), iovecs) + // Type-assert to syscall.Errno so the EAGAIN/EWOULDBLOCK/EINTR check doesn't box the errno + // constants into error interfaces on every call. + if errno, ok := callErr.(syscall.Errno); ok && errno.Temporary() { + return false + } + return true + }) + if err != nil { + return 0, err + } + if callErr != nil { + return 0, callErr + } + + return n - 4, nil +} + +// tunAF returns the utun address-family prefix byte for an IP packet. +func tunAF(pkt []byte) (byte, error) { + switch pkt[0] >> 4 { + case 4: + return syscall.AF_INET, nil + case 6: + return syscall.AF_INET6, nil + default: + return 0, fmt.Errorf("unable to determine IP version from packet") + } +} + +// WriteBatch writes pkts to the utun device with sendmsg_x, xnu's private batched sendmsg, up to +// tunWriteBatch packets per syscall. The kernel still hands each packet to utun on its own, so this +// saves syscalls, not per-packet kernel work. Safe for concurrent use. +// +// An empty packet or one with no IP version is skipped and reported, as Write would. Any other +// error from sendmsg_x means the kernel took an unknown prefix of that call and dropped the rest, so +// it is reported rather than retried, which could deliver packets twice. +// +// Under mbuf exhaustion, sendmsg_x silently drops packets that writev would have delivered: the +// kernel counts each packet as sent once it is copied in, and when a later allocation fails it +// frees that prefix unsent and returns its length as a short count, which is indistinguishable +// from a real one. +func (u *Utun) WriteBatch(pkts [][]byte) error { + if u.noSendmsgX.Load() { + return u.writeEach(pkts) + } + + u.batchMu.Lock() + defer u.batchMu.Unlock() + b := &u.batch + var firstErr error + for len(pkts) > 0 { + n := 0 + for len(pkts) > 0 && n < tunWriteBatch { + p := pkts[0] + pkts = pkts[1:] + var af byte + err := error(syscall.EIO) + if len(p) > 0 { + af, err = tunAF(p) + } + if err != nil { + if firstErr == nil { + firstErr = err + } + continue + } + b.heads[n] = [4]byte{3: af} + b.iovs[n] = [2]unix.Iovec{ + {Base: &b.heads[n][0], Len: 4}, + {Base: &p[0], Len: uint64(len(p))}, + } + b.hdrs[n] = msghdrX{Iov: &b.iovs[n][0], Iovlen: 2} + b.pkts[n] = p + n++ + } + if n == 0 { + continue + } + + for off := 0; off < n; { + sent, err := u.sendmsgX(off, n) + switch { + case err == nil && sent > 0: + // The kernel stops early without an error on a packet larger than the socket's send + // buffer, or when sending a later packet fails with ENOBUFS; resend the remainder. + off += sent + case err == nil: + if werr := u.writeEach(b.pkts[off:n]); werr != nil && firstErr == nil { + firstErr = werr + } + off = n + case err == unix.EMSGSIZE: + // Only the first packet being larger than the socket's send buffer fails the whole call, + // before the kernel takes any; Write reports that one and the rest go out in the next call. + if werr := u.writeEach(b.pkts[off : off+1]); werr != nil && firstErr == nil { + firstErr = werr + } + off++ + case err == unix.ENOSYS || err == unix.EPERM || err == unix.EOPNOTSUPP: + // sendmsg_x is private API: fall back to writev if a kernel or sandbox refuses it. + if u.noSendmsgX.CompareAndSwap(false, true) { + u.l.Warn("sendmsg_x unavailable on the tun device, writing one packet per syscall", "error", err) + } + if werr := u.writeEach(b.pkts[off:n]); werr != nil && firstErr == nil { + firstErr = werr + } + clear(b.iovs[:n]) + clear(b.pkts[:n]) + if werr := u.writeEach(pkts); werr != nil && firstErr == nil { + firstErr = werr + } + return firstErr + default: + if firstErr == nil { + firstErr = err + } + off = n + } + } + clear(b.iovs[:n]) + clear(b.pkts[:n]) + } + return firstErr +} + +// sendmsgX hands entries off through n-1 of u.batch to sendmsg_x and returns how many the kernel took. +func (u *Utun) sendmsgX(off, n int) (int, error) { + rc, err := u.f.SyscallConn() + if err != nil { + return 0, err + } + var sent uintptr + var errno syscall.Errno + err = rc.Write(func(fd uintptr) bool { + sent, _, errno = unix.Syscall6(unix.SYS_SENDMSG_X, fd, uintptr(unsafe.Pointer(&u.batch.hdrs[off])), uintptr(n-off), 0, 0, 0) + // sendmsg_x reports EAGAIN only when it took nothing; a partial batch returns its count. + return !errno.Temporary() + }) + if err != nil { + return 0, err + } + if errno != 0 { + return 0, errno + } + return int(sent), nil +} + +// writeEach writes pkts one at a time with Write. +func (u *Utun) writeEach(pkts [][]byte) error { + var firstErr error + for _, p := range pkts { + if _, err := u.Write(p); err != nil && firstErr == nil { + firstErr = err + } + } + return firstErr +} + +func (u *Utun) Close() error { + return u.f.Close() +} diff --git a/overlay/tun_darwin_test.go b/overlay/tio/tio_utun_darwin_test.go similarity index 66% rename from overlay/tun_darwin_test.go rename to overlay/tio/tio_utun_darwin_test.go index ee608576..2ab6a89d 100644 --- a/overlay/tun_darwin_test.go +++ b/overlay/tio/tio_utun_darwin_test.go @@ -1,37 +1,42 @@ -//go:build !ios && !e2e_testing +//go:build darwin && !ios -package overlay +package tio import ( "bytes" - "os" + "log/slog" "syscall" "testing" - "github.com/slackhq/nebula/test" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "golang.org/x/sys/unix" ) -// newSocketpairTun returns a tun writing to one end of a connected AF_UNIX datagram socketpair, which -// takes sendmsg_x the way a utun control socket does, and the other end's fd for reading back. -// sndbuf is the writer's send buffer, which caps the largest datagram it accepts. -func newSocketpairTun(t *testing.T, sndbuf int) (*tun, int) { +// newSocketpairUtun returns a Utun over one end of a connected AF_UNIX datagram socketpair, which +// takes sendmsg_x the way a utun control socket does, and the other end's fd for the test to use. +// sndbuf is the Utun's send buffer, which caps the largest datagram it accepts. +func newSocketpairUtun(t *testing.T, sndbuf int) (*Utun, int) { t.Helper() fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_DGRAM, 0) require.NoError(t, err) require.NoError(t, unix.SetsockoptInt(fds[0], unix.SOL_SOCKET, unix.SO_SNDBUF, sndbuf)) + // An AF_UNIX datagram send is bounded by the receiver's buffer, so size both ends' for the tests. + require.NoError(t, unix.SetsockoptInt(fds[0], unix.SOL_SOCKET, unix.SO_RCVBUF, 1<<20)) require.NoError(t, unix.SetsockoptInt(fds[1], unix.SOL_SOCKET, unix.SO_RCVBUF, 1<<20)) // A dropped datagram fails the read-back instead of hanging it. require.NoError(t, unix.SetsockoptTimeval(fds[1], unix.SOL_SOCKET, unix.SO_RCVTIMEO, &unix.Timeval{Sec: 2})) - require.NoError(t, unix.SetNonblock(fds[0], true)) - f := os.NewFile(uintptr(fds[0]), "socketpair") + u, err := NewUtun(fds[0], slog.New(slog.DiscardHandler)) + if err != nil { + _ = unix.Close(fds[0]) + _ = unix.Close(fds[1]) + t.Fatal(err) + } t.Cleanup(func() { - _ = f.Close() + _ = u.Close() _ = unix.Close(fds[1]) }) - return &tun{f: f, l: test.NewLogger()}, fds[1] + return u, fds[1] } // testTunPkt returns packet i of size bytes, alternating IPv4 and IPv6, and the datagram the utun @@ -75,12 +80,12 @@ func forEachTunWritePath(t *testing.T, fn func(t *testing.T, fallback bool)) { } } -// TestTunWriteBatch pins that WriteBatch delivers every packet whole, in order and behind the right +// TestUtunWriteBatch pins that WriteBatch delivers every packet whole, in order and behind the right // utun AF prefix, across several sendmsg_x calls, and returns nil. -func TestTunWriteBatch(t *testing.T) { +func TestUtunWriteBatch(t *testing.T) { forEachTunWritePath(t, func(t *testing.T, fallback bool) { - tn, r := newSocketpairTun(t, 1<<20) - tn.noSendmsgX.Store(fallback) + u, r := newSocketpairUtun(t, 1<<20) + u.noSendmsgX.Store(fallback) var pkts, want [][]byte for i := range 3*tunWriteBatch + 5 { @@ -88,18 +93,18 @@ func TestTunWriteBatch(t *testing.T) { pkts = append(pkts, p) want = append(want, w) } - require.NoError(t, tn.WriteBatch(pkts)) - assert.Equal(t, fallback, tn.noSendmsgX.Load()) + require.NoError(t, u.WriteBatch(pkts)) + assert.Equal(t, fallback, u.noSendmsgX.Load()) readTunBack(t, r, want) }) } -// TestTunWriteBatchSkipsInvalid pins that an empty packet or one with no IP version is skipped +// TestUtunWriteBatchSkipsInvalid pins that an empty packet or one with no IP version is skipped // without shifting the packets after it, and that the first such error is the one reported. -func TestTunWriteBatchSkipsInvalid(t *testing.T) { +func TestUtunWriteBatchSkipsInvalid(t *testing.T) { forEachTunWritePath(t, func(t *testing.T, fallback bool) { - tn, r := newSocketpairTun(t, 1<<20) - tn.noSendmsgX.Store(fallback) + u, r := newSocketpairUtun(t, 1<<20) + u.noSendmsgX.Store(fallback) var pkts, want [][]byte for i := range 20 { @@ -109,16 +114,16 @@ func TestTunWriteBatchSkipsInvalid(t *testing.T) { } pkts = append(pkts[:10], append([][]byte{{0x10, 1, 2}, {}}, pkts[10:]...)...) - err := tn.WriteBatch(pkts) + err := u.WriteBatch(pkts) assert.ErrorContains(t, err, "IP version") readTunBack(t, r, want) }) } -// TestTunWriteBatchOversize pins that a packet larger than the socket's send buffer drops only +// TestUtunWriteBatchOversize pins that a packet larger than the socket's send buffer drops only // itself and is reported, whether sendmsg_x reaches it mid-call (a short count) or first (EMSGSIZE // for the whole call). The send buffer is 1024 bytes and the oversize packet 2000. -func TestTunWriteBatchOversize(t *testing.T) { +func TestUtunWriteBatchOversize(t *testing.T) { for _, tc := range []struct { name string at int @@ -128,8 +133,8 @@ func TestTunWriteBatchOversize(t *testing.T) { } { t.Run(tc.name, func(t *testing.T) { forEachTunWritePath(t, func(t *testing.T, fallback bool) { - tn, r := newSocketpairTun(t, 1024) - tn.noSendmsgX.Store(fallback) + u, r := newSocketpairUtun(t, 1024) + u.noSendmsgX.Store(fallback) var pkts, want [][]byte for i := range 2*tunWriteBatch + 10 { @@ -143,26 +148,19 @@ func TestTunWriteBatchOversize(t *testing.T) { want = append(want, w) } } - err := tn.WriteBatch(pkts) + err := u.WriteBatch(pkts) assert.ErrorIs(t, err, unix.EMSGSIZE) - assert.Equal(t, fallback, tn.noSendmsgX.Load()) + assert.Equal(t, fallback, u.noSendmsgX.Load()) readTunBack(t, r, want) }) }) } } -// TestTunQueueReadDrains pins that Read hands back every queued packet, AF prefix stripped and in +// TestUtunReadDrains pins that Read hands back every queued packet, AF prefix stripped and in // order, capped at tunReadBatch per call, and doesn't wait for more once the queue is empty. -func TestTunQueueReadDrains(t *testing.T) { - tn, w := newSocketpairTun(t, 1<<20) - // An AF_UNIX datagram send is bounded by the receiver's buffer, here the tun's end. - rc, err := tn.f.SyscallConn() - require.NoError(t, err) - require.NoError(t, rc.Control(func(fd uintptr) { - require.NoError(t, unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUF, 1<<20)) - })) - q := &tunQueue{t: tn, buf: make([]byte, tunReadArena)} +func TestUtunReadDrains(t *testing.T) { + u, w := newSocketpairUtun(t, 1<<20) total := tunReadBatch + 3 var want [][]byte for i := range total { @@ -174,7 +172,7 @@ func TestTunQueueReadDrains(t *testing.T) { var got [][]byte for _, size := range []int{tunReadBatch, 3} { - pkts, err := q.Read() + pkts, err := u.Read() require.NoError(t, err) require.Len(t, pkts, size) for _, p := range pkts { @@ -184,17 +182,16 @@ func TestTunQueueReadDrains(t *testing.T) { assert.Equal(t, want, got) } -// TestTunQueueReadClipsPackets pins that each returned packet's capacity ends at its own bytes, so +// TestUtunReadClipsPackets pins that each returned packet's capacity ends at its own bytes, so // appending to one can't overwrite the next. -func TestTunQueueReadClipsPackets(t *testing.T) { - tn, w := newSocketpairTun(t, 1<<20) - q := &tunQueue{t: tn, buf: make([]byte, tunReadArena)} +func TestUtunReadClipsPackets(t *testing.T) { + u, w := newSocketpairUtun(t, 1<<20) for i := range 2 { _, wire := testTunPkt(i, 100) _, err := unix.Write(w, wire) require.NoError(t, err) } - pkts, err := q.Read() + pkts, err := u.Read() require.NoError(t, err) require.Len(t, pkts, 2) assert.Equal(t, len(pkts[0].Bytes), cap(pkts[0].Bytes)) diff --git a/overlay/tun_darwin.go b/overlay/tun_darwin.go index d2c48b35..8a6adbc3 100644 --- a/overlay/tun_darwin.go +++ b/overlay/tun_darwin.go @@ -7,10 +7,7 @@ import ( "fmt" "log/slog" "net/netip" - "os" - "sync" "sync/atomic" - "syscall" "unsafe" "github.com/gaissmai/bart" @@ -23,7 +20,7 @@ import ( ) type tun struct { - f *os.File + queue *tio.Utun Device string vpnNetworks []netip.Prefix DefaultMTU int @@ -34,35 +31,6 @@ type tun struct { // and routes for it. NEPacketTunnelProvider on darwin does this. hostOwned bool l *slog.Logger - - // noSendmsgX is set once sendmsg_x is refused; WriteBatch then writes one packet at a time. - noSendmsgX atomic.Bool - batchMu sync.Mutex - batch tunBatch -} - -// tunWriteBatch is how many packets one sendmsg_x call may carry. -const tunWriteBatch = 64 - -// tunBatch is WriteBatch's scratch, guarded by tun.batchMu. The kernel reads every entry in place. -type tunBatch struct { - heads [tunWriteBatch][4]byte - iovs [tunWriteBatch][2]unix.Iovec - hdrs [tunWriteBatch]msghdrX - pkts [tunWriteBatch][]byte -} - -// msghdrX mirrors xnu's struct msghdr_x (bsd/sys/socket_private.h). -// Keep in sync with msghdrX in udp/udp_darwin.go. -type msghdrX struct { - Name *byte - Namelen uint32 - Iov *unix.Iovec - Iovlen int32 - Control *byte - Controllen uint32 - Flags int32 - Datalen uint64 } type ifReq struct { @@ -147,13 +115,13 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, _ bool) (*t return nil, fmt.Errorf("failed to retrieve tun name: %w", err) } - err = unix.SetNonblock(fd, true) + queue, err := tio.NewUtun(fd, l) if err != nil { - return nil, fmt.Errorf("SetNonblock: %v", err) + return nil, err } t := &tun{ - f: os.NewFile(uintptr(fd), ""), + queue: queue, Device: name, vpnNetworks: vpnNetworks, DefaultMTU: c.GetInt("tun.mtu", DefaultMTU), @@ -186,15 +154,15 @@ func (t *tun) deviceBytes() (o [16]byte) { // network extension is handed its device. Everything about moving packets is shared with newTun, // only the setup differs: the host owns addressing and routing here. func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip.Prefix) (*tun, error) { - if err := unix.SetNonblock(deviceFd, true); err != nil { + queue, err := tio.NewUtun(deviceFd, l) + if err != nil { // We own the fd from the moment it is handed to us _ = unix.Close(deviceFd) - return nil, fmt.Errorf("failed to set the tun fd to non-blocking mode: %w", err) + return nil, err } - file := os.NewFile(uintptr(deviceFd), "/dev/tun") t := &tun{ - f: file, + queue: queue, Device: utunNameFromFd(deviceFd), vpnNetworks: vpnNetworks, DefaultMTU: c.GetInt("tun.mtu", DefaultMTU), @@ -203,7 +171,7 @@ func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip } if err := t.reload(c, true); err != nil { - _ = file.Close() + _ = queue.Close() return nil, err } @@ -227,8 +195,8 @@ func utunNameFromFd(fd int) string { } func (t *tun) Close() error { - if t.f != nil { - return t.f.Close() + if t.queue != nil { + return t.queue.Close() } return nil } @@ -586,312 +554,6 @@ func delRoute(prefix netip.Prefix, gateway netroute.Addr) error { return nil } -// tunWritev and tunReadv are linkname'd to x/sys/unix's libc-routed writev/readv stubs so the -// calls go through libSystem's pinned trampoline. A raw syscall.Syscall(SYS_WRITEV/SYS_READV, ...) -// on darwin/arm64 emits an SVC #0x80 trap (see $GOROOT/src/syscall/asm_darwin_arm64.s), the path -// Apple keeps warning they will eventually disallow. We pull the low-level stubs instead of calling -// unix.Writev/unix.Readv because those take [][]byte and rebuild the []Iovec every call, which -// heap-allocates the header; linkname'ing the stubs lets us hand them our own stack-allocated -// iovecs. See golang/go#78049. - -//go:linkname tunWritev golang.org/x/sys/unix.writev -//go:noescape -func tunWritev(fd int, iovecs []unix.Iovec) (n int, err error) - -//go:linkname tunReadv golang.org/x/sys/unix.readv -//go:noescape -func tunReadv(fd int, iovecs []unix.Iovec) (n int, err error) - -// 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) Read(to []byte) (int, error) { - var head [4]byte - - rc, err := t.f.SyscallConn() - if err != nil { - return 0, err - } - - var n int - var callErr error - err = rc.Read(func(fd uintptr) bool { - iovecs := []unix.Iovec{ - {Base: &head[0], Len: 4}, - {Base: &to[0], Len: uint64(len(to))}, - } - n, callErr = tunReadv(int(fd), iovecs) - if errno, ok := callErr.(syscall.Errno); ok && errno.Temporary() { - return false - } - return true - }) - if err != nil { - return 0, err - } - if callErr != nil { - return 0, callErr - } - if n < 4 { - return 0, nil - } - return n - 4, nil -} - -// tunReadBatch is the most packets one tunQueue.Read returns. -const tunReadBatch = 64 - -// tunReadArena is tunQueue's receive buffer. Draining stops once less than defaultBatchBufSize of it -// is left, so every readv has room for the largest packet utun can return at any device MTU. -const tunReadArena = 4 * defaultBatchBufSize - -// tunQueue is the darwin tun's Queue. Read drains every packet already queued on the utun, up to -// tunReadBatch, so the caller encrypts and sends them as one batch rather than one per wakeup. -type tunQueue struct { - t *tun - buf []byte - pkts [tunReadBatch]tio.Packet -} - -// Read waits for the utun to become readable, then reads packets until it would block. An error after -// at least one packet is dropped in favor of returning those packets; a persistent one recurs on the -// next Read. -func (q *tunQueue) Read() ([]tio.Packet, error) { - rc, err := q.t.f.SyscallConn() - if err != nil { - return nil, err - } - - var head [4]byte - n, off := 0, 0 - var callErr error - err = rc.Read(func(fd uintptr) bool { - for n < tunReadBatch && len(q.buf)-off >= defaultBatchBufSize { - iovecs := [2]unix.Iovec{ - {Base: &head[0], Len: 4}, - {Base: &q.buf[off], Len: uint64(len(q.buf) - off)}, - } - l, e := tunReadv(int(fd), iovecs[:]) - if e != nil { - if errno, ok := e.(syscall.Errno); ok && errno.Temporary() { - // Park on the poller only while there is nothing to hand back. - return n > 0 - } - callErr = e - return true - } - if l < 4 { - // A datagram too short to carry the AF prefix, or end of file; stop rather than spin on it. - return true - } - end := off + l - 4 - q.pkts[n] = tio.Packet{Bytes: q.buf[off:end:end]} - n++ - off = end - } - return true - }) - if n > 0 { - return q.pkts[:n], nil - } - if err != nil { - return nil, err - } - return nil, callErr -} - -func (q *tunQueue) Write(p []byte) (int, error) { return q.t.Write(p) } - -func (q *tunQueue) WriteBatch(pkts [][]byte) error { return q.t.WriteBatch(pkts) } - -func (q *tunQueue) Close() error { return q.t.Close() } - -// Write pushes one IP packet onto the utun device. Safe for concurrent use: -// the AF prefix and iovecs are per-call stack state, and the fd write itself -// serializes on the runtime's fd mutex (see the Queue contract in tio.go). -func (t *tun) Write(from []byte) (int, error) { - if len(from) == 0 { - return 0, syscall.EIO - } - - var head [4]byte - af, err := tunAF(from) - if err != nil { - return 0, err - } - head[3] = af - - // Grab rc as a local so the compiler can devirtualize the call and keep the closure on the stack. - rc, err := t.f.SyscallConn() - if err != nil { - return 0, err - } - - var n int - var callErr error - err = rc.Write(func(fd uintptr) bool { - iovecs := []unix.Iovec{ - {Base: &head[0], Len: 4}, - {Base: &from[0], Len: uint64(len(from))}, - } - n, callErr = tunWritev(int(fd), iovecs) - // Type-assert to syscall.Errno so the EAGAIN/EWOULDBLOCK/EINTR check doesn't box the errno - // constants into error interfaces on every call. - if errno, ok := callErr.(syscall.Errno); ok && errno.Temporary() { - return false - } - return true - }) - if err != nil { - return 0, err - } - if callErr != nil { - return 0, callErr - } - - return n - 4, nil -} - -// tunAF returns the utun address-family prefix byte for an IP packet. -func tunAF(pkt []byte) (byte, error) { - switch pkt[0] >> 4 { - case 4: - return syscall.AF_INET, nil - case 6: - return syscall.AF_INET6, nil - default: - return 0, fmt.Errorf("unable to determine IP version from packet") - } -} - -// WriteBatch writes pkts to the utun device with sendmsg_x, xnu's private batched sendmsg, up to -// tunWriteBatch packets per syscall. The kernel still hands each packet to utun on its own, so this -// saves syscalls, not per-packet kernel work. Safe for concurrent use. -// -// An empty packet or one with no IP version is skipped and reported, as Write would. Any other -// error from sendmsg_x means the kernel took an unknown prefix of that call and dropped the rest, so -// it is reported rather than retried, which could deliver packets twice. -// -// Under mbuf exhaustion, sendmsg_x silently drops packets that writev would have delivered: the -// kernel counts each packet as sent once it is copied in, and when a later allocation fails it -// frees that prefix unsent and returns its length as a short count, which is indistinguishable -// from a real one. -func (t *tun) WriteBatch(pkts [][]byte) error { - if t.noSendmsgX.Load() { - return t.writeEach(pkts) - } - - t.batchMu.Lock() - defer t.batchMu.Unlock() - b := &t.batch - var firstErr error - for len(pkts) > 0 { - n := 0 - for len(pkts) > 0 && n < tunWriteBatch { - p := pkts[0] - pkts = pkts[1:] - var af byte - err := error(syscall.EIO) - if len(p) > 0 { - af, err = tunAF(p) - } - if err != nil { - if firstErr == nil { - firstErr = err - } - continue - } - b.heads[n] = [4]byte{3: af} - b.iovs[n] = [2]unix.Iovec{ - {Base: &b.heads[n][0], Len: 4}, - {Base: &p[0], Len: uint64(len(p))}, - } - b.hdrs[n] = msghdrX{Iov: &b.iovs[n][0], Iovlen: 2} - b.pkts[n] = p - n++ - } - if n == 0 { - continue - } - - for off := 0; off < n; { - sent, err := t.sendmsgX(off, n) - switch { - case err == nil && sent > 0: - // The kernel stops early without an error on a packet larger than the socket's send - // buffer, or when sending a later packet fails with ENOBUFS; resend the remainder. - off += sent - case err == nil: - if werr := t.writeEach(b.pkts[off:n]); werr != nil && firstErr == nil { - firstErr = werr - } - off = n - case err == unix.EMSGSIZE: - // Only the first packet being larger than the socket's send buffer fails the whole call, - // before the kernel takes any; Write reports that one and the rest go out in the next call. - if werr := t.writeEach(b.pkts[off : off+1]); werr != nil && firstErr == nil { - firstErr = werr - } - off++ - case err == unix.ENOSYS || err == unix.EPERM || err == unix.EOPNOTSUPP: - // sendmsg_x is private API: fall back to writev if a kernel or sandbox refuses it. - if t.noSendmsgX.CompareAndSwap(false, true) { - t.l.Warn("sendmsg_x unavailable on the tun device, writing one packet per syscall", "error", err) - } - if werr := t.writeEach(b.pkts[off:n]); werr != nil && firstErr == nil { - firstErr = werr - } - clear(b.iovs[:n]) - clear(b.pkts[:n]) - if werr := t.writeEach(pkts); werr != nil && firstErr == nil { - firstErr = werr - } - return firstErr - default: - if firstErr == nil { - firstErr = err - } - off = n - } - } - clear(b.iovs[:n]) - clear(b.pkts[:n]) - } - return firstErr -} - -// sendmsgX hands entries off through n-1 of t.batch to sendmsg_x and returns how many the kernel took. -func (t *tun) sendmsgX(off, n int) (int, error) { - rc, err := t.f.SyscallConn() - if err != nil { - return 0, err - } - var sent uintptr - var errno syscall.Errno - err = rc.Write(func(fd uintptr) bool { - sent, _, errno = unix.Syscall6(unix.SYS_SENDMSG_X, fd, uintptr(unsafe.Pointer(&t.batch.hdrs[off])), uintptr(n-off), 0, 0, 0) - // sendmsg_x reports EAGAIN only when it took nothing; a partial batch returns its count. - return !errno.Temporary() - }) - if err != nil { - return 0, err - } - if errno != 0 { - return 0, errno - } - return int(sent), nil -} - -// writeEach writes pkts one at a time with Write. -func (t *tun) writeEach(pkts [][]byte) error { - var firstErr error - for _, p := range pkts { - if _, err := t.Write(p); err != nil && firstErr == nil { - firstErr = err - } - } - return firstErr -} - func (t *tun) Networks() []netip.Prefix { return t.vpnNetworks } @@ -901,5 +563,5 @@ func (t *tun) Name() string { } func (t *tun) Queues(int) ([]tio.Queue, error) { - return []tio.Queue{&tunQueue{t: t, buf: make([]byte, tunReadArena)}}, nil + return []tio.Queue{t.queue}, nil } diff --git a/udp/udp_darwin.go b/udp/udp_darwin.go index e4e17856..d40f18c4 100644 --- a/udp/udp_darwin.go +++ b/udp/udp_darwin.go @@ -332,7 +332,7 @@ const recvBatch = 64 // msghdrX mirrors xnu's struct msghdr_x (bsd/sys/socket_private.h). recvmsg_x reports each // datagram's length in Datalen rather than in its return value. -// Keep in sync with msghdrX in overlay/tun_darwin.go. +// Keep in sync with msghdrX in overlay/tio/tio_utun_darwin.go. type msghdrX struct { Name *byte Namelen uint32