overlay/darwin: batch tun writes with sendmsg_x

Passthrough.Flush hands its whole lane to a new tio.BatchWriter in one
WriteBatch call when the queue implements it, and NewSingleQueue
exposes WriteBatch when its source does. On darwin, tun.WriteBatch
writes up to 64 packets per sendmsg_x on the utun control socket,
which is a connected datagram socket, so xnu takes the batched
sendit_x path rather than one sendit per message. utun registers no
send_list handler, so the kernel still passes each packet to utun on
its own: this saves syscalls and fd locking, not per-packet kernel work.

Error handling follows what sendmsg_x reports:

- A short count with no error (a packet past the send buffer
  mid-call, or a later send failing with ENOBUFS): the remainder goes
  out in the next sendmsg_x.
- EMSGSIZE: only returned when the first packet is too large, before
  the kernel takes any, so that packet alone goes through Write, which
  reports it, and the rest are batched again.
- ENOSYS, EPERM, EOPNOTSUPP: sendmsg_x is private API, so WriteBatch
  logs once and falls back to writev per packet for good.
- Anything else: sendmsg_x returns the error without the count it
  took, so the call is reported rather than retried, which could
  deliver packets twice.

One loss mode has no userspace fix. sendit_x counts each packet as
sent once it is copied into an mbuf; if a later allocation fails, it
frees the copied prefix unsent, returns its length, and sendmsg_x
drops the ENOBUFS because the count is nonzero. That looks exactly
like a real short count, so under mbuf exhaustion WriteBatch silently
drops packets that writev would have delivered.

The iOS tun path (tun_ios.go) is unchanged.

Inside a macOS Network Extension, download goes from 640 to 832 Mbps
against the recvmsg_x change alone in three paired rounds, at 15% less
CPU per Mbps; the likely cause is that its utun charges more per write
than a raw one. Raw dnclient already runs download near line rate,
and there the CPU difference is within noise.

TestTunWriteBatch* round-trip through a connected AF_UNIX datagram
socketpair, which takes sendmsg_x the same way, over both the batched
and the fallback path: order, AF prefixes, skipped invalid packets,
and an oversize packet mid-call and first-in-call. Dropping Namelen
from msghdrX fails them. Mutation testing with gremlins leaves only
equivalent survivors in WriteBatch, plus which error is reported first
when several fail.

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

overlay/batch/multi_coalesce_test.go

@@ -435,3 +435,65 @@ func testPP(pkt []byte) *firewall.ParsedPacket {
 	}
 	return pp
 }
