personal_interest_mirrors/nebula
1
0
Fork 0
mirror of https://github.com/slackhq/nebula.git synced 2026-08-15 11:47:02 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

stuff

Browse Source
This commit is contained in:
JackDoan
2026-08-04 09:03:40 -05:00
parent 3b1004588d
commit a3eef407b2
11 changed files with 163 additions and 366 deletions
Download Patch File Download Diff File Expand all files Collapse all files
overlay/batch/batch.go
+4 -29
View File
@@ -1,36 +1,11 @@
package batch
import "github.com/slackhq/nebula/firewall"
// SortKey identifies a packet's position in its sender's transmission order.
// Epoch is a receiver-local ordinal for the tunnel (ConnectionState) that
// decrypted the packet. A re-handshake replaces the tunnel outright — new
// hostinfo, new keys, a fresh counter space — and the replacement's epoch is
// higher, so during the cutover overlap the old tunnel's packets sort first.
// Counter is the packet's AEAD message counter within that tunnel. The replay
// window has already rejected duplicates by Commit time, so keys are unique
// per tunnel and (Epoch, Counter) is a total order with no ties.
// Epoch is a receiver-local ordinal for the tunnel (ConnectionState) that decrypted the packet:
// a re-handshake replaces the tunnel outright and the replacement's epoch is higher,
// so the old tunnel's packets sort first during the cutover overlap.
// Counter is the packet's AEAD message counter within that tunnel.
type SortKey struct {
Epoch uint64
Counter uint64
}
type RxBatcher interface {
// Commit stages pkt to be flushed by the batch. key must carry the
// packet's session epoch and message counter; pp must be the firewall's
// parse of this same packet. The caller must keep pkt valid until the
// next Flush, and not re-use it. pp, by contrast, is borrowed only for
// the duration of the call — the caller reuses one ParsedPacket per
// receive loop — so implementations must copy what they need from it.
Commit(pkt []byte, key SortKey, pp *firewall.ParsedPacket) error
// Flush emits every staged packet. Packets are first sorted by key, so
// within each protocol lane emission follows the sender's transmission
// order regardless of arrival order. One shape may legally be overtaken
// by later same-flow data: a pure TCP ACK, which does not close its
// flow's open coalesce chain (a late ACK is just a stale ACK). Cross-lane
// order (TCP vs UDP vs everything else) is not preserved.
// Returns the first error observed; keeps draining so one bad packet
// doesn't hold up the rest.
// After Flush returns, committed payload slices may be recycled.
Flush() error
}
overlay/batch/coalesce_core.go
+12 -49
View File
@@ -20,7 +20,7 @@ type flowKey struct {
// so this matches a typical carrier-side recvmmsg batch on the UDP socket.
const initialSlots = 64
// parsedIP is the IP-level result of parseIPPrologue.
// 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
@@ -31,46 +31,11 @@ type parsedIP struct {
pkt []byte
}
// parseIPPrologue extracts the IP-level fields the coalescers care about:
// IHL/payload length, version, src/dst addresses, and the L4 protocol byte.
// Returns ok=false for malformed input, IPv4 with options or fragmentation,
// or IPv6 with extension headers (all rejected by both coalescers in
// identical ways before this refactor).
//
// On success, p.pkt is len-trimmed to the IP-declared length so callers
// don't have to repeat the trim. wantProto is the IANA protocol number to
// require (6 for TCP, 17 for UDP); ok=false for any other value.
// This is the standalone-lane-Commit entry; the dispatcher path uses
// parseIPAt, where the protocol was already resolved upstream.
func parseIPPrologue(pkt []byte, wantProto byte) (parsedIP, bool) {
var p parsedIP
if len(pkt) < 20 {
return p, false
}
switch pkt[0] >> 4 {
case 4:
if pkt[9] != wantProto {
return p, false
}
return parseIPv4Prologue(pkt)
case 6:
if len(pkt) < 40 {
return p, false
}
if pkt[6] != wantProto {
return p, false
}
return parseIPv6Prologue(pkt)
}
return p, false
}
// parseIPAt is the dispatcher-path prologue: newPacket already resolved the
// L4 protocol and header offset once for the firewall, so the proto sniff is
// replaced by a cross-check of the caller's ipHdrLen. A plain header (v4:
// IHL 20, v6: exactly 40 — no options, no extension headers) is the only
// coalesceable shape, which is the same rule parseIPPrologue enforces
// through its own reads.
// 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
// 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
// byte 40. On success p.pkt is trimmed to the IP-declared length.
func parseIPAt(pkt []byte, ipHdrLen int) (parsedIP, bool) {
var p parsedIP
if len(pkt) < 20 {
@@ -91,18 +56,16 @@ func parseIPAt(pkt []byte, ipHdrLen int) (parsedIP, bool) {
return p, false
}
// parseIPv4Prologue is the shared IPv4 tail of the two prologue entries.
// The caller has verified len(pkt) >= 20 and either the protocol
// (parseIPPrologue) or the upstream-resolved header length (parseIPAt).
// parseIPv4Prologue is the shared IPv4 tail of the prologue entries; the
// caller has verified len(pkt) >= 20 and the version.
func parseIPv4Prologue(pkt []byte) (parsedIP, bool) {
var p parsedIP
ihl := int(pkt[0]&0x0f) * 4
if ihl != 20 {
return p, false
}
// Reject actual fragmentation (MF or non-zero frag offset). On the
// dispatcher path FragAny was already gated; kept as defense in depth —
// a fragment folded into a superpacket would corrupt reassembly.
// 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.
if binary.BigEndian.Uint16(pkt[6:8])&0x3fff != 0 {
return p, false
}
@@ -118,8 +81,8 @@ func parseIPv4Prologue(pkt []byte) (parsedIP, bool) {
return p, true
}
// parseIPv6Prologue is the shared IPv6 tail; caller has verified
// len(pkt) >= 40 and version/proto-or-offset.
// parseIPv6Prologue is the shared IPv6 tail; the caller has verified
// len(pkt) >= 40, the version, and that the L4 header sits at byte 40.
func parseIPv6Prologue(pkt []byte) (parsedIP, bool) {
var p parsedIP
payloadLen := int(binary.BigEndian.Uint16(pkt[4:6]))
overlay/batch/multi_coalesce.go
+35 -58
View File
@@ -1,6 +1,7 @@
package batch
import (
"cmp"
"errors"
"io"
"log/slog"
@@ -9,47 +10,34 @@ import (
"github.com/slackhq/nebula/firewall"
)
// MultiCoalescer stages plaintext packets with their (epoch, counter) sort
// keys, and at Flush replays them in sender-transmission order into
// lane-specific batchers selected by the IP/L4 protocol of the packet.
// MultiCoalescer stages plaintext packets with their (epoch, counter) sort keys and, at Flush,
// replays them in sender-transmission order into lane-specific batchers selected by L4 protocol.
//
// Sorting *before* the lanes see anything is what makes the ordering story
// simple: each lane consumes packets in transmission order, builds its slots
// in that order, and emits them in creation order. Wire reorder inside a
// flush batch is repaired here, before it can fragment a lane's coalesce
// chains, so the lanes carry no reorder-repair machinery of their own.
// Sorting before dispatch keeps the ordering story simple: each lane consumes packets in
// transmission order, builds slots in that order, and emits them in creation order. Wire reorder
// inside a flush batch is repaired here, before it can fragment a lane's coalesce chains, so the
// lanes carry no reorder-repair machinery.
//
// The ordering contract is per-tunnel transmission order within each lane:
// a sender's packets are emitted in the order it encrypted them. Two
// qualifications:
// - a pure TCP ACK may be overtaken by later same-flow data, because it
// does not close the flow's open coalesce chain (a late ACK is just a
// stale ACK; see TCPCoalescer.commitParsed);
// - an unparseable shape (fragment, IP options) seals every open chain in
// its lane — its flow is unknowable, so this is the only way to keep
// later data from extending a chain that would emit ahead of it. The
// packet then rides its lane as an in-lane passthrough, still in
// transmission order.
// The contract is per-tunnel transmission order within each lane, with two exceptions: a pure TCP
// ACK may be overtaken by later same-flow data (it does not close the flow's open chain; a late
// ACK is just a stale ACK), and an unparseable shape seals every open chain in its lane (its flow
// is unknown) and rides the lane as an in-lane verbatim, still in transmission order. Routing
// follows the flow: a flow's non-coalesceable shapes ride its protocol lane rather than falling
// to the later-flushed pt lane.
//
// Routing follows the flow, not the coalesceability: IPv4 fragments keep
// their L4 proto visible and IPv6 extension chains are walked to the
// terminal proto, so a flow's non-coalesceable shapes ride its lane rather
// than falling to the later-flushed pt lane.
//
// Cross-lane order is intentionally NOT preserved across the TCP/UDP/verbatim split.
// Cross-lane order (TCP vs UDP vs everything else) is not preserved.
type MultiCoalescer struct {
tcp *TCPCoalescer
udp *UDPCoalescer
pt *Passthrough
// staged holds this batch's packets and sort keys until Flush. Borrowed:
// the caller keeps each pkt alive until Flush returns.
// staged holds this batch's packets and sort keys until Flush. Borrowed: the caller keeps
// each pkt alive until Flush returns.
staged []stagedPacket
}
// stagedPacket also carries the scalars dispatch needs from the firewall's
// ParsedPacket: pp itself is reused by the caller per packet and must not be
// retained past Commit, so the relevant fields are copied by value here.
// stagedPacket carries the scalars dispatch needs from the firewall's ParsedPacket, copied by
// value: pp is reused by the caller per packet and must not be retained past Commit.
type stagedPacket struct {
pkt []byte
key SortKey
@@ -58,10 +46,10 @@ type stagedPacket struct {
ipHdrLen uint16
}
// NewMultiCoalescer builds a multi-lane batcher over w, based on available
// protocol support. The staging sort applies even when no GSO lane is
// available: passthrough-only platforms still get transmission-order repair.
func NewMultiCoalescer(w io.Writer, l *slog.Logger) RxBatcher {
// NewMultiCoalescer builds a multi-lane batcher over w, based on available protocol support. The
// staging sort applies even when no GSO lane is available: passthrough-only platforms still get
// transmission-order repair.
func NewMultiCoalescer(w io.Writer, l *slog.Logger) *MultiCoalescer {
m := &MultiCoalescer{
pt: NewPassthrough(w),
staged: make([]stagedPacket, 0, initialSlots),
@@ -71,10 +59,10 @@ func NewMultiCoalescer(w io.Writer, l *slog.Logger) RxBatcher {
return m
}
// Commit stages pkt for the next Flush. All lane dispatch is deferred to
// Flush so it runs on packets already in transmission order. pp is the
// firewall's parse of pkt — the single source of truth for the packet's
// protocol and L4 offset — and is only borrowed for this call.
// 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.
func (m *MultiCoalescer) Commit(pkt []byte, key SortKey, pp *firewall.ParsedPacket) error {
m.staged = append(m.staged, stagedPacket{
pkt: pkt,
@@ -86,24 +74,12 @@ func (m *MultiCoalescer) Commit(pkt []byte, key SortKey, pp *firewall.ParsedPack
return nil
}
// compareStaged orders staged packets by (epoch, counter): sender
// transmission order within a tunnel, tunnel-creation order across a
// re-handshake cutover. Keys are unique (see SortKey), so this is a total
// order and sort stability doesn't matter.
// compareStaged orders staged packets by (epoch, counter)
func compareStaged(a, b stagedPacket) int {
if a.key.Epoch != b.key.Epoch {
if a.key.Epoch < b.key.Epoch {
return -1
}
return 1
if c := cmp.Compare(a.key.Epoch, b.key.Epoch); c != 0 {
return c
}
if a.key.Counter == b.key.Counter {
return 0
}
if a.key.Counter < b.key.Counter {
return -1
}
return 1
return cmp.Compare(a.key.Counter, b.key.Counter)
}
// dispatch routes one staged packet to its lane.
@@ -145,11 +121,12 @@ func (m *MultiCoalescer) dispatch(sp stagedPacket) error {
return m.pt.enqueue(sp.pkt)
}
// Flush sorts the staged batch into transmission order, replays it into the
// lanes, then flushes each lane.
// Flush sorts the staged batch into transmission order, replays it into the lanes, then flushes each lane.
// Drains everything and returns the joined errors; one bad packet does not hold up the rest.
// After Flush returns, committed payload slices may be recycled.
func (m *MultiCoalescer) Flush() error {
// Arrival order is already almost sorted (reorder is the exception, not
// the rule), which pdqsort detects and handles in near-linear time.
// Arrival order is already almost sorted (reorder is the exception), which pdqsort detects
// and handles in near-linear time.
slices.SortFunc(m.staged, compareStaged)
var errs []error
overlay/batch/multi_coalesce_test.go
+2 -8
View File
@@ -21,16 +21,10 @@ func (k *keySeq) next() SortKey {
return SortKey{Epoch: k.epoch, Counter: k.counter}
}
// newTestMultiCoalescer builds a batcher over w and asserts the concrete
// type so tests can reach into the lanes.
// newTestMultiCoalescer builds a batcher over w.
func newTestMultiCoalescer(tb testing.TB, w io.Writer) *MultiCoalescer {
tb.Helper()
b := NewMultiCoalescer(w, test.NewLogger())
m, ok := b.(*MultiCoalescer)
if !ok {
tb.Fatalf("want a *MultiCoalescer, got %T", b)
}
return m
return NewMultiCoalescer(w, test.NewLogger())
}
// TestMultiCoalescerRoutesByProto confirms TCP/UDP/other land in the right
overlay/batch/passthrough.go
+3 -9
View File
@@ -2,11 +2,10 @@ package batch
import (
"io"
"github.com/slackhq/nebula/firewall"
)
// Passthrough is a RxBatcher that doesn't batch anything, it just accumulates and then sends packets.
// Passthrough is MultiCoalescer's verbatim lane: no batching, packets are written at Flush in the
// order enqueued.
type Passthrough struct {
out io.Writer
slots [][]byte
@@ -19,12 +18,7 @@ func NewPassthrough(w io.Writer) *Passthrough {
}
}
func (p *Passthrough) Commit(pkt []byte, _ SortKey, _ *firewall.ParsedPacket) error {
return p.enqueue(pkt)
}
// enqueue is the lane-facing half of Commit: MultiCoalescer.dispatch hands
// packets here already sorted into transmission order.
// enqueue accepts one packet, already sorted into transmission order by dispatch.
func (p *Passthrough) enqueue(pkt []byte) error {
p.slots = append(p.slots, pkt)
return nil
overlay/batch/tcp_coalesce.go
+70 -147
View File
@@ -26,24 +26,21 @@ const tcpCoalesceMaxSegs = 64
// into. IPv6 (40) + TCP with full options (60) = 100 bytes.
const tcpCoalesceHdrCap = 100
// coalesceSlot is one entry in the coalescer's ordered event queue.
// When verbatim is true the slot holds a single borrowed packet that is
// emitted as-is (pure ACK, non-admissible TCP, unparseable, or oversize seed).
// When verbatim is false the slot is an in-progress coalesced superpacket.
// hdrBuf is a mutable copy of the seed's IP+TCP header, populated on the
// first append (we patch total length and pseudo-header partial at flush;
// a slot that never grows flushes from rawPkt and never touches hdrBuf)
// payIovs are *borrowed* slices from the caller's plaintext buffers.
// The caller (listenOut) must keep those buffers alive until Flush.
// 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
// caller's plaintext buffers; the caller must keep them alive until Flush.
type coalesceSlot struct {
verbatim bool
// rawPkt is borrowed: the whole packet for verbatim 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
// checksum ships DATA_VALID instead of making the kernel recompute it.
// 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.
rawPkt []byte
fk flowKey
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
hdrLen int
ipHdrLen int
@@ -55,24 +52,20 @@ type coalesceSlot struct {
payIovs [][]byte
}
// 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.
// It expects its input in sender-transmission order (MultiCoalescer sorts the
// staged batch by (epoch, counter) before dispatching here) and emits slots in
// creation order, which therefore reproduces transmission order — modulo the
// pure-ACK allowance in commitParsed.
// Owns no locks; one coalescer per TUN write queue.
// 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. Input must be in sender
// transmission order (MultiCoalescer sorts by (epoch, counter) before dispatch); slots are emitted
// in creation order, so emission reproduces transmission order except for the pure-ACK case in
// commitParsed. Owns no locks; one coalescer per TUN write queue.
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 w.Write (verbatim).
slots []*coalesceSlot
// openSlots maps a flow key to its still-open slot, so new segments can
// extend an in-progress superpacket in O(1). Membership here is what
// keeps a chain extendable: slots are removed when they close (PSH or
// short-last-segment), when a non-admissible packet for that flow
// arrives, or in Flush.
// openSlots maps a flow key to its open slot so new segments can extend an in-progress
// superpacket in O(1). Removal is what closes a chain: on PSH or a short last segment, on a
// non-admissible packet for the flow, or in Flush.
openSlots map[flowKey]*coalesceSlot
// lastSlot caches the most recently touched open slot. Bulk traffic
// arrives in same-flow runs (single-flow steady state, or GRO bursts
@@ -104,28 +97,17 @@ func NewTCPCoalescer(w io.Writer, l *slog.Logger) *TCPCoalescer {
// parsedTCP holds the fields extracted from a single parse so later steps
// (admission, slot lookup, canAppend) don't re-walk the header.
type parsedTCP struct {
fk flowKey
ipHdrLen int
tcpHdrLen int
hdrLen int
payLen int
seq uint32
flags byte
fk flowKey
ipHdrLen int
hdrLen int
payLen int
seq uint32
flags byte
}
// parseTCPBase extracts the flow key and IP/TCP offsets for any TCP packet,
// regardless of whether it's admissible for coalescing. Returns ok=false for non-TCP or malformed input.
// Accepts IPv4 (no options or fragmentation) and IPv6 (no extension headers).
func parseTCPBase(pkt []byte) (parsedTCP, bool) {
ip, ok := parseIPPrologue(pkt, ipProtoTCP)
if !ok {
return parsedTCP{}, false
}
return parseTCPTail(ip)
}
// parseTCPAt is parseTCPBase for the dispatcher path: the packet is already
// known to be TCP and ipHdrLen is the upstream-resolved L4 offset (see parseIPAt).
// parseTCPAt 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
// input or any shape that must not coalesce (IPv4 options/fragmentation, IPv6 extension headers).
func parseTCPAt(pkt []byte, ipHdrLen int) (parsedTCP, bool) {
ip, ok := parseIPAt(pkt, ipHdrLen)
if !ok {
@@ -151,7 +133,6 @@ func parseTCPTail(ip parsedIP) (parsedTCP, bool) {
if len(pkt) < p.ipHdrLen+tcpOff {
return p, false
}
p.tcpHdrLen = tcpOff
p.hdrLen = p.ipHdrLen + tcpOff
p.payLen = len(pkt) - p.hdrLen
p.fk.sport = binary.BigEndian.Uint16(pkt[p.ipHdrLen : p.ipHdrLen+2])
@@ -161,85 +142,31 @@ func parseTCPTail(ip parsedIP) (parsedTCP, bool) {
return p, true
}
// TCP flag bits (byte 13 of the TCP header). Only the bits actually consulted
// by the coalescer are named; FIN/SYN/RST/URG/CWR are rejected via the
// negative mask in coalesceable, not by name.
// TCP flag bits (byte 13 of the TCP header). Only the bits the coalescer consults are named;
// FIN/SYN/RST/URG/CWR are rejected by the negative mask in commitParsed.
const (
tcpFlagPsh = 0x08
tcpFlagAck = 0x10
tcpFlagEce = 0x40
)
// coalesceable reports whether a parsed TCP segment is eligible for
// coalescing. Accepts ACK, ACK|PSH, ACK|ECE, ACK|PSH|ECE with a
// non-empty payload. CWR is excluded because it marks a one-shot
// congestion-window-reduced transition the receiver must observe at a
// segment boundary.
func (p parsedTCP) coalesceable() bool {
if p.flags&tcpFlagAck == 0 {
return false
}
if p.flags&^(tcpFlagAck|tcpFlagPsh|tcpFlagEce) != 0 {
return false
}
return p.payLen > 0
}
// pureAck reports whether a parsed segment is a bare acknowledgment: no
// payload and nothing beyond ACK|PSH|ECE in the flags. These are the only
// non-coalesceable shape that may safely pass through WITHOUT sealing the
// flow's open slot — a late-delivered stale ACK is ignored by the receiver,
// whereas SYN/FIN/RST/CWR mark transitions the flow must observe in order.
func (p parsedTCP) pureAck() bool {
return p.payLen == 0 &&
p.flags&tcpFlagAck != 0 &&
p.flags&^(tcpFlagAck|tcpFlagPsh|tcpFlagEce) == 0
}
func (c *TCPCoalescer) Commit(pkt []byte) error {
info, ok := parseTCPBase(pkt)
if !ok {
// Unparseable shape: flow key unknowable, so seal every open chain to
// keep later data from extending a chain that would emit ahead of it.
c.sealAllOpen()
c.addVerbatim(pkt)
return nil
}
return c.commitParsed(pkt, info)
}
// sealAllOpen closes every open coalesce chain: nothing committed after this
// call can extend a slot created before it. Called when an unparseable packet
// arrives — its flow is unknown, so any open chain might be the one whose
// later data would otherwise leapfrog it.
// sealAllOpen closes every open coalesce chain. Called for unparseable packets: the flow key is
// unknown, so any open chain could otherwise absorb later data and emit it ahead of this packet.
func (c *TCPCoalescer) sealAllOpen() {
clear(c.openSlots)
c.lastSlot = nil
}
// commitParsed is the post-parse half of Commit. The caller must have
// already verified parseTCPBase succeeded (info is a valid TCP parse).
// Used by MultiCoalescer.Commit to avoid re-walking the IP/TCP header
// after the dispatcher has already done so.
// commitParsed commits one parsed TCP packet. The caller (dispatch, via parseTCPAt) supplies a
// valid parse so the header is not re-walked here.
func (c *TCPCoalescer) commitParsed(pkt []byte, info parsedTCP) error {
if !info.coalesceable() {
if info.pureAck() {
// A bare window/ack update carries no ordering obligation toward
// the flow's data: delivering it after later-transmitted data only
// makes it a stale ACK, which receivers ignore. Skipping the
// 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. This is the one place
// emission can deviate from transmission order.
c.addVerbatim(pkt)
return nil
}
// TCP but not admissible (SYN/FIN/RST/URG/CWR or a shape the flow
// must observe in sequence). Seal this flow's open slot so later
// in-flow packets don't extend it and emit ahead of this verbatim;
// with input in transmission order that pins the verbatim's exact
// in-flow position. The len guard skips hashing the 38-byte key on
// ack-dominant queues, where the map is almost always empty.
// 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
// 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
@@ -249,6 +176,14 @@ func (c *TCPCoalescer) commitParsed(pkt []byte, info parsedTCP) error {
c.addVerbatim(pkt)
return nil
}
if info.payLen == 0 {
// Pure ACK: no ordering obligation toward the flow's data. Delivering it after
// later-transmitted data only makes it a stale ACK, which receivers ignore. Not sealing
// keeps a bidirectional flow's data run coalescing across interleaved peer ACKs, matching
// kernel GRO. This is the only place emission deviates from transmission order.
c.addVerbatim(pkt)
return nil
}
// Cached-slot fast path. Arrival isn't per-packet interleaved even with
// many flows: wire-side GRO delivers runs of same-flow packets
@@ -291,11 +226,9 @@ func (c *TCPCoalescer) Flush() error {
for _, s := range c.slots {
var err error
if s.verbatim || 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
// software csum. appendPayload only touches hdrBuf once
// numSeg >= 2, so rawPkt is still pristine here.
// 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.
_, err = c.w.Write(s.rawPkt)
} else {
err = c.flushSlot(s)
@@ -328,8 +261,8 @@ func (c *TCPCoalescer) seed(pkt []byte, info parsedTCP) {
}
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 and is the header source for canAppend until
// the first append copies it into hdrBuf.
s.rawPkt = pkt
s.hdrLen = info.hdrLen
s.ipHdrLen = info.ipHdrLen
@@ -345,21 +278,16 @@ func (c *TCPCoalescer) seed(pkt []byte, info parsedTCP) {
c.openSlots[info.fk] = s
c.lastSlot = s
} else if last := c.lastSlot; last != nil && last.fk == info.fk {
// PSH-on-seed closes the chain immediately: never registered as
// open. Any prior cached open slot for this flow has just been
// closed-and-replaced by this seed, so drop the cache too.
// 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
}
}
// 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, and
// openSlots/lastSlot are the only paths in.
// Header reads go through rawPkt, not hdrBuf: hdrBuf is populated lazily on
// the first append, and every field consulted here is one the pre-flush
// patches never touch (headersMatch skips the flags byte, and PSH is the
// only bit patched before flush).
// 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.
func (c *TCPCoalescer) canAppend(s *coalesceSlot, pkt []byte, info parsedTCP) bool {
if info.hdrLen != s.hdrLen {
return false
@@ -391,14 +319,13 @@ func (c *TCPCoalescer) canAppend(s *coalesceSlot, pkt []byte, info parsedTCP) bo
return true
}
// 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 (a semantic delimiter). The caller
// must deregister a closed slot from openSlots.
// 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.
// 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 packet. Deferred out
// of seed so solo slots, which flush from rawPkt, never pay the copy.
// 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])
@@ -406,8 +333,7 @@ func (c *TCPCoalescer) appendPayload(s *coalesceSlot, pkt []byte, info parsedTCP
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. Without this the sender's push signal is dropped.
// Propagate PSH into the seed header so kernel TSO sets it on the last segment.
s.hdrBuf[s.ipHdrLen+13] |= tcpFlagPsh
}
return info.payLen < s.gsoSize || info.flags&tcpFlagPsh != 0
@@ -497,13 +423,10 @@ func headersMatch(a, b []byte, isV6 bool, ipHdrLen int) bool {
return true
}
// logSeqGaps reports same-flow seq discontinuities between consecutively
// created data slots. Input arrives in transmission order (MultiCoalescer
// sorts by (epoch, counter) before dispatch), so a gap here is traffic this
// batch never contained: loss upstream of nebula, a reorder spanning a flush
// boundary (which no intra-batch mechanism can repair), or a retransmit
// (negative gap). Logged so the operator can quantify how often that happens.
// The caller gates on debug level, so the map only allocates when asked for.
// logSeqGaps reports same-flow seq discontinuities between consecutively created data slots. Input
// is in transmission order, so a gap is traffic this batch never contained: loss upstream of
// nebula, reorder across a flush boundary, or a retransmit (negative gap). The caller gates on
// debug level, so the map only allocates when enabled.
func (c *TCPCoalescer) logSeqGaps() {
prevByFlow := make(map[flowKey]*coalesceSlot, len(c.slots))
for _, s := range c.slots {
overlay/batch/udp_coalesce.go
+19 -45
View File
@@ -83,19 +83,9 @@ type parsedUDP struct {
payLen int
}
// parseUDP extracts the flow key and IP/UDP offsets for a UDP packet.
// Returns ok=false for non-UDP, malformed, or unsupported header shapes
// (IPv4 with options/fragmentation, IPv6 with extension headers).
func parseUDP(pkt []byte) (parsedUDP, bool) {
ip, ok := parseIPPrologue(pkt, ipProtoUDP)
if !ok {
return parsedUDP{}, false
}
return parseUDPTail(ip)
}
// parseUDPAt is parseUDP for the dispatcher path: the packet is already
// known to be UDP and ipHdrLen is the upstream-resolved L4 offset (see parseIPAt).
// parseUDPAt 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
// input or any shape that must not coalesce (IPv4 options/fragmentation, IPv6 extension headers).
func parseUDPAt(pkt []byte, ipHdrLen int) (parsedUDP, bool) {
ip, ok := parseIPAt(pkt, ipHdrLen)
if !ok {
@@ -126,23 +116,11 @@ func parseUDPTail(ip parsedIP) (parsedUDP, bool) {
return p, true
}
// Commit borrows pkt. The caller must keep pkt valid until the next Flush.
func (c *UDPCoalescer) Commit(pkt []byte) error {
info, ok := parseUDP(pkt)
if !ok {
c.addVerbatim(pkt)
return nil
}
return c.commitParsed(pkt, info)
}
// commitParsed is the post-parse half of Commit. The caller must have
// already verified parseUDP succeeded. Used by MultiCoalescer.Commit to
// avoid re-walking the IP/UDP header.
// commitParsed commits one parsed UDP packet. The caller (dispatch, via parseUDPAt) 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 (UDP `length` == 8) is legal and must still
// reach the TUN, but it can't be coalesced. The len guard skips hashing
// the key when no flow is open.
// 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.
if info.payLen == 0 {
if len(c.openSlots) != 0 {
if last := c.lastSlot; last != nil && last.fk == info.fk {
@@ -206,10 +184,8 @@ func (c *UDPCoalescer) Flush() error {
return first
}
// sealAllOpen closes every open coalesce chain: nothing committed after this
// call can extend a slot created before it. Called when an unparseable packet
// arrives — its flow is unknown, so any open chain might be the one whose
// later data would otherwise leapfrog it.
// sealAllOpen closes every open coalesce chain. Called for unparseable packets: the flow key is
// unknown, so any open chain could otherwise absorb later data and emit it ahead of this packet.
func (c *UDPCoalescer) sealAllOpen() {
clear(c.openSlots)
c.lastSlot = nil
@@ -229,8 +205,8 @@ func (c *UDPCoalescer) seed(pkt []byte, info parsedUDP) {
}
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 and is the header source for canAppend until
// the first append copies it into hdrBuf.
s.rawPkt = pkt
s.hdrLen = info.hdrLen
s.ipHdrLen = info.ipHdrLen
@@ -261,10 +237,9 @@ func (c *UDPCoalescer) canAppend(s *udpSlot, pkt []byte, info parsedUDP) bool {
if s.hdrLen+s.totalPay+info.payLen > udpCoalesceBufSize {
return false
}
// Header reads go through rawPkt: hdrBuf is populated lazily on the
// first append, and 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 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.
if !s.isV6 && !ipv4CanCoalesceID(s.rawPkt, pkt, s.numSeg) {
return false
}
@@ -274,14 +249,13 @@ func (c *UDPCoalescer) canAppend(s *udpSlot, pkt []byte, info parsedUDP) bool {
return true
}
// 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 the final one. The caller
// must deregister a closed slot from openSlots.
// 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
// 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 packet. Deferred out
// of seed so solo slots, which flush from rawPkt, never pay the copy.
// 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])