diff --git a/overlay/batch/coalesce_core.go b/overlay/batch/coalesce_core.go index f9d89f18..b64db92d 100644 --- a/overlay/batch/coalesce_core.go +++ b/overlay/batch/coalesce_core.go @@ -3,16 +3,40 @@ package batch import ( "bytes" "encoding/binary" + "math/bits" + "math/rand/v2" ) -// 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 +// 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)) } // initialSlots is the starting capacity of the slot pool. @@ -26,11 +50,10 @@ const initialSlots = 64 // shape. The v6 check is load-bearing: it rejects extension-header packets whose L4 is not at // byte 40. // -// 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. +// 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. func (fk *flowKey) parseIPAt(pkt []byte, ipHdrLen int) ([]byte, bool) { if len(pkt) < 20 { return nil, false @@ -66,9 +89,7 @@ func (fk *flowKey) parseIPv4Prologue(pkt []byte) ([]byte, bool) { if totalLen > len(pkt) || totalLen < ihl { return nil, false } - fk.isV6 = false - copy(fk.src[:4], pkt[12:16]) - copy(fk.dst[:4], pkt[16:20]) + *fk = flowKey(mix64(binary.LittleEndian.Uint64(pkt[12:20])^flowKeySeed[0], flowKeyM1)) return pkt[:totalLen], true } @@ -79,9 +100,11 @@ func (fk *flowKey) parseIPv6Prologue(pkt []byte) ([]byte, bool) { if 40+payloadLen > len(pkt) { return nil, false } - fk.isV6 = true - copy(fk.src[:], pkt[8:24]) - copy(fk.dst[:], pkt[24:40]) + 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)) return pkt[:40+payloadLen], true } diff --git a/overlay/batch/dispatch_bench_test.go b/overlay/batch/dispatch_bench_test.go index b523be22..131a3a26 100644 --- a/overlay/batch/dispatch_bench_test.go +++ b/overlay/batch/dispatch_bench_test.go @@ -110,3 +110,48 @@ 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 c78fe0b7..579b9bed 100644 --- a/overlay/batch/tcp_coalesce.go +++ b/overlay/batch/tcp_coalesce.go @@ -103,6 +103,7 @@ 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 @@ -133,8 +134,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.sport = binary.BigEndian.Uint16(pkt[ipHdrLen : ipHdrLen+2]) - p.fk.dport = binary.BigEndian.Uint16(pkt[ipHdrLen+2 : ipHdrLen+4]) + p.fk = p.fk.withPorts(binary.LittleEndian.Uint32(pkt[ipHdrLen : ipHdrLen+4])) + p.isV6 = ipHdrLen == 40 p.seq = binary.BigEndian.Uint32(pkt[ipHdrLen+4 : ipHdrLen+8]) p.flags = pkt[ipHdrLen+13] return true @@ -264,7 +265,7 @@ func (c *TCPCoalescer) seed(pkt []byte, info *parsedTCP) { s.rawPkt = pkt s.hdrLen = info.hdrLen s.ipHdrLen = info.ipHdrLen - s.isV6 = info.fk.isV6 + s.isV6 = info.isV6 s.fk = info.fk s.gsoSize = info.payLen s.numSeg = 1 @@ -357,7 +358,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 = flowKey{} + s.fk = 0 s.hdrLen = 0 s.ipHdrLen = 0 s.isV6 = false @@ -433,11 +434,12 @@ 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", flowKeyAddr(s.fk, false), - "dst", flowKeyAddr(s.fk, true), - "sport", s.fk.sport, - "dport", s.fk.dport, + "src", src, + "dst", dst, + "sport", sport, + "dport", dport, "prev_seed_seq", slotSeedSeq(prev), "prev_next_seq", prev.nextSeq, "this_seed_seq", slotSeedSeq(s), @@ -450,20 +452,20 @@ func (c *TCPCoalescer) logSeqGaps() { } } -// 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 +// 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])) } - if fk.isV6 { - return netip.AddrFrom16(src) - } - var v4 [4]byte - copy(v4[:], src[:4]) - return netip.AddrFrom4(v4) + sport = binary.BigEndian.Uint16(pkt[s.ipHdrLen : s.ipHdrLen+2]) + dport = binary.BigEndian.Uint16(pkt[s.ipHdrLen+2 : s.ipHdrLen+4]) + return } // 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 3df07b4e..60154241 100644 --- a/overlay/batch/udp_coalesce.go +++ b/overlay/batch/udp_coalesce.go @@ -81,6 +81,7 @@ 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 @@ -109,8 +110,8 @@ func (p *parsedUDP) parseTail(pkt []byte, ipHdrLen int) bool { p.ipHdrLen = ipHdrLen p.hdrLen = ipHdrLen + 8 p.payLen = udpLen - 8 - p.fk.sport = binary.BigEndian.Uint16(pkt[ipHdrLen : ipHdrLen+2]) - p.fk.dport = binary.BigEndian.Uint16(pkt[ipHdrLen+2 : ipHdrLen+4]) + p.fk = p.fk.withPorts(binary.LittleEndian.Uint32(pkt[ipHdrLen : ipHdrLen+4])) + p.isV6 = ipHdrLen == 40 return true } @@ -208,7 +209,7 @@ func (c *UDPCoalescer) seed(pkt []byte, info *parsedUDP) { s.rawPkt = pkt s.hdrLen = info.hdrLen s.ipHdrLen = info.ipHdrLen - s.isV6 = info.fk.isV6 + s.isV6 = info.isV6 s.fk = info.fk s.gsoSize = info.payLen s.numSeg = 1 @@ -280,7 +281,7 @@ func (c *UDPCoalescer) release(s *udpSlot) { s.numSeg = 0 s.totalPay = 0 // Zero the identity fields too; see TCPCoalescer.release. - s.fk = flowKey{} + s.fk = 0 s.hdrLen = 0 s.ipHdrLen = 0 s.isV6 = false