udp/darwin: batch receives and connected-socket sends with recvmsg_x/sendmsg_x

internal/msgx is the one copy of xnu's private batched socket calls: the
msghdr_x layout, Send/Recv through the raw syscall (EINTR retried) and
Unsupported for the errnos a kernel or sandbox refuses them with.

Every socket the listener owns reads with recvmsg_x: the listener and each
connected socket run their own reader goroutine, and hand whole batches to
ListenOut over a channel. ListenOut runs the EncReader over a batch and
flushes once, so the reader stays single-threaded without a lock that
readers contend on, and the next recvmsg_x overlaps the decrypt of the last
batch. Batch buffers (64 x MTU) come from a per-listener free list that
keeps up to 11; a reader takes one only while its socket is readable, so
idle connected sockets hold none. A batch with a header intact msghdr_x
entries can't produce, or ENOSYS/EPERM/EOPNOTSUPP, switches every reader to
one datagram per read for good.

Connected sockets send runs with sendmsg_x, which xnu batches through the
stack on a connected socket. EAGAIN/ENOBUFS keep #1901's single-datagram
stall handling; a refused sendmsg_x retries the datagram with write and
turns sendmsg_x off once that goes through.

The connected-socket knobs lose their NEBULA_CONNSOCK* environment
overrides and keep #1901's defaults.

Based on Caleb Jasik's #1895 (recvmsg_x) and #1897 (sendmsg_x), and Jack
Doan's internal/msgx shape from #1898.

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

udp/connsock_bench_darwin_test.go

@@ -10,19 +10,10 @@ import (
 	"time"
 )
 
