overlay: replace per-fd tun readers with a batched Queue interface

Device loses io.ReadWriteCloser + NewMultiQueueReader in favor of
Queues(n), which returns up to n tio.Queue objects; platforms without
multiqueue hand back their single queue and the interface sizes its
reader routines to what it actually got. Queue.Read returns a batch of
borrowed packets (single-element for every current backend) so a future
backend can deliver more than one packet per syscall without another
interface change.

The Linux poll/eventfd machinery moves out of tun_linux.go into the new
overlay/tio package: nonblocking fds, a shared shutdown eventfd owned by
the queue set, and pollfd arrays built on the stack so concurrent
writers parked in blockOnWrite no longer share Revents storage. Other
platforms wrap their existing one-datagram Read/Write in a singleQueue
adapter that owns a private scratch buffer, so multiqueue-by-sharing
devices (user, disabled) no longer race concurrent readers on one
buffer.

This is the tun-interface subset of better-tun-interface-ordering,
extracted at 18dc13b with none of the GSO/GRO offload mechanics and no
udp/sendmmsg changes.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

overlay/tio/blockon_linux.go

@@ -0,0 +1,45 @@
+//go:build linux && !android
+// +build linux,!android
+
+package tio
+
+import (
+	"os"
+
+	"golang.org/x/sys/unix"
+)
+
+// blockOn parks the calling goroutine until fd is ready (events is POLLIN for
+// reads, POLLOUT for writes) or shutdownFd signals teardown. It builds the
+// pollfd array on the stack every call, so concurrent callers on the same
+// Queue never share Revents storage.
+//
+// Returns os.ErrClosed when shutdown was signaled (POLLIN on shutdownFd)
+// or either fd reported a problem condition (POLLHUP|POLLNVAL|POLLERR).
+func blockOn(fd, shutdownFd int32, events int16) error {
+	const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR
+	pfds := [2]unix.PollFd{
+		{Fd: fd, Events: events},
+		{Fd: shutdownFd, Events: unix.POLLIN},
+	}
+	var err error
+	for {
+		_, err = unix.Poll(pfds[:], -1)
+		if err != unix.EINTR {
+			break
+		}
+	}
+	tunEvents := pfds[0].Revents
+	shutdownEvents := pfds[1].Revents
+	// Check err before trusting the potentially bogus bits we just got.
+	if err != nil {
+		return err
+	}
+	if shutdownEvents&(unix.POLLIN|problemFlags) != 0 {
+		return os.ErrClosed
+	}
+	if tunEvents&problemFlags != 0 {
+		return os.ErrClosed
+	}
+	return nil
+}

overlay/tio/queueset_poll_linux.go

@@ -0,0 +1,90 @@
+//go:build linux && !android
+// +build linux,!android
+
+package tio
+
+import (
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"sync/atomic"
+
+	"golang.org/x/sys/unix"
+)
+
+type pollQueueSet struct {
+	pq []*Poll
+	// pqi is exactly the same as pq, but stored as the interface type
+	pqi        []Queue
+	shutdownFd int
+	closed     atomic.Bool
+}
+
+func NewPollQueueSet() (QueueSet, error) {
+	shutdownFd, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create eventfd: %w", err)
+	}
+
+	out := &pollQueueSet{
+		pq:         []*Poll{},
+		pqi:        []Queue{},
+		shutdownFd: shutdownFd,
+	}
+
+	return out, nil
+}
+
+func (c *pollQueueSet) Queues() []Queue {
+	return c.pqi
+}
+
+func (c *pollQueueSet) Add(fd int) error {
+	x, err := newPoll(fd, c.shutdownFd)
+	if err != nil {
+		return err
+	}
+	c.pq = append(c.pq, x)
+	c.pqi = append(c.pqi, x)
+
+	return nil
+}
+
+func (c *pollQueueSet) wakeForShutdown() error {
+	var buf [8]byte
+	binary.NativeEndian.PutUint64(buf[:], 1)
+	_, err := unix.Write(int(c.shutdownFd), buf[:])
+	return err
+}
+
+func (c *pollQueueSet) Close() error {
+	if c.closed.Swap(true) {
+		return nil
+	}
+
+	errs := []error{}
+
+	// Wake any reader blocked in poll so it observes POLLIN on the shutdown
+	// eventfd and returns os.ErrClosed.
+	if err := c.wakeForShutdown(); err != nil {
+		errs = append(errs, err)
+	}
+
+	// Close the per-queue tun fds; this also unblocks any in-flight reads.
+	// The per-queue Close deliberately leaves shutdownFd alone - it belongs
+	// to this container.
+	for _, x := range c.pq {
+		if err := x.Close(); err != nil {
+			errs = append(errs, err)
+		}
+	}
+
+	// Close the shutdown eventfd last: every reader's pollfd set references
+	// it, so it must outlive the wake + per-queue teardown above.
+	if err := unix.Close(c.shutdownFd); err != nil {
+		errs = append(errs, err)
+	}
+	c.shutdownFd = -1
+
+	return errors.Join(errs...)
+}

