diff --git a/overlay/batch/coalesce_core.go b/overlay/batch/coalesce_core.go index b64db92d..f9d89f18 100644 --- a/overlay/batch/coalesce_core.go +++ b/overlay/batch/coalesce_core.go @@ -3,40 +3,16 @@ package batch import ( "bytes" "encoding/binary" - "math/bits" - "math/rand/v2" ) -// flowKey is a keyed 64-bit digest of a flow's {src, dst, sport, dport, family}. It is an index, -// not an identity: canAppend's headersMatch compares the real header bytes before any merge, so a -// key collision costs at most one lost merge or a prematurely closed chain, never a cross-flow -// merge. 64 bits keeps openSlots on the runtime's fast 8-byte-key map path and makes the lastSlot -// compare a single instruction; flowKeySeed keys the digest so crafted traffic cannot -// deterministically collide with a victim flow. Shared by the TCP and UDP coalescers; each keeps -// its own openSlots map, so a TCP and UDP flow on the same 5-tuple never alias. -type flowKey uint64 - -// flowKeySeed is the per-process random key for the flow digest. -var flowKeySeed = [2]uint64{rand.Uint64(), rand.Uint64()} - -// Mixing constants (from wyhash). Distinct constants per position separate the v4 and v6 domains. -const ( - flowKeyM1 = 0xa0761d6478bd642f - flowKeyM2 = 0xe7037ed1a0b428db - flowKeyM3 = 0x8ebc6af09c88c6e3 -) - -// mix64 is a wyhash-style multiply-fold: both halves of the 128-bit product, XORed. One mulx and -// one xor on amd64. -func mix64(a, b uint64) uint64 { - hi, lo := bits.Mul64(a, b) - return hi ^ lo -} - -// withPorts folds the L4 port pair (the raw 4 bytes at the L4 offset) into the digest. Called by -// the transport tails once the L4 header bounds are checked. -func (fk flowKey) withPorts(ports uint32) flowKey { - return flowKey(mix64(uint64(fk)^uint64(ports), flowKeySeed[1]|1)) +// 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. +type flowKey struct { + src, dst [16]byte + sport, dport uint16 + isV6 bool } // initialSlots is the starting capacity of the slot pool. @@ -50,10 +26,11 @@ const initialSlots = 64 // shape. The v6 check is load-bearing: it rejects extension-header packets whose L4 is not at // byte 40. // -// The prologues write the address portion of the digest into fk (the transport tails fold the -// ports in via withPorts) and return pkt trimmed to the IP-declared length. The -// receiver-as-out-pointer shape is deliberate: these functions are too big to inline, and -// returning parse structs by value put five 64-byte copies on the per-packet path. +// The prologues fill fk's addresses and family in place (ports belong to the L4 parser; fk must +// be zero on entry so the v4 path leaves src[4:]/dst[4:] clear for map equality) and return pkt +// trimmed to the IP-declared length. The receiver-as-out-pointer shape is deliberate: these +// functions are too big to inline, and returning structs by value put five 64-byte copies on the +// per-packet path. func (fk *flowKey) parseIPAt(pkt []byte, ipHdrLen int) ([]byte, bool) { if len(pkt) < 20 { return nil, false @@ -89,7 +66,9 @@ func (fk *flowKey) parseIPv4Prologue(pkt []byte) ([]byte, bool) { if totalLen > len(pkt) || totalLen < ihl { return nil, false } - *fk = flowKey(mix64(binary.LittleEndian.Uint64(pkt[12:20])^flowKeySeed[0], flowKeyM1)) + fk.isV6 = false + copy(fk.src[:4], pkt[12:16]) + copy(fk.dst[:4], pkt[16:20]) return pkt[:totalLen], true } @@ -100,11 +79,9 @@ func (fk *flowKey) parseIPv6Prologue(pkt []byte) ([]byte, bool) { if 40+payloadLen > len(pkt) { return nil, false } - s0 := binary.LittleEndian.Uint64(pkt[8:16]) - s1 := binary.LittleEndian.Uint64(pkt[16:24]) - d0 := binary.LittleEndian.Uint64(pkt[24:32]) - d1 := binary.LittleEndian.Uint64(pkt[32:40]) - *fk = flowKey(mix64(s0^flowKeySeed[0], s1^flowKeyM2) ^ mix64(d0^flowKeyM3, d1^flowKeyM1)) + fk.isV6 = true + copy(fk.src[:], pkt[8:24]) + copy(fk.dst[:], pkt[24:40]) return pkt[:40+payloadLen], true } diff --git a/overlay/batch/dispatch_bench_test.go b/overlay/batch/dispatch_bench_test.go index 131a3a26..b523be22 100644 --- a/overlay/batch/dispatch_bench_test.go +++ b/overlay/batch/dispatch_bench_test.go @@ -110,48 +110,3 @@ func BenchmarkDispatchSeedHeavy(b *testing.B) { } runDispatchBench(b, pkts, len(pkts)) } - -// TestFlowKeyDigestDistinct pins digest quality: distinct flows must produce distinct keys across -// a large sample, and the v4/v6 domains must not alias. Collisions are tolerated by construction -// (headersMatch gates every merge), so this is a quality canary, not a correctness requirement. -func TestFlowKeyDigestDistinct(t *testing.T) { - seen := make(map[flowKey]struct{}, 1<<17) - v4 := make([]byte, 40) - v4[0] = 0x45 - v4[3] = 40 // total length - add := func(fk flowKey) { - if _, dup := seen[fk]; dup { - t.Fatal("flow digest collision in small sample") - } - seen[fk] = struct{}{} - } - var fk flowKey - for a := range 256 { - for b := range 128 { - v4[15] = byte(a) // src low byte - v4[19] = byte(b) // dst low byte - v4[21] = byte(a) - v4[23] = byte(b) - trimmed, ok := fk.parseIPv4Prologue(v4) - if !ok { - t.Fatal("v4 prologue rejected synthetic packet") - } - _ = trimmed - add(fk.withPorts(uint32(a)<<16 | uint32(b))) - } - } - v6 := make([]byte, 60) - v6[0] = 0x60 - for a := range 256 { - for b := range 128 { - v6[23] = byte(a) - v6[39] = byte(b) - trimmed, ok := fk.parseIPv6Prologue(v6) - if !ok { - t.Fatal("v6 prologue rejected synthetic packet") - } - _ = trimmed - add(fk.withPorts(uint32(a)<<16 | uint32(b))) - } - } -} diff --git a/overlay/batch/tcp_coalesce.go b/overlay/batch/tcp_coalesce.go index 579b9bed..c78fe0b7 100644 --- a/overlay/batch/tcp_coalesce.go +++ b/overlay/batch/tcp_coalesce.go @@ -103,7 +103,6 @@ type parsedTCP struct { payLen int seq uint32 flags byte - isV6 bool } // parseAt extracts the flow key and IP/TCP offsets for a packet the dispatcher already knows is @@ -134,8 +133,8 @@ func (p *parsedTCP) parseTail(pkt []byte, ipHdrLen int) bool { p.ipHdrLen = ipHdrLen p.hdrLen = ipHdrLen + tcpOff p.payLen = len(pkt) - p.hdrLen - p.fk = p.fk.withPorts(binary.LittleEndian.Uint32(pkt[ipHdrLen : ipHdrLen+4])) - p.isV6 = ipHdrLen == 40 + p.fk.sport = binary.BigEndian.Uint16(pkt[ipHdrLen : ipHdrLen+2]) + p.fk.dport = binary.BigEndian.Uint16(pkt[ipHdrLen+2 : ipHdrLen+4]) p.seq = binary.BigEndian.Uint32(pkt[ipHdrLen+4 : ipHdrLen+8]) p.flags = pkt[ipHdrLen+13] return true @@ -265,7 +264,7 @@ func (c *TCPCoalescer) seed(pkt []byte, info *parsedTCP) { s.rawPkt = pkt s.hdrLen = info.hdrLen s.ipHdrLen = info.ipHdrLen - s.isV6 = info.isV6 + s.isV6 = info.fk.isV6 s.fk = info.fk s.gsoSize = info.payLen s.numSeg = 1 @@ -358,7 +357,7 @@ func (c *TCPCoalescer) release(s *coalesceSlot) { // 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 = 0 + s.fk = flowKey{} s.hdrLen = 0 s.ipHdrLen = 0 s.isV6 = false @@ -434,12 +433,11 @@ func (c *TCPCoalescer) logSeqGaps() { } if prev, ok := prevByFlow[s.fk]; ok && prev.nextSeq != slotSeedSeq(s) { gap := int64(slotSeedSeq(s)) - int64(prev.nextSeq) - src, dst, sport, dport := slotFlowAddrs(s) c.l.Debug("tcp coalesce: cross-slot seq gap", - "src", src, - "dst", dst, - "sport", sport, - "dport", dport, + "src", flowKeyAddr(s.fk, false), + "dst", flowKeyAddr(s.fk, true), + "sport", s.fk.sport, + "dport", s.fk.dport, "prev_seed_seq", slotSeedSeq(prev), "prev_next_seq", prev.nextSeq, "this_seed_seq", slotSeedSeq(s), @@ -452,20 +450,20 @@ func (c *TCPCoalescer) logSeqGaps() { } } -// slotFlowAddrs extracts the addresses and ports from the slot's seed packet for the debug log; -// the flow digest cannot be reversed. Cold path only. -func slotFlowAddrs(s *coalesceSlot) (src, dst netip.Addr, sport, dport uint16) { - pkt := s.rawPkt - if s.isV6 { - src = netip.AddrFrom16([16]byte(pkt[8:24])) - dst = netip.AddrFrom16([16]byte(pkt[24:40])) - } else { - src = netip.AddrFrom4([4]byte(pkt[12:16])) - dst = netip.AddrFrom4([4]byte(pkt[16:20])) +// flowKeyAddr returns the src or dst address from fk as a netip.Addr for +// logging. Only used on the cold gap-log path so the netip allocation +// doesn't matter. +func flowKeyAddr(fk flowKey, dst bool) netip.Addr { + src := fk.src + if dst { + src = fk.dst } - sport = binary.BigEndian.Uint16(pkt[s.ipHdrLen : s.ipHdrLen+2]) - dport = binary.BigEndian.Uint16(pkt[s.ipHdrLen+2 : s.ipHdrLen+4]) - return + if fk.isV6 { + return netip.AddrFrom16(src) + } + var v4 [4]byte + copy(v4[:], src[:4]) + return netip.AddrFrom4(v4) } // slotSeedSeq returns the TCP seq of the slot's seed (first segment). diff --git a/overlay/batch/udp_coalesce.go b/overlay/batch/udp_coalesce.go index 60154241..3df07b4e 100644 --- a/overlay/batch/udp_coalesce.go +++ b/overlay/batch/udp_coalesce.go @@ -81,7 +81,6 @@ type parsedUDP struct { ipHdrLen int hdrLen int // ipHdrLen + 8 payLen int - isV6 bool } // parseAt extracts the flow key and IP/UDP offsets for a packet the dispatcher already knows is @@ -110,8 +109,8 @@ func (p *parsedUDP) parseTail(pkt []byte, ipHdrLen int) bool { p.ipHdrLen = ipHdrLen p.hdrLen = ipHdrLen + 8 p.payLen = udpLen - 8 - p.fk = p.fk.withPorts(binary.LittleEndian.Uint32(pkt[ipHdrLen : ipHdrLen+4])) - p.isV6 = ipHdrLen == 40 + p.fk.sport = binary.BigEndian.Uint16(pkt[ipHdrLen : ipHdrLen+2]) + p.fk.dport = binary.BigEndian.Uint16(pkt[ipHdrLen+2 : ipHdrLen+4]) return true } @@ -209,7 +208,7 @@ func (c *UDPCoalescer) seed(pkt []byte, info *parsedUDP) { s.rawPkt = pkt s.hdrLen = info.hdrLen s.ipHdrLen = info.ipHdrLen - s.isV6 = info.isV6 + s.isV6 = info.fk.isV6 s.fk = info.fk s.gsoSize = info.payLen s.numSeg = 1 @@ -281,7 +280,7 @@ func (c *UDPCoalescer) release(s *udpSlot) { s.numSeg = 0 s.totalPay = 0 // Zero the identity fields too; see TCPCoalescer.release. - s.fk = 0 + s.fk = flowKey{} s.hdrLen = 0 s.ipHdrLen = 0 s.isV6 = false