diff --git a/internal/msgx/msgx_darwin.go b/internal/msgx/msgx_darwin.go new file mode 100644 index 00000000..4f558d50 --- /dev/null +++ b/internal/msgx/msgx_darwin.go @@ -0,0 +1,58 @@ +// Package msgx calls xnu's private batched socket calls, sendmsg_x and recvmsg_x, which move many datagrams per +// syscall. They are not public API, so a kernel or sandbox may refuse them; callers fall back to one datagram per +// syscall when Unsupported reports that. +package msgx + +import ( + "syscall" + "unsafe" + + "golang.org/x/sys/unix" +) + +// Batch is the most messages the callers hand one sendmsg_x or recvmsg_x call. +const Batch = 64 + +// Hdr mirrors xnu's struct msghdr_x (bsd/sys/socket_private.h), which the SDK doesn't ship. recvmsg_x reports each +// message's length in Datalen rather than in its return value. +type Hdr struct { + Name *byte + Namelen uint32 + Iov *unix.Iovec + Iovlen int32 + Control *byte + Controllen uint32 + Flags int32 + Datalen uint64 +} + +// Send passes hdrs to sendmsg_x on fd and returns how many messages the kernel took. EINTR is retried. +func Send(fd uintptr, hdrs []Hdr) (int, syscall.Errno) { + return call(unix.SYS_SENDMSG_X, fd, hdrs) +} + +// Recv passes hdrs to recvmsg_x on fd and returns how many messages it filled. EINTR is retried. +func Recv(fd uintptr, hdrs []Hdr) (int, syscall.Errno) { + return call(unix.SYS_RECVMSG_X, fd, hdrs) +} + +func call(trap, fd uintptr, hdrs []Hdr) (int, syscall.Errno) { + if len(hdrs) == 0 { + return 0, syscall.EINVAL + } + for { + r, _, errno := unix.Syscall6(trap, fd, uintptr(unsafe.Pointer(&hdrs[0])), uintptr(len(hdrs)), 0, 0, 0) + switch errno { + case 0: + return int(r), 0 + case syscall.EINTR: + default: + return 0, errno + } + } +} + +// Unsupported reports whether errno means the kernel or a sandbox refuses the private calls. +func Unsupported(errno syscall.Errno) bool { + return errno == syscall.ENOSYS || errno == syscall.EPERM || errno == syscall.EOPNOTSUPP +} diff --git a/internal/msgx/msgx_darwin_test.go b/internal/msgx/msgx_darwin_test.go new file mode 100644 index 00000000..70fe1a67 --- /dev/null +++ b/internal/msgx/msgx_darwin_test.go @@ -0,0 +1,79 @@ +package msgx + +import ( + "bytes" + "syscall" + "testing" + "unsafe" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "golang.org/x/sys/unix" +) + +// TestHdrLayout pins Hdr to struct msghdr_x's LP64 layout: 56 bytes, Datalen last at 48. +func TestHdrLayout(t *testing.T) { + var h Hdr + assert.EqualValues(t, 56, unsafe.Sizeof(h)) + assert.EqualValues(t, 48, unsafe.Offsetof(h.Datalen)) +} + +// TestSendRecvRoundTrip sends a batch through sendmsg_x in one call and reads it back through recvmsg_x in one +// call, each message whole, in order, with Datalen set. +func TestSendRecvRoundTrip(t *testing.T) { + fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_DGRAM, 0) + require.NoError(t, err) + t.Cleanup(func() { + _ = unix.Close(fds[0]) + _ = unix.Close(fds[1]) + }) + // A dropped datagram fails the receive instead of hanging it. + require.NoError(t, unix.SetsockoptTimeval(fds[1], unix.SOL_SOCKET, unix.SO_RCVTIMEO, &unix.Timeval{Sec: 2})) + + const n = 5 + var want [n][]byte + var iovs [n]unix.Iovec + hdrs := make([]Hdr, n) + for i := range n { + want[i] = bytes.Repeat([]byte{byte(i + 1)}, 10+i) + iovs[i] = unix.Iovec{Base: &want[i][0]} + iovs[i].SetLen(len(want[i])) + hdrs[i] = Hdr{Iov: &iovs[i], Iovlen: 1} + } + sent, errno := Send(uintptr(fds[0]), hdrs) + require.Zero(t, errno) + require.Equal(t, n, sent) + + var bufs [n + 1][64]byte + var riovs [n + 1]unix.Iovec + rhdrs := make([]Hdr, n+1) + for i := range rhdrs { + riovs[i] = unix.Iovec{Base: &bufs[i][0]} + riovs[i].SetLen(len(bufs[i])) + rhdrs[i] = Hdr{Iov: &riovs[i], Iovlen: 1} + } + got, errno := Recv(uintptr(fds[1]), rhdrs) + require.Zero(t, errno) + require.Equal(t, n, got) + for i := range n { + l := int(rhdrs[i].Datalen) + assert.Equal(t, want[i], bufs[i][:l], "message %d", i) + } +} + +// TestEmptyBatch pins that an empty batch is refused before it reaches the kernel. +func TestEmptyBatch(t *testing.T) { + _, errno := Send(0, nil) + assert.Equal(t, syscall.EINVAL, errno) + _, errno = Recv(0, nil) + assert.Equal(t, syscall.EINVAL, errno) +} + +func TestUnsupported(t *testing.T) { + for _, e := range []syscall.Errno{syscall.ENOSYS, syscall.EPERM, syscall.EOPNOTSUPP} { + assert.True(t, Unsupported(e), "%v", e) + } + for _, e := range []syscall.Errno{0, syscall.EAGAIN, syscall.ENOBUFS, syscall.EINVAL} { + assert.False(t, Unsupported(e), "%v", e) + } +} diff --git a/udp/connsock_bench_darwin_test.go b/udp/connsock_bench_darwin_test.go index a35003b1..cafe83fd 100644 --- a/udp/connsock_bench_darwin_test.go +++ b/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") diff --git a/udp/connsock_darwin.go b/udp/connsock_darwin.go index e4c9fb62..d74d84ab 100644 --- a/udp/connsock_darwin.go +++ b/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 + } } } diff --git a/udp/connsock_darwin_test.go b/udp/connsock_darwin_test.go index f568dd67..6b3d3e21 100644 --- a/udp/connsock_darwin_test.go +++ b/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") diff --git a/udp/udp_darwin.go b/udp/udp_darwin.go index fad4ca5a..7e15d152 100644 --- a/udp/udp_darwin.go +++ b/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 } } diff --git a/udp/udp_recvmsgx_darwin_test.go b/udp/udp_recvmsgx_darwin_test.go new file mode 100644 index 00000000..3a2773b4 --- /dev/null +++ b/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) + }) + } +}