nebula/udp/udp_linux_offloads_test.go

//go:build linux && !android && !e2e_testing

package udp

import (
	"net"
	"net/netip"
	"slices"
	"testing"
	"time"

	"golang.org/x/sys/unix"
)

// These tests pin the listen.udp_offloads=false behavior: no GSO/GRO probes,
// no cmsg scratch, and — critically — a still-functional send/receive path.
// The sockaddr name buffers are needed for every sendmmsg entry whether or
// not offloads are on, so prepareWriteMessages must allocate them even when
// it skips the cmsg slab (a nil name buffer panics in writeSockaddr on the
// first WriteBatch).

// TestPrepareWriteMessagesAlwaysAllocatesNames covers all four
// (offloadsEnabled, gsoSupported) combinations: the sockaddr name buffers
// must exist in every one, and the cmsg slab only when both are true.
// gsoSupported=false with offloads enabled is the old-kernel path where the
// UDP_SEGMENT probe fails — not just a config choice.
func TestPrepareWriteMessagesAlwaysAllocatesNames(t *testing.T) {
	cases := []struct {
		name     string
		offloads bool
		gso      bool
	}{
		{"offloads-off", false, false},
		{"offloads-on-probe-failed", true, false},
		{"offloads-off-gso-flag-set", false, true},
		{"offloads-on", true, true},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			w := &batchWriter{fd: -1, isV4: true, l: testLogger()}
			w.gsoSupported = tc.gso
			w.prepareWriteMessages(MaxWriteBatch, tc.offloads)

			for i := range w.msgs {
				if len(w.names[i]) != unix.SizeofSockaddrInet6 {
					t.Fatalf("names[%d] len=%d, want %d", i, len(w.names[i]), unix.SizeofSockaddrInet6)
				}
				if w.msgs[i].Hdr.Name == nil {
					t.Fatalf("msgs[%d].Hdr.Name is nil", i)
				}
			}

			wantCmsg := tc.offloads && tc.gso
			if (w.cmsg != nil) != wantCmsg {
				t.Errorf("cmsg allocated = %v, want %v", w.cmsg != nil, wantCmsg)
			}
		})
	}
}

// TestWriteBatchOffloadsDisabledScripted drives WriteBatch through a
// batchWriter built with offloads disabled and a scripted sendFn: every
// packet must become its own sendmmsg entry (no GSO coalescing to plan),
// packed correctly despite the missing cmsg slab.
func TestWriteBatchOffloadsDisabledScripted(t *testing.T) {
	w := &batchWriter{fd: -1, isV4: true, l: testLogger()}
	w.prepareWriteMessages(MaxWriteBatch, false)

	var entryCounts []int
	w.sendFn = func(start, n int) (int, error) {
		entryCounts = append(entryCounts, n)
		return n, nil
	}

	// Same destination, equal sizes: prime coalescing bait that must not
	// coalesce with offloads off.
	dst := netip.MustParseAddrPort("127.0.0.1:4242")
	const numPkts = 4
	bufs := make([][]byte, numPkts)
	addrs := make([]netip.AddrPort, numPkts)
	for i := range bufs {
		bufs[i] = make([]byte, 1200)
		for j := range bufs[i] {
			bufs[i][j] = byte(i)
		}
		addrs[i] = dst
	}

	written, err := w.WriteBatch(bufs, addrs)
	if err != nil {
		t.Fatalf("WriteBatch: %v", err)
	}
	if written != numPkts {
		t.Errorf("written = %d, want %d", written, numPkts)
	}
	if len(entryCounts) != 1 || entryCounts[0] != numPkts {
		t.Errorf("sendmmsg entry counts = %v, want [%d]: packets must be one entry each", entryCounts, numPkts)
	}

	// Each prepared entry must carry exactly its own packet's bytes.
	got := capturePrepared(w, 0, numPkts)
	if len(got) != numPkts {
		t.Fatalf("prepared %d packets, want %d", len(got), numPkts)
	}
	for i, pkt := range got {
		if !slices.Equal(pkt, bufs[i]) {
			t.Errorf("entry %d bytes differ from bufs[%d]", i, i)
		}
	}
}

