udp: resume partial sendmmsg in place instead of repacking

A partial success left the remaining entries' iovecs, sockaddrs, and cmsgs fully intact, then threw them away and replanned the remainder from bufs -- doubling the packing work exactly when the socket is congested. Give sendFn a start offset so the drain resumes the same prepared array at the first unsent entry, and skip a kernel-rejected entry in place the same way. Only the GSO-disable path still replans, since its entries change shape; it now rewinds precisely to the failed run instead of the whole chunk, so entries already sent are never duplicated. New scripted tests pin the two paths that didn't exist before: a mid- chunk rejected entry (drop it, resume the rest, start offsets advance) and a mid-chunk EIO (GSO off, replay only the failed run, no dup of already-sent packets). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014ugV2edVqoz3tBvq9J6yWp
2026-08-16 01:16:58 +02:00 · 2026-07-29 17:55:13 -05:00
parent d13c53db43
commit e8d6be1dd9
2 changed files with 177 additions and 49 deletions
@@ -8,6 +8,7 @@ import (
 	"log/slog"
 	"net"
 	"net/netip"
+	"slices"
 	"syscall"
 	"testing"
 	"time"
@@ -410,13 +411,13 @@ func newRewindTestWriter() *batchWriter {
 	return w
 }

-// capturePrepared decodes the first n prepared mmsghdr entries straight
-// from their iovecs -- ground truth, deliberately not the entryEnd
-// bookkeeping the rewind logic itself relies on. Returns one []byte per
+// capturePrepared decodes n prepared mmsghdr entries beginning at start
+// straight from their iovecs -- ground truth, deliberately not the entryEnd
+// bookkeeping the resume logic itself relies on. Returns one []byte per
 // packed packet, in entry order.
-func capturePrepared(w *batchWriter, n int) [][]byte {
+func capturePrepared(w *batchWriter, start, n int) [][]byte {
 	var out [][]byte
-	for e := 0; e < n; e++ {
+	for e := start; e < start+n; e++ {
 		hdr := &w.msgs[e].Hdr
 		iovs := unsafe.Slice(hdr.Iov, int(hdr.Iovlen))
 		for _, iov := range iovs {
@@ -470,13 +471,13 @@ func TestWriteBatchPartialSendRewind(t *testing.T) {
 			w := newRewindTestWriter()
 			var wire [][]byte
 			call := 0
-			w.sendFn = func(n int) (int, error) {
+			w.sendFn = func(start, n int) (int, error) {
 				accept := n
 				if call < len(script) && script[call] < n {
 					accept = script[call]
 				}
 				call++
-				wire = append(wire, capturePrepared(w, accept)...)
+				wire = append(wire, capturePrepared(w, start, accept)...)
 				return accept, nil
 			}

@@ -523,13 +524,13 @@ func TestWriteBatchSkipUnroutableRunAccounting(t *testing.T) {
 			w := newRewindTestWriter()
 			var wire [][]byte
 			call := 0
-			w.sendFn = func(n int) (int, error) {
+			w.sendFn = func(start, n int) (int, error) {
 				accept := n
 				if call < len(script) && script[call] < n {
 					accept = script[call]
 				}
 				call++
-				wire = append(wire, capturePrepared(w, accept)...)
+				wire = append(wire, capturePrepared(w, start, accept)...)
 				return accept, nil
 			}

@@ -553,11 +554,126 @@ func TestWriteBatchSkipUnroutableRunAccounting(t *testing.T) {
 	}
 }

+// TestWriteBatchMidChunkRejectResumes: after a partial success, a zero-sent
+// error on the FIRST REMAINING entry (done > 0) must drop only that entry's
+// run and resume the rest of the chunk in place -- no repacking, no packets
+// lost from entries before or after the rejected one.
+func TestWriteBatchMidChunkRejectResumes(t *testing.T) {
+	dstA := netip.MustParseAddrPort("127.0.0.1:4242")
+	dstB := netip.MustParseAddrPort("127.0.0.2:4242")
+	dstC := netip.MustParseAddrPort("127.0.0.3:4242")
+
+	mk := func(tag byte, n int) []byte {
+		b := make([]byte, n)
+		b[0] = tag
+		return b
+	}
+	// Three entries: a 2-packet GSO run to A, a 2-packet run to B, one to C.
+	bufs := [][]byte{mk(1, 1200), mk(2, 1200), mk(3, 900), mk(4, 900), mk(5, 600)}
+	addrs := []netip.AddrPort{dstA, dstA, dstB, dstB, dstC}
+
+	w := newRewindTestWriter()
+	var wire [][]byte
+	var starts []int
+	call := 0
+	w.sendFn = func(start, n int) (int, error) {
+		starts = append(starts, start)
+		call++
+		switch call {
+		case 1: // accept only entry 0 (the run to A)
+			wire = append(wire, capturePrepared(w, start, 1)...)
+			return 1, nil
+		case 2: // reject entry 1 (the run to B) outright
+			return -1, &net.OpError{Op: "sendmmsg", Err: unix.EPERM}
+		default: // accept the rest
+			wire = append(wire, capturePrepared(w, start, n)...)
+			return n, nil
+		}
+	}
+
+	written, err := w.WriteBatch(bufs, addrs, nil)
+	if err != nil {
+		t.Fatalf("WriteBatch: %v", err)
+	}
+	if written != 3 {
+		t.Errorf("written = %d, want 3 (B's rejected run is the only casualty)", written)
+	}
+	wantTags := []byte{1, 2, 5}
+	if len(wire) != len(wantTags) {
+		t.Fatalf("wire got %d packets, want %d (dup or loss around the mid-chunk reject)", len(wire), len(wantTags))
+	}
+	for i, b := range wire {
+		if b[0] != wantTags[i] {
+			t.Errorf("wire[%d] tag = %d, want %d", i, b[0], wantTags[i])
+		}
+	}
+	// The resume must reuse the prepared entries: same chunk, advancing
+	// start offsets, no repack (which would restart at 0 with fresh entries).
+	if want := []int{0, 1, 2}; !slices.Equal(starts, want) {
+		t.Errorf("sendFn start offsets = %v, want %v", starts, want)
+	}
+}
+
+// TestWriteBatchMidChunkEIODisablesGSOWithoutDup: an EIO on a GSO entry
+// after earlier entries in the chunk already went out must replay ONLY from
+// the failed run (replanned as single-packet entries) -- the already-sent
+// entries must not be duplicated.
+func TestWriteBatchMidChunkEIODisablesGSOWithoutDup(t *testing.T) {
+	dstA := netip.MustParseAddrPort("127.0.0.1:4242")
+	dstB := netip.MustParseAddrPort("127.0.0.2:4242")
+
+	mk := func(tag byte, n int) []byte {
+		b := make([]byte, n)
+		b[0] = tag
+		return b
+	}
+	// Entry 0: single packet to A. Entry 1: 2-packet GSO run to B.
+	bufs := [][]byte{mk(1, 600), mk(2, 1200), mk(3, 1200)}
+	addrs := []netip.AddrPort{dstA, dstB, dstB}
+
+	w := newRewindTestWriter()
+	var wire [][]byte
+	call := 0
+	w.sendFn = func(start, n int) (int, error) {
+		call++
+		switch call {
+		case 1: // accept entry 0 only
+			wire = append(wire, capturePrepared(w, start, 1)...)
+			return 1, nil
+		case 2: // EIO on the GSO run to B
+			return -1, &net.OpError{Op: "sendmmsg", Err: unix.EIO}
+		default: // replanned single-packet replay
+			wire = append(wire, capturePrepared(w, start, n)...)
+			return n, nil
+		}
+	}
+
+	written, err := w.WriteBatch(bufs, addrs, nil)
+	if err != nil {
+		t.Fatalf("WriteBatch: %v", err)
+	}
+	if w.gsoSupported {
+		t.Error("gsoSupported still true after EIO on a GSO entry")
+	}
+	if written != len(bufs) {
+		t.Errorf("written = %d, want %d", written, len(bufs))
+	}
+	wantTags := []byte{1, 2, 3}
+	if len(wire) != len(wantTags) {
+		t.Fatalf("wire got %d packets, want %d (packet 1 duplicated, or B's run lost)", len(wire), len(wantTags))
+	}
+	for i, b := range wire {
+		if b[0] != wantTags[i] {
+			t.Errorf("wire[%d] tag = %d, want %d", i, b[0], wantTags[i])
+		}
+	}
+}
+
 // TestWriteBatchZeroProgress: sent == 0 with no error must abort with an
 // error rather than spin forever replaying the same chunk.
 func TestWriteBatchZeroProgress(t *testing.T) {
 	w := newRewindTestWriter()
-	w.sendFn = func(n int) (int, error) { return 0, nil }
+	w.sendFn = func(start, n int) (int, error) { return 0, nil }
 	bufs := [][]byte{make([]byte, 100)}
 	addrs := []netip.AddrPort{netip.MustParseAddrPort("127.0.0.1:4242")}
 	if _, err := w.WriteBatch(bufs, addrs, nil); err == nil {
@@ -577,7 +693,7 @@ func TestWriteBatchEIODisablesGSOAndReplays(t *testing.T) {
 	w := newRewindTestWriter()
 	var entryCounts []int
 	call := 0
-	w.sendFn = func(n int) (int, error) {
+	w.sendFn = func(start, n int) (int, error) {
 		entryCounts = append(entryCounts, n)
 		call++
 		if call == 1 {
@@ -640,9 +756,9 @@ func TestGSOEngagesOnLoopback(t *testing.T) {
 	// changing what hits the kernel.
 	var entryCounts []int
 	real := sc.bw.sendFn
-	sc.bw.sendFn = func(n int) (int, error) {
+	sc.bw.sendFn = func(start, n int) (int, error) {
 		entryCounts = append(entryCounts, n)
-		return real(n)
+		return real(start, n)
 	}

 	const numPkts = 8
@@ -68,16 +68,17 @@ type batchWriter struct {
 	cmsgEcnSpace int

 	// entryEnd[e] is the bufs index after the last packet packed into entry
-	// e. Used to rewind i on partial sendmmsg success.
+	// e. entryEnd[e]-entryPkts[e] recovers the bufs index the entry's run
+	// started at, used to rewind i for the GSO-disable replay.
 	entryEnd []int

 	// entryPkts[e] is the number of packets packed into entry e. Not
 	// derivable from entryEnd: skipped runs leave holes in the bufs index space.
 	entryPkts []int

-	// sendFn sends the first n prepared entries. The real syscall in
-	// production; tests inject partial-success and error scripts.
-	sendFn func(n int) (int, error)
+	// sendFn sends n prepared entries beginning at w.msgs[start]. The real
+	// syscall in production; tests inject partial-success and error scripts.
+	sendFn func(start, n int) (int, error)
 }

 func newBatchWriter(fd int, isV4 bool, l *slog.Logger) *batchWriter {
@@ -189,9 +190,10 @@ func parseRelease(r string) (major, minor int) {
 // entries, so one syscall can mix GSO superpackets and plain datagrams.
 // Without GSO support every packet is its own entry.
 //
-// Batches larger than the scratch take one sendmmsg per chunk. A zero-sent
-// error means the kernel rejected entry 0: its packets are dropped and the
-// rest of the chunk is replayed. A partial success replays the remainder.
+// Batches larger than the scratch take one sendmmsg per chunk. A partial
+// success resumes the same prepared entries at the first unsent one — no
+// repacking. A zero-sent error means the kernel rejected the first remaining
+// entry: its packets are dropped and the rest of the chunk resumes in place.
 //
 // Returns the number of packets sent. An error means the call itself
 // failed; a short count means some destinations were undeliverable.
@@ -212,7 +214,6 @@ func (w *batchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []b

 	i := 0
 	for i < len(bufs) {
-		baseI := i
 		entry := 0
 		iovIdx := 0
 		for entry < len(w.msgs) && i < len(bufs) {
@@ -273,25 +274,44 @@ func (w *batchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []b
 			break
 		}

-		sent, serr := w.sendFn(entry)
-		if serr != nil && sent <= 0 {
-			// sent<=0 means entry 0 itself failed. EIO on a superpacket
-			// means the route cannot carry a GSO send even though the
-			// setsockopt probe passed: udp_send_skb() returns EIO when the
+		// Drain the packed entries without repacking: everything the packing
+		// loop wired (iovecs, names, cmsgs) stays intact until the next chunk
+		// overwrites it, so a partial success resumes the same sendmmsg array
+		// at the first unsent entry, and a rejected entry is skipped in place.
+		// Only the GSO-disable path replans, since its entries change shape.
+		done := 0
+		for done < entry {
+			sent, serr := w.sendFn(done, entry-done)
+			if sent > 0 {
+				// Count packets per entry; the bufs index span would
+				// overcount across holes left by skipped runs.
+				for e := done; e < done+sent; e++ {
+					written += w.entryPkts[e]
+				}
+				done += sent
+				continue
+			}
+			if serr == nil {
+				return written, fmt.Errorf("sendmmsg made no progress")
+			}
+			// sent<=0 means the first remaining entry itself failed. EIO on a
+			// superpacket means the route cannot carry a GSO send even though
+			// the setsockopt probe passed: udp_send_skb() returns EIO when the
 			// egress device lacks TX checksum offload (kernels through
 			// 6.10) or when an xfrm policy covers the route. Persistent, so
 			// disable GSO (socket-wide, though the kernel condition is
-			// per-route) and replay the chunk as one-packet entries, still batched.
-			if w.gsoSupported && w.entryPkts[0] >= 2 && errors.Is(serr, unix.EIO) {
+			// per-route) and replay from the failed run as one-packet
+			// entries, still batched.
+			if w.gsoSupported && w.entryPkts[done] >= 2 && errors.Is(serr, unix.EIO) {
 				w.gsoSupported = false
 				w.l.Warn("udp: kernel rejected GSO send, disabling GSO", "error", serr)
 				recordCapability("udp.gso.enabled", false)
-				i = baseI
-				continue
+				i = w.entryEnd[done] - w.entryPkts[done]
+				break
 			}
 			// TODO: a transient zero-sent errno (ENOBUFS under socket-memory
 			// pressure, or a theoretical EINTR) lands here too and drops
-			// entry 0's entire run (up to 63/127 packets). The RX path
+			// the failed entry's run (up to 63/127 packets). The RX path
 			// retries EINTR; consider a bounded retry for those two before
 			// falling through to the per-entry drop.
 			//
@@ -299,27 +319,19 @@ func (w *batchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []b
 			// an unreachable destination, a firewall EPERM, or a PMTU shrink after a roam
 			// (EINVAL, or EMSGSIZE since kernel 6.14, once gso_size no longer fits the path).
 			// Retrying the packets individually cannot succeed where the entry did not, and
-			// disabling GSO cannot make oversized segments fit, so drop the entry and replay whatever was packed after it.
+			// disabling GSO cannot make oversized segments fit, so skip the entry and resume with the rest.
 			// Small-segment entries still pass, so the tunnel stays up while full-size packets drop.
 			w.l.Debug("sendmmsg rejected entry",
 				"error", serr,
-				"udpAddr", addrs[w.entryEnd[0]-w.entryPkts[0]],
-				"packets", w.entryPkts[0],
+				"udpAddr", addrs[w.entryEnd[done]-w.entryPkts[done]],
+				"packets", w.entryPkts[done],
 				"gso", w.gsoSupported,
 			)
-			i = w.entryEnd[0]
-			continue
+			done++
 		}
-		if sent == 0 {
-			return written, fmt.Errorf("sendmmsg made no progress")
-		}
-		// Rewind i to the end of the last sent entry: a no-op on full
-		// success, a replay of the remainder on partial success. Count
-		// packets per entry; the bufs index span would overcount across holes.
-		for e := 0; e < sent; e++ {
-			written += w.entryPkts[e]
-		}
-		i = w.entryEnd[sent-1]
+		// When the drain finished every entry, i already sits past the whole
+		// chunk (including any trailing skipped runs); the GSO-disable break
+		// above rewound it to the failed run for the replanned retry.
 	}
 	return written, nil
 }
@@ -419,10 +431,10 @@ func (w *batchWriter) writeEntryCmsg(entry, runLen, segSize int, ecn byte, dstIs
 	}
 }

-// sendmmsg issues sendmmsg(2) against the first n entries of w.msgs.
-func (w *batchWriter) sendmmsg(n int) (int, error) {
+// sendmmsg issues sendmmsg(2) against n entries of w.msgs starting at start.
+func (w *batchWriter) sendmmsg(start, n int) (int, error) {
 	r1, _, errno := unix.Syscall6(unix.SYS_SENDMMSG, uintptr(w.fd),
-		uintptr(unsafe.Pointer(&w.msgs[0])), uintptr(n),
+		uintptr(unsafe.Pointer(&w.msgs[start])), uintptr(n),
 		0, 0, 0,
 	)
 	sent := int(r1)