diff --git a/overlay/batch/tcp_coalesce.go b/overlay/batch/tcp_coalesce.go index 9d445563..8fd9b0ae 100644 --- a/overlay/batch/tcp_coalesce.go +++ b/overlay/batch/tcp_coalesce.go @@ -35,8 +35,17 @@ const tcpCoalesceHdrCap = 100 // (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. + +const ( + passthroughFalse = iota + // passthroughTrue means a sync-point packet, that may not be re-ordered + passthroughTrue + // passthroughACK packets are "passed through" without coalescing, but traffic "after" them may be pulled forward to facilitate coalescing. + passthroughACK +) + type coalesceSlot struct { - passthrough bool + passthrough uint8 // rawPkt is borrowed: the whole packet for passthrough slots, the seed // packet for coalesce slots. A coalesce slot that never grows past one // segment is emitted from rawPkt so its original (already valid) L4 @@ -52,6 +61,11 @@ type coalesceSlot struct { numSeg int totalPay int nextSeq uint32 + // tsVal is the TCP timestamp of the slot's seed segment (uniform across + // the slot: headersMatch requires byte-equal options for every append and + // merge). Sort key only, see compareCoalesceSlots. + tsVal uint32 + hasTS bool // 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 eviction out of openSlots (e.g. on seq mismatch), @@ -60,6 +74,10 @@ type coalesceSlot struct { payIovs [][]byte } +func (c *coalesceSlot) isPassthrough() bool { + return c.passthrough != passthroughFalse +} + // 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. @@ -68,7 +86,7 @@ type TCPCoalescer struct { 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). + // entry as either a WriteGSO (coalesced) or a w.Write (passthrough). slots []*coalesceSlot // openSlots maps a flow key to its most recent non-sealed slot, so new // segments can extend an in-progress superpacket in O(1). Slots are @@ -112,6 +130,7 @@ type parsedTCP struct { payLen int seq uint32 flags byte + options []byte } // parseTCPBase extracts the flow key and IP/TCP offsets for any TCP packet, @@ -140,10 +159,14 @@ func parseTCPBase(pkt []byte) (parsedTCP, bool) { p.tcpHdrLen = tcpOff p.hdrLen = p.ipHdrLen + tcpOff p.payLen = len(pkt) - p.hdrLen - p.seq = binary.BigEndian.Uint32(pkt[p.ipHdrLen+4 : p.ipHdrLen+8]) - p.flags = pkt[p.ipHdrLen+13] p.fk.sport = binary.BigEndian.Uint16(pkt[p.ipHdrLen : p.ipHdrLen+2]) p.fk.dport = binary.BigEndian.Uint16(pkt[p.ipHdrLen+2 : p.ipHdrLen+4]) + p.seq = binary.BigEndian.Uint32(pkt[p.ipHdrLen+4 : p.ipHdrLen+8]) + p.flags = pkt[p.ipHdrLen+13] + //window: 14, 15 + //csum: 16, 17 + //urg: 18, 19 + p.options = pkt[p.ipHdrLen+20 : p.ipHdrLen+p.tcpHdrLen : p.ipHdrLen+p.tcpHdrLen] return p, true } @@ -204,7 +227,7 @@ func (c *TCPCoalescer) commitParsed(pkt []byte, info parsedTCP) error { // evict keeps a bidirectional flow's inbound data run coalescing // across the peer ACKs interleaved into it — kernel GRO likewise // doesn't flush held data on a pure ACK. - c.addPassthrough(pkt) + c.addPassthroughACK(pkt, info) return nil } // TCP but not admissible (SYN/FIN/RST/URG/CWR or a shape the flow @@ -260,7 +283,7 @@ func (c *TCPCoalescer) Flush() error { var first error for _, s := range c.slots { var err error - if s.passthrough || s.numSeg == 1 { + if s.isPassthrough() || s.numSeg == 1 { // A slot that never grew (nor absorbed a merge) is byte-identical // to the packet it was seeded from; ship the original so its valid // checksum rides the DATA_VALID path instead of paying a kernel @@ -285,11 +308,27 @@ func (c *TCPCoalescer) Flush() error { func (c *TCPCoalescer) addPassthrough(pkt []byte) { s := c.take() - s.passthrough = true + s.passthrough = passthroughTrue s.rawPkt = pkt c.slots = append(c.slots, s) } +// addPassthroughACK commits a pure ACK as a passthrough slot that keeps its +// flow identity and sort keys. Unlike addPassthrough slots it does not split +// sort runs, so reorderForFlush may sort same-flow data across it (the +// contract allows data to overtake a bare ACK). A pure ACK's seq is the +// sender's snd_nxt, which orders it after all data the peer sent before it, +// and the TSval-first comparator keeps it behind any older-timestamp data. +func (c *TCPCoalescer) addPassthroughACK(pkt []byte, info parsedTCP) { + s := c.take() + s.passthrough = passthroughACK + s.rawPkt = pkt + s.fk = info.fk + s.nextSeq = info.seq // totalPay stays 0, so slotSeedSeq yields info.seq + s.tsVal, _, s.hasTS = parseTCPOptions(info.options) + c.slots = append(c.slots, s) +} + 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. @@ -297,7 +336,7 @@ func (c *TCPCoalescer) seed(pkt []byte, info parsedTCP) { return } s := c.take() - s.passthrough = false + s.passthrough = passthroughFalse s.rawPkt = pkt // kept for the numSeg==1 fast path in Flush copy(s.hdrBuf[:], pkt[:info.hdrLen]) s.hdrLen = info.hdrLen @@ -308,6 +347,7 @@ func (c *TCPCoalescer) seed(pkt []byte, info parsedTCP) { s.numSeg = 1 s.totalPay = info.payLen s.nextSeq = info.seq + uint32(info.payLen) + s.tsVal, _, s.hasTS = parseTCPOptions(info.options) s.sealed = info.flags&tcpFlagPsh != 0 s.payIovs = append(s.payIovs[:0], pkt[info.hdrLen:info.hdrLen+info.payLen]) c.slots = append(c.slots, s) @@ -384,7 +424,7 @@ func (c *TCPCoalescer) take() *coalesceSlot { } func (c *TCPCoalescer) release(s *coalesceSlot) { - s.passthrough = false + s.passthrough = passthroughFalse s.rawPkt = nil clear(s.payIovs) s.payIovs = s.payIovs[:0] @@ -400,6 +440,8 @@ func (c *TCPCoalescer) release(s *coalesceSlot) { s.isV6 = false s.gsoSize = 0 s.nextSeq = 0 + s.tsVal = 0 + s.hasTS = false c.pool = append(c.pool, s) } @@ -484,7 +526,11 @@ func (c *TCPCoalescer) reorderForFlush() { } runStart := 0 for i := 0; i <= len(c.slots); i++ { - if i < len(c.slots) && !c.slots[i].passthrough { + // Only hard passthroughs (unparseable, SYN/FIN/RST/CWR, oversized) + // split sort runs. Pure-ACK passthroughs stay inside the run so + // same-flow data separated by an interleaved ACK can still sort + // adjacent and merge; their own sort keys keep them ordered. + if i < len(c.slots) && c.slots[i].passthrough != passthroughTrue { continue } c.sortRun(c.slots[runStart:i]) @@ -494,13 +540,11 @@ func (c *TCPCoalescer) reorderForFlush() { for _, s := range c.slots { if n := len(out); n > 0 { prev := out[n-1] - if !prev.passthrough && !s.passthrough && prev.fk == s.fk { + if !prev.isPassthrough() && !s.isPassthrough() && prev.fk == s.fk { // Same-flow neighbors after sort. If they aren't seq- - // contiguous it's a real gap — packets the wire reordered + // contiguous it's a real gap: packets the wire reordered // across batches, or actual loss before nebula. Log it so - // the operator can quantify how often it happens; the data - // itself still emits in seq order, kernel TCP handles the - // gap via its OOO queue. + // the operator can quantify how often it happens if c.l.Enabled(context.Background(), slog.LevelDebug) { if prev.nextSeq != slotSeedSeq(s) { gap := int64(slotSeedSeq(s)) - int64(prev.nextSeq) @@ -564,6 +608,22 @@ func compareCoalesceSlots(a, b *coalesceSlot) int { if cmp := flowKeyCompare(a.fk, b.fk); cmp != 0 { return cmp } + // A retransmit carries a lower seq but a newer TCP timestamp than + // in-flight original data. Emitting it first would advance the + // receiver's ts_recent past the original's TSval, and PAWS would then + // drop the original as an old duplicate. So order by TSval before seq: + // TSval order approximates transmission order (which wire reordering + // never changed), and slots whose TSvals tie still get seq-repaired below. + // Flows without timestamps fall through to pure seq order, where PAWS cannot apply. + // tcpSeqLess is reused for the TSval compare: RFC 7323 defines TSval + // comparison in the same serial-number arithmetic. + if a.hasTS && b.hasTS && a.tsVal != b.tsVal { + if tcpSeqLess(a.tsVal, b.tsVal) { + return -1 + } + return 1 + } + aSeq, bSeq := slotSeedSeq(a), slotSeedSeq(b) if aSeq == bSeq { return 0 @@ -574,6 +634,44 @@ func compareCoalesceSlots(a, b *coalesceSlot) int { return 1 } +// parseTCPOptions attempts to locate timestamps. If it finds them, it returns tsval, secr, true. 0,0,false otherwise. +func parseTCPOptions(opts []byte) (uint32, uint32, bool) { + const timeStampOptionSize = 1 + 1 + 4 + 4 + const timeStampOptionCode = 0x8 + const nopOptionCode = 0x1 + const eolOptionCode = 0x0 + // Inclusive bound: a timestamp ending exactly at len(opts) is the common + // case (Linux emits NOP,NOP,TS as the whole option block). It also + // guards opts[i+1] in every arm, since timeStampOptionSize >= 2. + for i := 0; i+timeStampOptionSize <= len(opts); /* no increment */ { + switch opts[i] { + case eolOptionCode: + // End-of-option-list: everything after is padding. + return 0, 0, false + case nopOptionCode: + i++ + case timeStampOptionCode: + // we found it! + length := opts[i+1] + if length != timeStampOptionSize { + return 0, 0, false //weird, wrong option? + } + tsval := binary.BigEndian.Uint32(opts[i+2 : i+2+4]) + secr := binary.BigEndian.Uint32(opts[i+2+4 : i+2+4+4]) + return tsval, secr, true + default: + length := int(opts[i+1]) + if length < 2 { + // Malformed: a non-NOP option shorter than its own + // kind+length bytes would loop forever. + return 0, 0, false + } + i += length + } + } + return 0, 0, false +} + // slotSeedSeq returns the TCP seq of the slot's seed (first segment). // nextSeq tracks the seq just past the last appended byte; subtracting // totalPay walks back to the seed. uint32 wraparound is the right TCP @@ -668,10 +766,6 @@ func canMergeSlots(prev, s *coalesceSlot) bool { if (prevFlags^sFlags)&tcpFlagEce != 0 { return false } - // Same IPv4 ID rule as canAppend: s becomes segment prev.numSeg of the - // merged chain, so its seed ID must continue prev's sequence (or DF must - // make the IDs meaningless). s's own interior segments already passed - // this check against s's seed when they were appended. if !prev.isV6 && !ipv4CanCoalesceID(prev.hdrBuf[:], s.hdrBuf[:], prev.numSeg) { return false } diff --git a/overlay/batch/tcp_coalesce_test.go b/overlay/batch/tcp_coalesce_test.go index bbbdcd84..5d396687 100644 --- a/overlay/batch/tcp_coalesce_test.go +++ b/overlay/batch/tcp_coalesce_test.go @@ -1437,3 +1437,179 @@ func TestCoalescerAtomicRandomIDsCoalesce(t *testing.T) { t.Fatalf("DF=1 chain with arbitrary IDs must coalesce: gso=%d", len(w.gsoWrites)) } } + +// buildTCPv4TS is buildTCPv4 with a TCP timestamp option in the standard +// Linux layout (NOP,NOP,TS — a 32-byte TCP header). +func buildTCPv4TS(seq uint32, flags byte, tsVal, tsEcr uint32, payload []byte) []byte { + const ipHdrLen = 20 + const tcpHdrLen = 32 + total := ipHdrLen + tcpHdrLen + len(payload) + pkt := make([]byte, total) + + pkt[0] = 0x45 + pkt[1] = 0x00 + binary.BigEndian.PutUint16(pkt[2:4], uint16(total)) + binary.BigEndian.PutUint16(pkt[4:6], 0) + binary.BigEndian.PutUint16(pkt[6:8], 0x4000) + pkt[8] = 64 + pkt[9] = ipProtoTCP + copy(pkt[12:16], []byte{10, 0, 0, 1}) + copy(pkt[16:20], []byte{10, 0, 0, 2}) + + binary.BigEndian.PutUint16(pkt[20:22], 1000) + binary.BigEndian.PutUint16(pkt[22:24], 2000) + binary.BigEndian.PutUint32(pkt[24:28], seq) + binary.BigEndian.PutUint32(pkt[28:32], 12345) + pkt[32] = 0x80 // doff=8: 32-byte TCP header + pkt[33] = flags + binary.BigEndian.PutUint16(pkt[34:36], 0xffff) + pkt[40] = 0x01 // NOP + pkt[41] = 0x01 // NOP + pkt[42] = 0x08 // TS kind + pkt[43] = 10 // TS length + binary.BigEndian.PutUint32(pkt[44:48], tsVal) + binary.BigEndian.PutUint32(pkt[48:52], tsEcr) + + copy(pkt[52:], payload) + return pkt +} + +func TestParseTCPOptions(t *testing.T) { + ts := func(val, ecr uint32) []byte { + b := make([]byte, 10) + b[0], b[1] = 0x08, 10 + binary.BigEndian.PutUint32(b[2:6], val) + binary.BigEndian.PutUint32(b[6:10], ecr) + return b + } + cases := []struct { + name string + opts []byte + wantVal uint32 + wantEcr uint32 + wantOK bool + }{ + {"empty", nil, 0, 0, false}, + {"bare TS filling the block exactly", ts(100, 200), 100, 200, true}, + {"standard linux NOP,NOP,TS", append([]byte{1, 1}, ts(7, 9)...), 7, 9, true}, + {"unknown option then TS", append([]byte{254, 4, 0, 0}, ts(3, 4)...), 3, 4, true}, + {"EOL terminates before garbage", append([]byte{0, 0}, ts(1, 2)...), 0, 0, false}, + {"zero-length option must not hang", []byte{254, 0, 8, 10, 0, 0, 0, 1, 0, 0, 0, 2}, 0, 0, false}, + {"TS with wrong length", []byte{8, 4, 0, 0, 1, 1, 1, 1, 1, 1, 1, 1}, 0, 0, false}, + {"truncated TS", append([]byte{1, 1, 1}, ts(5, 6)[:9]...), 0, 0, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + val, ecr, ok := parseTCPOptions(tc.opts) + if val != tc.wantVal || ecr != tc.wantEcr || ok != tc.wantOK { + t.Fatalf("parseTCPOptions(%v) = (%d, %d, %v), want (%d, %d, %v)", + tc.opts, val, ecr, ok, tc.wantVal, tc.wantEcr, tc.wantOK) + } + }) + } +} + +// TestCompareCoalesceSlotsAntisymmetric pins the comparator contract for the +// retransmit shape: a lower seq with a newer TSval (retransmit) versus a +// higher seq with an older TSval (delayed original). The TSval must win in +// BOTH directions — an asymmetric comparator gives SortStableFunc an +// inconsistent order and unspecified output. +func TestCompareCoalesceSlotsAntisymmetric(t *testing.T) { + mk := func(seq, tsVal uint32, hasTS bool) *coalesceSlot { + return &coalesceSlot{nextSeq: seq, tsVal: tsVal, hasTS: hasTS} + } + original := mk(5000, 100, true) // sent first, delayed in flight + retransmit := mk(1000, 105, true) // sent later, lower seq + + if got := compareCoalesceSlots(original, retransmit); got != -1 { + t.Fatalf("compare(original, retransmit) = %d, want -1 (older TSval first)", got) + } + if got := compareCoalesceSlots(retransmit, original); got != 1 { + t.Fatalf("compare(retransmit, original) = %d, want 1", got) + } + + // Equal TSvals (a burst within one tick) fall back to seq order, + // still antisymmetrically. + a, b := mk(1000, 50, true), mk(2000, 50, true) + if compareCoalesceSlots(a, b) != -1 || compareCoalesceSlots(b, a) != 1 { + t.Fatal("equal-TSval slots must order by seq in both directions") + } + + // Timestamp-less flows keep pure seq order. + c, d := mk(2000, 0, false), mk(1000, 99, true) + if compareCoalesceSlots(c, d) != 1 || compareCoalesceSlots(d, c) != -1 { + t.Fatal("mixed/absent timestamps must fall back to seq in both directions") + } +} + +// TestCoalescerRetransmitEmitsAfterDelayedOriginal: a retransmit (lower seq, +// newer TSval) and a delayed original (higher seq, older TSval) land in one +// flush window. Seq-only sorting would emit the retransmit first; the +// receiver would advance ts_recent past the original's TSval and PAWS would +// drop the original. TSval-first ordering must emit the original first. +func TestCoalescerRetransmitEmitsAfterDelayedOriginal(t *testing.T) { + w := &fakeTunWriter{gsoEnabled: true} + c := newTestTCPCoalescer(t, w) + pay := make([]byte, 100) + + original := buildTCPv4TS(5000, tcpAck, 100, 1, pay) + retransmit := buildTCPv4TS(1000, tcpAck, 105, 1, pay) + + if err := c.Commit(original); err != nil { + t.Fatal(err) + } + if err := c.Commit(retransmit); err != nil { + t.Fatal(err) + } + if err := c.Flush(); err != nil { + t.Fatal(err) + } + if len(w.writes) != 2 { + t.Fatalf("want 2 plain writes (non-contiguous single-segment slots), got %d writes, %d gso", len(w.writes), len(w.gsoWrites)) + } + firstSeq := binary.BigEndian.Uint32(w.writes[0][24:28]) + secondSeq := binary.BigEndian.Uint32(w.writes[1][24:28]) + if firstSeq != 5000 || secondSeq != 1000 { + t.Fatalf("emission order (%d, %d), want (5000, 1000): retransmit must not overtake the older-TSval original", firstSeq, secondSeq) + } +} + +// TestCoalescerACKDoesNotSplitSortRun: an interleaved pure ACK must not stop +// wire-reordered same-flow data on either side of it from sorting adjacent +// and merging — the contract explicitly allows data to overtake a bare ACK. +// Arrival is D2, ACK, D1; the two data slots must still merge into one +// superpacket, with the ACK emitted after (its seq is the peer's snd_nxt, +// which orders it behind the data it followed). +func TestCoalescerACKDoesNotSplitSortRun(t *testing.T) { + w := &fakeTunWriter{gsoEnabled: true} + c := newTestTCPCoalescer(t, w) + pay := make([]byte, 1200) + + d2 := buildTCPv4(2200, tcpAck, pay) + ack := buildTCPv4(3400, tcpAck, nil) + d1 := buildTCPv4(1000, tcpAck, pay) + + for _, pkt := range [][]byte{d2, ack, d1} { + if err := c.Commit(pkt); err != nil { + t.Fatal(err) + } + } + if err := c.Flush(); err != nil { + t.Fatal(err) + } + if len(w.gsoWrites) != 1 { + t.Fatalf("want the two data slots merged into 1 gso write across the ACK, got %d gso + %d plain", len(w.gsoWrites), len(w.writes)) + } + if got := w.gsoWrites[0].payLen(); got != 2400 { + t.Fatalf("merged payload = %d, want 2400", got) + } + if len(w.writes) != 1 { + t.Fatalf("want the ACK as 1 plain write, got %d", len(w.writes)) + } + if got := binary.BigEndian.Uint32(w.writes[0][24:28]); got != 3400 { + t.Fatalf("plain write seq = %d, want the ACK (3400)", got) + } + if len(w.order) != 2 || w.order[0] != "gso" || w.order[1] != "write" { + t.Fatalf("emission order = %v, want [gso write]", w.order) + } +} diff --git a/overlay/tio/virtio/segment_linux.go b/overlay/tio/virtio/segment_linux.go index c4e88313..0509b600 100644 --- a/overlay/tio/virtio/segment_linux.go +++ b/overlay/tio/virtio/segment_linux.go @@ -62,6 +62,8 @@ const ( udpChecksumOff = 6 ) +var errPacketTooShort = errors.New("packet too short") + // tcpFinPshMask is cleared on every segment except the last of a TSO burst. const tcpFinPshMask = 0x09 // FIN(0x01) | PSH(0x08) @@ -78,9 +80,12 @@ func CheckValid(pkt []byte, hdr Hdr) error { return fmt.Errorf("virtio RSC_INFO flag not supported on TUN reads") } if len(pkt) < ipv4HeaderMinLen { - return fmt.Errorf("packet too short") + return errPacketTooShort } ipVersion := pkt[0] >> 4 + if ipVersion == 6 && len(pkt) < ipv6FixedLen { + return errPacketTooShort + } gsoType := hdr.GSOType() if gsoType != unix.VIRTIO_NET_HDR_GSO_NONE && hdr.GSOSize == 0 { diff --git a/udp/udp_linux_writebatch.go b/udp/udp_linux_writebatch.go index 35ddbec6..912ed619 100644 --- a/udp/udp_linux_writebatch.go +++ b/udp/udp_linux_writebatch.go @@ -321,12 +321,14 @@ func (w *batchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []b // Retrying the packets individually cannot succeed where the entry did not, and // 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[done]-w.entryPkts[done]], - "packets", w.entryPkts[done], - "gso", w.gsoSupported, - ) + if w.l.Enabled(context.Background(), slog.LevelDebug) { + w.l.Debug("sendmmsg rejected entry", + "error", serr, + "udpAddr", addrs[w.entryEnd[done]-w.entryPkts[done]], + "packets", w.entryPkts[done], + "gso", w.gsoSupported, + ) + } done++ } // When the drain finished every entry, i already sits past the whole