// TestOffloadsDisabledOnLoopback is the offloads-off smoke test, the mirror
// of TestGSOEngagesOnLoopback: a real socket built with Offloads=false must
// skip the GSO/GRO probes entirely (even on kernels that support them) and
// still deliver a same-destination batch as plain per-packet datagrams.
func TestOffloadsDisabledOnLoopback(t *testing.T) {
	rx, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
	if err != nil {
		t.Fatalf("listen rx: %v", err)
	}
	defer rx.Close()
	dst := rx.LocalAddr().(*net.UDPAddr).AddrPort()

	udpSettings := Settings{
		Listen:   netip.MustParseAddrPort("127.0.0.1:0"),
		Multi:    false,
		Batch:    8, // batch > 1 would enable GRO if Offloads did not gate it
		Offloads: false,
	}
	uc, err := NewListener(testLogger(), udpSettings)
	if err != nil {
		t.Fatalf("NewListener: %v", err)
	}
	sc := uc.(*StdConn)
	defer sc.Close()

	if sc.bw.gsoSupported {
		t.Error("gsoSupported true with offloads disabled: probe was not skipped")
	}
	if sc.bw.cmsg != nil {
		t.Error("cmsg slab allocated with offloads disabled")
	}
	if sc.groSupported {
		t.Error("groSupported true with offloads disabled: probe was not skipped")
	}

	var entryCounts []int
	real := sc.bw.sendFn
	sc.bw.sendFn = func(start, n int) (int, error) {
		entryCounts = append(entryCounts, n)
		return real(start, n)
	}

	const numPkts = 8
	const pktLen = 1200
	bufs := make([][]byte, numPkts)
	addrs := make([]netip.AddrPort, numPkts)
	for i := range bufs {
		bufs[i] = make([]byte, pktLen)
		for j := range bufs[i] {
			bufs[i][j] = byte(i)
		}
		addrs[i] = dst
	}

	written, err := sc.WriteBatch(bufs, addrs)
	if err != nil {
		t.Fatalf("WriteBatch: %v", err)
	}
	if written != numPkts {
		t.Fatalf("written = %d, want %d", written, numPkts)
	}
	// One sendmmsg call with one entry per packet: a single-entry call here
	// means GSO engaged despite being disabled.
	if len(entryCounts) != 1 || entryCounts[0] != numPkts {
		t.Fatalf("sendmmsg entry counts = %v, want [%d]", entryCounts, numPkts)
	}

	_ = rx.SetReadDeadline(time.Now().Add(5 * time.Second))
	got := make([]byte, pktLen+1)
	for i := range numPkts {
		n, _, err := rx.ReadFromUDP(got)
		if err != nil {
			t.Fatalf("rx read %d: %v", i, err)
		}
		if n != pktLen {
			t.Fatalf("rx read %d: len=%d want %d", i, n, pktLen)
		}
		for j := range n {
			if got[j] != byte(i) {
				t.Fatalf("rx read %d: byte %d = %#x, want %#x", i, j, got[j], byte(i))
			}
		}
	}
}

// TestOffloadsDisabledRxDelivers exercises the receive path with GRO gated
// off but batch reads still on: ListenOut must deliver plain datagrams via
// the MTU-sized buffer layout (no cmsg slots).
func TestOffloadsDisabledRxDelivers(t *testing.T) {
	udpSettings := Settings{
		Listen:   netip.MustParseAddrPort("127.0.0.1:0"),
		Multi:    false,
		Batch:    8,
		Offloads: false,
	}
	uc, err := NewListener(testLogger(), udpSettings)
	if err != nil {
		t.Fatalf("NewListener: %v", err)
	}
	sc := uc.(*StdConn)

	addr, err := sc.LocalAddr()
	if err != nil {
		t.Fatalf("LocalAddr: %v", err)
	}

	type rxPkt struct {
		from    netip.AddrPort
		payload []byte
	}
	rxCh := make(chan rxPkt, 16)
	listenDone := make(chan struct{})
	go func() {
		defer close(listenDone)
		_ = sc.ListenOut(func(from netip.AddrPort, payload []byte) {
			// payload aliases the shared recv buffer row; copy before handing off.
			rxCh <- rxPkt{from, slices.Clone(payload)}
		}, func(bufs [][]byte) [][]byte { return bufs })
	}()

	tx, err := net.DialUDP("udp4", nil, net.UDPAddrFromAddrPort(addr))
	if err != nil {
		t.Fatalf("dial tx: %v", err)
	}
	defer tx.Close()

	want := [][]byte{
		[]byte("one"),
		make([]byte, 1200),
		make([]byte, 9000), // near-MTU datagram must fit the non-GRO buffer size
	}
	for i := range want[1] {
		want[1][i] = 0xAB
	}
	for i := range want[2] {
		want[2][i] = 0xCD
	}
	for i, p := range want {
		if _, err := tx.Write(p); err != nil {
			t.Fatalf("tx write %d: %v", i, err)
		}
	}

	for i, p := range want {
		select {
		case got := <-rxCh:
			if !slices.Equal(got.payload, p) {
				t.Errorf("packet %d: payload differs (len=%d want %d)", i, len(got.payload), len(p))
			}
			if got.from.Port() != tx.LocalAddr().(*net.UDPAddr).AddrPort().Port() {
				t.Errorf("packet %d: from=%v, want sender port %d", i, got.from, tx.LocalAddr().(*net.UDPAddr).Port)
			}
		case <-time.After(5 * time.Second):
			t.Fatalf("timed out waiting for packet %d", i)
		}
	}

	if err := sc.Close(); err != nil {
		t.Fatalf("Close: %v", err)
	}
	select {
	case <-listenDone:
	case <-time.After(5 * time.Second):
		t.Fatal("ListenOut did not return after Close")
	}
}