overlay/tio/single.go

@@ -0,0 +1,50 @@
+package tio
+
+import "io"
+
+// singleQueue adapts a legacy one-datagram-per-Read source into a Queue.
+// Read fills a private scratch buffer and returns exactly one Packet whose
+// Bytes borrow from that buffer, valid only until the next Read, per the
+// Queue contract. Single-reader like every Queue; Write is exactly as safe
+// for concurrent use as the underlying source's Write.
+type singleQueue struct {
+	rw     io.ReadWriter
+	closer io.Closer // nil: Close is a no-op (the source is shared and owned elsewhere)
+	buf    []byte
+	ret    [1]Packet
+}
+
+// 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.
+func NewSingleQueue(rwc io.ReadWriteCloser, bufSize int) Queue {
+	return &singleQueue{rw: rwc, closer: rwc, buf: make([]byte, bufSize)}
+}
+
+// NewSingleQueueNoClose is NewSingleQueue for a source owned by someone else,
+// e.g. several queues sharing one device. Close on the returned Queue is a
+// no-op so one queue can't tear the shared source out from under its
+// siblings; the owner remains responsible for closing the source itself.
+func NewSingleQueueNoClose(rw io.ReadWriter, bufSize int) Queue {
+	return &singleQueue{rw: rw, buf: make([]byte, bufSize)}
+}
+
+func (q *singleQueue) Read() ([]Packet, error) {
+	n, err := q.rw.Read(q.buf)
+	if err != nil {
+		return nil, err
+	}
+	q.ret[0] = Packet{Bytes: q.buf[:n]}
+	return q.ret[:], nil
+}
+
+func (q *singleQueue) Write(p []byte) (int, error) {
+	return q.rw.Write(p)
+}
+
+func (q *singleQueue) Close() error {
+	if q.closer == nil {
+		return nil
+	}
+	return q.closer.Close()
+}

overlay/tio/tio.go

