From 81f991f1fdfff13ab0b7293319e12df9b2d03d01 Mon Sep 17 00:00:00 2001 From: Ryan Huber Date: Wed, 30 Sep 2026 10:56:35 +0000 Subject: [PATCH] overlay/darwin: batch tun writes with sendmsg_x, drain reads, 64 pending packets tun.WriteBatch writes up to 64 packets per sendmsg_x on the utun control socket, and Passthrough.Flush hands its lane to it whole through the new tio.BatchWriter (Caleb Jasik's #1896). Only the darwin tun implements BatchWriter, so every other platform keeps one Write per packet. tunQueue.Read drains every packet already queued on the utun, up to 64, so a burst reaches the send batch together. xnu lets only one packet wait on a utun control socket by default, so each read found one packet. The tun now raises UTUN_OPT_MAX_PENDING_PACKETS to 64, growing SO_RCVBUF to hold that many packets first; on failure it logs and keeps xnu's default. Tests: #1896's WriteBatch tests over sendmsg_x and the writev fallback, the drain tests, the pending-packet setsockopts, and #1906's framing and zero-alloc Read/Write tests. Co-Authored-By: Claude Opus 5.5 --- overlay/batch/multi_coalesce_test.go | 62 +++++ overlay/batch/passthrough.go | 23 +- overlay/tio/tio.go | 11 + overlay/tun_darwin.go | 286 ++++++++++++++++++++- overlay/tun_darwin_test.go | 359 +++++++++++++++++++++++++++ 5 files changed, 726 insertions(+), 15 deletions(-) create mode 100644 overlay/tun_darwin_test.go 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/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..b7592ddd 100644 --- a/overlay/tun_darwin.go +++ b/overlay/tun_darwin.go @@ -8,12 +8,14 @@ import ( "log/slog" "net/netip" "os" + "sync" "sync/atomic" "syscall" "unsafe" "github.com/gaissmai/bart" "github.com/slackhq/nebula/config" + "github.com/slackhq/nebula/internal/msgx" "github.com/slackhq/nebula/overlay/tio" "github.com/slackhq/nebula/routing" "github.com/slackhq/nebula/util" @@ -33,8 +35,23 @@ type tun struct { // and routes for it. NEPacketTunnelProvider on darwin does this. hostOwned bool l *slog.Logger + + // batchMu guards batch, WriteBatch's sendmsg_x scratch. + batchMu sync.Mutex + batch tunBatch } +// tunBatch is WriteBatch's scratch. The kernel reads every entry in place. +type tunBatch struct { + heads [msgx.Batch][4]byte + iovs [msgx.Batch][2]unix.Iovec + hdrs [msgx.Batch]msgx.Hdr + pkts [msgx.Batch][]byte +} + +// noTunSendX is set once sendmsg_x is refused on the tun; WriteBatch then writes one packet at a time. +var noTunSendX atomic.Bool + type ifReq struct { Name [unix.IFNAMSIZ]byte Flags uint16 @@ -615,16 +632,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 +670,129 @@ 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, up to msgx.Batch packets per syscall. utun is a +// connected datagram socket, so xnu takes the batched send path, though it still hands utun each packet on its own: +// 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. A short count goes on with the +// rest; EMSGSIZE, which sendmsg_x returns only when the first packet is past the send buffer, sends that one through +// Write, which reports it. Any other error means the kernel took an unknown prefix and dropped the rest, so it is +// reported rather than retried, which could deliver packets twice. Under mbuf exhaustion sendmsg_x reports packets it +// freed unsent as a short count, so they are lost where writev would have delivered them. +func (t *tun) WriteBatch(pkts [][]byte) error { + if noTunSendX.Load() { + return t.writeEach(pkts) + } + + t.batchMu.Lock() + defer t.batchMu.Unlock() + b := &t.batch + var firstErr error + keep := func(err error) { + if err != nil && firstErr == nil { + firstErr = err + } + } + for len(pkts) > 0 { + n := 0 + for len(pkts) > 0 && n < msgx.Batch { + p := pkts[0] + pkts = pkts[1:] + if len(p) == 0 { + keep(syscall.EIO) + continue + } + af, err := tunAF(p) + if err != nil { + keep(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] = msgx.Hdr{Iov: &b.iovs[n][0], Iovlen: 2} + b.pkts[n] = p + n++ + } + + for off := 0; off < n; { + sent, err := t.sendmsgX(b.hdrs[off:n]) + switch { + case err == nil && sent > 0: + off += sent + case err == nil: + keep(t.writeEach(b.pkts[off:n])) + off = n + case err == unix.EMSGSIZE: + keep(t.writeEach(b.pkts[off : off+1])) + off++ + case isUnsupported(err): + if noTunSendX.CompareAndSwap(false, true) { + t.l.Warn("sendmsg_x unavailable on the tun device, writing one packet per syscall", "error", err) + } + keep(t.writeEach(b.pkts[off:n])) + keep(t.writeEach(pkts)) + pkts, off = nil, n + default: + keep(err) + off = n + } + } + clear(b.iovs[:n]) + clear(b.pkts[:n]) + } + return firstErr +} + +// sendmsgX hands hdrs to sendmsg_x, waiting while the socket is full, and returns how many the kernel took. +func (t *tun) sendmsgX(hdrs []msgx.Hdr) (int, error) { + rc, err := t.f.SyscallConn() + if err != nil { + return 0, err + } + var sent int + var errno syscall.Errno + err = rc.Write(func(fd uintptr) bool { + sent, errno = msgx.Send(fd, hdrs) + // sendmsg_x reports EAGAIN only when it took nothing; a partial batch returns its count. + return errno != syscall.EAGAIN + }) + if err != nil { + return 0, err + } + if errno != 0 { + return 0, errno + } + return sent, nil +} + +func isUnsupported(err error) bool { + errno, ok := err.(syscall.Errno) + return ok && msgx.Unsupported(errno) +} + +// 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 } @@ -666,5 +802,137 @@ func (t *tun) Name() string { } func (t *tun) Queues(int) ([]tio.Queue, error) { - return []tio.Queue{tio.NewSingleQueue(t, defaultBatchBufSize)}, nil + t.raisePendingPackets() + return []tio.Queue{&tunQueue{t: t, buf: make([]byte, tunReadArena)}}, nil +} + +// tunReadBatch is the most packets one tunQueue.Read returns. +const tunReadBatch = 64 + +// 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 + +// 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. +type tunQueue struct { + t *tun + buf []byte + pkts [tunReadBatch]tio.Packet +} + +// 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. +func (q *tunQueue) Read() ([]tio.Packet, error) { + rc, err := q.t.f.SyscallConn() + if err != nil { + return nil, err + } + + var head [4]byte + n, off := 0, 0 + var callErr error + err = rc.Read(func(fd uintptr) bool { + for n < tunReadBatch && len(q.buf)-off >= defaultBatchBufSize { + iovecs := [2]unix.Iovec{ + {Base: &head[0], Len: 4}, + {Base: &q.buf[off], Len: uint64(len(q.buf) - off)}, + } + l, e := tunReadv(int(fd), iovecs[:]) + if e != nil { + if errno, ok := e.(syscall.Errno); ok && errno.Temporary() { + // Park on the poller only while there is nothing to hand back. + return n > 0 + } + callErr = e + return true + } + if l < 4 { + // A datagram too short to carry the AF prefix, or end of file; stop rather than spin on it. + return true + } + end := off + l - 4 + q.pkts[n] = tio.Packet{Bytes: q.buf[off:end:end]} + n++ + off = end + } + return true + }) + if n > 0 { + return q.pkts[:n], nil + } + if err != nil { + return nil, err + } + return nil, callErr +} + +func (q *tunQueue) Write(p []byte) (int, error) { return q.t.Write(p) } + +func (q *tunQueue) WriteBatch(pkts [][]byte) error { return q.t.WriteBatch(pkts) } + +func (q *tunQueue) Close() error { return q.t.Close() } + +// xnu lets one packet wait on a utun control socket by default (if_utun.c utun_ctl_bind): utun_start stops moving +// packets from the interface's send queue to the socket once one is waiting, and resumes only when a read wakes the +// interface's starter thread. So every read finds one packet. tunPendingPackets lets a whole read batch wait +// instead (_UTUN_OPT_MAX_PENDING_PACKETS, bsd/net/if_utun.h, set at SYSPROTO_CONTROL level). +const ( + tunPendingPackets = tunReadBatch + _UTUN_OPT_MAX_PENDING_PACKETS = 16 + _SYSPROTO_CONTROL = 2 +) + +// tunSetsockoptInt and tunGetsockoptInt are setsockopt and getsockopt; tests swap them. +var ( + tunSetsockoptInt = unix.SetsockoptInt + tunGetsockoptInt = unix.GetsockoptInt +) + +// raisePendingPackets lets up to tunPendingPackets packets wait on the utun control socket, first growing its +// receive buffer to hold that many packets and their 4 byte AF prefix: utun drops a packet the buffer has no room +// for, while one over the limit waits in the interface's queue. It sizes for the larger of tun.mtu and the device's +// current MTU. Failures are logged and leave xnu's default; the tun works either way. +func (t *tun) raisePendingPackets() { + if t.f == nil { + return + } + rc, err := t.f.SyscallConn() + if err != nil { + return + } + mtu := max(t.DefaultMTU, t.currentMTU()) + _ = rc.Control(func(fd uintptr) { + pending := tunPendingPackets + need := pending * (mtu + 4) + if have, err := tunGetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUF); err == nil && have < need { + if err := tunSetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUF, need); err != nil { + pending = max(have/(mtu+4), 1) + t.l.Warn("could not grow the tun control socket's receive buffer, capping pending packets", + "error", err, "rcvbuf", have, "wanted", need, "pending", pending) + } + } + if err := tunSetsockoptInt(int(fd), _SYSPROTO_CONTROL, _UTUN_OPT_MAX_PENDING_PACKETS, pending); err != nil { + t.l.Warn("could not raise the tun's pending packet limit, reads will return one packet each", + "error", err, "pending", pending) + } + }) +} + +// currentMTU is the utun's MTU as the kernel has it now, or 0 if that can't be read. A host-owned device's MTU +// comes from the host, not tun.mtu. +func (t *tun) currentMTU() int { + if t.Device == "" { + return 0 + } + s, err := unix.Socket(unix.AF_INET, unix.SOCK_DGRAM, unix.IPPROTO_IP) + if err != nil { + return 0 + } + defer unix.Close(s) + ifr, err := unix.IoctlGetIfreqMTU(s, t.Device) + if err != nil { + return 0 + } + return int(ifr.MTU) } diff --git a/overlay/tun_darwin_test.go b/overlay/tun_darwin_test.go new file mode 100644 index 00000000..126c2080 --- /dev/null +++ b/overlay/tun_darwin_test.go @@ -0,0 +1,359 @@ +//go:build !ios && !e2e_testing + +package overlay + +import ( + "bytes" + "os" + "syscall" + "testing" + + "github.com/slackhq/nebula/internal/msgx" + "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) { + noTunSendX.Store(fallback) + t.Cleanup(func() { noTunSendX.Store(false) }) + 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) + + var pkts, want [][]byte + for i := range 3*msgx.Batch + 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, noTunSendX.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) + + 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", msgx.Batch + 6}, + {"first-in-call", msgx.Batch}, + } { + t.Run(tc.name, func(t *testing.T) { + forEachTunWritePath(t, func(t *testing.T, fallback bool) { + tn, r := newSocketpairTun(t, 1024) + + var pkts, want [][]byte + for i := range 2*msgx.Batch + 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, noTunSendX.Load()) + readTunBack(t, r, want) + }) + }) + } +} + +// 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) { + tn, w := newSocketpairTun(t, 1<<20) + // An AF_UNIX datagram send is bounded by the receiver's buffer, here the tun's end. + 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) + } + + var got [][]byte + for _, size := range []int{tunReadBatch, 3} { + pkts, err := q.Read() + require.NoError(t, err) + require.Len(t, pkts, size) + for _, p := range pkts { + got = append(got, p.Clone().Bytes) + } + } + assert.Equal(t, want, got) +} + +// 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) + require.NoError(t, err) + } + 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 +// grown to hold that many packets of the tun's MTU, and capped to what the buffer holds when it can't grow. +func TestTunRaisePendingPackets(t *testing.T) { + type call struct{ level, opt, value int } + stub := func(t *testing.T, rcvbuf int, failLevel, failOpt int) *[]call { + set, get := tunSetsockoptInt, tunGetsockoptInt + var calls []call + tunSetsockoptInt = func(fd, level, opt, value int) error { + calls = append(calls, call{level, opt, value}) + if level == failLevel && opt == failOpt { + return unix.ENOBUFS + } + return nil + } + tunGetsockoptInt = func(fd, level, opt int) (int, error) { return rcvbuf, nil } + t.Cleanup(func() { tunSetsockoptInt, tunGetsockoptInt = set, get }) + return &calls + } + for _, tc := range []struct { + name string + rcvbuf int + mtu int + failLevel int + failOpt int + want []call + }{ + {"grows rcvbuf first", 1 << 16, 9000, -1, -1, []call{ + {unix.SOL_SOCKET, unix.SO_RCVBUF, 64 * 9004}, + {2, 16, 64}, + }}, + {"rcvbuf already big enough", 1 << 20, 1300, -1, -1, []call{{2, 16, 64}}}, + {"rcvbuf refused caps pending", 10 * 1304, 1300, unix.SOL_SOCKET, unix.SO_RCVBUF, []call{ + {unix.SOL_SOCKET, unix.SO_RCVBUF, 64 * 1304}, + {2, 16, 10}, + }}, + } { + t.Run(tc.name, func(t *testing.T) { + calls := stub(t, tc.rcvbuf, tc.failLevel, tc.failOpt) + tn, _ := newSocketpairTun(t, 1<<20) + tn.DefaultMTU = tc.mtu + _, err := tn.Queues(1) + require.NoError(t, err) + assert.Equal(t, tc.want, *calls) + }) + } +} + +func TestTun_WriteFraming(t *testing.T) { + r, w, err := os.Pipe() + require.NoError(t, err) + defer r.Close() + defer w.Close() + tn := &tun{f: w} + + for _, tc := range []struct { + name string + pkt []byte + af byte + }{ + // Literal families rather than syscall.AF_*, so a wrong constant in Write can't also fix the expectation. + {"ipv4", []byte{0x45, 1, 2, 3, 4, 5}, 2}, + {"ipv6", []byte{0x60, 1, 2, 3, 4, 5}, 30}, + } { + t.Run(tc.name, func(t *testing.T) { + n, err := tn.Write(tc.pkt) + require.NoError(t, err) + require.Equal(t, len(tc.pkt), n) + + got := make([]byte, 64) + n, err = r.Read(got) + require.NoError(t, err) + require.Equal(t, append([]byte{0, 0, 0, tc.af}, tc.pkt...), got[:n]) + }) + } + + _, err = tn.Write([]byte{0x10, 1, 2, 3}) + require.Error(t, err) + _, err = tn.Write([]byte{}) + require.ErrorIs(t, err, syscall.EIO) + + // The rejected writes must not have put anything on the fd, so the next read sees only this packet. + pkt := []byte{0x45, 9, 9} + _, err = tn.Write(pkt) + require.NoError(t, err) + got := make([]byte, 64) + n, err := r.Read(got) + require.NoError(t, err) + require.Equal(t, append([]byte{0, 0, 0, 2}, pkt...), got[:n]) +} + +func TestTun_ReadStripsHeader(t *testing.T) { + r, w, err := os.Pipe() + require.NoError(t, err) + defer r.Close() + defer w.Close() + tn := &tun{f: r} + + pkt := []byte{0x45, 1, 2, 3, 4, 5} + _, err = w.Write(append([]byte{0, 0, 0, syscall.AF_INET}, pkt...)) + require.NoError(t, err) + + got := make([]byte, 64) + n, err := tn.Read(got) + require.NoError(t, err) + require.Equal(t, pkt, got[:n]) +} + +func TestTun_ReadShort(t *testing.T) { + r, w, err := os.Pipe() + require.NoError(t, err) + defer r.Close() + defer w.Close() + tn := &tun{f: r} + + // Fewer bytes than the protocol header is not a packet. + _, err = w.Write([]byte{0, 0}) + require.NoError(t, err) + n, err := tn.Read(make([]byte, 64)) + require.NoError(t, err) + require.Zero(t, n) + + // The next packet reads cleanly only if the short read consumed exactly what was written. + pkt := []byte{0x45, 1, 2, 3} + _, err = w.Write(append([]byte{0, 0, 0, syscall.AF_INET}, pkt...)) + require.NoError(t, err) + got := make([]byte, 64) + n, err = tn.Read(got) + require.NoError(t, err) + require.Equal(t, pkt, got[:n]) +} + +// A heap allocation per packet rarely shows in a short throughput run, only across GC pauses, so +// this pins Read and Write to zero allocations directly. +func TestTun_ReadWriteDoNotAllocate(t *testing.T) { + r, w, err := os.Pipe() + require.NoError(t, err) + defer r.Close() + defer w.Close() + tw := &tun{f: w} + tr := &tun{f: r} + + pkts := [][]byte{bytes.Repeat([]byte{0x45}, 1300), bytes.Repeat([]byte{0x60}, 1300)} + buf := make([]byte, 1500) + i := 0 + allocs := testing.AllocsPerRun(1000, func() { + i++ + if _, err := tw.Write(pkts[i%len(pkts)]); err != nil { + t.Fatal(err) + } + if _, err := tr.Read(buf); err != nil { + t.Fatal(err) + } + }) + require.Zero(t, allocs) +}