This commit is contained in:
JackDoan
2026-08-04 09:27:51 -05:00
parent a3eef407b2
commit d8d5ce344d
8 changed files with 283 additions and 111 deletions
+32 -44
View File
@@ -20,81 +20,69 @@ type flowKey struct {
// so this matches a typical carrier-side recvmmsg batch on the UDP socket. // so this matches a typical carrier-side recvmmsg batch on the UDP socket.
const initialSlots = 64 const initialSlots = 64
// parsedIP is the IP-level result of the prologue parsers.
// The caller layers L4-specific parsing (TCP / UDP) on top.
type parsedIP struct {
fk flowKey
ipHdrLen int
// pkt is the original buffer trimmed to the IP-declared total length.
// Anything below the IP layer (transport parsers) should slice into
// pkt rather than the unbounded original.
pkt []byte
}
// parseIPAt validates the IP header for lane parsing. newPacket already resolved the L4 protocol // parseIPAt validates the IP header for lane parsing. newPacket already resolved the L4 protocol
// and offset for the firewall, so there is no proto sniff here; the caller's ipHdrLen is // and offset for the firewall, so there is no proto sniff here; the caller's ipHdrLen is
// cross-checked instead. A plain header (v4 IHL 20, v6 exactly 40) is the only coalesceable // cross-checked instead. A plain header (v4 IHL 20, v6 exactly 40) is the only coalesceable
// shape. The v6 check is load-bearing: it rejects extension-header packets whose L4 is not at // shape. The v6 check is load-bearing: it rejects extension-header packets whose L4 is not at
// byte 40. On success p.pkt is trimmed to the IP-declared length. // byte 40.
func parseIPAt(pkt []byte, ipHdrLen int) (parsedIP, bool) { //
var p parsedIP // 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 { if len(pkt) < 20 {
return p, false return nil, false
} }
switch pkt[0] >> 4 { switch pkt[0] >> 4 {
case 4: case 4:
if ipHdrLen != 20 { if ipHdrLen != 20 {
return p, false return nil, false
} }
return parseIPv4Prologue(pkt) return fk.parseIPv4Prologue(pkt)
case 6: case 6:
if ipHdrLen != 40 || len(pkt) < 40 { if ipHdrLen != 40 || len(pkt) < 40 {
return p, false return nil, false
} }
return parseIPv6Prologue(pkt) return fk.parseIPv6Prologue(pkt)
} }
return p, false return nil, false
} }
// parseIPv4Prologue is the shared IPv4 tail of the prologue entries; the // parseIPv4Prologue is the shared IPv4 tail of the prologue entries; the caller has verified
// caller has verified len(pkt) >= 20 and the version. // len(pkt) >= 20 and the version.
func parseIPv4Prologue(pkt []byte) (parsedIP, bool) { func (fk *flowKey) parseIPv4Prologue(pkt []byte) ([]byte, bool) {
var p parsedIP
ihl := int(pkt[0]&0x0f) * 4 ihl := int(pkt[0]&0x0f) * 4
if ihl != 20 { if ihl != 20 {
return p, false return nil, false
} }
// Reject any fragmentation (MF or nonzero offset). The dispatcher already gated FragAny; kept // Reject any fragmentation (MF or nonzero offset). The dispatcher already gated FragAny; kept
// as defense in depth, since a fragment folded into a superpacket would corrupt reassembly. // as defense in depth, since a fragment folded into a superpacket would corrupt reassembly.
if binary.BigEndian.Uint16(pkt[6:8])&0x3fff != 0 { if binary.BigEndian.Uint16(pkt[6:8])&0x3fff != 0 {
return p, false return nil, false
} }
totalLen := int(binary.BigEndian.Uint16(pkt[2:4])) totalLen := int(binary.BigEndian.Uint16(pkt[2:4]))
if totalLen > len(pkt) || totalLen < ihl { if totalLen > len(pkt) || totalLen < ihl {
return p, false return nil, false
} }
p.ipHdrLen = 20 fk.isV6 = false
p.fk.isV6 = false copy(fk.src[:4], pkt[12:16])
copy(p.fk.src[:4], pkt[12:16]) copy(fk.dst[:4], pkt[16:20])
copy(p.fk.dst[:4], pkt[16:20]) return pkt[:totalLen], true
p.pkt = pkt[:totalLen]
return p, true
} }
// parseIPv6Prologue is the shared IPv6 tail; the caller has verified // parseIPv6Prologue is the shared IPv6 tail; the caller has verified len(pkt) >= 40, the version,
// len(pkt) >= 40, the version, and that the L4 header sits at byte 40. // and that the L4 header sits at byte 40.
func parseIPv6Prologue(pkt []byte) (parsedIP, bool) { func (fk *flowKey) parseIPv6Prologue(pkt []byte) ([]byte, bool) {
var p parsedIP
payloadLen := int(binary.BigEndian.Uint16(pkt[4:6])) payloadLen := int(binary.BigEndian.Uint16(pkt[4:6]))
if 40+payloadLen > len(pkt) { if 40+payloadLen > len(pkt) {
return p, false return nil, false
} }
p.ipHdrLen = 40 fk.isV6 = true
p.fk.isV6 = true copy(fk.src[:], pkt[8:24])
copy(p.fk.src[:], pkt[8:24]) copy(fk.dst[:], pkt[24:40])
copy(p.fk.dst[:], pkt[24:40]) return pkt[:40+payloadLen], true
p.pkt = pkt[:40+payloadLen]
return p, true
} }
// ipHeadersMatch compares the IP portion of two packet header prefixes for // ipHeadersMatch compares the IP portion of two packet header prefixes for
+112
View File
@@ -0,0 +1,112 @@
package batch
import (
"testing"
"github.com/slackhq/nebula/test"
)
// stagePackets builds the stagedPacket entries Commit would have produced, so dispatch benchmarks
// bypass staging and the sort entirely.
func stagePackets(pkts [][]byte) []stagedPacket {
staged := make([]stagedPacket, len(pkts))
for i, p := range pkts {
pp := testPP(p)
staged[i] = stagedPacket{
pkt: p,
key: SortKey{Epoch: 1, Counter: uint64(i + 1)},
proto: pp.Protocol,
fragAny: pp.FragAny,
ipHdrLen: uint16(pp.IPHdrLen),
}
}
return staged
}
func flushLanes(b *testing.B, m *MultiCoalescer) {
b.Helper()
if m.tcp != nil {
if err := m.tcp.Flush(); err != nil {
b.Fatal(err)
}
}
if m.udp != nil {
if err := m.udp.Flush(); err != nil {
b.Fatal(err)
}
}
if err := m.pt.Flush(); err != nil {
b.Fatal(err)
}
}
// runDispatchBench measures dispatch plus the per-batch lane flush: the post-sort half of the
// batcher, which is where the production profile concentrates.
func runDispatchBench(b *testing.B, pkts [][]byte, batchSize int) {
b.Helper()
m := NewMultiCoalescer(nopTunWriter{}, test.NewLogger())
staged := stagePackets(pkts)
b.ReportAllocs()
b.SetBytes(int64(len(pkts[0])))
b.ResetTimer()
for i := 0; i < b.N; i++ {
if err := m.dispatch(staged[i%len(staged)]); err != nil {
b.Fatal(err)
}
if (i+1)%batchSize == 0 {
flushLanes(b, m)
}
}
b.StopTimer()
flushLanes(b, m)
}
// BenchmarkDispatchSingleFlow is the bulk steady state: every packet past the seed appends.
func BenchmarkDispatchSingleFlow(b *testing.B) {
runDispatchBench(b, buildTCPv4BulkFlow(tcpCoalesceMaxSegs, 1200), tcpCoalesceMaxSegs)
}
// BenchmarkDispatchInterleaved16 stresses the openSlots map: 16 flows round-robined defeats the
// lastSlot cache on every packet.
func BenchmarkDispatchInterleaved16(b *testing.B) {
pkts := buildTCPv4Interleaved(16, tcpCoalesceMaxSegs, 1200)
runDispatchBench(b, pkts, len(pkts))
}
// BenchmarkDispatchAckHeavy alternates MSS data with pure ACKs on one flow — the RX shape of a
// bidirectional transfer (the peer's data and its ACKs of our data share the tunnel direction).
func BenchmarkDispatchAckHeavy(b *testing.B) {
pay := make([]byte, 1200)
var pkts [][]byte
seq := uint32(1000)
for range tcpCoalesceMaxSegs / 2 {
pkts = append(pkts, buildTCPv4(seq, tcpAck, pay))
seq += uint32(len(pay))
pkts = append(pkts, buildTCPv4(seq, tcpAck, nil))
}
runDispatchBench(b, pkts, len(pkts))
}
// BenchmarkDispatchUDPFlow is the QUIC-ish bulk UDP shape.
func BenchmarkDispatchUDPFlow(b *testing.B) {
pay := make([]byte, 1200)
pkts := make([][]byte, udpCoalesceMaxSegs)
for i := range pkts {
pkts[i] = buildUDPv4(2000, 443, pay)
}
runDispatchBench(b, pkts, len(pkts))
}
// BenchmarkDispatchSeedHeavy sets PSH on every packet so each one seeds and immediately closes
// its own slot — the small-write RPC shape, and the upper bound on what the seed path (including
// the parsedTCP-to-slot field transfer) can cost.
func BenchmarkDispatchSeedHeavy(b *testing.B) {
pay := make([]byte, 1200)
pkts := make([][]byte, tcpCoalesceMaxSegs)
seq := uint32(1000)
for i := range pkts {
pkts[i] = buildTCPv4(seq, tcpAckPsh, pay)
seq += uint32(len(pay))
}
runDispatchBench(b, pkts, len(pkts))
}
+76
View File
@@ -0,0 +1,76 @@
package batch
//TODO refactor this away
// This file holds the lanes' self-parsing Commit entries and the proto-checking parsers behind
// them. Production traffic enters the lanes only through MultiCoalescer.dispatch and the At
// parsers; these wrappers reproduce that path (including seal-all on unparseable shapes) on top
// of a local parse, so tests and benches can drive one lane with nothing but a packet.
// parseIPPrologue resolves the IP version, requires the L4 protocol to match wantProto (6 TCP,
// 17 UDP), and defers to the shared per-version cores. Returns the trimmed packet and the L4
// offset; fk must be zero on entry and is filled in place.
func (fk *flowKey) parseIPPrologue(pkt []byte, wantProto byte) ([]byte, int, bool) {
if len(pkt) < 20 {
return nil, 0, false
}
switch pkt[0] >> 4 {
case 4:
if pkt[9] != wantProto {
return nil, 0, false
}
trimmed, ok := fk.parseIPv4Prologue(pkt)
return trimmed, 20, ok
case 6:
if len(pkt) < 40 {
return nil, 0, false
}
if pkt[6] != wantProto {
return nil, 0, false
}
trimmed, ok := fk.parseIPv6Prologue(pkt)
return trimmed, 40, ok
}
return nil, 0, false
}
// parseBase extracts the flow key and IP/TCP offsets for any TCP packet, admissible for
// coalescing or not. Returns false for non-TCP or malformed input.
func (p *parsedTCP) parseBase(pkt []byte) bool {
trimmed, ipHdrLen, ok := p.fk.parseIPPrologue(pkt, ipProtoTCP)
if !ok {
return false
}
return p.parseTail(trimmed, ipHdrLen)
}
// parseBase extracts the flow key and IP/UDP offsets for a UDP packet.
func (p *parsedUDP) parseBase(pkt []byte) bool {
trimmed, ipHdrLen, ok := p.fk.parseIPPrologue(pkt, ipProtoUDP)
if !ok {
return false
}
return p.parseTail(trimmed, ipHdrLen)
}
// Commit borrows pkt. The caller must keep pkt valid until the next Flush.
func (c *TCPCoalescer) Commit(pkt []byte) error {
var info parsedTCP
if !info.parseBase(pkt) {
// Unparseable: flow key unknown, seal everything so later data cannot emit ahead of it.
c.sealAllOpen()
c.addVerbatim(pkt)
return nil
}
return c.commitParsed(pkt, &info)
}
// Commit borrows pkt. The caller must keep pkt valid until the next Flush.
func (c *UDPCoalescer) Commit(pkt []byte) error {
var info parsedUDP
if !info.parseBase(pkt) {
c.sealAllOpen()
c.addVerbatim(pkt)
return nil
}
return c.commitParsed(pkt, &info)
}
+6 -6
View File
@@ -94,13 +94,13 @@ func (m *MultiCoalescer) dispatch(sp stagedPacket) error {
m.tcp.addVerbatim(sp.pkt) m.tcp.addVerbatim(sp.pkt)
return nil return nil
} }
info, ok := parseTCPAt(sp.pkt, int(sp.ipHdrLen)) var info parsedTCP
if !ok { if !info.parseAt(sp.pkt, int(sp.ipHdrLen)) {
m.tcp.sealAllOpen() m.tcp.sealAllOpen()
m.tcp.addVerbatim(sp.pkt) m.tcp.addVerbatim(sp.pkt)
return nil return nil
} }
return m.tcp.commitParsed(sp.pkt, info) return m.tcp.commitParsed(sp.pkt, &info)
} }
case ipProtoUDP: case ipProtoUDP:
if m.udp != nil { if m.udp != nil {
@@ -109,13 +109,13 @@ func (m *MultiCoalescer) dispatch(sp stagedPacket) error {
m.udp.addVerbatim(sp.pkt) m.udp.addVerbatim(sp.pkt)
return nil return nil
} }
info, ok := parseUDPAt(sp.pkt, int(sp.ipHdrLen)) var info parsedUDP
if !ok { if !info.parseAt(sp.pkt, int(sp.ipHdrLen)) {
m.udp.sealAllOpen() m.udp.sealAllOpen()
m.udp.addVerbatim(sp.pkt) m.udp.addVerbatim(sp.pkt)
return nil return nil
} }
return m.udp.commitParsed(sp.pkt, info) return m.udp.commitParsed(sp.pkt, &info)
} }
} }
return m.pt.enqueue(sp.pkt) return m.pt.enqueue(sp.pkt)
+29 -31
View File
@@ -105,41 +105,39 @@ type parsedTCP struct {
flags byte flags byte
} }
// parseTCPAt extracts the flow key and IP/TCP offsets for a packet the dispatcher already knows is // parseAt extracts the flow key and IP/TCP offsets for a packet the dispatcher already knows is
// TCP; ipHdrLen is the upstream-resolved L4 offset (see parseIPAt). Returns ok=false for malformed // TCP; ipHdrLen is the upstream-resolved L4 offset (see flowKey.parseIPAt). p must be zero on
// input or any shape that must not coalesce (IPv4 options/fragmentation, IPv6 extension headers). // entry and is filled in place; see flowKey.parseIPAt for why. Returns false for malformed input
func parseTCPAt(pkt []byte, ipHdrLen int) (parsedTCP, bool) { // or any shape that must not coalesce (IPv4 options/fragmentation, IPv6 extension headers).
ip, ok := parseIPAt(pkt, ipHdrLen) func (p *parsedTCP) parseAt(pkt []byte, ipHdrLen int) bool {
trimmed, ok := p.fk.parseIPAt(pkt, ipHdrLen)
if !ok { if !ok {
return parsedTCP{}, false return false
} }
return parseTCPTail(ip) return p.parseTail(trimmed, ipHdrLen)
} }
// parseTCPTail layers the TCP-header parse on a validated IP prologue. // parseTail layers the TCP-header parse on a validated IP prologue. pkt is the trimmed packet;
func parseTCPTail(ip parsedIP) (parsedTCP, bool) { // fk's addresses are already filled.
var p parsedTCP func (p *parsedTCP) parseTail(pkt []byte, ipHdrLen int) bool {
pkt := ip.pkt if len(pkt) < ipHdrLen+20 {
p.fk = ip.fk return false
p.ipHdrLen = ip.ipHdrLen
if len(pkt) < p.ipHdrLen+20 {
return p, false
} }
tcpOff := int(pkt[p.ipHdrLen+12]>>4) * 4 tcpOff := int(pkt[ipHdrLen+12]>>4) * 4
if tcpOff < 20 || tcpOff > 60 { if tcpOff < 20 || tcpOff > 60 {
return p, false return false
} }
if len(pkt) < p.ipHdrLen+tcpOff { if len(pkt) < ipHdrLen+tcpOff {
return p, false return false
} }
p.hdrLen = p.ipHdrLen + tcpOff p.ipHdrLen = ipHdrLen
p.hdrLen = ipHdrLen + tcpOff
p.payLen = len(pkt) - p.hdrLen p.payLen = len(pkt) - p.hdrLen
p.fk.sport = binary.BigEndian.Uint16(pkt[p.ipHdrLen : p.ipHdrLen+2]) p.fk.sport = binary.BigEndian.Uint16(pkt[ipHdrLen : ipHdrLen+2])
p.fk.dport = binary.BigEndian.Uint16(pkt[p.ipHdrLen+2 : p.ipHdrLen+4]) p.fk.dport = binary.BigEndian.Uint16(pkt[ipHdrLen+2 : ipHdrLen+4])
p.seq = binary.BigEndian.Uint32(pkt[p.ipHdrLen+4 : p.ipHdrLen+8]) p.seq = binary.BigEndian.Uint32(pkt[ipHdrLen+4 : ipHdrLen+8])
p.flags = pkt[p.ipHdrLen+13] p.flags = pkt[ipHdrLen+13]
return p, true return true
} }
// TCP flag bits (byte 13 of the TCP header). Only the bits the coalescer consults are named; // TCP flag bits (byte 13 of the TCP header). Only the bits the coalescer consults are named;
@@ -157,9 +155,9 @@ func (c *TCPCoalescer) sealAllOpen() {
c.lastSlot = nil c.lastSlot = nil
} }
// commitParsed commits one parsed TCP packet. The caller (dispatch, via parseTCPAt) supplies a // commitParsed commits one parsed TCP packet. The caller (dispatch, via parseAt) supplies a
// valid parse so the header is not re-walked here. // valid parse so the header is not re-walked here.
func (c *TCPCoalescer) commitParsed(pkt []byte, info parsedTCP) error { func (c *TCPCoalescer) commitParsed(pkt []byte, info *parsedTCP) error {
// Admission: only ACK, ACK|PSH, ACK|ECE, ACK|PSH|ECE may ride a coalesce chain. CWR marks a // Admission: only ACK, ACK|PSH, ACK|ECE, ACK|PSH|ECE may ride a coalesce chain. CWR marks a
// one-shot congestion transition the receiver must observe at a segment boundary. NB: AccECN // one-shot congestion transition the receiver must observe at a segment boundary. NB: AccECN
// reuses CWR as ACE counter bits; revisit this check if inner hosts adopt AccECN. // reuses CWR as ACE counter bits; revisit this check if inner hosts adopt AccECN.
@@ -253,7 +251,7 @@ func (c *TCPCoalescer) addVerbatim(pkt []byte) {
c.slots = append(c.slots, s) c.slots = append(c.slots, s)
} }
func (c *TCPCoalescer) seed(pkt []byte, info parsedTCP) { func (c *TCPCoalescer) seed(pkt []byte, info *parsedTCP) {
if info.hdrLen > tcpCoalesceHdrCap || info.hdrLen+info.payLen > tcpCoalesceBufSize { 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.addVerbatim(pkt) c.addVerbatim(pkt)
@@ -288,7 +286,7 @@ func (c *TCPCoalescer) seed(pkt []byte, info parsedTCP) {
// contents, adjacent seq, not oversized. A closed chain never reaches here; closing removes the // 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 // 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. // lazily on the first append; every field consulted here is one the pre-flush patches never touch.
func (c *TCPCoalescer) canAppend(s *coalesceSlot, pkt []byte, info parsedTCP) bool { func (c *TCPCoalescer) canAppend(s *coalesceSlot, pkt []byte, info *parsedTCP) bool {
if info.hdrLen != s.hdrLen { if info.hdrLen != s.hdrLen {
return false return false
} }
@@ -322,7 +320,7 @@ func (c *TCPCoalescer) canAppend(s *coalesceSlot, pkt []byte, info parsedTCP) bo
// appendPayload folds info's packet into s and reports whether the chain is now closed: the // appendPayload folds info's packet into s and reports whether the chain is now closed: the
// segment was sub-gsoSize (kernel TSO allows only the final segment to be short) or carried PSH. // 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. // The caller must deregister a closed slot from openSlots.
func (c *TCPCoalescer) appendPayload(s *coalesceSlot, pkt []byte, info parsedTCP) bool { func (c *TCPCoalescer) appendPayload(s *coalesceSlot, pkt []byte, info *parsedTCP) bool {
if s.numSeg == 1 { if s.numSeg == 1 {
// First append: populate hdrBuf from the seed. Deferred out of seed so solo slots, which // First append: populate hdrBuf from the seed. Deferred out of seed so solo slots, which
// flush from rawPkt, never pay the copy. // flush from rawPkt, never pay the copy.
+1 -1
View File
@@ -145,7 +145,7 @@ func BenchmarkCommitRunInterleaved4(b *testing.B) {
runCommitBench(b, pkts, len(pkts)) runCommitBench(b, pkts, len(pkts))
} }
// BenchmarkCommitPassthrough exercises the non-TCP branch: parseTCPBase // BenchmarkCommitPassthrough exercises the non-TCP branch: parseBase
// bails early and addVerbatim is the only work. // bails early and addVerbatim is the only work.
func BenchmarkCommitPassthrough(b *testing.B) { func BenchmarkCommitPassthrough(b *testing.B) {
pkt := buildICMPv4() pkt := buildICMPv4()
+1 -1
View File
@@ -1252,7 +1252,7 @@ func TestCoalescerUnparseableSealsAllChains(t *testing.T) {
if err := c.Commit(buildTCPv4(2200, tcpAck, pay)); err != nil { if err := c.Commit(buildTCPv4(2200, tcpAck, pay)); err != nil {
t.Fatal(err) t.Fatal(err)
} }
// IHL=6 fakes IP options: parseTCPBase bails, flow key unknown. // IHL=6 fakes IP options: the parse bails, flow key unknown.
opts := buildTCPv4(5000, tcpAck, make([]byte, 500)) opts := buildTCPv4(5000, tcpAck, make([]byte, 500))
opts[0] = 0x46 opts[0] = 0x46
if err := c.Commit(opts); err != nil { if err := c.Commit(opts); err != nil {
+26 -28
View File
@@ -83,42 +83,40 @@ type parsedUDP struct {
payLen int payLen int
} }
// parseUDPAt extracts the flow key and IP/UDP offsets for a packet the dispatcher already knows is // parseAt extracts the flow key and IP/UDP offsets for a packet the dispatcher already knows is
// UDP; ipHdrLen is the upstream-resolved L4 offset (see parseIPAt). Returns ok=false for malformed // UDP; ipHdrLen is the upstream-resolved L4 offset (see flowKey.parseIPAt). p must be zero on
// input or any shape that must not coalesce (IPv4 options/fragmentation, IPv6 extension headers). // entry and is filled in place. Returns false for malformed input or any shape that must not
func parseUDPAt(pkt []byte, ipHdrLen int) (parsedUDP, bool) { // coalesce (IPv4 options/fragmentation, IPv6 extension headers).
ip, ok := parseIPAt(pkt, ipHdrLen) func (p *parsedUDP) parseAt(pkt []byte, ipHdrLen int) bool {
trimmed, ok := p.fk.parseIPAt(pkt, ipHdrLen)
if !ok { if !ok {
return parsedUDP{}, false return false
} }
return parseUDPTail(ip) return p.parseTail(trimmed, ipHdrLen)
} }
// parseUDPTail layers the UDP-header parse on a validated IP prologue. // parseTail layers the UDP-header parse on a validated IP prologue. pkt is the trimmed packet;
func parseUDPTail(ip parsedIP) (parsedUDP, bool) { // fk's addresses are already filled.
var p parsedUDP func (p *parsedUDP) parseTail(pkt []byte, ipHdrLen int) bool {
pkt := ip.pkt if len(pkt) < ipHdrLen+8 {
p.fk = ip.fk return false
p.ipHdrLen = ip.ipHdrLen
if len(pkt) < p.ipHdrLen+8 {
return p, false
} }
p.hdrLen = p.ipHdrLen + 8
// UDP `length` field: must equal IP-derived length-of-UDP-header-plus-payload. // UDP `length` field: must equal IP-derived length-of-UDP-header-plus-payload.
udpLen := int(binary.BigEndian.Uint16(pkt[p.ipHdrLen+4 : p.ipHdrLen+6])) udpLen := int(binary.BigEndian.Uint16(pkt[ipHdrLen+4 : ipHdrLen+6]))
if udpLen < 8 || udpLen > len(pkt)-p.ipHdrLen { if udpLen < 8 || udpLen > len(pkt)-ipHdrLen {
return p, false return false
} }
p.ipHdrLen = ipHdrLen
p.hdrLen = ipHdrLen + 8
p.payLen = udpLen - 8 p.payLen = udpLen - 8
p.fk.sport = binary.BigEndian.Uint16(pkt[p.ipHdrLen : p.ipHdrLen+2]) p.fk.sport = binary.BigEndian.Uint16(pkt[ipHdrLen : ipHdrLen+2])
p.fk.dport = binary.BigEndian.Uint16(pkt[p.ipHdrLen+2 : p.ipHdrLen+4]) p.fk.dport = binary.BigEndian.Uint16(pkt[ipHdrLen+2 : ipHdrLen+4])
return p, true return true
} }
// commitParsed commits one parsed UDP packet. The caller (dispatch, via parseUDPAt) supplies a // commitParsed commits one parsed UDP packet. The caller (dispatch, via parseAt) supplies a
// valid parse so the header is not re-walked here. // valid parse so the header is not re-walked here.
func (c *UDPCoalescer) commitParsed(pkt []byte, info parsedUDP) error { 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 // 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. The len guard skips hashing the key when no flow is open.
if info.payLen == 0 { if info.payLen == 0 {
@@ -198,7 +196,7 @@ func (c *UDPCoalescer) addVerbatim(pkt []byte) {
c.slots = append(c.slots, s) c.slots = append(c.slots, s)
} }
func (c *UDPCoalescer) seed(pkt []byte, info parsedUDP) { func (c *UDPCoalescer) seed(pkt []byte, info *parsedUDP) {
if info.hdrLen > udpCoalesceHdrCap || info.hdrLen+info.payLen > udpCoalesceBufSize { if info.hdrLen > udpCoalesceHdrCap || info.hdrLen+info.payLen > udpCoalesceBufSize {
c.addVerbatim(pkt) c.addVerbatim(pkt)
return return
@@ -224,7 +222,7 @@ func (c *UDPCoalescer) seed(pkt []byte, info parsedUDP) {
// canAppend reports whether info's packet extends the slot's seed. // canAppend reports whether info's packet extends the slot's seed.
// Kernel UDP-GSO requires every segment except possibly the last to be // Kernel UDP-GSO requires every segment except possibly the last to be
// exactly gsoSize, and the last may be shorter (≤ gsoSize). // exactly gsoSize, and the last may be shorter (≤ gsoSize).
func (c *UDPCoalescer) canAppend(s *udpSlot, pkt []byte, info parsedUDP) bool { func (c *UDPCoalescer) canAppend(s *udpSlot, pkt []byte, info *parsedUDP) bool {
if info.hdrLen != s.hdrLen { if info.hdrLen != s.hdrLen {
return false return false
} }
@@ -252,7 +250,7 @@ func (c *UDPCoalescer) canAppend(s *udpSlot, pkt []byte, info parsedUDP) bool {
// appendPayload folds info's packet into s and reports whether the chain is now closed: kernel // appendPayload folds info's packet into s and reports whether the chain is now closed: kernel
// UDP-GSO requires every segment but the last to be exactly gsoSize, so a short segment must be // 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. // the final one. The caller must deregister a closed slot from openSlots.
func (c *UDPCoalescer) appendPayload(s *udpSlot, pkt []byte, info parsedUDP) bool { func (c *UDPCoalescer) appendPayload(s *udpSlot, pkt []byte, info *parsedUDP) bool {
if s.numSeg == 1 { if s.numSeg == 1 {
// First append: populate hdrBuf from the seed. Deferred out of seed so solo slots, which // First append: populate hdrBuf from the seed. Deferred out of seed so solo slots, which
// flush from rawPkt, never pay the copy. // flush from rawPkt, never pay the copy.