diff --git a/overlay/tun_darwin.go b/overlay/tun_darwin.go index b7592ddd..01b93ae2 100644 --- a/overlay/tun_darwin.go +++ b/overlay/tun_darwin.go @@ -802,28 +802,155 @@ func (t *tun) Name() string { } func (t *tun) Queues(int) ([]tio.Queue, error) { - t.raisePendingPackets() - return []tio.Queue{&tunQueue{t: t, buf: make([]byte, tunReadArena)}}, nil + return []tio.Queue{newTunQueue(t)}, nil } // tunReadBatch is the most packets one tunQueue.Read returns. -const tunReadBatch = 64 +const tunReadBatch = msgx.Batch -// tunReadArena is tunQueue's receive buffer. Draining stops once less than defaultBatchBufSize of it is left, so -// every readv has room for the largest packet utun can return at any device MTU. -const tunReadArena = 4 * defaultBatchBufSize +// tunRecvXSlot is each recvmsg_x entry's packet buffer. recvmsg_x lays its buffers out before it knows the packet +// sizes, so each gets room for the largest IP packet; the kernel only touches what it writes. +const tunRecvXSlot = defaultBatchBufSize -// tunQueue is the darwin tun's Queue. Read drains every packet already queued on the utun, up to tunReadBatch, so -// the caller encrypts and sends them as one batch rather than one per wakeup. +// tunRecvXStride is how far apart the slots start: 64 byte aligned, and one cache line more than 64KiB so the packet +// starts don't all land in the same cache sets. +const tunRecvXStride = 65536 + 64 + +// tunQueue is the darwin tun's Queue. Read hands back every packet already queued on the utun, up to tunReadBatch, so +// the caller encrypts and sends them as one batch rather than one per wakeup. It reads them with one recvmsg_x when +// it can and one readv per packet otherwise. type tunQueue struct { - t *tun + t *tun + // buf holds the packets Read returns: recvmsg_x's slots, or, for readv, one arena. readv stops once less than + // defaultBatchBufSize of it is left, so every readv has room for the largest packet utun can return. buf []byte pkts [tunReadBatch]tio.Packet + // x is the recvmsg_x scratch, nil once recvmsg_x is off for good. + x *tunRecvX } -// Read waits for the utun to become readable, then reads packets until it would block. An error after at least one -// packet is dropped in favor of returning those packets; a persistent one recurs on the next Read. +// tunRecvX is the recvmsg_x scratch. Entry i scatters its AF prefix into heads[i] and its packet into slot i of the +// queue's buf; the kernel reads and writes every entry in place. +type tunRecvX struct { + heads [tunReadBatch][4]byte + iovs [tunReadBatch][2]unix.Iovec + hdrs [tunReadBatch]msgx.Hdr +} + +// newTunQueue returns t's Queue, raising t's pending packet limit first so a read can find more than one. +func newTunQueue(t *tun) *tunQueue { + t.raisePendingPackets() + q := &tunQueue{t: t, buf: make([]byte, tunReadBatch*tunRecvXStride), x: &tunRecvX{}} + x := q.x + for i := range x.hdrs { + x.iovs[i] = [2]unix.Iovec{ + {Base: &x.heads[i][0], Len: 4}, + {Base: &q.buf[i*tunRecvXStride], Len: tunRecvXSlot}, + } + x.hdrs[i] = msgx.Hdr{Iov: &x.iovs[i][0], Iovlen: 2} + } + return q +} + +// Read waits for the utun to become readable, then returns the packets queued on it. func (q *tunQueue) Read() ([]tio.Packet, error) { + if q.x != nil { + if pkts, err, ok := q.readX(); ok { + return pkts, err + } + } + return q.readv() +} + +// tunRecvmsgX is msgx.Recv; tests swap it to inject results. +var tunRecvmsgX = msgx.Recv + +// readX reads up to tunReadBatch packets with one recvmsg_x. ok is false when the caller should read with readv +// instead: recvmsg_x failed this time, or has been turned off for good. A batch that fails tunRecvXCheck is dropped: +// its packets are off the socket, but their lengths can't be trusted. +func (q *tunQueue) readX() (pkts []tio.Packet, err error, ok bool) { + rc, err := q.t.f.SyscallConn() + if err != nil { + return nil, err, true + } + + x := q.x + var n int + var errno syscall.Errno + err = rc.Read(func(fd uintptr) bool { + for i := range x.hdrs { + x.hdrs[i].Flags = 0 + x.hdrs[i].Datalen = 0 + } + n, errno = tunRecvmsgX(fd, x.hdrs[:]) + return errno != unix.EAGAIN + }) + if err != nil { + return nil, err, true + } + if tunRecvXRefused(errno) { + q.disableX("unavailable", errno) + return nil, nil, false + } + if errno != 0 { + // recvmsg_x can fail where readv wouldn't (a kernel allocation, say), so let readv read this wakeup and + // report the error if it is real. + return nil, nil, false + } + if cerr := tunRecvXCheck(n, x.hdrs[:], x.heads[:], tunRecvXSlot); cerr != nil { + q.disableX("returned an unexpected batch", cerr) + return nil, nil, false + } + for i := range n { + off := i * tunRecvXStride + end := off + int(x.hdrs[i].Datalen) - 4 + q.pkts[i] = tio.Packet{Bytes: q.buf[off:end:end]} + } + // n is 0 at end of file, as readv's short read is. + return q.pkts[:n], nil, true +} + +// tunRecvXRefused reports whether a recvmsg_x errno means it will never work here: msgx.Unsupported's refusals, and +// EINVAL and EMSGSIZE, which with the headers newTunQueue builds come only from xnu rejecting their layout. +func tunRecvXRefused(errno syscall.Errno) bool { + return msgx.Unsupported(errno) || errno == unix.EINVAL || errno == unix.EMSGSIZE +} + +// disableX turns recvmsg_x off for q for good. readv keeps reading into buf. +func (q *tunQueue) disableX(why string, err error) { + q.x = nil + q.t.l.Warn("recvmsg_x "+why+" on the tun device, reading one packet per syscall", "error", err) +} + +// tunRecvXCheck rejects a recvmsg_x batch of n entries that intact msghdr_x entries can't produce, which would mean +// the private layout changed under us: more entries than headers, one past its 4 byte head plus slot byte buffer or +// too short to hold the head, or an AF prefix utun never writes. A slot holds the largest IP packet, so nothing is +// ever truncated; xnu's default recvmsg_x path doesn't report MSG_TRUNC per entry anyway, only its SO_DONTTRUNC path +// does, and that is checked too. +func tunRecvXCheck(n int, hdrs []msgx.Hdr, heads [][4]byte, slot int) error { + if n < 0 || n > len(hdrs) || n > len(heads) { + return fmt.Errorf("returned %d packets for %d headers", n, len(hdrs)) + } + for i := range n { + h := &hdrs[i] + if h.Flags&unix.MSG_TRUNC != 0 { + return fmt.Errorf("packet %d came back truncated, flags %#x", i, h.Flags) + } + if h.Datalen < 4 || h.Datalen > uint64(4+slot) { + return fmt.Errorf("packet %d is %d bytes, outside the 4 to %d the buffers allow", i, h.Datalen, 4+slot) + } + switch heads[i] { + case [4]byte{3: syscall.AF_INET}, [4]byte{3: syscall.AF_INET6}: + default: + return fmt.Errorf("packet %d has AF prefix % x", i, heads[i]) + } + } + return nil +} + +// readv reads packets one readv at a time until the utun would block. An error after at least one packet is dropped +// in favor of returning those packets; a persistent one recurs on the next Read. +func (q *tunQueue) readv() ([]tio.Packet, error) { rc, err := q.t.f.SyscallConn() if err != nil { return nil, err diff --git a/overlay/tun_darwin_test.go b/overlay/tun_darwin_test.go index 126c2080..c306c432 100644 --- a/overlay/tun_darwin_test.go +++ b/overlay/tun_darwin_test.go @@ -5,8 +5,10 @@ package overlay import ( "bytes" "os" + "sync/atomic" "syscall" "testing" + "time" "github.com/slackhq/nebula/internal/msgx" "github.com/slackhq/nebula/test" @@ -154,28 +156,59 @@ func TestTunWriteBatchOversize(t *testing.T) { } } -// TestTunQueueReadDrains pins that Read hands back every queued packet, AF prefix stripped and in -// order, capped at tunReadBatch per call, and doesn't wait for more once the queue is empty. -func TestTunQueueReadDrains(t *testing.T) { +// forEachTunReadPath runs fn once over recvmsg_x and once over the per-packet readv drain. +func forEachTunReadPath(t *testing.T, fn func(t *testing.T, newQ func(*tun) *tunQueue)) { + for _, recvX := range []bool{true, false} { + name := "recvmsg_x" + if !recvX { + name = "readv" + } + t.Run(name, func(t *testing.T) { + fn(t, func(tn *tun) *tunQueue { + q := newTunQueue(tn) + if !recvX { + q.x = nil + } + return q + }) + }) + } +} + +// writeTunPkts writes n test packets of size bytes to the peer end w and returns the packets the tun should hand back +// for them. +func writeTunPkts(t *testing.T, w, n, size int) [][]byte { + t.Helper() + var want [][]byte + for i := range n { + pkt, wire := testTunPkt(i, size+i) + _, err := unix.Write(w, wire) + require.NoError(t, err) + want = append(want, pkt) + } + return want +} + +// newReadTun is newSocketpairTun with the tun's receive buffer big enough for every test burst, since an AF_UNIX +// datagram send is bounded by the receiver's buffer, and the writer's send buffer big enough for the largest IP packet. +func newReadTun(t *testing.T) (*tun, int) { + t.Helper() tn, w := newSocketpairTun(t, 1<<20) - // An AF_UNIX datagram send is bounded by the receiver's buffer, here the tun's end. + require.NoError(t, unix.SetsockoptInt(w, unix.SOL_SOCKET, unix.SO_SNDBUF, 1<<18)) rc, err := tn.f.SyscallConn() require.NoError(t, err) require.NoError(t, rc.Control(func(fd uintptr) { require.NoError(t, unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUF, 1<<20)) })) - q := &tunQueue{t: tn, buf: make([]byte, tunReadArena)} - total := tunReadBatch + 3 - var want [][]byte - for i := range total { - pkt, wire := testTunPkt(i, 60+i) - _, err := unix.Write(w, wire) - require.NoError(t, err) - want = append(want, pkt) - } + return tn, w +} +// readTunPkts calls q.Read once per entry of sizes, requires each to return that many packets, and returns copies of +// them all. +func readTunPkts(t *testing.T, q *tunQueue, sizes ...int) [][]byte { + t.Helper() var got [][]byte - for _, size := range []int{tunReadBatch, 3} { + for _, size := range sizes { pkts, err := q.Read() require.NoError(t, err) require.Len(t, pkts, size) @@ -183,23 +216,268 @@ func TestTunQueueReadDrains(t *testing.T) { got = append(got, p.Clone().Bytes) } } - assert.Equal(t, want, got) + return got } -// TestTunQueueReadClipsPackets pins that each returned packet's capacity ends at its own bytes, so -// appending to one can't overwrite the next. +// stubTunRecvmsgX swaps tunRecvmsgX for fn until the test ends and counts its calls. fn gets the real msgx.Recv to +// delegate to. +func stubTunRecvmsgX(t *testing.T, fn func(sys func(uintptr, []msgx.Hdr) (int, syscall.Errno), fd uintptr, hdrs []msgx.Hdr) (int, syscall.Errno)) *atomic.Int64 { + t.Helper() + sys := tunRecvmsgX + var calls atomic.Int64 + tunRecvmsgX = func(fd uintptr, hdrs []msgx.Hdr) (int, syscall.Errno) { + calls.Add(1) + return fn(sys, fd, hdrs) + } + t.Cleanup(func() { tunRecvmsgX = sys }) + return &calls +} + +// TestTunQueueReadDrains pins that Read hands back every queued packet, AF prefix stripped and in order, capped at +// tunReadBatch per call, and doesn't wait for more once the queue is empty. +func TestTunQueueReadDrains(t *testing.T) { + forEachTunReadPath(t, func(t *testing.T, newQ func(*tun) *tunQueue) { + tn, w := newReadTun(t) + q := newQ(tn) + want := writeTunPkts(t, w, tunReadBatch+3, 60) + assert.Equal(t, want, readTunPkts(t, q, tunReadBatch, 3)) + }) +} + +// TestTunQueueReadUsesRecvX pins that a burst is read with one recvmsg_x, not one call per packet. +func TestTunQueueReadUsesRecvX(t *testing.T) { + calls := stubTunRecvmsgX(t, func(sys func(uintptr, []msgx.Hdr) (int, syscall.Errno), fd uintptr, hdrs []msgx.Hdr) (int, syscall.Errno) { + return sys(fd, hdrs) + }) + tn, w := newReadTun(t) + q := newTunQueue(tn) + want := writeTunPkts(t, w, 10, 60) + assert.Equal(t, want, readTunPkts(t, q, 10)) + assert.EqualValues(t, 1, calls.Load()) + assert.NotNil(t, q.x) +} + +// TestTunQueueReadLargePackets pins that packets up to the largest IP packet come back whole, which on the recvmsg_x +// path means each lands in its own slot and none spills into the next. +func TestTunQueueReadLargePackets(t *testing.T) { + forEachTunReadPath(t, func(t *testing.T, newQ func(*tun) *tunQueue) { + tn, w := newReadTun(t) + q := newQ(tn) + var want [][]byte + for i, size := range []int{65535, 1, 9001, 65535} { + pkt, wire := testTunPkt(i, max(size, 3)) + pkt, wire = pkt[:size], wire[:4+size] + _, err := unix.Write(w, wire) + require.NoError(t, err) + want = append(want, pkt) + } + assert.Equal(t, want, readTunPkts(t, q, 4)) + }) +} + +// TestTunQueueReadClipsPackets pins that each returned packet's capacity ends at its own bytes, so appending to one +// can't overwrite the next. func TestTunQueueReadClipsPackets(t *testing.T) { - tn, w := newSocketpairTun(t, 1<<20) - q := &tunQueue{t: tn, buf: make([]byte, tunReadArena)} - for i := range 2 { - _, wire := testTunPkt(i, 100) - _, err := unix.Write(w, wire) + forEachTunReadPath(t, func(t *testing.T, newQ func(*tun) *tunQueue) { + tn, w := newSocketpairTun(t, 1<<20) + q := newQ(tn) + writeTunPkts(t, w, 2, 100) + pkts, err := q.Read() require.NoError(t, err) + require.Len(t, pkts, 2) + assert.Equal(t, len(pkts[0].Bytes), cap(pkts[0].Bytes)) + assert.Equal(t, len(pkts[1].Bytes), cap(pkts[1].Bytes)) + }) +} + +// TestTunQueueReadDoesNotAllocate pins both read paths to zero allocations per Read once the queue exists. +func TestTunQueueReadDoesNotAllocate(t *testing.T) { + forEachTunReadPath(t, func(t *testing.T, newQ func(*tun) *tunQueue) { + tn, w := newReadTun(t) + q := newQ(tn) + _, wire := testTunPkt(0, 1300) + allocs := testing.AllocsPerRun(1000, func() { + if _, err := unix.Write(w, wire); err != nil { + t.Fatal(err) + } + if pkts, err := q.Read(); err != nil || len(pkts) != 1 { + t.Fatalf("read %d packets: %v", len(pkts), err) + } + }) + assert.Zero(t, allocs) + }) +} + +// TestTunQueueReadXRefused pins that recvmsg_x being refused turns it off for good, and that the same Read still +// hands back the queued packets through readv. +func TestTunQueueReadXRefused(t *testing.T) { + for _, errno := range []syscall.Errno{unix.ENOSYS, unix.EPERM, unix.EOPNOTSUPP, unix.EINVAL} { + t.Run(errno.Error(), func(t *testing.T) { + calls := stubTunRecvmsgX(t, func(func(uintptr, []msgx.Hdr) (int, syscall.Errno), uintptr, []msgx.Hdr) (int, syscall.Errno) { + return 0, errno + }) + tn, w := newReadTun(t) + q := newTunQueue(tn) + want := writeTunPkts(t, w, 5, 60) + assert.Equal(t, want, readTunPkts(t, q, 5)) + assert.Nil(t, q.x) + assert.EqualValues(t, 1, calls.Load()) + + want = writeTunPkts(t, w, 2, 60) + assert.Equal(t, want, readTunPkts(t, q, 2)) + assert.EqualValues(t, 1, calls.Load(), "recvmsg_x was tried again after it was refused") + }) + } +} + +// TestTunQueueReadXTransientError pins that an error from recvmsg_x that doesn't mean it is refused leaves this Read +// to readv but keeps recvmsg_x on for the next. +func TestTunQueueReadXTransientError(t *testing.T) { + var failed atomic.Bool + calls := stubTunRecvmsgX(t, func(sys func(uintptr, []msgx.Hdr) (int, syscall.Errno), fd uintptr, hdrs []msgx.Hdr) (int, syscall.Errno) { + if !failed.Swap(true) { + return 0, unix.ENOBUFS + } + return sys(fd, hdrs) + }) + tn, w := newReadTun(t) + q := newTunQueue(tn) + want := writeTunPkts(t, w, 3, 60) + assert.Equal(t, want, readTunPkts(t, q, 3)) + assert.NotNil(t, q.x) + assert.EqualValues(t, 1, calls.Load()) + + want = writeTunPkts(t, w, 2, 60) + assert.Equal(t, want, readTunPkts(t, q, 2)) + assert.EqualValues(t, 2, calls.Load(), "recvmsg_x wasn't tried again after a transient error") +} + +// TestTunQueueReadXParks pins that EAGAIN from recvmsg_x parks the Read until the utun is readable, rather than +// falling back or returning an empty batch. +func TestTunQueueReadXParks(t *testing.T) { + var failed atomic.Bool + calls := stubTunRecvmsgX(t, func(sys func(uintptr, []msgx.Hdr) (int, syscall.Errno), fd uintptr, hdrs []msgx.Hdr) (int, syscall.Errno) { + if !failed.Swap(true) { + return 0, unix.EAGAIN + } + return sys(fd, hdrs) + }) + tn, w := newReadTun(t) + q := newTunQueue(tn) + // A real EAGAIN means nothing is queued, so the packet arrives only once Read has parked. One packet, so the + // wakeup can't split it across Reads. + done := make(chan struct{}) + go func() { + defer close(done) + time.Sleep(50 * time.Millisecond) + _, wire := testTunPkt(0, 60) + _, _ = unix.Write(w, wire) + }() + t.Cleanup(func() { <-done }) + pkt, _ := testTunPkt(0, 60) + assert.Equal(t, [][]byte{pkt}, readTunPkts(t, q, 1)) + assert.NotNil(t, q.x) + // A spurious wakeup costs another EAGAIN, so only a floor. + assert.GreaterOrEqual(t, calls.Load(), int64(2)) +} + +// TestTunQueueReadXBadBatch pins that a recvmsg_x batch no intact header could produce turns recvmsg_x off for good +// and hands the Read to readv. The stub reports a bad batch without taking any packets, so readv then reads them all. +func TestTunQueueReadXBadBatch(t *testing.T) { + v4 := [4]byte{3: syscall.AF_INET} + for _, tc := range []struct { + name string + mess func(x *tunRecvX) int + }{ + {"too many", func(x *tunRecvX) int { + for i := range x.hdrs { + x.hdrs[i].Datalen, x.heads[i] = 64, v4 + } + return len(x.hdrs) + 1 + }}, + {"past the buffer", func(x *tunRecvX) int { + x.hdrs[0].Datalen, x.heads[0] = 4+tunRecvXSlot+1, v4 + return 1 + }}, + {"bad AF prefix", func(x *tunRecvX) int { + x.hdrs[0].Datalen, x.heads[0] = 64, [4]byte{syscall.AF_INET} + return 1 + }}, + } { + t.Run(tc.name, func(t *testing.T) { + var q *tunQueue + stubTunRecvmsgX(t, func(func(uintptr, []msgx.Hdr) (int, syscall.Errno), uintptr, []msgx.Hdr) (int, syscall.Errno) { + return tc.mess(q.x), 0 + }) + tn, w := newReadTun(t) + q = newTunQueue(tn) + want := writeTunPkts(t, w, 4, 60) + assert.Equal(t, want, readTunPkts(t, q, 4)) + assert.Nil(t, q.x) + }) + } +} + +// TestTunRecvXCheck pins which recvmsg_x batches tunRecvXCheck accepts. +func TestTunRecvXCheck(t *testing.T) { + const slot = 1000 + v4, v6 := [4]byte{3: syscall.AF_INET}, [4]byte{3: syscall.AF_INET6} + good := func() ([]msgx.Hdr, [][4]byte) { + hdrs := make([]msgx.Hdr, 4) + heads := make([][4]byte, 4) + for i := range hdrs { + hdrs[i].Datalen = uint64(4 + 20 + i) + heads[i] = v4 + if i%2 == 1 { + heads[i] = v6 + } + } + return hdrs, heads + } + for _, tc := range []struct { + name string + n int + mess func(hdrs []msgx.Hdr, heads [][4]byte) + ok bool + }{ + {"all", 4, nil, true}, + {"none", 0, nil, true}, + {"prefix", 2, nil, true}, + {"garbage past n is ignored", 2, func(h []msgx.Hdr, hd [][4]byte) { h[3].Datalen, hd[3] = 1<<40, [4]byte{9} }, true}, + {"just the AF prefix", 1, func(h []msgx.Hdr, _ [][4]byte) { h[0].Datalen = 4 }, true}, + {"fills the slot", 1, func(h []msgx.Hdr, _ [][4]byte) { h[0].Datalen = 4 + slot }, true}, + {"more than headers", 5, nil, false}, + {"negative", -1, nil, false}, + {"past the slot", 2, func(h []msgx.Hdr, _ [][4]byte) { h[1].Datalen = 4 + slot + 1 }, false}, + {"shorter than the AF prefix", 1, func(h []msgx.Hdr, _ [][4]byte) { h[0].Datalen = 3 }, false}, + {"truncated", 3, func(h []msgx.Hdr, _ [][4]byte) { h[2].Flags = unix.MSG_TRUNC }, false}, + {"zero AF prefix", 1, func(_ []msgx.Hdr, hd [][4]byte) { hd[0] = [4]byte{} }, false}, + {"host order AF prefix", 1, func(_ []msgx.Hdr, hd [][4]byte) { hd[0] = [4]byte{syscall.AF_INET} }, false}, + {"other AF", 2, func(_ []msgx.Hdr, hd [][4]byte) { hd[1] = [4]byte{3: syscall.AF_UNIX} }, false}, + } { + t.Run(tc.name, func(t *testing.T) { + hdrs, heads := good() + if tc.mess != nil { + tc.mess(hdrs, heads) + } + err := tunRecvXCheck(tc.n, hdrs, heads, slot) + if tc.ok { + assert.NoError(t, err) + } else { + assert.Error(t, err) + } + }) + } +} + +// TestTunRecvXRefused pins which recvmsg_x errnos turn it off for good. +func TestTunRecvXRefused(t *testing.T) { + for _, errno := range []syscall.Errno{unix.ENOSYS, unix.EPERM, unix.EOPNOTSUPP, unix.EINVAL, unix.EMSGSIZE} { + assert.True(t, tunRecvXRefused(errno), "%v", errno) + } + for _, errno := range []syscall.Errno{0, unix.EAGAIN, unix.EINTR, unix.ENOBUFS, unix.ENOMEM, unix.EBADF, unix.ENOTCONN} { + assert.False(t, tunRecvXRefused(errno), "%v", errno) } - pkts, err := q.Read() - require.NoError(t, err) - require.Len(t, pkts, 2) - assert.Equal(t, len(pkts[0].Bytes), cap(pkts[0].Bytes)) } // TestTunRaisePendingPackets pins that the pending limit is set at SYSPROTO_CONTROL level after the receive buffer is