From e8d6be1dd9ee1e7cce78876ae6a7133f1a02dc02 Mon Sep 17 00:00:00 2001 From: JackDoan Date: Wed, 29 Jul 2026 17:55:13 -0500 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_014ugV2edVqoz3tBvq9J6yWp --- udp/udp_linux_fixes_test.go | 142 ++++++++++++++++++++++++++++++++---- udp/udp_linux_writebatch.go | 84 ++++++++++++--------- 2 files changed, 177 insertions(+), 49 deletions(-) diff --git a/udp/udp_linux_fixes_test.go b/udp/udp_linux_fixes_test.go index 6a000393..03696b76 100644 --- a/udp/udp_linux_fixes_test.go +++ b/udp/udp_linux_fixes_test.go @@ -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 diff --git a/udp/udp_linux_writebatch.go b/udp/udp_linux_writebatch.go index dc1e1726..35ddbec6 100644 --- a/udp/udp_linux_writebatch.go +++ b/udp/udp_linux_writebatch.go @@ -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)