+
+// fakeTunBatchWriter records each WriteBatch call and each plain Write separately.
+type fakeTunBatchWriter struct {
+	writes  [][]byte
+	batches [][][]byte
+}
+
+func (w *fakeTunBatchWriter) Write(p []byte) (int, error) {
+	w.writes = append(w.writes, bytes.Clone(p))
+	return len(p), nil
+}
+
+func (w *fakeTunBatchWriter) WriteBatch(pkts [][]byte) error {
+	var b [][]byte
+	for _, p := range pkts {
+		b = append(b, bytes.Clone(p))
+	}
+	w.batches = append(w.batches, b)
+	return nil
+}
+
+// TestPassthroughUsesWriteBatch pins that a writer implementing tio.BatchWriter gets the whole
+// passthrough lane in one WriteBatch per Flush, in order, and no plain Writes; an empty Flush
+// makes no call.
+func TestPassthroughUsesWriteBatch(t *testing.T) {
+	w := &fakeTunBatchWriter{}
+	m := newTestMultiCoalescer(t, w)
+	k := &keySeq{epoch: 1}
+
+	var want [][]byte
+	for i := range 5 {
+		icmp := make([]byte, 28)
+		icmp[0] = 0x45
+		icmp[3] = 28
+		icmp[9] = 1
+		icmp[20] = byte(i)
+		want = append(want, icmp)
+		if err := m.Commit(icmp, k.next(), testPP(icmp)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := m.Flush(); err != nil {
+		t.Fatal(err)
+	}
+	if err := m.Flush(); err != nil {
+		t.Fatal(err)
+	}
+	if len(w.writes) != 0 {
+		t.Fatalf("got %d plain Writes, want 0", len(w.writes))
+	}
+	if len(w.batches) != 1 {
+		t.Fatalf("got %d WriteBatch calls, want 1", len(w.batches))
+	}
+	for i, p := range w.batches[0] {
+		if !bytes.Equal(p, want[i]) {
+			t.Fatalf("packet %d: got % x, want % x", i, p, want[i])
+		}
+	}
+	if len(w.batches[0]) != len(want) {
+		t.Fatalf("got %d packets, want %d", len(w.batches[0]), len(want))
+	}
+}

overlay/batch/passthrough.go

@@ -2,18 +2,23 @@ package batch
 
 import (
 	"io"
+
+	"github.com/slackhq/nebula/overlay/tio"
 )
 
-// Passthrough is MultiCoalescer's verbatim lane: no batching, packets are written at Flush in the
-// order enqueued.
+// Passthrough is MultiCoalescer's verbatim lane: no coalescing, packets are written at Flush in the
+// order enqueued, in one WriteBatch when out implements tio.BatchWriter.
 type Passthrough struct {
 	out   io.Writer
+	bw    tio.BatchWriter // nil: Flush writes one packet per Write
 	slots [][]byte
 }
 
 func NewPassthrough(w io.Writer) *Passthrough {
+	bw, _ := w.(tio.BatchWriter)
 	return &Passthrough{
 		out:   w,
+		bw:    bw,
 		slots: make([][]byte, 0, 128),
 	}
 }
@@ -26,10 +31,16 @@ func (p *Passthrough) enqueue(pkt []byte) error {
 
 func (p *Passthrough) Flush() error {
 	var firstErr error
-	for _, s := range p.slots {
-		_, err := p.out.Write(s)
-		if err != nil && firstErr == nil {
-			firstErr = err
+	if p.bw != nil {
+		if len(p.slots) > 0 {
+			firstErr = p.bw.WriteBatch(p.slots)
+		}
+	} else {
+		for _, s := range p.slots {
+			_, err := p.out.Write(s)
+			if err != nil && firstErr == nil {
+				firstErr = err
+			}
 		}
 	}
 	clear(p.slots)

overlay/tio/single.go

@@ -15,9 +15,13 @@ 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.
+// Close closes rwc. The Queue implements BatchWriter when rwc does.
 func NewSingleQueue(rwc io.ReadWriteCloser, bufSize int) Queue {
-	return &singleQueue{rw: rwc, closer: rwc, buf: make([]byte, bufSize)}
+	q := &singleQueue{rw: rwc, closer: rwc, buf: make([]byte, bufSize)}
+	if bw, ok := rwc.(BatchWriter); ok {
+		return &batchSingleQueue{singleQueue: q, bw: bw}
+	}
+	return q
 }
 
 // NewSingleQueueNoClose is NewSingleQueue for a source owned by someone else,
@@ -47,3 +51,14 @@ 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)
+}

overlay/tio/tio.go

@@ -145,3 +145,14 @@ func SupportsGSO(w io.Writer, want GSOProto) (GSOWriter, bool) {
 		return gw, false
 	}
 }
+
+// BatchWriter is implemented by Queues that can emit several already-segmented packets in one call.
+// Backends without a batched write do not implement it and callers loop over Write.
+//
+// pkts are borrowed: WriteBatch must not retain them past its return.
+// A returned error doesn't say which packets were delivered, so callers must not retry the batch.
+// Safe for concurrent use, like Write.
+type BatchWriter interface {
+	io.Writer
+	WriteBatch(pkts [][]byte) error
+}

overlay/tun_darwin.go

@@ -8,6 +8,7 @@ import (
 	"log/slog"
 	"net/netip"
 	"os"
+	"sync"
 	"sync/atomic"
 	"syscall"
 	"unsafe"
@@ -33,6 +34,35 @@ 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 {
@@ -615,16 +645,12 @@ func (t *tun) Write(from []byte) (int, error) {
 		return 0, syscall.EIO
 	}
 
-	ipVer := from[0] >> 4
 	var head [4]byte
-	switch ipVer {
-	case 4:
-		head[3] = syscall.AF_INET
-	case 6:
-		head[3] = syscall.AF_INET6
-	default:
-		return 0, fmt.Errorf("unable to determine IP version from packet")
+	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()
@@ -657,6 +683,147 @@ func (t *tun) Write(from []byte) (int, error) {
 	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
 }

overlay/tun_darwin_test.go

@@ -0,0 +1,153 @@
+//go:build !ios && !e2e_testing
+
+package overlay
+
+import (
+	"bytes"
+	"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 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) {
+	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 datagram the utun
+// device should receive for it.
+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...)
+}
+
+// readTunBack asserts r receives exactly want, in order.
+func readTunBack(t *testing.T, r int, want [][]byte) {
+	t.Helper()
+	buf := make([]byte, 4096)
+	for i, w := range want {
+		n, _, err := unix.Recvfrom(r, buf, 0)
+		require.NoError(t, err, "datagram %d", i)
+		if !bytes.Equal(w, buf[:n]) {
+			t.Fatalf("datagram %d: got % x, want % x", i, buf[:n], w)
+		}
+	}
+	_, _, err := unix.Recvfrom(r, buf, unix.MSG_DONTWAIT)
+	assert.ErrorIs(t, err, unix.EAGAIN, "extra datagrams delivered")
+}
+
+// forEachTunWritePath runs fn once over sendmsg_x and once over the per-packet writev fallback.
+func forEachTunWritePath(t *testing.T, fn func(t *testing.T, fallback bool)) {
+	for _, fallback := range []bool{false, true} {
+		name := "sendmsg_x"
+		if fallback {
+			name = "writev"
+		}
+		t.Run(name, func(t *testing.T) { fn(t, fallback) })
+	}
+}
+
+// TestTunWriteBatch 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) {
+	forEachTunWritePath(t, func(t *testing.T, fallback bool) {
+		tn, r := newSocketpairTun(t, 1<<20)
+		tn.noSendmsgX.Store(fallback)
+
+		var pkts, want [][]byte
+		for i := range 3*tunWriteBatch + 5 {
+			p, w := testTunPkt(i, 60+i)
+			pkts = append(pkts, p)
+			want = append(want, w)
+		}
+		require.NoError(t, tn.WriteBatch(pkts))
+		assert.Equal(t, fallback, tn.noSendmsgX.Load())
+		readTunBack(t, r, want)
+	})
+}
+
+// TestTunWriteBatchSkipsInvalid 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) {
+	forEachTunWritePath(t, func(t *testing.T, fallback bool) {
+		tn, r := newSocketpairTun(t, 1<<20)
+		tn.noSendmsgX.Store(fallback)
+
+		var pkts, want [][]byte
+		for i := range 20 {
+			p, w := testTunPkt(i, 60)
+			pkts = append(pkts, p)
+			want = append(want, w)
+		}
+		pkts = append(pkts[:10], append([][]byte{{0x10, 1, 2}, {}}, pkts[10:]...)...)
+
+		err := tn.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
+// 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) {
+	for _, tc := range []struct {
+		name string
+		at   int
+	}{
+		{"mid-call", tunWriteBatch + 6},
+		{"first-in-call", tunWriteBatch},
+	} {
+		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)
+
+				var pkts, want [][]byte
+				for i := range 2*tunWriteBatch + 10 {
+					size := 100
+					if i == tc.at {
+						size = 2000
+					}
+					p, w := testTunPkt(i, size)
+					pkts = append(pkts, p)
+					if i != tc.at {
+						want = append(want, w)
+					}
+				}
+				err := tn.WriteBatch(pkts)
+				assert.ErrorIs(t, err, unix.EMSGSIZE)
+				assert.Equal(t, fallback, tn.noSendmsgX.Load())
+				readTunBack(t, r, want)
+			})
+		})
+	}
+}

udp/udp_darwin.go

@@ -185,6 +185,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.
 type msghdrX struct {
 	Name       *byte
 	Namelen    uint32