diff --git a/overlay/batch/coalesce_core.go b/overlay/batch/coalesce_core.go index f9d89f18..ae67092e 100644 --- a/overlay/batch/coalesce_core.go +++ b/overlay/batch/coalesce_core.go @@ -95,40 +95,14 @@ func (fk *flowKey) parseIPv6Prologue(pkt []byte) ([]byte, bool) { // 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], - // bytes [2:4] = flow[15:0], [6:8] = next_hdr/hop, [8:40] = src+dst. - // Compare byte 1 fully so ECN (TC[1:0]) must match. Skip [4:6] payload_len. - if a[0] != b[0] { - return false - } - if a[1] != b[1] { - return false - } - if !bytes.Equal(a[2:4], b[2:4]) { - return false - } - if !bytes.Equal(a[6:40], b[6:40]) { - return false - } - return true + // IPv6: [0:4] = version/TC/flow label (TC[1:0] is ECN, so the full TC byte must match), + // [6:40] = next_hdr/hop + src + dst. Skip [4:6] payload_len. + return bytes.Equal(a[:4], b[:4]) && bytes.Equal(a[6:40], b[6:40]) } - // IPv4: byte 0 = version/IHL, byte 1 = DSCP(6)|ECN(2), - // [6:10] flags/fragoff/TTL/proto, [12:20] src+dst. - // Compare byte 1 fully so ECN must match. + // IPv4: [0:2] = version/IHL + DSCP|ECN (full ECN byte must match), + // [6:10] = flags/fragoff/TTL/proto, [12:20] = src+dst. // Skip [2:4] total len, [4:6] id, [10:12] csum. - if a[0] != b[0] { - return false - } - if a[1] != b[1] { - return false - } - if !bytes.Equal(a[6:10], b[6:10]) { - return false - } - if !bytes.Equal(a[12:20], b[12:20]) { - return false - } - return true + return bytes.Equal(a[:2], b[:2]) && bytes.Equal(a[6:10], b[6:10]) && bytes.Equal(a[12:20], b[12:20]) } // ipv4FlagDF is the Don't Fragment bit in the IPv4 flags byte (header byte 6). diff --git a/overlay/batch/multi_coalesce.go b/overlay/batch/multi_coalesce.go index 2c8034a5..c8457e2d 100644 --- a/overlay/batch/multi_coalesce.go +++ b/overlay/batch/multi_coalesce.go @@ -61,8 +61,9 @@ func NewMultiCoalescer(w io.Writer, l *slog.Logger) *MultiCoalescer { // Commit stages pkt for the next Flush; dispatch is deferred so it runs on packets already in // transmission order. key carries the packet's tunnel epoch and message counter. pkt is borrowed: -// the caller must keep it valid until the next Flush and not re-use it. pp is the firewall's -// parse of pkt and is borrowed only for this call, so the fields dispatch needs are copied here. +// the caller must keep it valid until the next Flush and not re-use it, and Flush may patch a +// coalesced packet's headers in place. pp is the firewall's parse of pkt and is borrowed only +// for this call, so the fields dispatch needs are copied here. func (m *MultiCoalescer) Commit(pkt []byte, key SortKey, pp *firewall.ParsedPacket) error { m.staged = append(m.staged, stagedPacket{ pkt: pkt, @@ -82,40 +83,17 @@ func compareStaged(a, b stagedPacket) int { return cmp.Compare(a.key.Counter, b.key.Counter) } -// dispatch routes one staged packet to its lane. -// The protocol and L4 offset come from the firewall's parse of the same packet. -// Any shape a lane can't coalesce seals every open chain in its lane +// dispatch routes one staged packet to its protocol lane (see commitStaged), or to the verbatim +// passthrough when the lane has no GSO support. func (m *MultiCoalescer) dispatch(sp stagedPacket) error { switch sp.proto { case ipProtoTCP: if m.tcp != nil { - if sp.fragAny { - m.tcp.sealAllOpen() - m.tcp.addVerbatim(sp.pkt) - return nil - } - var info parsedTCP - if !info.parseAt(sp.pkt, int(sp.ipHdrLen)) { - m.tcp.sealAllOpen() - m.tcp.addVerbatim(sp.pkt) - return nil - } - return m.tcp.commitParsed(sp.pkt, &info) + return m.tcp.commitStaged(sp) } case ipProtoUDP: if m.udp != nil { - if sp.fragAny { - m.udp.sealAllOpen() - m.udp.addVerbatim(sp.pkt) - return nil - } - var info parsedUDP - if !info.parseAt(sp.pkt, int(sp.ipHdrLen)) { - m.udp.sealAllOpen() - m.udp.addVerbatim(sp.pkt) - return nil - } - return m.udp.commitParsed(sp.pkt, &info) + return m.udp.commitStaged(sp) } } return m.pt.enqueue(sp.pkt) diff --git a/overlay/batch/tcp_coalesce.go b/overlay/batch/tcp_coalesce.go index d218bf5a..e6f78021 100644 --- a/overlay/batch/tcp_coalesce.go +++ b/overlay/batch/tcp_coalesce.go @@ -20,10 +20,6 @@ const tcpCoalesceBufSize = 65535 // superpacket. Keeping this well below the kernel's TSO ceiling bounds latency. const tcpCoalesceMaxSegs = 64 -// tcpCoalesceHdrCap is the scratch space we copy a seed's IP+TCP header -// into. IPv6 (40) + TCP with full options (60) = 100 bytes. -const tcpCoalesceHdrCap = 100 - // coalesceSlot is one entry in the coalescer's ordered event queue. A verbatim slot holds a single // borrowed packet emitted as-is (pure ACK, non-admissible TCP, unparseable, or oversize seed); a // non-verbatim slot is an in-progress coalesced superpacket. payIovs are borrowed slices of the @@ -33,13 +29,10 @@ type coalesceSlot struct { // rawPkt is borrowed: the whole packet for verbatim slots, the seed packet for coalesce // slots. A slot that never grows past one segment is emitted from rawPkt so its original // (already valid) L4 checksum ships DATA_VALID instead of making the kernel recompute it. + // A multi-segment slot's superpacket header is rawPkt's, patched in place at flush. rawPkt []byte - fk flowKey - // hdrBuf is a mutable copy of the seed's IP+TCP header, populated on the first append. Total - // length and the pseudo-header checksum partial are patched at flush. A slot that never grows - // flushes from rawPkt and never touches hdrBuf. - hdrBuf [tcpCoalesceHdrCap]byte + fk flowKey hdrLen int ipHdrLen int isV6 bool @@ -153,6 +146,36 @@ func (c *TCPCoalescer) sealAllOpen() { c.lastSlot = nil } +// sealFlow closes fk's open chain, if any, keeping lastSlot in lockstep. The len guard skips +// hashing the 38-byte key when no chains are open (e.g. ack-dominant queues). +func (c *TCPCoalescer) sealFlow(fk flowKey) { + if len(c.openSlots) == 0 { + return + } + if last := c.lastSlot; last != nil && last.fk == fk { + c.lastSlot = nil + } + delete(c.openSlots, fk) +} + +// commitStaged commits one staged packet dispatch routed to this lane. A shape the lane cannot +// coalesce (any fragmentation, unparseable header) seals every open chain +// and rides the lane as an in-lane verbatim, still in transmission order. +func (c *TCPCoalescer) commitStaged(sp stagedPacket) error { + if sp.fragAny { + c.sealAllOpen() + c.addVerbatim(sp.pkt) + return nil + } + var info parsedTCP + if !info.parseAt(sp.pkt, int(sp.ipHdrLen)) { + c.sealAllOpen() + c.addVerbatim(sp.pkt) + return nil + } + return c.commitParsed(sp.pkt, &info) +} + // commitParsed commits one parsed TCP packet. The caller (dispatch, via parseAt) supplies a // valid parse so the header is not re-walked here. func (c *TCPCoalescer) commitParsed(pkt []byte, info *parsedTCP) error { @@ -161,14 +184,8 @@ func (c *TCPCoalescer) commitParsed(pkt []byte, info *parsedTCP) error { // reuses CWR as ACE counter bits; revisit this check if inner hosts adopt AccECN. if info.flags&tcpFlagAck == 0 || info.flags&^(tcpFlagAck|tcpFlagPsh|tcpFlagEce) != 0 { // SYN/FIN/RST/URG/CWR must be observed in sequence. Seal the flow's open slot so later - // in-flow packets cannot extend it and emit ahead of this verbatim. The len guard skips - // hashing the 38-byte key on ack-dominant queues, where the map is almost always empty. - if len(c.openSlots) != 0 { - if last := c.lastSlot; last != nil && last.fk == info.fk { - c.lastSlot = nil - } - delete(c.openSlots, info.fk) - } + // in-flow packets cannot extend it and emit ahead of this verbatim. + c.sealFlow(info.fk) c.addVerbatim(pkt) return nil } @@ -196,8 +213,7 @@ func (c *TCPCoalescer) commitParsed(pkt []byte, info *parsedTCP) error { if c.canAppend(open, pkt, info) { if c.appendPayload(open, pkt, info) { // Chain closed (PSH or short segment): stop extending it. - delete(c.openSlots, info.fk) - c.lastSlot = nil + c.sealFlow(info.fk) } else { c.lastSlot = open } @@ -205,10 +221,7 @@ func (c *TCPCoalescer) commitParsed(pkt []byte, info *parsedTCP) error { } // Can't extend (seq gap from upstream loss, header change, or a full // chain): evict it from openSlots and fall through to seed a fresh slot. - delete(c.openSlots, info.fk) - if c.lastSlot == open { - c.lastSlot = nil - } + c.sealFlow(info.fk) } c.seed(pkt, info) return nil @@ -221,7 +234,8 @@ func (c *TCPCoalescer) Flush() error { if s.verbatim || s.numSeg == 1 { // A slot that never grew is byte-identical to its seed packet; ship the original so // its valid checksum rides the DATA_VALID path instead of a kernel software csum. - // appendPayload only touches hdrBuf once numSeg >= 2, so rawPkt is pristine here. + // rawPkt is only mutated once numSeg >= 2 (PSH propagate, flush patches), so it is + // pristine here. _, err = c.w.Write(s.rawPkt) } else { err = c.flushSlot(s) @@ -247,15 +261,18 @@ func (c *TCPCoalescer) addVerbatim(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. + if info.hdrLen+info.payLen > tcpCoalesceBufSize { + // Pathological shape that can't ride a superpacket; emit as-is. No chain for this flow can + // be open here (commitParsed evicts before seeding), so sealFlow is defense in depth + // against a stale cache entry absorbing later data. + c.sealFlow(info.fk) c.addVerbatim(pkt) return } s := c.take() s.verbatim = false - // rawPkt serves the numSeg==1 fast path in Flush and is the header source for canAppend until - // the first append copies it into hdrBuf. + // rawPkt serves the numSeg==1 fast path in Flush, is the header source for canAppend, and is + // the superpacket header flushSlot patches in place. s.rawPkt = pkt s.hdrLen = info.hdrLen s.ipHdrLen = info.ipHdrLen @@ -270,17 +287,17 @@ func (c *TCPCoalescer) seed(pkt []byte, info *parsedTCP) { if info.flags&tcpFlagPsh == 0 { c.openSlots[info.fk] = s c.lastSlot = s - } else if last := c.lastSlot; last != nil && last.fk == info.fk { - // PSH on the seed closes the chain immediately; it is never registered as open. Drop any - // stale cache entry for this flow too. - c.lastSlot = nil + } else { + // PSH on the seed closes the chain immediately; it is never registered as open. + // Drop any stale entry for this flow too (defense in depth, unreachable if lastSlot's lockstep invariant holds). + c.sealFlow(info.fk) } } // canAppend reports whether info's packet extends the slot's seed: same header shape and stable // contents, adjacent seq, not oversized. A closed chain never reaches here; closing removes the -// slot from openSlots, the only path in. Header reads use rawPkt because hdrBuf is populated -// lazily on the first append; every field consulted here is one the pre-flush patches never touch. +// slot from openSlots, the only path in. The header fields read from rawPkt are always pristine: +// the only pre-flush mutation is the PSH propagate, which also closes the chain. func (c *TCPCoalescer) canAppend(s *coalesceSlot, pkt []byte, info *parsedTCP) bool { if info.hdrLen != s.hdrLen { return false @@ -316,18 +333,14 @@ func (c *TCPCoalescer) canAppend(s *coalesceSlot, pkt []byte, info *parsedTCP) b // segment was sub-gsoSize (kernel TSO allows only the final segment to be short) or carried PSH. // The caller must deregister a closed slot from openSlots. func (c *TCPCoalescer) appendPayload(s *coalesceSlot, pkt []byte, info *parsedTCP) bool { - if s.numSeg == 1 { - // First append: populate hdrBuf from the seed. Deferred out of seed so solo slots, which - // flush from rawPkt, never pay the copy. - copy(s.hdrBuf[:s.hdrLen], s.rawPkt[:s.hdrLen]) - } s.payIovs = append(s.payIovs, pkt[info.hdrLen:info.hdrLen+info.payLen]) s.numSeg++ s.totalPay += info.payLen s.nextSeq = info.seq + uint32(info.payLen) if info.flags&tcpFlagPsh != 0 { - // Propagate PSH into the seed header so kernel TSO sets it on the last segment. - s.hdrBuf[s.ipHdrLen+13] |= tcpFlagPsh + // Propagate PSH into the seed header so kernel TSO sets it on the last segment. Mutating + // rawPkt is safe: PSH also closes the chain, so no admission check re-reads this header. + s.rawPkt[s.ipHdrLen+13] |= tcpFlagPsh } return info.payLen < s.gsoSize || info.flags&tcpFlagPsh != 0 } @@ -343,29 +356,18 @@ func (c *TCPCoalescer) take() *coalesceSlot { } func (c *TCPCoalescer) release(s *coalesceSlot) { - s.verbatim = false - s.rawPkt = nil clear(s.payIovs) - s.payIovs = s.payIovs[:0] - s.numSeg = 0 - s.totalPay = 0 - // Zero the identity fields too: addVerbatim doesn't set them, so a - // pooled slot reused as a verbatim must not carry a stale flow key - // that a future refactor could mistake for real. - s.fk = flowKey{} - s.hdrLen = 0 - s.ipHdrLen = 0 - s.isV6 = false - s.gsoSize = 0 - s.nextSeq = 0 + *s = coalesceSlot{payIovs: s.payIovs[:0]} c.pool = append(c.pool, s) } -// flushSlot patches the header and calls WriteGSO. Does not remove the slot from c.slots. +// flushSlot patches the superpacket header in place in rawPkt (total length, IPv4 header +// checksum, pseudo-header checksum seed) and calls WriteGSO. The slot is released right after, +// so nothing re-reads the patched header. Does not remove the slot from c.slots. func (c *TCPCoalescer) flushSlot(s *coalesceSlot) error { total := s.hdrLen + s.totalPay l4Len := total - s.ipHdrLen - hdr := s.hdrBuf[:s.hdrLen] + hdr := s.rawPkt[:s.hdrLen] if s.isV6 { binary.BigEndian.PutUint16(hdr[4:6], uint16(l4Len)) diff --git a/overlay/batch/udp_coalesce.go b/overlay/batch/udp_coalesce.go index 3df07b4e..851bb59b 100644 --- a/overlay/batch/udp_coalesce.go +++ b/overlay/batch/udp_coalesce.go @@ -1,6 +1,7 @@ package batch import ( + "bytes" "encoding/binary" "io" @@ -18,10 +19,6 @@ const udpCoalesceBufSize = 65535 // accepts up to 64 segments per skb (UDP_MAX_SEGMENTS); stay under that. const udpCoalesceMaxSegs = 64 -// udpCoalesceHdrCap is the scratch space we copy a seed's IP+UDP header -// into. IPv6 (40) + UDP (8) = 48; round up for safety. -const udpCoalesceHdrCap = 64 - // udpSlot is one entry in the UDPCoalescer's ordered event queue. type udpSlot struct { verbatim bool @@ -29,10 +26,10 @@ type udpSlot struct { // packet for coalesce slots. A coalesce slot that never grows past one // segment is emitted from rawPkt so its original (already valid) L4 // checksum ships DATA_VALID instead of making the kernel recompute it. + // A multi-segment slot's superpacket header is rawPkt's, patched in place at flush. rawPkt []byte fk flowKey - hdrBuf [udpCoalesceHdrCap]byte hdrLen int ipHdrLen int isV6 bool @@ -114,18 +111,43 @@ func (p *parsedUDP) parseTail(pkt []byte, ipHdrLen int) bool { return true } +// sealFlow closes fk's open chain, if any, keeping lastSlot in lockstep. The len guard skips +// hashing the 38-byte key when no chains are open. +func (c *UDPCoalescer) sealFlow(fk flowKey) { + if len(c.openSlots) == 0 { + return + } + if last := c.lastSlot; last != nil && last.fk == fk { + c.lastSlot = nil + } + delete(c.openSlots, fk) +} + +// commitStaged commits one staged packet dispatch routed to this lane. A shape the lane cannot +// coalesce (any fragmentation, unparseable header) seals every open chain — its flow is unknown — +// and rides the lane as an in-lane verbatim, still in transmission order. +func (c *UDPCoalescer) commitStaged(sp stagedPacket) error { + if sp.fragAny { + c.sealAllOpen() + c.addVerbatim(sp.pkt) + return nil + } + var info parsedUDP + if !info.parseAt(sp.pkt, int(sp.ipHdrLen)) { + c.sealAllOpen() + c.addVerbatim(sp.pkt) + return nil + } + return c.commitParsed(sp.pkt, &info) +} + // commitParsed commits one parsed UDP packet. The caller (dispatch, via parseAt) supplies a // valid parse so the header is not re-walked here. func (c *UDPCoalescer) commitParsed(pkt []byte, info *parsedUDP) error { // A zero-length UDP datagram (length == 8) is legal and must reach the TUN, but cannot be - // coalesced. The len guard skips hashing the key when no flow is open. + // coalesced. if info.payLen == 0 { - if len(c.openSlots) != 0 { - if last := c.lastSlot; last != nil && last.fk == info.fk { - c.lastSlot = nil - } - delete(c.openSlots, info.fk) - } + c.sealFlow(info.fk) c.addVerbatim(pkt) return nil } @@ -140,8 +162,7 @@ func (c *UDPCoalescer) commitParsed(pkt []byte, info *parsedUDP) error { if c.canAppend(open, pkt, info) { if c.appendPayload(open, pkt, info) { // Chain closed (short segment): stop extending it. - delete(c.openSlots, info.fk) - c.lastSlot = nil + c.sealFlow(info.fk) } else { c.lastSlot = open } @@ -149,10 +170,7 @@ func (c *UDPCoalescer) commitParsed(pkt []byte, info *parsedUDP) error { } // Can't extend: evict it from openSlots and fall through to seed a // fresh slot. - delete(c.openSlots, info.fk) - if c.lastSlot == open { - c.lastSlot = nil - } + c.sealFlow(info.fk) } c.seed(pkt, info) return nil @@ -197,14 +215,18 @@ func (c *UDPCoalescer) addVerbatim(pkt []byte) { } func (c *UDPCoalescer) seed(pkt []byte, info *parsedUDP) { - if info.hdrLen > udpCoalesceHdrCap || info.hdrLen+info.payLen > udpCoalesceBufSize { + if info.hdrLen+info.payLen > udpCoalesceBufSize { + // Pathological shape that can't ride a superpacket; emit as-is. No chain for this flow can + // be open here (commitParsed evicts before seeding), so sealFlow is defense in depth + // against a stale cache entry absorbing later data. + c.sealFlow(info.fk) c.addVerbatim(pkt) return } s := c.take() s.verbatim = false - // rawPkt serves the numSeg==1 fast path in Flush and is the header source for canAppend until - // the first append copies it into hdrBuf. + // rawPkt serves the numSeg==1 fast path in Flush, is the header source for canAppend, and is + // the superpacket header flushSlot patches in place. s.rawPkt = pkt s.hdrLen = info.hdrLen s.ipHdrLen = info.ipHdrLen @@ -235,9 +257,8 @@ func (c *UDPCoalescer) canAppend(s *udpSlot, pkt []byte, info *parsedUDP) bool { if s.hdrLen+s.totalPay+info.payLen > udpCoalesceBufSize { return false } - // Header reads use rawPkt because hdrBuf is populated lazily on the first append; the fields - // consulted here are never patched before flush. A closed chain never reaches here; closing - // removes the slot from openSlots, the only path in. + // Header reads use rawPkt, which is never mutated before flush. A closed chain never reaches + // here; closing removes the slot from openSlots, the only path in. if !s.isV6 && !ipv4CanCoalesceID(s.rawPkt, pkt, s.numSeg) { return false } @@ -251,11 +272,6 @@ func (c *UDPCoalescer) canAppend(s *udpSlot, pkt []byte, info *parsedUDP) bool { // UDP-GSO requires every segment but the last to be exactly gsoSize, so a short segment must be // the final one. The caller must deregister a closed slot from openSlots. func (c *UDPCoalescer) appendPayload(s *udpSlot, pkt []byte, info *parsedUDP) bool { - if s.numSeg == 1 { - // First append: populate hdrBuf from the seed. Deferred out of seed so solo slots, which - // flush from rawPkt, never pay the copy. - copy(s.hdrBuf[:s.hdrLen], s.rawPkt[:s.hdrLen]) - } s.payIovs = append(s.payIovs, pkt[info.hdrLen:info.hdrLen+info.payLen]) s.numSeg++ s.totalPay += info.payLen @@ -273,27 +289,19 @@ func (c *UDPCoalescer) take() *udpSlot { } func (c *UDPCoalescer) release(s *udpSlot) { - s.verbatim = false - s.rawPkt = nil + // Reset every field, identity ones included; see TCPCoalescer.release. clear(s.payIovs) - s.payIovs = s.payIovs[:0] - s.numSeg = 0 - s.totalPay = 0 - // Zero the identity fields too; see TCPCoalescer.release. - s.fk = flowKey{} - s.hdrLen = 0 - s.ipHdrLen = 0 - s.isV6 = false - s.gsoSize = 0 + *s = udpSlot{payIovs: s.payIovs[:0]} c.pool = append(c.pool, s) } // flushSlot patches the IP header total length / IPv6 payload length and // the UDP length to the *total* across all coalesced segments, then seeds // the UDP checksum field with the pseudo-header partial (single-fold, not -// inverted) per virtio NEEDS_CSUM. +// inverted) per virtio NEEDS_CSUM. The patches land in place in rawPkt; the +// slot is released right after, so nothing re-reads the patched header. func (c *UDPCoalescer) flushSlot(s *udpSlot) error { - hdr := s.hdrBuf[:s.hdrLen] + hdr := s.rawPkt[:s.hdrLen] total := s.hdrLen + s.totalPay // full IP+UDP+all_payloads bytes l4Len := total - s.ipHdrLen // total UDP (8 + sum of payloads) @@ -330,11 +338,8 @@ 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] { - return false - } - return true + return bytes.Equal(a[udp:udp+4], b[udp:udp+4]) }