@@ -0,0 +1,52 @@
+package tio
+
+import (
+	"io"
+)
+
+// QueueSet holds one or many Queue objects and helps close them in an orderly way.
+type QueueSet interface {
+	io.Closer
+	Queues() []Queue
+
+	// Add takes a tun fd, adds it to the set, and prepares it for use as a Queue.
+	Add(fd int) error
+}
+
+// Queue is a readable/writable packet queue. Concurrency contract: a single
+// read goroutine drives Read; plain Write is safe for concurrent callers.
+type Queue interface {
+	io.Closer
+
+	// Read returns one or more packets. The returned Packet.Bytes slices
+	// are borrowed from the Queue's internal buffer and are only valid
+	// until the next Read or Close on this Queue - callers must encrypt
+	// or copy each slice before the next call. Single-reader only: not
+	// safe for concurrent Reads (it reuses per-queue rx scratch each call).
+	Read() ([]Packet, error)
+
+	// Write emits a single packet on the plaintext (outside→inside)
+	// delivery path. Safe for concurrent use.
+	Write(p []byte) (int, error)
+}
+
+// Packet is the unit Queue.Read returns. Bytes points into the queue's
+// internal buffer and is only valid until the next Read or Close on the
+// queue that produced it.
+type Packet struct {
+	Bytes []byte
+}
+
+// Clone returns a Packet whose Bytes is a freshly allocated copy of p.Bytes,
+// safe to retain past the next Read or Close on the originating Queue.
+// Use this only when a caller genuinely needs to outlive the borrowed-slice
+// contract — the hot path reads should continue to consume the borrow
+// synchronously to avoid the allocation.
+func (p Packet) Clone() Packet {
+	if p.Bytes == nil {
+		return p
+	}
+	cp := make([]byte, len(p.Bytes))
+	copy(cp, p.Bytes)
+	return Packet{Bytes: cp}
+}

overlay/tio/tio_poll_linux.go

@@ -0,0 +1,116 @@
+//go:build linux && !android
+// +build linux,!android
+
+package tio
+
+import (
+	"fmt"
+	"os"
+	"sync/atomic"
+
+	"golang.org/x/sys/unix"
+)
+
+// Maximum size we accept for a single read from a TUN. 65535 covers any
+// single IP packet.
+const tunReadBufSize = 65535
+
+type Poll struct {
+	fd         int
+	shutdownFd int
+	closed     atomic.Bool
+
+	readBuf  []byte
+	batchRet [1]Packet
+}
+
+// newPoll wraps an existing tun fd. On failure it does NOT close fd: the
+// caller owns fd and is the sole closer (see pollQueueSet.Add callers in
+// overlay/tun_linux.go, which unix.Close on Add error). This keeps closes
+// at exactly one on every path.
+func newPoll(fd int, shutdownFd int) (*Poll, error) {
+	if err := unix.SetNonblock(fd, true); err != nil {
+		return nil, fmt.Errorf("failed to set Poll device as nonblocking: %w", err)
+	}
+
+	out := &Poll{
+		fd:         fd,
+		shutdownFd: shutdownFd,
+		readBuf:    make([]byte, tunReadBufSize),
+	}
+	return out, nil
+}
+
+// blockOnRead waits until the Poll fd is readable or shutdown has been signaled.
+// Returns os.ErrClosed if Close was called.
+func (t *Poll) blockOnRead() error {
+	return blockOn(int32(t.fd), int32(t.shutdownFd), unix.POLLIN)
+}
+
+func (t *Poll) blockOnWrite() error {
+	return blockOn(int32(t.fd), int32(t.shutdownFd), unix.POLLOUT)
+}
+
+func (t *Poll) Read() ([]Packet, error) {
+	n, err := t.readOne(t.readBuf)
+	if err != nil {
+		return nil, err
+	}
+	t.batchRet[0] = Packet{Bytes: t.readBuf[:n]}
+	return t.batchRet[:], nil
+}
+
+func (t *Poll) readOne(to []byte) (int, error) {
+	for {
+		n, errno := unix.Read(t.fd, to)
+		if errno == nil {
+			return n, nil
+		}
+		switch errno {
+		case unix.EAGAIN:
+			if err := t.blockOnRead(); err != nil {
+				return 0, err
+			}
+		case unix.EINTR:
+			// retry
+		case unix.EBADF:
+			return 0, os.ErrClosed
+		default:
+			return 0, errno
+		}
+	}
+}
+
+// Write is safe for concurrent use
+func (t *Poll) Write(from []byte) (int, error) {
+	for {
+		n, errno := unix.Write(t.fd, from)
+		if errno == nil {
+			return n, nil
+		}
+		switch errno {
+		case unix.EAGAIN:
+			if err := t.blockOnWrite(); err != nil {
+				return 0, err
+			}
+		case unix.EINTR:
+			// retry
+		case unix.EBADF:
+			return 0, os.ErrClosed
+		default:
+			return 0, errno
+		}
+	}
+}
+
+func (t *Poll) Close() error {
+	if t.closed.Swap(true) {
+		return nil
+	}
+	//shutdownFd is owned by the container, so we should not close it
+	// Close the underlying fd but do NOT null t.fd: a reader may still be
+	// loading it in readOne, and mutating the field would race that load.
+	// It gets EBADF -> os.ErrClosed (or wakes via the shutdown eventfd's
+	// ppoll first). closed.Swap already guarantees we only close once.
+	return unix.Close(t.fd)
+}

