From 5c0f6e2b5f4ba356c5ac9fb34965dafecd2f0e44 Mon Sep 17 00:00:00 2001 From: JackDoan Date: Tue, 28 Jul 2026 16:20:14 -0500 Subject: [PATCH] unslop, improve the tio interface --- interface.go | 11 +-- overlay/batch/coalesce_core.go | 44 ++++------ overlay/batch/multi_coalesce.go | 33 +++---- overlay/batch/multi_coalesce_test.go | 84 +++++++++++++++--- overlay/batch/tcp_coalesce.go | 70 +++++++-------- overlay/batch/tcp_coalesce_bench_test.go | 4 +- overlay/batch/tcp_coalesce_test.go | 107 ++++++++++++++--------- overlay/batch/udp_coalesce.go | 51 ++++------- overlay/batch/udp_coalesce_test.go | 59 +++++++------ overlay/tio/segment_other.go | 6 -- overlay/tio/single.go | 11 ++- overlay/tio/tio.go | 13 +-- overlay/tio/tio_gso_linux.go | 2 +- 13 files changed, 255 insertions(+), 240 deletions(-) diff --git a/interface.go b/interface.go index 4508e6ad..98984816 100644 --- a/interface.go +++ b/interface.go @@ -301,16 +301,7 @@ func (f *Interface) activate() error { metrics.GetOrRegisterGauge("routines", nil).Update(int64(f.routines)) for i := range f.queues { - caps := tio.QueueCapabilities(f.queues[i]) - if caps.TSO || caps.USO { - // Multi-lane: TCP gets coalesced when TSO is on, UDP when USO - // is on, everything else (and either lane disabled) falls - // through to passthrough so non-IP / non-TCP-UDP traffic still - // reaches the TUN. - f.batchers[i] = batch.NewMultiCoalescer(f.queues[i], f.l, caps.TSO, caps.USO) - } else { - f.batchers[i] = batch.NewPassthrough(f.queues[i]) - } + f.batchers[i] = batch.NewMultiCoalescer(f.queues[i], f.l) } // On error the caller owns the cleanup, Control.Start cancels the service context diff --git a/overlay/batch/coalesce_core.go b/overlay/batch/coalesce_core.go index e0a8331b..ab098263 100644 --- a/overlay/batch/coalesce_core.go +++ b/overlay/batch/coalesce_core.go @@ -8,21 +8,20 @@ import ( // flowKey identifies a transport flow by {src, dst, sport, dport, family}. // Comparable, so map lookups and linear scans over the slot list stay tight. // Shared by the TCP and UDP coalescers; each coalescer keeps its own -// openSlots map, so a TCP and UDP flow on the same 5-tuple-without-proto -// never alias. +// openSlots map, so a TCP and UDP flow on the same 5-tuple-without-proto never alias. type flowKey struct { src, dst [16]byte sport, dport uint16 isV6 bool } -// initialSlots is the starting capacity of the slot pool. One flow per -// packet is the worst case so this matches a typical carrier-side -// recvmmsg batch on the encrypted UDP socket. +// initialSlots is the starting capacity of the slot pool. +// One flow per packet is the worst case, +// so this matches a typical carrier-side recvmmsg batch on the UDP socket. const initialSlots = 64 -// parsedIP is the IP-level result of parseIPPrologue. The caller layers -// L4-specific parsing (TCP / UDP) on top. +// parsedIP is the IP-level result of parseIPPrologue. +// The caller layers L4-specific parsing (TCP / UDP) on top. type parsedIP struct { fk flowKey ipHdrLen int @@ -92,15 +91,13 @@ func parseIPPrologue(pkt []byte, wantProto byte) (parsedIP, bool) { } // ipHeadersMatch compares the IP portion of two packet header prefixes for -// byte-for-byte equality on every field that must be identical across -// coalesced segments. Size/IPID/IPCsum are masked out. The full DSCP/ECN -// byte (IPv4 ToS / IPv6 traffic class) is compared, matching Linux kernel -// GRO: segments with differing ECN codepoints must not coalesce, otherwise -// ORing e.g. ECT(0) with ECT(1) would fabricate a false CE (congestion) -// mark or mark a Not-ECT flow as ECN-capable. +// byte-for-byte equality on every field that must be identical across coalesced segments. +// Size/IPID/IPCsum are masked out. +// The full DSCP/ECN byte (IPv4 ToS / IPv6 traffic class) is compared, matching Linux kernel GRO: +// segments with differing ECN codepoints must not coalesce, +// otherwise ORing e.g. ECT(0) with ECT(1) would fabricate a false CE (congestion) mark or mark a Not-ECT flow as ECN-capable. // -// The transport (L4) portion of the header is checked separately by the -// per-protocol matcher. +// The transport (L4) portion of the header is checked separately by the per-protocol matcher. func ipHeadersMatch(a, b []byte, isV6 bool) bool { if isV6 { // IPv6: byte 0 = version/TC[7:4], byte 1 = TC[3:0]/flow[19:16], @@ -145,17 +142,14 @@ type Arena struct { buf []byte } -// NewArena returns an Arena with a pre-allocated backing of the given -// capacity. Pass 0 if you don't intend to call Reserve (e.g. a test that -// only feeds the coalescer pre-made []byte packets via Commit). +// NewArena returns an Arena with a pre-allocated backing of the given capacity. func NewArena(capacity int) *Arena { return &Arena{buf: make([]byte, 0, capacity)} } -// Reserve hands out a non-overlapping sz-byte slice from the arena. If the -// request doesn't fit the current backing, a fresh, larger backing is -// allocated; already-borrowed slices reference the old backing and remain -// valid until Reset. +// Reserve hands out a non-overlapping sz-byte slice from the arena. +// If the request doesn't fit the current backing, a fresh, larger backing is allocated. +// Already-borrowed slices reference the old backing and remain valid until Reset. func (a *Arena) Reserve(sz int) []byte { if len(a.buf)+sz > cap(a.buf) { newCap := max(cap(a.buf)*2, sz) @@ -166,9 +160,9 @@ func (a *Arena) Reserve(sz int) []byte { return a.buf[start : start+sz : start+sz] } -// Reset releases every slice handed out since the last Reset. Callers must -// not use any previously-borrowed slice after this returns. The underlying -// backing array is retained so subsequent Reserves don't re-allocate. +// Reset releases every slice handed out since the last Reset. +// Callers must not use any previously-borrowed slice after this returns. +// The underlying backing array is retained so subsequent Reserves don't re-allocate. func (a *Arena) Reset() { a.buf = a.buf[:0] } diff --git a/overlay/batch/multi_coalesce.go b/overlay/batch/multi_coalesce.go index 559f73dc..2ee7724c 100644 --- a/overlay/batch/multi_coalesce.go +++ b/overlay/batch/multi_coalesce.go @@ -10,46 +10,33 @@ import ( // on the IP/L4 protocol of the packet. // // Lanes are processed independently: the TCP coalescer only sees TCP, the -// UDP coalescer only sees UDP, and the passthrough lane handles everything -// else. Per-flow arrival order is preserved because a single 5-tuple only +// UDP coalescer only sees UDP, and the passthrough lane handles everything else. +// Per-flow delivery order is preserved because a single 5-tuple only // ever lands in one lane and each lane preserves its own slot order. // -// Cross-lane order is NOT preserved across the TCP/UDP/passthrough split. -// This is acceptable because the carrier-side recvmmsg path already -// stable-sorts by (peer, message counter) before delivering plaintext -// here, so replay-window invariants are unaffected, and apps observe -// correct per-flow ordering; which is all the IP layer guarantees anyway. -// Do not "fix" this by interleaving lane outputs at flush time; that -// negates the entire point of coalescing (each lane needs to see runs of -// adjacent same-flow packets to coalesce them). +// Cross-lane order is intentionally NOT preserved across the TCP/UDP/passthrough split. type MultiCoalescer struct { tcp *TCPCoalescer udp *UDPCoalescer pt *Passthrough } -// NewMultiCoalescer builds a multi-lane batcher. tcpEnabled lets the caller -// opt out of TCP coalescing (e.g. when the queue can't do TSO); udpEnabled -// likewise gates UDP coalescing (only enable when USO was negotiated). -// Either lane disabled redirects its traffic into the passthrough lane. -func NewMultiCoalescer(w io.Writer, l *slog.Logger, tcpEnabled, udpEnabled bool) *MultiCoalescer { +// NewMultiCoalescer builds a multi-lane batcher over w, based on available protocol support. +func NewMultiCoalescer(w io.Writer, l *slog.Logger) RxBatcher { m := &MultiCoalescer{ pt: NewPassthrough(w), } - if tcpEnabled { - m.tcp = NewTCPCoalescer(w, l) - } - if udpEnabled { - m.udp = NewUDPCoalescer(w) + m.tcp = NewTCPCoalescer(w, l) + m.udp = NewUDPCoalescer(w) + if m.tcp == nil && m.udp == nil { + return m.pt //no offloads? Use passthrough directly. } return m } // Commit dispatches pkt to the appropriate lane based on IP version + L4 proto. -// // On the success path the IP/TCP-or-UDP parse happens here once and the -// parsed struct is handed to the lane via commitParsed so the lane doesn't -// re-walk the header. +// parsed struct is handed to the lane via commitParsed so the lane doesn't re-walk the header. func (m *MultiCoalescer) Commit(pkt []byte) error { if len(pkt) < 20 { return m.pt.Commit(pkt) diff --git a/overlay/batch/multi_coalesce_test.go b/overlay/batch/multi_coalesce_test.go index 655cb708..d92e4ab1 100644 --- a/overlay/batch/multi_coalesce_test.go +++ b/overlay/batch/multi_coalesce_test.go @@ -1,17 +1,33 @@ package batch import ( + "bytes" + "io" "testing" "github.com/slackhq/nebula/test" ) +// newTestMultiCoalescer builds a batcher over w and asserts it really is +// multi-lane. NewMultiCoalescer collapses to a bare Passthrough when w can +// offload neither protocol, and a test that meant to exercise a lane would +// otherwise pass vacuously. +func newTestMultiCoalescer(tb testing.TB, w io.Writer) *MultiCoalescer { + tb.Helper() + b := NewMultiCoalescer(w, test.NewLogger()) + m, ok := b.(*MultiCoalescer) + if !ok { + tb.Fatalf("want a *MultiCoalescer, got %T", b) + } + return m +} + // TestMultiCoalescerRoutesByProto confirms TCP/UDP/other land in the right // lane: TCP and UDP get coalesced when their lanes are enabled, anything // else (ICMP here) falls through to plain Write. func TestMultiCoalescerRoutesByProto(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - m := NewMultiCoalescer(w, test.NewLogger(), true, true) + m := newTestMultiCoalescer(t, w) tcpPay := make([]byte, 1200) udpPay := make([]byte, 1200) @@ -48,12 +64,15 @@ func TestMultiCoalescerRoutesByProto(t *testing.T) { } } -// TestMultiCoalescerDisabledUDPFallsThrough verifies that when the UDP lane -// is disabled (e.g. kernel doesn't support USO), UDP packets still reach -// the kernel via the passthrough lane rather than being lost. -func TestMultiCoalescerDisabledUDPFallsThrough(t *testing.T) { - w := &fakeTunWriter{gsoEnabled: true} - m := NewMultiCoalescer(w, test.NewLogger(), true, false) // TSO on, USO off +// TestMultiCoalescerNoUSOFallsThrough verifies that on a queue without USO +// (older kernel: TSO but no GSO_UDP_L4) the UDP lane never comes up and UDP +// packets still reach the kernel via passthrough rather than being lost. +func TestMultiCoalescerNoUSOFallsThrough(t *testing.T) { + w := &fakeTunWriter{gsoEnabled: true, noUSO: true} + m := newTestMultiCoalescer(t, w) + if m.udp != nil { + t.Fatal("UDP lane must not come up without USO") + } if err := m.Commit(buildUDPv4(1000, 53, make([]byte, 800))); err != nil { t.Fatal(err) @@ -72,10 +91,53 @@ func TestMultiCoalescerDisabledUDPFallsThrough(t *testing.T) { } } -// TestMultiCoalescerDisabledTCPFallsThrough mirrors the TSO=off case. -func TestMultiCoalescerDisabledTCPFallsThrough(t *testing.T) { - w := &fakeTunWriter{gsoEnabled: true} - m := NewMultiCoalescer(w, test.NewLogger(), false, true) // TSO off, USO on +// TestMultiCoalescerNoOffloadsIsPassthrough covers a queue that can't offload +// anything. Both lane constructors refuse, so there's nothing left to +// dispatch between and NewMultiCoalescer hands back the passthrough lane +// itself — no wrapper, no per-packet protocol demux, and every packet reaches +// the kernel in arrival order. This is the case Interface.activate used to +// special-case with a bare Passthrough. +func TestMultiCoalescerNoOffloadsIsPassthrough(t *testing.T) { + w := &fakeTunWriter{gsoEnabled: false} + m := NewMultiCoalescer(w, test.NewLogger()) + + if _, ok := m.(*Passthrough); !ok { + t.Fatalf("want a bare *Passthrough when neither offload is available, got %T", m) + } + pkts := [][]byte{ + buildTCPv4(1000, tcpAck, make([]byte, 1200)), + buildUDPv4(1000, 53, make([]byte, 800)), + buildTCPv4(2200, tcpAck, make([]byte, 1200)), + } + for _, p := range pkts { + if err := m.Commit(p); err != nil { + t.Fatal(err) + } + } + if err := m.Flush(); err != nil { + t.Fatal(err) + } + if len(w.gsoWrites) != 0 { + t.Errorf("no GSO writes possible, got %d", len(w.gsoWrites)) + } + if len(w.writes) != len(pkts) { + t.Fatalf("want %d plain writes, got %d", len(pkts), len(w.writes)) + } + // One lane for everything means arrival order survives end to end. + for i, want := range pkts { + if !bytes.Equal(w.writes[i], want) { + t.Errorf("write %d out of order or corrupt", i) + } + } +} + +// TestMultiCoalescerNoTSOFallsThrough mirrors the no-TSO case. +func TestMultiCoalescerNoTSOFallsThrough(t *testing.T) { + w := &fakeTunWriter{gsoEnabled: true, noTSO: true} + m := newTestMultiCoalescer(t, w) + if m.tcp != nil { + t.Fatal("TCP lane must not come up without TSO") + } pay := make([]byte, 1200) if err := m.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { diff --git a/overlay/batch/tcp_coalesce.go b/overlay/batch/tcp_coalesce.go index 9926a4a0..ab7239c9 100644 --- a/overlay/batch/tcp_coalesce.go +++ b/overlay/batch/tcp_coalesce.go @@ -27,13 +27,13 @@ const tcpCoalesceMaxSegs = 64 // into. IPv6 (40) + TCP with full options (60) = 100 bytes. const tcpCoalesceHdrCap = 100 -// coalesceSlot is one entry in the coalescer's ordered event queue. When -// passthrough is true the slot holds a single borrowed packet that must be -// emitted verbatim (non-TCP, non-admissible TCP, or oversize seed). When -// passthrough is false the slot is an in-progress coalesced superpacket: -// hdrBuf is a mutable copy of the seed's IP+TCP header (we patch total -// length and pseudo-header partial at flush), and payIovs are *borrowed* -// slices from the caller's plaintext buffers — no payload is ever copied. +// coalesceSlot is one entry in the coalescer's ordered event queue. +// When passthrough is true the slot holds a single borrowed packet that must be +// emitted verbatim (non-TCP, non-admissible TCP, or oversize seed). +// When passthrough is false the slot is an in-progress coalesced superpacket. +// hdrBuf is a mutable copy of the seed's IP+TCP header +// (we patch total length and pseudo-header partial at flush) +// payIovs are *borrowed* slices from the caller's plaintext buffers. // The caller (listenOut) must keep those buffers alive until Flush. type coalesceSlot struct { passthrough bool @@ -50,20 +50,18 @@ type coalesceSlot struct { nextSeq uint32 // sealed marks the chain permanently closed: the last-accepted segment had PSH or was sub-gsoSize, // so no append or flush-time merge may follow. - // Distinct from mere eviction out of openSlots (e.g. on seq mismatch), + // Distinct from eviction out of openSlots (e.g. on seq mismatch), // which leaves sealed=false so reorderForFlush can still merge the slot. sealed bool payIovs [][]byte } -// TCPCoalescer accumulates adjacent in-flow TCP data segments across -// multiple concurrent flows and emits each flow's run as a single TSO -// superpacket via tio.GSOWriter. All output — coalesced or not — is -// deferred until Flush so arrival order is preserved on the wire. Owns -// no locks; one coalescer per TUN write queue. +// TCPCoalescer accumulates adjacent in-flow TCP data segments across multiple concurrent flows +// and emits each flow's run as a single TSO superpacket via tio.GSOWriter. +// All output, coalesced or not, is deferred until Flush so arrival order is preserved on the wire. +// Owns no locks; one coalescer per TUN write queue. type TCPCoalescer struct { - plainW io.Writer - gsoW tio.GSOWriter // nil when the queue doesn't support TSO + w tio.GSOWriter // slots is the ordered event queue. Flush walks it once and emits each // entry as either a WriteGSO (coalesced) or a plainW.Write (passthrough). @@ -84,18 +82,19 @@ type TCPCoalescer struct { l *slog.Logger } +// NewTCPCoalescer wraps w, returning nil if w can't accept GSO_TCP writes. func NewTCPCoalescer(w io.Writer, l *slog.Logger) *TCPCoalescer { - c := &TCPCoalescer{ - plainW: w, + gw, ok := tio.SupportsGSO(w, tio.GSOProtoTCP) + if !ok { + return nil + } + return &TCPCoalescer{ + w: gw, slots: make([]*coalesceSlot, 0, initialSlots), openSlots: make(map[flowKey]*coalesceSlot, initialSlots), pool: make([]*coalesceSlot, 0, initialSlots), l: l, } - if gw, ok := tio.SupportsGSO(w, tio.GSOProtoTCP); ok { - c.gsoW = gw - } - return c } // parsedTCP holds the fields extracted from a single parse so later steps @@ -111,8 +110,7 @@ type parsedTCP struct { } // parseTCPBase extracts the flow key and IP/TCP offsets for any TCP packet, -// regardless of whether it's admissible for coalescing. Returns ok=false -// for non-TCP or malformed input. +// regardless of whether it's admissible for coalescing. Returns ok=false for non-TCP or malformed input. // Accepts IPv4 (no options or fragmentation) and IPv6 (no extension headers). func parseTCPBase(pkt []byte) (parsedTCP, bool) { var p parsedTCP @@ -169,10 +167,6 @@ func (p parsedTCP) coalesceable() bool { } func (c *TCPCoalescer) Commit(pkt []byte) error { - if c.gsoW == nil { - c.addPassthrough(pkt) - return nil - } info, ok := parseTCPBase(pkt) if !ok { c.addPassthrough(pkt) @@ -186,10 +180,6 @@ func (c *TCPCoalescer) Commit(pkt []byte) error { // Used by MultiCoalescer.Commit to avoid re-walking the IP/TCP header // after the dispatcher has already done so. func (c *TCPCoalescer) commitParsed(pkt []byte, info parsedTCP) error { - if c.gsoW == nil { - c.addPassthrough(pkt) - return nil - } if !info.coalesceable() { // TCP but not admissible (SYN/FIN/RST/URG/CWR or zero-payload). // Seal this flow's open slot so later in-flow packets don't extend @@ -240,7 +230,7 @@ func (c *TCPCoalescer) Flush() error { for _, s := range c.slots { var err error if s.passthrough { - _, err = c.plainW.Write(s.rawPkt) + _, err = c.w.Write(s.rawPkt) } else { err = c.flushSlot(s) } @@ -266,7 +256,7 @@ func (c *TCPCoalescer) addPassthrough(pkt []byte) { func (c *TCPCoalescer) seed(pkt []byte, info parsedTCP) { if info.hdrLen > tcpCoalesceHdrCap || info.hdrLen+info.payLen > tcpCoalesceBufSize { - // Pathological shape — can't fit our scratch, emit as-is. + // Pathological shape. Can't fit our scratch, emit as-is. c.addPassthrough(pkt) return } @@ -317,8 +307,8 @@ func (c *TCPCoalescer) canAppend(s *coalesceSlot, pkt []byte, info parsedTCP) bo if s.hdrLen+s.totalPay+info.payLen > tcpCoalesceBufSize { return false } - // ECE state must be stable across a burst — receivers expect the - // flag set on every segment of a CE-echoing window or none. + // ECE state must be stable across a burst. + // Receivers expect the flag set on every segment of a CE-echoing window or none. seedFlags := s.hdrBuf[s.ipHdrLen+13] if (seedFlags^info.flags)&tcpFlagEce != 0 { return false @@ -389,7 +379,7 @@ func (c *TCPCoalescer) flushSlot(s *coalesceSlot) error { tcsum := s.ipHdrLen + 16 binary.BigEndian.PutUint16(hdr[tcsum:tcsum+2], foldOnceNoInvert(psum)) - return c.gsoW.WriteGSO(hdr[:s.ipHdrLen], hdr[s.ipHdrLen:], s.payIovs, tio.GSOProtoTCP) + return c.w.WriteGSO(hdr[:s.ipHdrLen], hdr[s.ipHdrLen:], s.payIovs, tio.GSOProtoTCP) } // headersMatch compares two IP+TCP header prefixes for byte-for-byte @@ -421,9 +411,9 @@ func headersMatch(a, b []byte, isV6 bool, ipHdrLen int) bool { } // reorderForFlush neutralizes wire-side reorder that the rxOrder buffer -// couldn't catch (anything crossing a recvmmsg batch boundary). Without -// this pass a small wire reorder — counter 250 arriving in batch K when -// 200..249 are coming in batch K+1 — would seed an out-of-seq slot first +// couldn't catch (anything crossing a recvmmsg batch boundary). +// Without this pass a small wire reorder, counter 250 arriving in batch K when +// 200..249 are coming in batch K+1, would seed an out-of-seq slot first // and emit it ahead of the lower-seq slot, manifesting at the inner TCP // receiver as a much larger reorder than the wire actually had. // @@ -434,7 +424,7 @@ func headersMatch(a, b []byte, isV6 bool, ipHdrLen int) bool { // 2. Sweep once and merge adjacent same-flow slots whose ranges are now // contiguous AND whose tail is gsoSize-aligned. The tail constraint // matters because the kernel TSO splitter chops at gsoSize from the -// start of the merged payload — a short segment in the middle would +// start of the merged payload. A short segment in the middle would // desynchronize every later segment. // // Passthrough slots act as barriers: the merge check skips them on either diff --git a/overlay/batch/tcp_coalesce_bench_test.go b/overlay/batch/tcp_coalesce_bench_test.go index 833a11e3..1226523c 100644 --- a/overlay/batch/tcp_coalesce_bench_test.go +++ b/overlay/batch/tcp_coalesce_bench_test.go @@ -71,7 +71,7 @@ func buildICMPv4() []byte { // between batches, and reports per-packet cost. func runCommitBench(b *testing.B, pkts [][]byte, batchSize int) { b.Helper() - c := NewTCPCoalescer(nopTunWriter{}, test.NewLogger()) + c := newTestTCPCoalescer(b, nopTunWriter{}) b.ReportAllocs() b.SetBytes(int64(len(pkts[0]))) b.ResetTimer() @@ -140,7 +140,7 @@ func BenchmarkCommitNonCoalesceableTCP(b *testing.B) { // is the bench that shows the savings of skipping the lane's re-parse. func runMultiCommitBench(b *testing.B, pkts [][]byte, batchSize int) { b.Helper() - m := NewMultiCoalescer(nopTunWriter{}, test.NewLogger(), true, true) + m := NewMultiCoalescer(nopTunWriter{}, test.NewLogger()) b.ReportAllocs() b.SetBytes(int64(len(pkts[0]))) b.ResetTimer() diff --git a/overlay/batch/tcp_coalesce_test.go b/overlay/batch/tcp_coalesce_test.go index 47e02007..f3100fae 100644 --- a/overlay/batch/tcp_coalesce_test.go +++ b/overlay/batch/tcp_coalesce_test.go @@ -2,6 +2,7 @@ package batch import ( "encoding/binary" + "io" "testing" "github.com/slackhq/nebula/overlay/tio" @@ -11,8 +12,13 @@ import ( // fakeTunWriter records plain Writes and WriteGSO calls without touching a // real TUN fd. WriteGSO records the IP header, transport header, and // borrowed payload fragments separately so tests can inspect each. +// noTSO / noUSO withhold one offload from an otherwise GSO-capable writer, so +// tests can build the half-capable queues real kernels hand us (USO needs a +// newer kernel than TSO). type fakeTunWriter struct { gsoEnabled bool + noTSO bool + noUSO bool writes [][]byte gsoWrites []fakeGSOWrite } @@ -79,7 +85,7 @@ func (w *fakeTunWriter) WriteGSO(hdr []byte, transportHdr []byte, pays [][]byte, } func (w *fakeTunWriter) Capabilities() tio.Capabilities { - return tio.Capabilities{TSO: w.gsoEnabled, USO: w.gsoEnabled} + return tio.Capabilities{TSO: w.gsoEnabled && !w.noTSO, USO: w.gsoEnabled && !w.noUSO} } // buildTCPv4 constructs a minimal IPv4+TCP packet with the given payload, @@ -126,28 +132,43 @@ const ( tcpAckPsh = tcpAck | tcpPsh ) -func TestCoalescerPassthroughWhenGSOUnavailable(t *testing.T) { - w := &fakeTunWriter{gsoEnabled: false} +// newTestTCPCoalescer builds a coalescer over w and fails the test if w can't +// do TSO. Every test but TestNewTCPCoalescerRefusesWhenGSOUnavailable wants the +// GSO path, and the constructor now hands back a nil coalescer otherwise. +func newTestTCPCoalescer(tb testing.TB, w io.Writer) *TCPCoalescer { + tb.Helper() c := NewTCPCoalescer(w, test.NewLogger()) - pkt := buildTCPv4(1000, tcpAck, []byte("hello")) - if err := c.Commit(pkt); err != nil { - t.Fatal(err) + if c == nil { + tb.Fatal("NewTCPCoalescer: writer does not support TSO") } - // No sync write — passthrough is deferred to Flush. - if len(w.writes) != 0 || len(w.gsoWrites) != 0 { - t.Fatalf("no Add-time writes: got writes=%d gso=%d", len(w.writes), len(w.gsoWrites)) + return c +} + +// TestNewTCPCoalescerRefusesWhenGSOUnavailable pins the constructor +// precondition: no TSO, no coalescer. There's no degraded mode — the caller +// (MultiCoalescer) sends TCP down the passthrough lane instead. +func TestNewTCPCoalescerRefusesWhenGSOUnavailable(t *testing.T) { + if c := NewTCPCoalescer(&fakeTunWriter{gsoEnabled: false}, test.NewLogger()); c != nil { + t.Fatalf("want nil for a non-TSO writer, got %v", c) } - if err := c.Flush(); err != nil { - t.Fatal(err) - } - if len(w.writes) != 1 || len(w.gsoWrites) != 0 { - t.Fatalf("want single plain write, got writes=%d gso=%d", len(w.writes), len(w.gsoWrites)) + // A writer that isn't a GSOWriter at all is refused the same way. + if c := NewTCPCoalescer(&plainOnlyWriter{}, test.NewLogger()); c != nil { + t.Fatalf("want nil for a plain writer, got %v", c) } } +// plainOnlyWriter is an io.Writer with no GSO support at all — the +// single-packet Queue shape. +type plainOnlyWriter struct{ writes int } + +func (w *plainOnlyWriter) Write(p []byte) (int, error) { + w.writes++ + return len(p), nil +} + func TestCoalescerNonTCPPassthrough(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pkt := make([]byte, 28) pkt[0] = 0x45 binary.BigEndian.PutUint16(pkt[2:4], 28) @@ -167,7 +188,7 @@ func TestCoalescerNonTCPPassthrough(t *testing.T) { func TestCoalescerSeedThenFlushAlone(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pkt := buildTCPv4(1000, tcpAck, make([]byte, 1000)) if err := c.Commit(pkt); err != nil { t.Fatal(err) @@ -194,7 +215,7 @@ func TestCoalescerSeedThenFlushAlone(t *testing.T) { func TestCoalescerCoalescesAdjacentACKs(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) if err := c.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { t.Fatal(err) @@ -234,7 +255,7 @@ func TestCoalescerCoalescesAdjacentACKs(t *testing.T) { func TestCoalescerRejectsSeqGap(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) if err := c.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { t.Fatal(err) @@ -253,7 +274,7 @@ func TestCoalescerRejectsSeqGap(t *testing.T) { func TestCoalescerRejectsFlagMismatch(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) if err := c.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { t.Fatal(err) @@ -274,7 +295,7 @@ func TestCoalescerRejectsFlagMismatch(t *testing.T) { func TestCoalescerRejectsFIN(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) fin := buildTCPv4(1000, tcpAck|tcpFin, []byte("x")) if err := c.Commit(fin); err != nil { t.Fatal(err) @@ -290,7 +311,7 @@ func TestCoalescerRejectsFIN(t *testing.T) { func TestCoalescerShortLastSegmentClosesChain(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) full := make([]byte, 1200) half := make([]byte, 500) if err := c.Commit(buildTCPv4(1000, tcpAck, full)); err != nil { @@ -325,7 +346,7 @@ func TestCoalescerShortLastSegmentClosesChain(t *testing.T) { func TestCoalescerPSHFinalizesChain(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) if err := c.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { t.Fatal(err) @@ -355,7 +376,7 @@ func TestCoalescerPSHFinalizesChain(t *testing.T) { // coalescer drops it the sender's push signal never reaches the receiver. func TestCoalescerPropagatesPSHFromAppended(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) // Seed has no PSH; second segment carries PSH and seals the chain. if err := c.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { @@ -383,7 +404,7 @@ func TestCoalescerPropagatesPSHFromAppended(t *testing.T) { func TestCoalescerRejectsDifferentFlow(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) p1 := buildTCPv4(1000, tcpAck, pay) p2 := buildTCPv4(2200, tcpAck, pay) @@ -405,7 +426,7 @@ func TestCoalescerRejectsDifferentFlow(t *testing.T) { func TestCoalescerRejectsIPOptions(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 500) pkt := buildTCPv4(1000, tcpAck, pay) // Bump IHL to 6 to simulate 4 bytes of IP options. Don't actually add @@ -425,7 +446,7 @@ func TestCoalescerRejectsIPOptions(t *testing.T) { func TestCoalescerCapBySegments(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 512) seq := uint32(1000) for i := 0; i < tcpCoalesceMaxSegs+5; i++ { @@ -449,7 +470,7 @@ func TestCoalescerCapBySegments(t *testing.T) { // flows coalesce independently in a single Flush. func TestCoalescerMultipleFlowsInSameBatch(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) // Flow A: sport 1000. Flow B: sport 3000. @@ -506,7 +527,7 @@ func TestCoalescerMultipleFlowsInSameBatch(t *testing.T) { // writing passthrough packets synchronously. func TestCoalescerPreservesArrivalOrder(t *testing.T) { w := &orderedFakeWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) // Sequence: coalesceable TCP, ICMP (passthrough), coalesceable TCP on // a different flow. Expected emit order: gso(X), plain(ICMP), gso(Y). pay := make([]byte, 1200) @@ -574,7 +595,7 @@ func stringSliceEq(a, b []string) bool { // packet (SYN) mid-flow only flushes its own flow, not others. func TestCoalescerInterleavedFlowsPreserveOrdering(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) // Flow A two segments. @@ -679,7 +700,7 @@ func buildTCPv6(tcLow byte, seq uint32, flags byte, payload []byte) []byte { // retains ECE on the wire. func TestCoalescerCoalescesEceFlow(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) flags := byte(tcpAck | tcpEce) if err := c.Commit(buildTCPv4(1000, flags, pay)); err != nil { @@ -708,7 +729,7 @@ func TestCoalescerCoalescesEceFlow(t *testing.T) { // in-flow segment seeds a new slot rather than extending the prior burst. func TestCoalescerCwrSealsFlow(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) if err := c.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { t.Fatal(err) @@ -741,7 +762,7 @@ func TestCoalescerCwrSealsFlow(t *testing.T) { // a CE-echoing window or none. func TestCoalescerEceMismatchReseeds(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) if err := c.Commit(buildTCPv4(1000, tcpAck|tcpEce, pay)); err != nil { t.Fatal(err) @@ -771,7 +792,7 @@ func TestCoalescerEceMismatchReseeds(t *testing.T) { // across the whole burst. func TestCoalescerDifferingECNReseeds(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) if err := c.Commit(buildTCPv4WithToS(ecnECT0, 1000, tcpAck, pay)); err != nil { t.Fatal(err) @@ -816,7 +837,7 @@ func TestCoalescerDifferingECNReseeds(t *testing.T) { // codepoint, and neither may end up CE-marked. func TestCoalescerECT0ThenECT1NoCE(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) if err := c.Commit(buildTCPv4WithToS(ecnECT0, 1000, tcpAck, pay)); err != nil { t.Fatal(err) @@ -846,7 +867,7 @@ func TestCoalescerECT0ThenECT1NoCE(t *testing.T) { // six DSCP bits must match too. func TestCoalescerDscpMismatchReseeds(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) // Same ECN (Not-ECT), different DSCP (0x10 vs 0x20 in upper 6 bits). tosA := byte(0x10<<2) | ecnNotECT @@ -869,7 +890,7 @@ func TestCoalescerDscpMismatchReseeds(t *testing.T) { // TestCoalescerCoalescesEceFlow. func TestCoalescerIPv6CoalescesEceFlow(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) flags := byte(tcpAck | tcpEce) if err := c.Commit(buildTCPv6(0, 1000, flags, pay)); err != nil { @@ -900,7 +921,7 @@ func TestCoalescerIPv6CoalescesEceFlow(t *testing.T) { // seen had the wire never reordered. func TestCoalescerSortsReorderedSeedsAndMerges(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) // Arrival order: seq 1000, 3400, 2200. The 3400 seeds a separate slot // because 3400 != nextSeq=2200, then 2200 fails to extend the 3400 slot @@ -936,7 +957,7 @@ func TestCoalescerSortsReorderedSeedsAndMerges(t *testing.T) { // without any cross-flow contamination. func TestCoalescerSortAcrossFlowsMergesEachIndependently(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) // Flow A (sport 1000) seq 100, 1300; flow B (sport 3000) seq 500, 1700. // Arrival: A.1300, B.1700, A.100, B.500 — every flow reordered. @@ -987,7 +1008,7 @@ func TestCoalescerSortAcrossFlowsMergesEachIndependently(t *testing.T) { // boundary by an arbitrary number of segments. func TestCoalescerSortKeepsPSHBoundary(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) // Seq 1000 (no PSH) + 2200 (PSH) → seal one slot with PSH set. // Seq 3400 (no PSH) is contiguous to 3400 from seq 2200+1200; without @@ -1015,7 +1036,7 @@ func TestCoalescerSortKeepsPSHBoundary(t *testing.T) { // is sorted/merged independently. func TestCoalescerSortKeepsPassthroughBarrier(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) // First two segments seed S1 (then a 3400 reorder seeds S2). if err := c.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { @@ -1049,7 +1070,7 @@ func TestCoalescerSortKeepsPassthroughBarrier(t *testing.T) { // 0x30, so ipHeadersMatch (comparing byte 1 fully) still splits them. func TestCoalescerIPv6DifferingECNReseeds(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) // tcLow is the low 4 bits of TC; ECN occupies the bottom 2 of those. if err := c.Commit(buildTCPv6(ecnECT0, 1000, tcpAck, pay)); err != nil { @@ -1124,7 +1145,7 @@ func TestSortRunZeroAllocs(t *testing.T) { // synthesize it from the seal bool. func TestCoalescerMergeShortTailDoesNotFabricatePSH(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) short := make([]byte, 600) // Arrival: seq 3400 (full), 4600 (short, seals the slot), then the @@ -1162,7 +1183,7 @@ func TestCoalescerMergeShortTailDoesNotFabricatePSH(t *testing.T) { // source slot's tail really carried PSH, the merged header must keep it. func TestCoalescerMergePreservesRealPSH(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewTCPCoalescer(w, test.NewLogger()) + c := newTestTCPCoalescer(t, w) pay := make([]byte, 1200) short := make([]byte, 600) if err := c.Commit(buildTCPv4(3400, tcpAck, pay)); err != nil { diff --git a/overlay/batch/udp_coalesce.go b/overlay/batch/udp_coalesce.go index d8aac5e6..fe5350d9 100644 --- a/overlay/batch/udp_coalesce.go +++ b/overlay/batch/udp_coalesce.go @@ -36,44 +36,35 @@ type udpSlot struct { numSeg int totalPay int // sealed closes the chain: set when a sub-gsoSize segment is appended - // (kernel UDP-GSO requires every segment but the last to be exactly - // gsoSize) or when limits are hit. No further appends after. + // (kernel UDP-GSO requires every segment but the last to be exactly gsoSize) + // or when limits are hit. No further appends after. sealed bool payIovs [][]byte } // UDPCoalescer accumulates adjacent in-flow UDP datagrams across multiple -// concurrent flows and emits each flow's run as a single GSO_UDP_L4 -// superpacket via tio.GSOWriter. Falls back to per-packet writes when the -// underlying writer doesn't support USO. +// concurrent flows and emits each flow's run as a single GSO_UDP_L4 superpacket via tio.GSOWriter. // Preserves the in-flow order of packets as they are Commit-ed // // Owns no locks; one coalescer per TUN write queue. type UDPCoalescer struct { - plainW io.Writer - gsoW tio.GSOWriter // nil when the queue can't accept GSO_UDP_L4 - + w tio.GSOWriter slots []*udpSlot openSlots map[flowKey]*udpSlot pool []*udpSlot } -// NewUDPCoalescer wraps w. The caller is responsible for only constructing -// this when the underlying Queue's Capabilities advertise USO; otherwise -// the kernel may reject GSO_UDP_L4 writes. If w does not implement -// tio.GSOWriter at all (single-packet Queue), the coalescer degrades to -// plain Writes — same defensive shape as the TCP coalescer. func NewUDPCoalescer(w io.Writer) *UDPCoalescer { - c := &UDPCoalescer{ - plainW: w, + gw, ok := tio.SupportsGSO(w, tio.GSOProtoUDP) + if !ok { + return nil + } + return &UDPCoalescer{ + w: gw, slots: make([]*udpSlot, 0, initialSlots), openSlots: make(map[flowKey]*udpSlot, initialSlots), pool: make([]*udpSlot, 0, initialSlots), } - if gw, ok := tio.SupportsGSO(w, tio.GSOProtoUDP); ok { - c.gsoW = gw - } - return c } // parsedUDP holds the fields extracted from a single parse so later steps @@ -115,10 +106,6 @@ func parseUDP(pkt []byte) (parsedUDP, bool) { // Commit borrows pkt. The caller must keep pkt valid until the next Flush. func (c *UDPCoalescer) Commit(pkt []byte) error { - if c.gsoW == nil { - c.addPassthrough(pkt) - return nil - } info, ok := parseUDP(pkt) if !ok { c.addPassthrough(pkt) @@ -131,16 +118,8 @@ func (c *UDPCoalescer) Commit(pkt []byte) error { // already verified parseUDP succeeded. Used by MultiCoalescer.Commit to // avoid re-walking the IP/UDP header. func (c *UDPCoalescer) commitParsed(pkt []byte, info parsedUDP) error { - if c.gsoW == nil { - c.addPassthrough(pkt) - return nil - } // A zero-length UDP datagram (UDP `length` == 8) is legal and must still - // reach the TUN, but it can't be coalesced: a GSO slot would store an - // empty payload iovec and the kernel has nothing to segment. Seal any - // open chain for this flow (so a later, non-empty datagram seeds fresh - // *after* this one and per-flow arrival order is preserved) and deliver - // it as a plain single datagram. + // reach the TUN, but it can't be coalesced. if info.payLen == 0 { delete(c.openSlots, info.fk) c.addPassthrough(pkt) @@ -154,7 +133,7 @@ func (c *UDPCoalescer) commitParsed(pkt []byte, info parsedUDP) error { } return nil } - // Can't extend — seal it and fall through to seed a fresh slot. + // Can't extend. Seal it and fall through to seed a fresh slot. delete(c.openSlots, info.fk) } c.seed(pkt, info) @@ -166,7 +145,7 @@ func (c *UDPCoalescer) Flush() error { for _, s := range c.slots { var err error if s.passthrough { - _, err = c.plainW.Write(s.rawPkt) + _, err = c.w.Write(s.rawPkt) } else { err = c.flushSlot(s) } @@ -296,7 +275,7 @@ func (c *UDPCoalescer) flushSlot(s *udpSlot) error { udpCsumOff := s.ipHdrLen + 6 binary.BigEndian.PutUint16(hdr[udpCsumOff:udpCsumOff+2], foldOnceNoInvert(psum)) - return c.gsoW.WriteGSO(hdr[:s.ipHdrLen], hdr[s.ipHdrLen:], s.payIovs, tio.GSOProtoUDP) + return c.w.WriteGSO(hdr[:s.ipHdrLen], hdr[s.ipHdrLen:], s.payIovs, tio.GSOProtoUDP) } // udpHeadersMatch compares two IP+UDP header prefixes for byte-equality on @@ -308,7 +287,7 @@ func udpHeadersMatch(a, b []byte, isV6 bool, ipHdrLen int) bool { if !ipHeadersMatch(a, b, isV6) { return false } - // UDP: compare sport+dport ([0:4]). Skip length [4:6] and checksum [6:8] — + // UDP: compare sport+dport ([0:4]). Skip length [4:6] and checksum [6:8] // length varies (we rewrite at flush) and the checksum will be redone. udp := ipHdrLen if a[udp] != b[udp] || a[udp+1] != b[udp+1] || a[udp+2] != b[udp+2] || a[udp+3] != b[udp+3] { diff --git a/overlay/batch/udp_coalesce_test.go b/overlay/batch/udp_coalesce_test.go index c563d03f..801bb99b 100644 --- a/overlay/batch/udp_coalesce_test.go +++ b/overlay/batch/udp_coalesce_test.go @@ -2,6 +2,7 @@ package batch import ( "encoding/binary" + "io" "testing" ) @@ -58,27 +59,31 @@ func buildUDPv6(sport, dport uint16, payload []byte) []byte { return pkt } -func TestUDPCoalescerPassthroughWhenGSOUnavailable(t *testing.T) { - w := &fakeTunWriter{gsoEnabled: false} +// newTestUDPCoalescer builds a coalescer over w and fails the test if w can't +// do USO. See newTestTCPCoalescer. +func newTestUDPCoalescer(tb testing.TB, w io.Writer) *UDPCoalescer { + tb.Helper() c := NewUDPCoalescer(w) - pkt := buildUDPv4(1000, 53, make([]byte, 100)) - if err := c.Commit(pkt); err != nil { - t.Fatal(err) + if c == nil { + tb.Fatal("NewUDPCoalescer: writer does not support USO") } - if len(w.writes) != 0 || len(w.gsoWrites) != 0 { - t.Fatalf("no Add-time writes: writes=%d gso=%d", len(w.writes), len(w.gsoWrites)) + return c +} + +// TestNewUDPCoalescerRefusesWhenGSOUnavailable mirrors the TCP precondition: +// no USO, no coalescer. +func TestNewUDPCoalescerRefusesWhenGSOUnavailable(t *testing.T) { + if c := NewUDPCoalescer(&fakeTunWriter{gsoEnabled: false}); c != nil { + t.Fatalf("want nil for a non-USO writer, got %v", c) } - if err := c.Flush(); err != nil { - t.Fatal(err) - } - if len(w.writes) != 1 || len(w.gsoWrites) != 0 { - t.Fatalf("want single plain write, got writes=%d gso=%d", len(w.writes), len(w.gsoWrites)) + if c := NewUDPCoalescer(&plainOnlyWriter{}); c != nil { + t.Fatalf("want nil for a plain writer, got %v", c) } } func TestUDPCoalescerNonUDPPassthrough(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewUDPCoalescer(w) + c := newTestUDPCoalescer(t, w) // ICMP packet pkt := make([]byte, 28) pkt[0] = 0x45 @@ -99,7 +104,7 @@ func TestUDPCoalescerNonUDPPassthrough(t *testing.T) { func TestUDPCoalescerSeedThenFlushAlone(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewUDPCoalescer(w) + c := newTestUDPCoalescer(t, w) pkt := buildUDPv4(1000, 53, make([]byte, 800)) if err := c.Commit(pkt); err != nil { t.Fatal(err) @@ -116,7 +121,7 @@ func TestUDPCoalescerSeedThenFlushAlone(t *testing.T) { func TestUDPCoalescerCoalescesEqualSized(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewUDPCoalescer(w) + c := newTestUDPCoalescer(t, w) pay := make([]byte, 1200) for i := 0; i < 3; i++ { if err := c.Commit(buildUDPv4(1000, 53, pay)); err != nil { @@ -156,7 +161,7 @@ func TestUDPCoalescerCoalescesEqualSized(t *testing.T) { // Last segment may be shorter, sealing the chain. func TestUDPCoalescerShortLastSegmentSeals(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewUDPCoalescer(w) + c := newTestUDPCoalescer(t, w) full := make([]byte, 1200) tail := make([]byte, 600) if err := c.Commit(buildUDPv4(1000, 53, full)); err != nil { @@ -189,7 +194,7 @@ func TestUDPCoalescerShortLastSegmentSeals(t *testing.T) { // A larger-than-gsoSize packet cannot extend the slot — it reseeds. func TestUDPCoalescerLargerThanSeedReseeds(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewUDPCoalescer(w) + c := newTestUDPCoalescer(t, w) if err := c.Commit(buildUDPv4(1000, 53, make([]byte, 800))); err != nil { t.Fatal(err) } @@ -207,7 +212,7 @@ func TestUDPCoalescerLargerThanSeedReseeds(t *testing.T) { // Different 5-tuples must not coalesce. func TestUDPCoalescerDifferentFlowsKeepSeparate(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewUDPCoalescer(w) + c := newTestUDPCoalescer(t, w) pay := make([]byte, 800) if err := c.Commit(buildUDPv4(1000, 53, pay)); err != nil { t.Fatal(err) @@ -238,7 +243,7 @@ func TestUDPCoalescerDifferentFlowsKeepSeparate(t *testing.T) { // Caps at udpCoalesceMaxSegs. func TestUDPCoalescerCapsAtMaxSegs(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewUDPCoalescer(w) + c := newTestUDPCoalescer(t, w) pay := make([]byte, 100) for i := 0; i < udpCoalesceMaxSegs+5; i++ { if err := c.Commit(buildUDPv4(1000, 53, pay)); err != nil { @@ -267,7 +272,7 @@ func TestUDPCoalescerCapsAtMaxSegs(t *testing.T) { // trailing Not-ECT datagram seeds another. func TestUDPCoalescerDifferingECNReseeds(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewUDPCoalescer(w) + c := newTestUDPCoalescer(t, w) pay := make([]byte, 800) pkt0 := buildUDPv4(1000, 53, pay) // ECN=00 (Not-ECT) pkt1 := buildUDPv4(1000, 53, pay) @@ -298,7 +303,7 @@ func TestUDPCoalescerDifferingECNReseeds(t *testing.T) { // IPv6 path: same flow, equal-sized → coalesced. func TestUDPCoalescerIPv6Coalesces(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewUDPCoalescer(w) + c := newTestUDPCoalescer(t, w) pay := make([]byte, 1200) for i := 0; i < 3; i++ { if err := c.Commit(buildUDPv6(1000, 53, pay)); err != nil { @@ -334,7 +339,7 @@ func TestUDPCoalescerIPv6Coalesces(t *testing.T) { // DSCP differences must reseed: udpHeadersMatch compares the full ToS byte. func TestUDPCoalescerDSCPMismatchReseeds(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewUDPCoalescer(w) + c := newTestUDPCoalescer(t, w) pay := make([]byte, 800) pkt0 := buildUDPv4(1000, 53, pay) pkt1 := buildUDPv4(1000, 53, pay) @@ -356,7 +361,7 @@ func TestUDPCoalescerDSCPMismatchReseeds(t *testing.T) { // Fragmented IPv4 must not be coalesced. func TestUDPCoalescerFragmentedIPv4PassesThrough(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewUDPCoalescer(w) + c := newTestUDPCoalescer(t, w) pkt := buildUDPv4(1000, 53, make([]byte, 200)) binary.BigEndian.PutUint16(pkt[6:8], 0x2000) // MF=1 if err := c.Commit(pkt); err != nil { @@ -377,7 +382,7 @@ func TestUDPCoalescerFragmentedIPv4PassesThrough(t *testing.T) { // reach the GSO path. Regression: must not panic and must be written. func TestUDPCoalescerZeroLengthPayloadPassesThrough(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewUDPCoalescer(w) + c := newTestUDPCoalescer(t, w) pkt := buildUDPv4(1000, 53, nil) // UDP length 8, zero payload if err := c.Commit(pkt); err != nil { t.Fatal(err) @@ -396,7 +401,7 @@ func TestUDPCoalescerZeroLengthPayloadPassesThrough(t *testing.T) { // IPv6 zero-length UDP datagram: same passthrough contract as v4. func TestUDPCoalescerZeroLengthPayloadIPv6PassesThrough(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewUDPCoalescer(w) + c := newTestUDPCoalescer(t, w) pkt := buildUDPv6(1000, 53, nil) // UDP length 8, zero payload if err := c.Commit(pkt); err != nil { t.Fatal(err) @@ -417,7 +422,7 @@ func TestUDPCoalescerZeroLengthPayloadIPv6PassesThrough(t *testing.T) { // wire — per-flow arrival order (full, empty, full) must be preserved. func TestUDPCoalescerZeroLengthMidFlowSealsAndPreservesOrder(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewUDPCoalescer(w) + c := newTestUDPCoalescer(t, w) full := make([]byte, 800) if err := c.Commit(buildUDPv4(1000, 53, full)); err != nil { t.Fatal(err) @@ -441,7 +446,7 @@ func TestUDPCoalescerZeroLengthMidFlowSealsAndPreservesOrder(t *testing.T) { // IPv4 with options is not admissible (we require IHL=5). func TestUDPCoalescerIPv4WithOptionsPassesThrough(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} - c := NewUDPCoalescer(w) + c := newTestUDPCoalescer(t, w) pkt := buildUDPv4(1000, 53, make([]byte, 200)) pkt[0] = 0x46 // IHL = 6 (24-byte IPv4 header — has options) if err := c.Commit(pkt); err != nil { diff --git a/overlay/tio/segment_other.go b/overlay/tio/segment_other.go index 6bae0030..993a967d 100644 --- a/overlay/tio/segment_other.go +++ b/overlay/tio/segment_other.go @@ -8,12 +8,6 @@ func protoFromGSOType(_ uint8) (GSOProto, error) { return 0, fmt.Errorf("GSO unsupported") } -// SegmentSuperpacket invokes fn once per segment of pkt. On non-Linux -// builds (and Android/e2e_testing) this package does not provide a Queue -// implementation, so any caller that does construct a Packet here can only -// be operating on non-superpacket bytes and the stub forwards them -// directly. A non-zero GSO field is a programming error from the caller -// and returns an explicit error rather than silently misbehaving. func SegmentSuperpacket(pkt Packet, fn func(seg []byte) error) error { if pkt.GSO.IsSuperpacket() { return fmt.Errorf("tio: GSO superpacket on platform without segmentation support") diff --git a/overlay/tio/single.go b/overlay/tio/single.go index 7a5be8bc..a24c2242 100644 --- a/overlay/tio/single.go +++ b/overlay/tio/single.go @@ -4,9 +4,8 @@ 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. +// 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) @@ -14,9 +13,9 @@ type singleQueue struct { 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. +// 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)} } diff --git a/overlay/tio/tio.go b/overlay/tio/tio.go index cac45617..e1e02cfd 100644 --- a/overlay/tio/tio.go +++ b/overlay/tio/tio.go @@ -93,14 +93,6 @@ type CapsProvider interface { Capabilities() Capabilities } -// QueueCapabilities returns q's negotiated offload capabilities, or the zero value when q does not advertise any. -func QueueCapabilities(q io.Writer) Capabilities { - if cp, ok := q.(CapsProvider); ok { - return cp.Capabilities() - } - return Capabilities{} -} - // GSOProto selects the L4 protocol for a GSO superpacket. // Determines which VIRTIO_NET_HDR_GSO_* type the writer stamps and which checksum offset // inside the transport header virtio NEEDS_CSUM expects. @@ -126,9 +118,10 @@ const ( // Every segment in pays except possibly the last is exactly the same size. // proto picks the L4 protocol so the writer knows which gsoType / CsumOffset to set. // -// Callers should also consult CapsProvider (via SupportsGSO or QueueCapabilities) -// for the per-protocol negotiated capability: USO may not have been negotiated even when TSO was. +// Callers should also consult CapsProvider (via SupportsGSO) for the per-protocol negotiated capability: +// USO may not have been negotiated even when TSO was. type GSOWriter interface { + io.Writer CapsProvider WriteGSO(hdr []byte, transportHdr []byte, pays [][]byte, proto GSOProto) error } diff --git a/overlay/tio/tio_gso_linux.go b/overlay/tio/tio_gso_linux.go index c4822894..2f62d468 100644 --- a/overlay/tio/tio_gso_linux.go +++ b/overlay/tio/tio_gso_linux.go @@ -173,7 +173,7 @@ func (r *Offload) Read() ([]Packet, error) { return nil, err } if err := r.decodeRead(n); err != nil { - // Drop and read again — a bad packet should not kill the reader. + // Drop and read again. A bad packet should not kill the reader. continue } break