overlay/darwin: drain several tun packets per read wakeup

The darwin tun returned one packet per Read, and listenIn flushes its
UDP send batch after every Read, so each outbound packet went to the
underlay as its own batch. tunQueue.Read now keeps calling readv until
the utun would block, up to tunReadBatch packets, so a burst reaches the
send batch together.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

overlay/tun_darwin.go

@@ -607,6 +607,72 @@ func (t *tun) Read(to []byte) (int, error) {
 	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) 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).
@@ -666,5 +732,5 @@ func (t *tun) Name() string {
 }
 
 func (t *tun) Queues(int) ([]tio.Queue, error) {
-	return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil
+	return []tio.Queue{&tunQueue{t: t, buf: make([]byte, tunReadArena)}}, nil
 }

overlay/tun_darwin_test.go

@@ -0,0 +1,97 @@
+//go:build !ios && !e2e_testing
+
+package overlay
+
+import (
+	"os"
+	"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 on one end of a connected AF_UNIX datagram socketpair, which keeps
+// datagram boundaries the way a utun control socket does, and the other end's fd.
+// sndbuf is the tun end's send buffer, which caps the largest datagram it accepts.
+func newSocketpairTun(t *testing.T, sndbuf int) (*tun, 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))
+	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")
+	t.Cleanup(func() {
+		_ = f.Close()
+		_ = unix.Close(fds[1])
+	})
+	return &tun{f: f, l: test.NewLogger()}, fds[1]
+}
+
+// testTunPkt returns packet i of size bytes, alternating IPv4 and IPv6, and the same packet as a utun
+// datagram, behind its AF prefix.
+func testTunPkt(i, size int) (pkt, wire []byte) {
+	pkt = make([]byte, size)
+	pkt[0] = 0x45
+	af := byte(syscall.AF_INET)
+	if i%2 == 1 {
+		pkt[0] = 0x60
+		af = syscall.AF_INET6
+	}
+	pkt[1] = byte(i)
+	pkt[2] = byte(i >> 8)
+	return pkt, append([]byte{0, 0, 0, af}, pkt...)
+}
+
+// TestTunQueueReadDrains 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)}
+	total := tunReadBatch + 3
+	var want [][]byte
+	for i := range total {
+		pkt, wire := testTunPkt(i, 60+i)
+		_, err := unix.Write(w, wire)
+		require.NoError(t, err)
+		want = append(want, pkt)
+	}
+
+	var got [][]byte
+	for _, size := range []int{tunReadBatch, 3} {
+		pkts, err := q.Read()
+		require.NoError(t, err)
+		require.Len(t, pkts, size)
+		for _, p := range pkts {
+			got = append(got, p.Clone().Bytes)
+		}
+	}
+	assert.Equal(t, want, got)
+}
+
+// TestTunQueueReadClipsPackets 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)}
+	for i := range 2 {
+		_, wire := testTunPkt(i, 100)
+		_, err := unix.Write(w, wire)
+		require.NoError(t, err)
+	}
+	pkts, err := q.Read()
+	require.NoError(t, err)
+	require.Len(t, pkts, 2)
+	assert.Equal(t, len(pkts[0].Bytes), cap(pkts[0].Bytes))
+}