overlay/tio/tun_file_linux_test.go

@@ -0,0 +1,208 @@
+//go:build linux && !android && !e2e_testing
+// +build linux,!android,!e2e_testing
+
+package tio
+
+import (
+	"errors"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sys/unix"
+)
+
+// newReadPipe returns a read fd. The matching write fd is registered for cleanup.
+// The caller takes ownership of the read fd (pass it into a QueueSet).
+func newReadPipe(t *testing.T) int {
+	t.Helper()
+	var fds [2]int
+	if err := unix.Pipe2(fds[:], unix.O_CLOEXEC); err != nil {
+		t.Fatalf("pipe2: %v", err)
+	}
+	t.Cleanup(func() { _ = unix.Close(fds[1]) })
+	return fds[0]
+}
+
+func TestPoll_WakeForShutdown_WakesFriends(t *testing.T) {
+	pipe1 := newReadPipe(t)
+	pipe2 := newReadPipe(t)
+	parent, err := NewPollQueueSet()
+	require.NoError(t, err)
+	require.NoError(t, parent.Add(pipe1))
+	require.NoError(t, parent.Add(pipe2))
+	t.Cleanup(func() {
+		_ = unix.Close(pipe1)
+		_ = unix.Close(pipe2)
+	})
+
+	readers := parent.Queues()
+	errs := make([]error, len(readers))
+	var wg sync.WaitGroup
+	for i, r := range readers {
+		wg.Add(1)
+		go func(i int, r Queue) {
+			defer wg.Done()
+			_, errs[i] = r.Read()
+		}(i, r)
+	}
+
+	time.Sleep(50 * time.Millisecond)
+
+	if err := parent.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+
+	done := make(chan struct{})
+	go func() { wg.Wait(); close(done) }()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("readers did not wake")
+	}
+
+	for i, err := range errs {
+		if !errors.Is(err, os.ErrClosed) {
+			t.Errorf("reader %d: expected os.ErrClosed, got %v", i, err)
+		}
+	}
+}
+
+// TestPoll_ConcurrentWrite_NoRace hammers a single Poll queue from two writer
+// goroutines while a reader drains the other end of the pipe. The writers
+// overflow the pipe buffer, so both repeatedly park in blockOnWrite at the same
+// time — the exact scenario that raced on the old shared writePoll member
+// array. Run under -race; a shared-array regression trips the detector here.
+func TestPoll_ConcurrentWrite_NoRace(t *testing.T) {
+	var fds [2]int
+	require.NoError(t, unix.Pipe2(fds[:], unix.O_CLOEXEC))
+	readFd, writeFd := fds[0], fds[1]
+
+	shutdownFd, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = unix.Close(shutdownFd) })
+
+	p, err := newPoll(writeFd, shutdownFd)
+	require.NoError(t, err)
+
+	const writers = 2
+	const perWriter = 4000
+	payload := make([]byte, 100)
+	total := writers * perWriter * len(payload)
+
+	// Reader: drain the read end (blocking) until every writer's bytes are
+	// consumed, so the writers keep making progress rather than wedging on a
+	// permanently full pipe.
+	readDone := make(chan struct{})
+	go func() {
+		defer close(readDone)
+		buf := make([]byte, 4096)
+		got := 0
+		for got < total {
+			n, rerr := unix.Read(readFd, buf)
+			got += n
+			if rerr != nil {
+				if rerr == unix.EINTR {
+					continue
+				}
+				return
+			}
+			if n == 0 { // EOF
+				return
+			}
+		}
+	}()
+
+	var wg sync.WaitGroup
+	for w := 0; w < writers; w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for i := 0; i < perWriter; i++ {
+				if _, werr := p.Write(payload); werr != nil {
+					t.Errorf("write: %v", werr)
+					return
+				}
+			}
+		}()
+	}
+	wg.Wait()
+
+	select {
+	case <-readDone:
+	case <-time.After(10 * time.Second):
+		t.Fatal("reader did not drain")
+	}
+
+	require.NoError(t, p.Close())
+	_ = unix.Close(readFd)
+}
+
+// TestPoll_NewPoll_DoesNotCloseFdOnFailure pins the ownership rule: when
+// newPoll fails, it must leave fd open so the caller (pollQueueSet.Add's
+// callers in tun_linux.go) is the sole closer. If newPoll also closed fd,
+// the poll path would double-close on Add error. We force the failure with
+// an O_PATH descriptor: fcntl(F_SETFL) — which SetNonblock performs — is not
+// permitted on O_PATH fds and fails with EBADF, while the fd itself stays
+// open so we can observe that newPoll left it alone.
+func TestPoll_NewPoll_DoesNotCloseFdOnFailure(t *testing.T) {
+	fd, err := unix.Open("/", unix.O_PATH|unix.O_CLOEXEC, 0)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = unix.Close(fd) })
+
+	p, err := newPoll(fd, 1)
+	require.Error(t, err, "SetNonblock on an O_PATH fd should fail")
+	require.Nil(t, p)
+
+	// If newPoll had closed fd, F_GETFD would report it closed. It staying
+	// open proves newPoll left the fd for the caller to close exactly once.
+	require.True(t, fdOpen(t, fd), "newPoll must not close fd on failure; caller is the sole closer")
+}
+
+func TestPoll_Close_Idempotent(t *testing.T) {
+	tf, err := newPoll(newReadPipe(t), 1)
+	require.NoError(t, err)
+	if err := tf.Close(); err != nil {
+		t.Fatalf("first Close: %v", err)
+	}
+	if err := tf.Close(); err != nil {
+		t.Fatalf("second Close should be a no-op, got %v", err)
+	}
+}
+
+// fdOpen reports whether fd currently refers to an open file description.
+// A closed (or never-allocated) fd makes F_GETFD fail with EBADF.
+func fdOpen(t *testing.T, fd int) bool {
+	t.Helper()
+	_, err := unix.FcntlInt(uintptr(fd), unix.F_GETFD, 0)
+	if err == nil {
+		return true
+	}
+	if errors.Is(err, unix.EBADF) {
+		return false
+	}
+	t.Fatalf("unexpected fcntl(F_GETFD) error on fd %d: %v", fd, err)
+	return false
+}
+
+// TestPollQueueSet_Close_ClosesShutdownFd is the regression test for the
+// leaked shutdown eventfd: the container that owns shutdownFd must close it in
+// Close, and a second Close must be a safe no-op.
+func TestPollQueueSet_Close_ClosesShutdownFd(t *testing.T) {
+	qs, err := NewPollQueueSet()
+	require.NoError(t, err)
+	c, ok := qs.(*pollQueueSet)
+	require.True(t, ok)
+	require.NoError(t, qs.Add(newReadPipe(t)))
+
+	shutdownFd := c.shutdownFd
+	require.True(t, fdOpen(t, shutdownFd), "shutdown eventfd should be open before Close")
+
+	require.NoError(t, qs.Close())
+	require.False(t, fdOpen(t, shutdownFd), "shutdown eventfd should be closed after Close")
+
+	// Second Close must not touch fds (shutdownFd is now -1) and must return nil.
+	require.NoError(t, qs.Close())
+}