-// benchConnSocks opens a wildcard listener with ListenOut running, with the connected socket knobs pinned to
-// their defaults and then cfg applied, and a loopback peer that never reads, so its sends cost only the sender.
+// benchConnSocks opens a wildcard listener with ListenOut running and cfg applied, and a loopback peer that never
+// reads, so its sends cost only the sender.
 func benchConnSocks(b *testing.B, s Settings, cfg func(*connSockConfig)) (*StdConn, netip.AddrPort) {
 	b.Helper()
-	for k, v := range map[string]string{
-		"NEBULA_CONNSOCKS":           "8",
-		"NEBULA_CONNSOCK_RUN":        "64",
-		"NEBULA_CONNSOCK_WINDOW":     "1s",
-		"NEBULA_CONNSOCK_IDLE":       "30s",
-		"NEBULA_CONNSOCK_OPEN_EVERY": "1s",
-	} {
-		b.Setenv(k, v)
-	}
 	c, err := NewListener(slog.New(slog.DiscardHandler), s)
 	if err != nil {
 		b.Fatal(err)
@@ -33,7 +24,7 @@ func benchConnSocks(b *testing.B, s Settings, cfg func(*connSockConfig)) (*StdCo
 		cfg(&u.socks.cfg)
 	}
 	go func() { _ = u.ListenOut(func(netip.AddrPort, []byte) {}, func() {}) }()
-	for u.reader.Load() == nil {
+	for !u.listening.Load() {
 		time.Sleep(time.Millisecond)
 	}
 	pc, err := net.ListenPacket("udp4", "127.0.0.1:0")

udp/connsock_darwin.go

@@ -12,12 +12,13 @@ import (
 	"net/netip"
 	"os"
 	"slices"
-	"strconv"
 	"sync"
 	"sync/atomic"
 	"syscall"
 	"time"
+	"unsafe"
 
+	"github.com/slackhq/nebula/internal/msgx"
 	"golang.org/x/sys/unix"
 )
 
@@ -52,13 +53,46 @@ type connSock struct {
 	// parkMu serializes the writes that wait for a full send buffer, since the write deadline bounding each wait
 	// is the descriptor's, not the writer's.
 	parkMu sync.Mutex
-	// sys writes one datagram on the descriptor: writeFd, or a stand-in for the interface in tests.
-	sys func(fd int, b []byte) error
+	// sys writes bufs on the descriptor and returns how many went out: sendFd, or a stand-in for the interface in
+	// tests.
+	sys func(fd int, bufs [][]byte, x *sendScratch) (int, error)
 }
 
-func writeFd(fd int, b []byte) error {
-	_, err := syscall.Write(fd, b)
-	return err
+// sendScratch is the msghdr_x array one sendmsg_x reads.
+type sendScratch struct {
+	hdrs [msgx.Batch]msgx.Hdr
+	iovs [msgx.Batch]unix.Iovec
+}
+
+// noSendX is set once sendmsg_x is refused; connected sockets then write one datagram per syscall.
+var noSendX atomic.Bool
+
+// sendFd writes bufs on fd, a lone datagram with write and more with one sendmsg_x, which xnu runs through the stack
+// as a batch on a connected socket. It returns how many went out, or none and the errno. sendmsg_x reports EAGAIN,
+// ENOBUFS and EMSGSIZE only when it took nothing, and any other error without the count it took, which a connected
+// socket gets only on its first datagram, as its peer's unreachable.
+func sendFd(fd int, bufs [][]byte, x *sendScratch) (int, error) {
+	if len(bufs) == 1 {
+		if _, err := syscall.Write(fd, bufs[0]); err != nil {
+			return 0, err
+		}
+		return 1, nil
+	}
+	n := min(len(bufs), msgx.Batch)
+	for i, b := range bufs[:n] {
+		x.iovs[i] = unix.Iovec{Base: unsafe.SliceData(b)}
+		x.iovs[i].SetLen(len(b))
+		x.hdrs[i] = msgx.Hdr{Iov: &x.iovs[i], Iovlen: 1}
+	}
+	sent, errno := msgx.Send(uintptr(fd), x.hdrs[:n])
+	clear(x.iovs[:n])
+	if errno != 0 {
+		return 0, errno
+	}
+	if sent == 0 {
+		return 0, syscall.EAGAIN
+	}
+	return sent, nil
 }
 
 // errStallSpent is a write that found the connected socket's stall budget spent, so its datagram is dropped.
@@ -67,11 +101,16 @@ var errStallSpent = errors.New("connected socket stall budget spent")
 // connSockWriter carries one write through RawConn.Write. They are pooled with fn bound once, since a closure
 // passed through the RawConn interface escapes and would allocate on every send.
 type connSockWriter struct {
-	s    *connSock
-	b    []byte
-	park bool
-	err  error
-	fn   func(fd uintptr) bool
+	s *connSock
+	// bufs holds a copy of the datagrams' slice headers rather than the caller's slice, which would escape
+	// through the pool and cost WriteTo's one-datagram array an allocation per send.
+	bufs  [msgx.Batch][]byte
+	nbufs int
+	park  bool
+	n     int
+	err   error
+	fn    func(fd uintptr) bool
+	x     sendScratch
 }
 
 var connSockWriters = sync.Pool{New: func() any {
@@ -82,7 +121,7 @@ var connSockWriters = sync.Pool{New: func() any {
 
 func (w *connSockWriter) write(fd uintptr) bool {
 	for {
-		w.err = w.s.sys(int(fd), w.b)
+		w.n, w.err = w.s.sys(int(fd), w.bufs[:w.nbufs], &w.x)
 		switch {
 		case w.err == syscall.EINTR:
 		case w.err == syscall.EAGAIN && w.park:
@@ -93,28 +132,28 @@ func (w *connSockWriter) write(fd uintptr) bool {
 	}
 }
 
-// try writes b on s once and returns nil, the write's errno, os.ErrDeadlineExceeded, or net.ErrClosed. With park
-// set, EAGAIN waits in the netpoller until the socket is writable or the write deadline passes; without it, EAGAIN
-// is returned.
-func (s *connSock) try(b []byte, park bool) error {
+// try writes bufs on s once and returns how many went out, or none and the write's errno,
+// os.ErrDeadlineExceeded, or net.ErrClosed. With park set, EAGAIN waits in the netpoller until the socket is
+// writable or the write deadline passes; without it, EAGAIN is returned.
+func (s *connSock) try(bufs [][]byte, park bool) (int, error) {
 	w := connSockWriters.Get().(*connSockWriter)
-	w.s, w.b, w.park = s, b, park
+	w.s, w.nbufs, w.park = s, copy(w.bufs[:], bufs), park
 	rerr := s.rc.Write(w.fn)
-	err := w.err
-	*w = connSockWriter{fn: w.fn}
+	n, err := w.n, w.err
+	clear(w.bufs[:w.nbufs])
+	w.s, w.nbufs, w.n, w.err = nil, 0, 0, nil
 	connSockWriters.Put(w)
 	switch {
 	case rerr == nil:
-		return err
+		return n, err
 	case errors.Is(rerr, os.ErrDeadlineExceeded):
-		return os.ErrDeadlineExceeded
+		return 0, os.ErrDeadlineExceeded
 	default:
-		return net.ErrClosed
+		return 0, net.ErrClosed
 	}
 }
 
-// connSockConfig holds the knobs. connSockConfigFromEnv reads all but backoff and enobufsMax from the environment,
-// so one binary can be tuned without a rebuild.
+// connSockConfig holds the knobs, which tests change.
 type connSockConfig struct {
 	// max is how many connected sockets the listen routines may have open at once; 0 turns connected sockets off.
 	max int
@@ -135,8 +174,8 @@ type connSockConfig struct {
 	enobufsMax time.Duration
 }
 
-func connSockConfigFromEnv() connSockConfig {
-	c := connSockConfig{
+func defaultConnSockConfig() connSockConfig {
+	return connSockConfig{
 		max:        8,
 		run:        64,
 		window:     time.Second,
@@ -145,22 +184,6 @@ func connSockConfigFromEnv() connSockConfig {
 		openEvery:  time.Second,
 		enobufsMax: 10 * time.Millisecond,
 	}
-	if v, err := strconv.Atoi(os.Getenv("NEBULA_CONNSOCKS")); err == nil {
-		c.max = v
-	}
-	if v, err := strconv.Atoi(os.Getenv("NEBULA_CONNSOCK_RUN")); err == nil && v > 0 {
-		c.run = v
-	}
-	if v, err := time.ParseDuration(os.Getenv("NEBULA_CONNSOCK_WINDOW")); err == nil && v > 0 {
-		c.window = v
-	}
-	if v, err := time.ParseDuration(os.Getenv("NEBULA_CONNSOCK_IDLE")); err == nil && v > 0 {
-		c.idle = v
-	}
-	if v, err := time.ParseDuration(os.Getenv("NEBULA_CONNSOCK_OPEN_EVERY")); err == nil && v > 0 {
-		c.openEvery = v
-	}
-	return c
 }
 
 // connSockGroups holds the connSocks that the listen routines bound to one address share, so the limits, the
@@ -182,7 +205,7 @@ func joinConnSocks(u *StdConn, la netip.AddrPort, s Settings) *connSocks {
 		cs.mu.Unlock()
 		return cs
 	}
-	cs := &connSocks{cfg: connSockConfigFromEnv(), key: la, members: []*StdConn{u}, clock: monoNow, sleep: time.Sleep}
+	cs := &connSocks{cfg: defaultConnSockConfig(), key: la, members: []*StdConn{u}, clock: monoNow, sleep: time.Sleep}
 	if cs.cfg.max > 0 && !u.listenHost.IsUnspecified() && !s.Multi {
 		// A connected socket on a specific listen.host binds the listener's own address, which xnu allows only when
 		// both sockets set SO_REUSEPORT, and a listener with it lets another uid bind a dual-stack wildcard on the
@@ -433,7 +456,7 @@ func (u *StdConn) connSockRun(p *connSockPeer, dst netip.AddrPort, bufs [][]byte
 // one out.
 func (u *StdConn) maybeOpenConnSock(p *connSockPeer, dst netip.AddrPort, n int) *connSock {
 	cs := u.socks
-	if n <= 0 || u.reader.Load() == nil || int(cs.reserved.Load()) >= cs.cfg.max || !p.busy(cs.clock(), n, &cs.cfg) {
+	if n <= 0 || !u.listening.Load() || int(cs.reserved.Load()) >= cs.cfg.max || !p.busy(cs.clock(), n, &cs.cfg) {
 		return nil
 	}
 	cs.mu.Lock()
@@ -527,7 +550,7 @@ func (u *StdConn) dialConnSock(p *connSockPeer, dst netip.AddrPort) (*connSock,
 		_ = uc.Close()
 		return nil, err
 	}
-	s := &connSock{dst: dst, peer: p, conn: uc, rc: rc, sys: writeFd}
+	s := &connSock{dst: dst, peer: p, conn: uc, rc: rc, sys: sendFd}
 	s.used.Store(u.socks.clock())
 	return s, nil
 }
@@ -677,7 +700,7 @@ func connSockDead(errno syscall.Errno) bool {
 // the 09-25 probes.
 const enobufsWait = 50 * time.Microsecond
 
-// connSockWrite writes bufs on s, one datagram per write. A full socket buffer parks the caller in the netpoller
+// connSockWrite writes bufs on s, up to msgx.Batch datagrams per sendmsg_x. A full socket buffer parks the caller in the netpoller
 // until the socket is writable, and a flow-controlled interface makes it wait out ENOBUFS, so a busy link slows
 // the tun reader instead of dropping. Once the connected socket has waited cfg.enobufsMax without a send going
 // through, either one drops the datagram instead, and a dropped datagram counts as sent, as the listener's
@@ -691,15 +714,30 @@ func (u *StdConn) connSockWrite(s *connSock, bufs [][]byte) (sent int, errno sys
 			s.used.Store(cs.clock())
 		}
 	}()
+	probe := false
 	for sent < len(bufs) {
-		err := s.try(bufs[sent], false)
+		n := 1
+		if !noSendX.Load() && !probe {
+			n = len(bufs) - sent
+		}
+		k, err := s.try(bufs[sent:sent+n], false)
+		if errno, ok := err.(syscall.Errno); ok && n > 1 && msgx.Unsupported(errno) {
+			// EPERM may be the datagram's own fault, so sendmsg_x is off for good only once a plain write of the same
+			// datagram goes through.
+			probe = true
+			continue
+		}
+		if probe && err == nil && noSendX.CompareAndSwap(false, true) {
+			u.l.Warn("sendmsg_x unavailable, writing one datagram per syscall")
+		}
+		probe = false
 		if err == syscall.EAGAIN || err == os.ErrDeadlineExceeded {
 			// os.ErrDeadlineExceeded here is another writer's parked deadline, passed but not yet cleared.
-			err = u.parkedWrite(s, bufs[sent])
+			k, err = u.parkedWrite(s, bufs[sent:sent+1])
 		}
 		switch err {
 		case nil:
-			sent++
+			sent += k
 			if s.stallAt.Load() != 0 {
 				s.stallAt.Store(0)
 				s.dropping.Store(false)
@@ -731,23 +769,23 @@ func (u *StdConn) connSockWrite(s *connSock, bufs [][]byte) (sent int, errno sys
 // parkedWrite writes b after EAGAIN, parked in the netpoller for what is left of s's stall budget, and returns
 // errStallSpent once none is. golang/go#73919: a darwin UDP write parked on EAGAIN can wait forever for a
 // writability event that never comes, so the wait always has a deadline.
-func (u *StdConn) parkedWrite(s *connSock, b []byte) error {
+func (u *StdConn) parkedWrite(s *connSock, b [][]byte) (int, error) {
 	cs := u.socks
 	s.parkMu.Lock()
 	defer s.parkMu.Unlock()
 	left := cs.cfg.enobufsMax - s.stalledFor(cs.clock())
 	if left <= 0 {
-		return errStallSpent
+		return 0, errStallSpent
 	}
 	if err := s.conn.SetWriteDeadline(time.Now().Add(left)); err != nil {
-		return net.ErrClosed
+		return 0, net.ErrClosed
 	}
-	err := s.try(b, true)
+	n, err := s.try(b, true)
 	_ = s.conn.SetWriteDeadline(time.Time{})
 	if err == os.ErrDeadlineExceeded {
-		return errStallSpent
+		return 0, errStallSpent
 	}
-	return err
+	return n, err
 }
 
 // dropStalled logs the first datagram a stall drops.
@@ -772,14 +810,13 @@ func (s *connSock) stalledFor(now int64) time.Duration {
 	return time.Duration(now - at)
 }
 
-// readConnSock delivers a connected socket's datagrams through the listener's reader, serialized with ListenOut,
-// until the connected socket is closed. Any receive error closes it: a connected socket reports its peer's
-// unreachables here, and retrying an unexpected error could spin.
+// readConnSock hands a connected socket's batches to ListenOut until the connected socket is closed. Any receive
+// error closes it: a connected socket reports its peer's unreachables here, and retrying an unexpected error could
+// spin.
 func (u *StdConn) readConnSock(s *connSock) {
-	rd := u.reader.Load()
-	buf := make([]byte, MTU)
+	r := u.newBatchReader(s.conn, s.rc)
 	for {
-		n, from, err := s.conn.ReadFromUDPAddrPort(buf)
+		b, err := u.read(r)
 		if err != nil {
 			if errors.Is(err, net.ErrClosed) {
 				return
@@ -795,9 +832,8 @@ func (u *StdConn) readConnSock(s *connSock) {
 			return
 		}
 		s.used.Store(u.socks.clock())
-		u.readMu.Lock()
-		rd.r(netip.AddrPortFrom(from.Addr().Unmap(), from.Port()), buf[:n:n])
-		rd.flush()
-		u.readMu.Unlock()
+		if !u.deliver(b) {
+			return
+		}
 	}
 }

udp/connsock_darwin_test.go

@@ -20,19 +20,9 @@ import (
 	"golang.org/x/sys/unix"
 )
 
-// newConnSockListener opens a listener with the connected socket knobs pinned to their defaults, so NEBULA_*
-// variables in the environment can't change what a test sees, applies cfg, and starts ListenOut with r.
+// newConnSockListener opens a listener, applies cfg to its connected socket knobs, and starts ListenOut with r.
 func newConnSockListener(t *testing.T, s Settings, h slog.Handler, r EncReader, cfg ...func(*connSockConfig)) *StdConn {
 	t.Helper()
-	for k, v := range map[string]string{
-		"NEBULA_CONNSOCKS":           "8",
-		"NEBULA_CONNSOCK_RUN":        "64",
-		"NEBULA_CONNSOCK_WINDOW":     "1s",
-		"NEBULA_CONNSOCK_IDLE":       "30s",
-		"NEBULA_CONNSOCK_OPEN_EVERY": "1s",
-	} {
-		t.Setenv(k, v)
-	}
 	if h == nil {
 		h = slog.DiscardHandler
 	}
@@ -47,10 +37,25 @@ func newConnSockListener(t *testing.T, s Settings, h slog.Handler, r EncReader,
 		f(&u.socks.cfg)
 	}
 	go func() { _ = u.ListenOut(r, func() {}) }()
-	require.Eventually(t, func() bool { return u.reader.Load() != nil }, 5*time.Second, time.Millisecond)
+	require.Eventually(t, func() bool { return u.listening.Load() }, 5*time.Second, time.Millisecond)
 	return u
 }
 
+// oneAtATime stands in for sendFd with f, which writes one datagram, so a test can fail chosen datagrams.
+func oneAtATime(f func(fd int, b []byte) error) func(int, [][]byte, *sendScratch) (int, error) {
+	return func(fd int, bufs [][]byte, _ *sendScratch) (int, error) {
+		if err := f(fd, bufs[0]); err != nil {
+			return 0, err
+		}
+		return 1, nil
+	}
+}
+
+func writeFd(fd int, b []byte) error {
+	_, err := syscall.Write(fd, b)
+	return err
+}
+
 func listen(s string) Settings { return Settings{Listen: netip.MustParseAddrPort(s)} }
 
 // run builds n datagrams to dst.
@@ -282,10 +287,10 @@ func TestConnSocksSharedAcrossRoutines(t *testing.T) {
 	s1 := u1.connSockFor(a)
 	require.NotNil(t, s1)
 	var sends atomic.Int64
-	s1.sys = func(fd int, b []byte) error {
+	s1.sys = oneAtATime(func(fd int, b []byte) error {
 		sends.Add(1)
 		return writeFd(fd, b)
-	}
+	})
 	require.NoError(t, u0.WriteTo([]byte("via u0"), a))
 	readFrom(t, peerA, 1)
 	assert.EqualValues(t, 1, sends.Load(), "routine 0 didn't send on routine 1's connected socket")
@@ -566,10 +571,10 @@ func TestConnSocksWriteBatchMixedDestinations(t *testing.T) {
 	peerB, b := newEstablishedPeer(t, u, "udp4", "127.0.0.1:0")
 	peerC, c := newPeer(t, "udp4", "127.0.0.1:0")
 	var onSock atomic.Int64
-	s.sys = func(fd int, b []byte) error {
+	s.sys = oneAtATime(func(fd int, b []byte) error {
 		onSock.Add(1)
 		return writeFd(fd, b)
-	}
+	})
 
 	var bufs [][]byte
 	var addrs []netip.AddrPort
@@ -598,12 +603,12 @@ func TestConnSocksWriteBatchMixedDestinations(t *testing.T) {
 func TestConnSocksWriteError(t *testing.T) {
 	u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil)
 	s, peer := openConnSock(t, u)
-	s.sys = func(fd int, b []byte) error {
+	s.sys = oneAtATime(func(fd int, b []byte) error {
 		if string(b) == "bad" {
 			return syscall.EMSGSIZE
 		}
 		return writeFd(fd, b)
-	}
+	})
 	bufs := [][]byte{[]byte("ok0"), []byte("bad"), []byte("ok2")}
 	n, err := u.WriteBatch(bufs, []netip.AddrPort{s.dst, s.dst, s.dst})
 	require.NoError(t, err)
@@ -621,12 +626,12 @@ func TestConnSocksWriteBatchFallsBackMidRun(t *testing.T) {
 			u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil)
 			s, peer := openConnSock(t, u)
 			var calls atomic.Int64
-			s.sys = func(fd int, b []byte) error {
+			s.sys = oneAtATime(func(fd int, b []byte) error {
 				if calls.Add(1) > 2 {
 					return fail
 				}
 				return writeFd(fd, b)
-			}
+			})
 			bufs, addrs := run(s.dst, 5, "r")
 			n, err := u.WriteBatch(bufs, addrs)
 			require.NoError(t, err)
@@ -673,12 +678,12 @@ func TestConnSockWriteENOBUFS(t *testing.T) {
 		now.Add(int64(d))
 	}
 	var enobufs atomic.Bool
-	s.sys = func(fd int, b []byte) error {
+	s.sys = oneAtATime(func(fd int, b []byte) error {
 		if enobufs.Load() {
 			return syscall.ENOBUFS
 		}
 		return writeFd(fd, b)
-	}
+	})
 	perStall := int(u.socks.cfg.enobufsMax / enobufsWait)
 	bufs, _ := run(s.dst, 3, "q")
 
@@ -718,13 +723,13 @@ func TestConnSockWriteENOBUFSConcurrentWriters(t *testing.T) {
 	// Datagrams starting with 'e' get ENOBUFS while enobufs is set; the rest go out.
 	var enobufs atomic.Bool
 	var wrote atomic.Int64
-	s.sys = func(fd int, b []byte) error {
+	s.sys = oneAtATime(func(fd int, b []byte) error {
 		if b[0] == 'e' && enobufs.Load() {
 			return syscall.ENOBUFS
 		}
 		wrote.Add(1)
 		return writeFd(fd, b)
-	}
+	})
 	write := func(tag string, done chan<- int) {
 		sent, _, _ := u.connSockWrite(s, [][]byte{[]byte(tag)})
 		done <- sent
@@ -782,12 +787,12 @@ func TestConnSockWriteEAGAINBounded(t *testing.T) {
 	s, peer := openConnSock(t, u)
 	var eagain atomic.Bool
 	eagain.Store(true)
-	s.sys = func(fd int, b []byte) error {
+	s.sys = oneAtATime(func(fd int, b []byte) error {
 		if eagain.Load() {
 			return syscall.EAGAIN
 		}
 		return writeFd(fd, b)
-	}
+	})
 	budget := u.socks.cfg.enobufsMax
 
 	start := time.Now()
@@ -853,12 +858,12 @@ func TestConnSockWriteDoesNotBlockClose(t *testing.T) {
 		<-wake
 	}
 	var gone atomic.Bool
-	s.sys = func(fd int, b []byte) error {
+	s.sys = oneAtATime(func(fd int, b []byte) error {
 		if gone.Load() {
 			return writeFd(fd, b)
 		}
 		return syscall.ENOBUFS
-	}
+	})
 	done := make(chan bool)
 	go func() {
 		bufs, _ := run(s.dst, 1, "q")

udp/udp_darwin.go

@@ -10,12 +10,12 @@ import (
 	"log/slog"
 	"net"
 	"net/netip"
-	"sync"
 	"sync/atomic"
 	"syscall"
 	"unsafe"
 
 	"github.com/slackhq/nebula/config"
+	"github.com/slackhq/nebula/internal/msgx"
 	"golang.org/x/sys/unix"
 )
 
@@ -31,15 +31,13 @@ type StdConn struct {
 	// rc reaches the listener's descriptor for another listen routine's Rebind.
 	rc    syscall.RawConn
 	socks *connSocks
-	// reader is ListenOut's reader, which connected socket readers share; readMu serializes them, since an
-	// EncReader isn't safe for concurrent use.
-	reader atomic.Pointer[readerPair]
-	readMu sync.Mutex
-}
-
-type readerPair struct {
-	r     EncReader
-	flush func()
+	// listening is set once ListenOut runs. The listener's and every connected socket's readers hand their batches
+	// to it on rx, so the EncReader runs on ListenOut's goroutine alone; done closes when ListenOut returns.
+	listening atomic.Bool
+	rx        chan *rxBatch
+	done      chan struct{}
+	// free holds the batches ListenOut has delivered, for the readers to fill again.
+	free chan *rxBatch
 }
 
 var _ Conn = &StdConn{}
@@ -53,7 +51,8 @@ func NewListener(l *slog.Logger, s Settings) (Conn, error) {
 	}
 
 	if uc, ok := pc.(*net.UDPConn); ok {
-		c := &StdConn{UDPConn: uc, l: l, listenHost: s.Listen.Addr()}
+		c := &StdConn{UDPConn: uc, l: l, listenHost: s.Listen.Addr(), rx: make(chan *rxBatch, 1), done: make(chan struct{}),
+			free: make(chan *rxBatch, rxBatchesKept)}
 
 		rc, err := uc.SyscallConn()
 		if err != nil {
@@ -247,14 +246,176 @@ func NewUDPStatsEmitter(udpConns []Conn) func() {
 	return func() {}
 }
 
+// ListenOut delivers the batches that the listener's reader and the connected sockets' readers hand it.
 func (u *StdConn) ListenOut(r EncReader, flush func()) error {
-	buffer := make([]byte, MTU)
-	u.reader.Store(&readerPair{r: r, flush: flush})
+	defer close(u.done)
+	u.listening.Store(true)
 	u.startConnSocks()
-
+	errc := make(chan error, 1)
+	go func() { errc <- u.readListener() }()
 	for {
-		// Just read one packet at a time
-		n, rua, err := u.ReadFromUDPAddrPort(buffer)
+		select {
+		case b := <-u.rx:
+			for i := range b.n {
+				r(b.addrs[i], b.pkts[i])
+			}
+			flush()
+			u.putBatch(b)
+		case err := <-errc:
+			return err
+		}
+	}
+}
+
+// rxBatch is one recvmsg_x batch: n datagrams in pkts, from addrs. A reader takes one only while its socket is
+// readable, so an idle connected socket holds none.
+type rxBatch struct {
+	n     int
+	pkts  [msgx.Batch][]byte
+	addrs [msgx.Batch]netip.AddrPort
+	buf   []byte
+	names [msgx.Batch]unix.RawSockaddrInet6
+	iovs  [msgx.Batch]unix.Iovec
+	hdrs  [msgx.Batch]msgx.Hdr
+}
+
+// rxBatchesKept is how many batches a listener keeps for reuse: one being delivered and one queued, plus one for
+// each socket reading at once, the listener and up to 8 connected sockets. Each is msgx.Batch*MTU, about 576KB.
+const rxBatchesKept = 11
+
+func (u *StdConn) getBatch() *rxBatch {
+	select {
+	case b := <-u.free:
+		return b
+	default:
+	}
+	b := &rxBatch{buf: make([]byte, msgx.Batch*MTU)}
+	for i := range b.hdrs {
+		b.iovs[i].Base = &b.buf[i*MTU]
+		b.iovs[i].SetLen(MTU)
+		b.hdrs[i] = msgx.Hdr{Name: (*byte)(unsafe.Pointer(&b.names[i])), Iov: &b.iovs[i], Iovlen: 1}
+	}
+	return b
+}
+
+func (u *StdConn) putBatch(b *rxBatch) {
+	select {
+	case u.free <- b:
+	default:
+	}
+}
+
+// noRecvX is set once recvmsg_x is refused or misbehaves; every reader then reads one datagram per syscall.
+var noRecvX atomic.Bool
+
+// parse checks the n entries recvmsg_x filled and sets pkts and addrs from them. It rejects a batch that intact
+// msghdr_x entries can't produce, which is how a change to xnu's private struct layout would show up.
+func (b *rxBatch) parse(n int) error {
+	if n > len(b.hdrs) {
+		return fmt.Errorf("returned %d datagrams for %d headers", n, len(b.hdrs))
+	}
+	for i := range n {
+		h := &b.hdrs[i]
+		if h.Datalen > MTU {
+			return fmt.Errorf("datagram %d is %d bytes, past the %d byte buffer", i, h.Datalen, MTU)
+		}
+		sa := &b.names[i]
+		switch {
+		case sa.Family == unix.AF_INET && h.Namelen == unix.SizeofSockaddrInet4:
+			sa4 := (*unix.RawSockaddrInet4)(unsafe.Pointer(sa))
+			b.addrs[i] = netip.AddrPortFrom(netip.AddrFrom4(sa4.Addr), binary.BigEndian.Uint16((*[2]byte)(unsafe.Pointer(&sa4.Port))[:]))
+		case sa.Family == unix.AF_INET6 && h.Namelen == unix.SizeofSockaddrInet6:
+			b.addrs[i] = netip.AddrPortFrom(netip.AddrFrom16(sa.Addr).Unmap(), binary.BigEndian.Uint16((*[2]byte)(unsafe.Pointer(&sa.Port))[:]))
+		default:
+			return fmt.Errorf("datagram %d has address family %d with length %d", i, sa.Family, h.Namelen)
+		}
+		b.pkts[i] = b.buf[i*MTU : i*MTU+int(h.Datalen) : i*MTU+int(h.Datalen)]
+	}
+	b.n = n
+	return nil
+}
+
+// batchReader reads one socket in batches. Its fn is bound once, since a closure passed through the RawConn
+// interface escapes and would allocate on every read.
+type batchReader struct {
+	u     *StdConn
+	conn  *net.UDPConn
+	rc    syscall.RawConn
+	b     *rxBatch
+	errno syscall.Errno
+	fn    func(fd uintptr) bool
+}
+
+func (u *StdConn) newBatchReader(conn *net.UDPConn, rc syscall.RawConn) *batchReader {
+	r := &batchReader{u: u, conn: conn, rc: rc}
+	r.fn = r.recv
+	return r
+}
+
+func (r *batchReader) recv(fd uintptr) bool {
+	b := r.u.getBatch()
+	for i := range b.hdrs {
+		b.hdrs[i].Namelen = unix.SizeofSockaddrInet6
+		b.hdrs[i].Flags = 0
+		b.hdrs[i].Datalen = 0
+	}
+	n, errno := msgx.Recv(fd, b.hdrs[:])
+	if errno == unix.EAGAIN {
+		r.u.putBatch(b)
+		return false
+	}
+	b.n, r.b, r.errno = n, b, errno
+	return true
+}
+
+// read returns the next batch: up to msgx.Batch datagrams from one recvmsg_x, or one datagram once recvmsg_x is off.
+func (u *StdConn) read(r *batchReader) (*rxBatch, error) {
+	if !noRecvX.Load() {
+		r.b = nil
+		err := r.rc.Read(r.fn)
+		b := r.b
+		switch {
+		case err != nil:
+			if b != nil {
+				u.putBatch(b)
+			}
+			return nil, err
+		case r.errno == 0:
+			perr := b.parse(b.n)
+			if perr == nil {
+				return b, nil
+			}
+			// The batch was read through a layout the kernel no longer writes, so none of it is trustworthy.
+			u.disableRecvX("returned an unexpected header", perr)
+		case msgx.Unsupported(r.errno):
+			u.disableRecvX("unavailable", r.errno)
+		default:
+			u.putBatch(b)
+			return nil, r.errno
+		}
+		u.putBatch(b)
+	}
+	b := u.getBatch()
+	n, from, err := r.conn.ReadFromUDPAddrPort(b.buf[:MTU])
+	if err != nil {
+		u.putBatch(b)
+		return nil, err
+	}
+	b.n, b.pkts[0], b.addrs[0] = 1, b.buf[:n:n], netip.AddrPortFrom(from.Addr().Unmap(), from.Port())
+	return b, nil
+}
+
+func (u *StdConn) disableRecvX(why string, err error) {
+	if noRecvX.CompareAndSwap(false, true) {
+		u.l.Warn("recvmsg_x "+why+", reading one datagram per syscall", "error", err)
+	}
+}
+
+// readListener hands the listener's batches to ListenOut until the listener closes.
+func (u *StdConn) readListener() error {
+	r := u.newBatchReader(u.UDPConn, u.rc)
+	for {
+		b, err := u.read(r)
 		if err != nil {
 			if errors.Is(err, net.ErrClosed) {
 				return err
@@ -262,11 +423,20 @@ func (u *StdConn) ListenOut(r EncReader, flush func()) error {
 			u.l.Error("unexpected udp socket receive error", "error", err)
 			continue
 		}
+		if !u.deliver(b) {
+			return net.ErrClosed
+		}
+	}
+}
 
-		u.readMu.Lock()
-		r(netip.AddrPortFrom(rua.Addr().Unmap(), rua.Port()), buffer[:n:n])
-		flush()
-		u.readMu.Unlock()
+// deliver hands b to ListenOut, and reports false once ListenOut has returned.
+func (u *StdConn) deliver(b *rxBatch) bool {
+	select {
+	case u.rx <- b:
+		return true
+	case <-u.done:
+		u.putBatch(b)
+		return false
 	}
 }
 

udp/udp_recvmsgx_darwin_test.go

@@ -0,0 +1,344 @@
+//go:build !e2e_testing
+
+package udp
+
+import (
+	"bytes"
+	"fmt"
+	"log/slog"
+	"net"
+	"net/netip"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+	"unsafe"
+
+	"github.com/slackhq/nebula/internal/msgx"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sys/unix"
+)
+
+// A burst larger than msgx.Batch arrives whole, with each datagram's own bytes and source address, and
+// flush runs once per recvmsg_x batch rather than once per datagram.
+func TestListenOutBatchesBurst(t *testing.T) {
+	for _, tc := range []struct{ name, listen string }{
+		{"v4", "127.0.0.1"},
+		{"v6", "::1"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			c, err := NewListener(slog.New(slog.DiscardHandler), Settings{Listen: netip.AddrPortFrom(netip.MustParseAddr(tc.listen), 0)})
+			require.NoError(t, err)
+			la, err := c.LocalAddr()
+			require.NoError(t, err)
+
+			tx, err := net.Dial("udp", la.String())
+			require.NoError(t, err)
+			defer tx.Close()
+			txAddr := netip.MustParseAddrPort(tx.LocalAddr().String())
+
+			const total = msgx.Batch*3 + 7
+			var mu sync.Mutex
+			var got []string
+			var froms []netip.AddrPort
+			flushes := 0
+			// Queued before the reader starts, so recvmsg_x sees full batches.
+			for i := 0; i < total; i++ {
+				_, err := tx.Write([]byte(fmt.Sprintf("pkt-%03d", i)))
+				require.NoError(t, err)
+			}
+
+			done := make(chan struct{})
+			go func() {
+				_ = c.ListenOut(func(addr netip.AddrPort, payload []byte) {
+					mu.Lock()
+					got = append(got, string(payload))
+					froms = append(froms, addr)
+					mu.Unlock()
+				}, func() {
+					mu.Lock()
+					flushes++
+					mu.Unlock()
+				})
+				close(done)
+			}()
+
+			require.Eventually(t, func() bool {
+				mu.Lock()
+				defer mu.Unlock()
+				return len(got) == total
+			}, 5*time.Second, 10*time.Millisecond)
+
+			require.NoError(t, c.Close())
+			<-done
+
+			for i, p := range got {
+				assert.Equal(t, fmt.Sprintf("pkt-%03d", i), p)
+				assert.Equal(t, txAddr, froms[i])
+			}
+			assert.Equal(t, (total+msgx.Batch-1)/msgx.Batch, flushes)
+		})
+	}
+}
+
+// Once recvmsg_x is off, every datagram still arrives, one per batch.
+func TestListenOutSingleFallback(t *testing.T) {
+	noRecvX.Store(true)
+	t.Cleanup(func() { noRecvX.Store(false) })
+	c, err := NewListener(slog.New(slog.DiscardHandler), Settings{Listen: netip.MustParseAddrPort("127.0.0.1:0")})
+	require.NoError(t, err)
+	la, err := c.LocalAddr()
+	require.NoError(t, err)
+	tx, err := net.Dial("udp", la.String())
+	require.NoError(t, err)
+	defer tx.Close()
+	txAddr := netip.MustParseAddrPort(tx.LocalAddr().String())
+
+	const total = 5
+	for i := 0; i < total; i++ {
+		_, err := tx.Write([]byte(fmt.Sprintf("pkt-%03d", i)))
+		require.NoError(t, err)
+	}
+
+	var mu sync.Mutex
+	var got []string
+	flushes := 0
+	done := make(chan struct{})
+	go func() {
+		_ = c.ListenOut(func(addr netip.AddrPort, payload []byte) {
+			assert.Equal(t, txAddr, addr)
+			mu.Lock()
+			got = append(got, string(payload))
+			mu.Unlock()
+		}, func() {
+			mu.Lock()
+			flushes++
+			mu.Unlock()
+		})
+		close(done)
+	}()
+
+	require.Eventually(t, func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(got) == total
+	}, 5*time.Second, 10*time.Millisecond)
+	require.NoError(t, c.Close())
+	<-done
+
+	for i, p := range got {
+		assert.Equal(t, fmt.Sprintf("pkt-%03d", i), p)
+	}
+	assert.Equal(t, total, flushes)
+}
+
+func TestRxBatchParse(t *testing.T) {
+	valid := func() *rxBatch {
+		b := (&StdConn{}).getBatch()
+		b.hdrs[0].Namelen, b.hdrs[0].Datalen = unix.SizeofSockaddrInet4, 1
+		b.names[0].Family = unix.AF_INET
+		sa4 := (*unix.RawSockaddrInet4)(unsafe.Pointer(&b.names[0]))
+		sa4.Addr = [4]byte{10, 0, 0, 1}
+		sa4.Port = 0x0110 // 4097 in network order
+		b.hdrs[1].Namelen, b.hdrs[1].Datalen = unix.SizeofSockaddrInet6, MTU
+		b.names[1].Family = unix.AF_INET6
+		b.names[1].Addr = netip.MustParseAddr("::ffff:10.0.0.2").As16()
+		b.names[1].Port = 0x0210
+		return b
+	}
+
+	b := valid()
+	require.NoError(t, b.parse(2))
+	assert.Equal(t, 2, b.n)
+	assert.Equal(t, netip.MustParseAddrPort("10.0.0.1:4097"), b.addrs[0])
+	assert.Equal(t, netip.MustParseAddrPort("10.0.0.2:4098"), b.addrs[1], "v4-mapped sources arrive unmapped")
+	assert.Len(t, b.pkts[0], 1)
+	assert.Equal(t, 1, cap(b.pkts[0]))
+	assert.Len(t, b.pkts[1], MTU)
+	require.NoError(t, valid().parse(0))
+
+	for _, tc := range []struct {
+		name   string
+		n      int
+		mutate func(*rxBatch)
+	}{
+		{"count past headers", msgx.Batch + 1, func(*rxBatch) {}},
+		{"datalen past buffer", 2, func(b *rxBatch) { b.hdrs[1].Datalen = MTU + 1 }},
+		{"unknown family", 2, func(b *rxBatch) { b.names[0].Family = unix.AF_UNIX }},
+		{"family and length disagree", 2, func(b *rxBatch) { b.hdrs[0].Namelen = unix.SizeofSockaddrInet6 }},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			b := valid()
+			tc.mutate(b)
+			assert.Error(t, b.parse(tc.n))
+		})
+	}
+}
+
+func TestMsghdrXKernelRoundTrip(t *testing.T) {
+	for _, v6 := range []bool{false, true} {
+		t.Run(fmt.Sprintf("v6=%v", v6), func(t *testing.T) {
+			// Loopback delivery goes through the dlil input thread, so the settle gives all three time to
+			// queue and come back from one call, which is what exercises the stride.
+			got := recvmsgXRoundTrip(t, v6, []int{1, 300, 1400}, 4, 50*time.Millisecond)
+			assert.Equal(t, 3, got, "largest batch from one recvmsg_x call")
+		})
+	}
+}
+
+// Any burst, of any datagram lengths up to the kernel's maxdgram, into any number of headers, comes back in order
+// with every entry's fields intact, and headers past the returned count untouched.
+func FuzzMsghdrXKernelRoundTrip(f *testing.F) {
+	f.Add(false, uint8(3), []byte{0, 1, 1, 44, 5, 120})
+	f.Add(true, uint8(0), []byte{0, 0, 0x23, 0x29})           // one empty datagram, then one at MTU
+	f.Add(false, uint8(1), []byte{0x23, 0x2a, 0x24, 0x00, 0}) // past MTU: cut to MTU, no MSG_TRUNC
+	f.Add(true, uint8(msgx.Batch-1), bytes.Repeat([]byte{0x05, 0xdc}, msgx.Batch))
+
+	f.Fuzz(func(t *testing.T, v6 bool, spare uint8, sizes []byte) {
+		// Two bytes per datagram, big-endian, folded into [0, maxdgram].
+		var lens []int
+		for i := 0; i+1 < len(sizes) && len(lens) < msgx.Batch; i += 2 {
+			lens = append(lens, (int(sizes[i])<<8|int(sizes[i+1]))%(maxDgram+1))
+		}
+		if len(lens) == 0 {
+			return
+		}
+		recvmsgXRoundTrip(t, v6, lens, 1+int(spare)%msgx.Batch, time.Millisecond)
+	})
+}
+
+// maxDgram is darwin's default net.inet.udp.maxdgram, the largest datagram a loopback sender may write.
+const maxDgram = 9216
+
+// recvmsgXRoundTrip sends one datagram per entry of lens to a fresh loopback socket, drains them with recvmsg_x
+// into nh headers, and checks every field the kernel writes. Neighboring datagrams differ in sender, length and
+// fill byte, so no field can pass by reading a neighbor's value. It returns the largest batch one call returned.
+func recvmsgXRoundTrip(t *testing.T, v6 bool, lens []int, nh int, settle time.Duration) int {
+	t.Helper()
+	network, lo := "udp4", netip.MustParseAddr("127.0.0.1")
+	if v6 {
+		network, lo = "udp6", netip.IPv6Loopback()
+	}
+	rx, err := net.ListenUDP(network, net.UDPAddrFromAddrPort(netip.AddrPortFrom(lo, 0)))
+	require.NoError(t, err)
+	defer rx.Close()
+
+	txs := make([]*net.UDPConn, 3)
+	for i := range txs {
+		txs[i], err = net.DialUDP(network, nil, rx.LocalAddr().(*net.UDPAddr))
+		require.NoError(t, err)
+		defer txs[i].Close()
+	}
+	senders := make([]netip.AddrPort, len(lens))
+	for i, l := range lens {
+		tx := txs[i%len(txs)]
+		senders[i] = tx.LocalAddr().(*net.UDPAddr).AddrPort()
+		_, err := tx.Write(bytes.Repeat([]byte{byte(i + 1)}, l))
+		require.NoError(t, err)
+	}
+	time.Sleep(settle)
+
+	// Namelen goes in as the v6 size on both families, so the v4 size coming back proves the kernel wrote it.
+	const sentinel = 0xa5a5a5a5a5a5a5a5
+	wantNamelen := uint32(unix.SizeofSockaddrInet4)
+	if v6 {
+		wantNamelen = unix.SizeofSockaddrInet6
+	}
+	bufs := make([][]byte, nh)
+	names := make([]unix.RawSockaddrInet6, nh)
+	iovs := make([]unix.Iovec, nh)
+	hdrs := make([]msgx.Hdr, nh)
+	for i := range hdrs {
+		bufs[i] = make([]byte, MTU)
+		iovs[i].Base = &bufs[i][0]
+		iovs[i].SetLen(MTU)
+	}
+
+	// A datagram dropped on loopback would otherwise block rc.Read forever.
+	require.NoError(t, rx.SetReadDeadline(time.Now().Add(5*time.Second)))
+	rc, err := rx.SyscallConn()
+	require.NoError(t, err)
+	largest := 0
+	for next := 0; next < len(lens); {
+		for i := range hdrs {
+			clear(bufs[i])
+			hdrs[i] = msgx.Hdr{
+				Name:    (*byte)(unsafe.Pointer(&names[i])),
+				Namelen: unix.SizeofSockaddrInet6,
+				Iov:     &iovs[i],
+				Iovlen:  1,
+				Flags:   -1,
+				Datalen: sentinel,
+			}
+		}
+		var n int
+		var errno syscall.Errno
+		require.NoError(t, rc.Read(func(fd uintptr) bool {
+			n, errno = msgx.Recv(fd, hdrs)
+			return errno != unix.EAGAIN
+		}), "received %d of %d datagrams", next, len(lens))
+		require.Zero(t, errno)
+		require.Positive(t, n)
+		require.LessOrEqual(t, n, min(nh, len(lens)-next))
+		largest = max(largest, n)
+		b := &rxBatch{buf: make([]byte, n*MTU)}
+		copy(b.hdrs[:], hdrs[:n])
+		copy(b.names[:], names[:n])
+		require.NoError(t, b.parse(n), "ListenOut would fall back on this batch")
+
+		for i := range n {
+			d := next + i
+			h := &hdrs[i]
+			want := min(lens[d], MTU)
+			require.EqualValues(t, want, h.Datalen, "datagram %d Datalen", d)
+			require.Equal(t, wantNamelen, h.Namelen, "datagram %d Namelen", d)
+			require.Equal(t, senders[d], b.addrs[i], "datagram %d sender", d)
+			// Flags goes in as -1 and must come back 0. xnu's list receive (soreceive_m_list) never sets
+			// MSG_TRUNC, so a datagram past MTU arrives silently cut to MTU, as recvfrom through net.UDPConn does.
+			require.Zero(t, h.Flags, "datagram %d Flags", d)
+			require.Equal(t, bytes.Repeat([]byte{byte(d + 1)}, want), bufs[i][:want], "datagram %d payload", d)
+			if want < MTU {
+				require.Zero(t, bufs[i][want], "datagram %d wrote past Datalen", d)
+			}
+		}
+		for i := n; i < nh; i++ {
+			require.EqualValues(t, uint64(sentinel), hdrs[i].Datalen, "spare header %d Datalen", i)
+			require.EqualValues(t, unix.SizeofSockaddrInet6, hdrs[i].Namelen, "spare header %d Namelen", i)
+		}
+		next += n
+	}
+	return largest
+}
+
+// Once a listener has batches to reuse, reading one allocates nothing, on recvmsg_x or on the fallback.
+func TestReadDoesNotAllocate(t *testing.T) {
+	for _, single := range []bool{false, true} {
+		t.Run(fmt.Sprintf("single=%v", single), func(t *testing.T) {
+			noRecvX.Store(single)
+			t.Cleanup(func() { noRecvX.Store(false) })
+			c, err := NewListener(slog.New(slog.DiscardHandler), Settings{Listen: netip.MustParseAddrPort("127.0.0.1:0")})
+			require.NoError(t, err)
+			t.Cleanup(func() { _ = c.Close() })
+			u := c.(*StdConn)
+			la, err := c.LocalAddr()
+			require.NoError(t, err)
+			tx, err := net.DialUDP("udp4", nil, net.UDPAddrFromAddrPort(la))
+			require.NoError(t, err)
+			t.Cleanup(func() { _ = tx.Close() })
+			r := u.newBatchReader(u.UDPConn, u.rc)
+			pkt := []byte("steady")
+			allocs := testing.AllocsPerRun(100, func() {
+				if _, err := tx.Write(pkt); err != nil {
+					t.Fatal(err)
+				}
+				b, err := u.read(r)
+				if err != nil || b.n != 1 {
+					t.Fatalf("read %v: %v", b, err)
+				}
+				u.putBatch(b)
+			})
+			assert.Zero(t, allocs)
+		})
+	}
+}