mirror of
https://github.com/slackhq/nebula.git
synced 2026-11-15 12:26:52 +01:00
overlay/darwin: read the utun with recvmsg_x
tunQueue.Read now takes up to 64 queued packets with one recvmsg_x through internal/msgx, each into its own 64KiB slot of a buffer allocated once per queue, and falls back to the readv drain for good when recvmsg_x is refused or returns a batch intact msghdr_x entries can't produce, or for one Read on a transient error. Ported from the pipeline bench build's tunRecvX, minus its env toggle and stats. With 64 pending packets on the utun this is the remaining send gain: on an M2 mini, one stream, pipeline + 64 pending reads 4.9 Gbps with recvmsg_x and 2.6 with the readv loop, the same as this branch before this commit. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
64bec56404
commit
21a7f1b711
2 files changed
+442
-37
No files matched your search
+138
-11
@@ -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
|
||||
|
||||
+304
-26
@@ -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
|
||||
|
||||
Reference in new issue
Block a user