diff --git a/overlay/batch/multi_coalesce_test.go b/overlay/batch/multi_coalesce_test.go index c40b739a..e55bbbfc 100644 --- a/overlay/batch/multi_coalesce_test.go +++ b/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)) + } +} diff --git a/overlay/batch/passthrough.go b/overlay/batch/passthrough.go index 3d3fc4de..740f18f1 100644 --- a/overlay/batch/passthrough.go +++ b/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) diff --git a/overlay/tio/single.go b/overlay/tio/single.go index a24c2242..61960c70 100644 --- a/overlay/tio/single.go +++ b/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) +} diff --git a/overlay/tio/tio.go b/overlay/tio/tio.go index c3c6b59d..a54659c0 100644 --- a/overlay/tio/tio.go +++ b/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 +} diff --git a/overlay/tun_darwin.go b/overlay/tun_darwin.go index c086eda2..23c43a79 100644 --- a/overlay/tun_darwin.go +++ b/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 } diff --git a/overlay/tun_darwin_test.go b/overlay/tun_darwin_test.go new file mode 100644 index 00000000..0795719f --- /dev/null +++ b/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) + }) + }) + } +} diff --git a/udp/udp_darwin.go b/udp/udp_darwin.go index 6b351bf8..e20065d9 100644 --- a/udp/udp_darwin.go +++ b/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