unslop some comments

This commit is contained in:
JackDoan
2026-07-14 13:38:39 -05:00
parent 5631346b07
commit 37b924945d
7 changed files with 24 additions and 63 deletions
+6 -6
View File
@@ -7,9 +7,9 @@ type RxBatcher interface {
Reserve(sz int) []byte
// Commit borrows pkt. The caller must keep pkt valid until the next Flush
Commit(pkt []byte) error
// Flush emits every queued packet in arrival order. Returns the
// first error observed; keeps draining so one bad packet doesn't hold up
// the rest. After Flush returns, borrowed payload slices may be recycled.
// Flush emits every queued packet in arrival order.
// Returns the first error observed; keeps draining so one bad packet doesn't hold up the rest.
// After Flush returns, borrowed payload slices may be recycled.
Flush() error
}
@@ -21,8 +21,8 @@ type TxBatcher interface {
// caller must keep pkt valid until the next Flush. Pass 0 (Not-ECT)
// to leave the outer ECN field unset.
Commit(pkt []byte, dst netip.AddrPort, outerECN byte)
// Flush emits every queued packet via the underlying batch writer in
// arrival order. Returns an errors.Join of one or more errors. After Flush returns,
// borrowed payload slices may be recycled.
// Flush emits every queued packet via the underlying batch writer in arrival order.
// Returns an errors.Join of one or more errors.
// After Flush returns, borrowed payload slices may be recycled.
Flush() error
}
+1 -8
View File
@@ -140,14 +140,7 @@ func ipHeadersMatch(a, b []byte, isV6 bool) bool {
}
// Arena is an injectable byte-slab that hands out non-overlapping borrowed
// slices via Reserve and releases them in bulk via Reset. Coalescers take
// an *Arena at construction so the caller controls the slab lifetime and
// can share one slab across multiple coalescers (MultiCoalescer hands the
// same *Arena to every lane so the lanes don't carry their own backings).
//
// Reserve borrows; the slice is valid until the next Reset. The slab grows
// (by allocating a fresh, larger backing array) if a Reserve doesn't fit;
// pre-size the arena via NewArena to avoid that path on the hot path.
// slices via Reserve and releases them in bulk via Reset.
type Arena struct {
buf []byte
}
+2 -3
View File
@@ -110,9 +110,8 @@ func (m *MultiCoalescer) Commit(pkt []byte) error {
return m.pt.Commit(pkt)
}
// Flush drains every lane in a fixed order — TCP, UDP, passthrough — then
// resets the shared arena once. A lane error doesn't stop the remaining
// lanes; the joined errors are returned.
// Flush drains every lane in a fixed order, then resets the shared arena once.
// A lane error doesn't stop the remaining lanes; the joined errors are returned.
func (m *MultiCoalescer) Flush() error {
var errs []error
if m.tcp != nil {
+9 -27
View File
@@ -12,10 +12,7 @@ import (
"github.com/slackhq/nebula/overlay/tio"
)
// ipProtoTCP is the IANA protocol number for TCP. Hardcoded instead of
// reaching for golang.org/x/sys/unix — that package doesn't define the
// constant on Windows, which would break cross-compiles even though this
// file runs unchanged on every platform.
// ipProtoTCP is the IANA protocol number for TCP. Defined here to help Windows out.
const ipProtoTCP = 6
// tcpCoalesceBufSize caps total bytes per superpacket. Mirrors the kernel's
@@ -23,8 +20,7 @@ const ipProtoTCP = 6
const tcpCoalesceBufSize = 65535
// tcpCoalesceMaxSegs caps how many segments we'll coalesce into a single
// superpacket. Keeping this well below the kernel's TSO ceiling bounds
// latency.
// 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
@@ -118,8 +114,8 @@ type parsedTCP struct {
// 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, no fragmentation)
// and IPv6 (no extension headers).
// for non-TCP or malformed input.
// Accepts IPv4 (no options or fragmentation) and IPv6 (no extension headers).
func parseTCPBase(pkt []byte) (parsedTCP, bool) {
var p parsedTCP
ip, ok := parseIPPrologue(pkt, ipProtoTCP)
@@ -178,9 +174,7 @@ func (c *TCPCoalescer) Reserve(sz int) []byte {
return c.reserver(sz)
}
// Commit borrows pkt. The caller must keep pkt valid until the next Flush,
// whether or not the packet was coalesced — passthrough (non-admissible)
// packets are queued and written at Flush time, not synchronously.
// Commit borrows pkt. The caller must keep pkt valid until the next Flush.
func (c *TCPCoalescer) Commit(pkt []byte) error {
if c.gsoW == nil {
c.addPassthrough(pkt)
@@ -246,15 +240,7 @@ func (c *TCPCoalescer) commitParsed(pkt []byte, info parsedTCP) error {
return nil
}
// Flush emits every queued event in (per-flow) seq order. Coalesced slots
// go out via WriteGSO; passthrough slots go out via plainW.Write.
// reorderForFlush first sorts each flow's slots into TCP-seq order within
// passthrough-bounded segments and merges contiguous adjacent slots, so
// any wire-side reorder that crossed an rxOrder batch boundary doesn't
// get amplified into kernel-visible reorder by the slot machinery.
// Returns the first error observed; keeps draining so one bad packet
// doesn't hold up the rest. After Flush returns, borrowed payload slices
// may be recycled.
// Flush emits every queued event in (per-flow) seq order.
func (c *TCPCoalescer) Flush() error {
first := c.drain()
if c.resetter != nil {
@@ -328,8 +314,7 @@ func (c *TCPCoalescer) seed(pkt []byte, info parsedTCP) {
}
// canAppend reports whether info's packet extends the slot's seed: same
// header shape and stable contents, adjacent seq, not oversized, chain not
// closed.
// header shape and stable contents, adjacent seq, not oversized, chain not closed.
func (c *TCPCoalescer) canAppend(s *coalesceSlot, pkt []byte, info parsedTCP) bool {
if s.psh {
return false
@@ -397,8 +382,7 @@ func (c *TCPCoalescer) release(s *coalesceSlot) {
c.pool = append(c.pool, s)
}
// flushSlot patches the header and calls WriteGSO. Does not remove the
// slot from c.slots.
// flushSlot patches the header and calls WriteGSO. Does not remove the slot from c.slots.
func (c *TCPCoalescer) flushSlot(s *coalesceSlot) error {
total := s.hdrLen + s.totalPay
l4Len := total - s.ipHdrLen
@@ -427,9 +411,7 @@ func (c *TCPCoalescer) flushSlot(s *coalesceSlot) error {
// headersMatch compares two IP+TCP header prefixes for byte-for-byte
// equality on every field that must be identical across coalesced
// segments. Size/IPID/IPCsum/seq/flags/tcpCsum are masked out. The IP-level
// ECN codepoint is compared (via ipHeadersMatch) so segments with differing
// ECN don't coalesce, matching kernel GRO.
// segments. Size/IPID/IPCsum/seq/flags/tcpCsum are masked out.
func headersMatch(a, b []byte, isV6 bool, ipHdrLen int) bool {
if len(a) != len(b) {
return false
+1 -4
View File
@@ -22,10 +22,7 @@ const udpCoalesceMaxSegs = 64
// into. IPv6 (40) + UDP (8) = 48; round up for safety.
const udpCoalesceHdrCap = 64
// udpSlot is one entry in the UDPCoalescer's ordered event queue. Same
// passthrough-vs-coalesced shape as the TCP coalescer's slot, but no
// seq/PSH/CWR bookkeeping — UDP segments only need 5-tuple + length
// matching to coalesce.
// udpSlot is one entry in the UDPCoalescer's ordered event queue.
type udpSlot struct {
passthrough bool
rawPkt []byte // borrowed when passthrough
+3 -8
View File
@@ -12,15 +12,10 @@ import (
// blockOn parks the calling goroutine until fd is ready (events is POLLIN for
// reads, POLLOUT for writes) or shutdownFd signals teardown. It builds the
// pollfd array on the stack every call, so concurrent callers on the same
// Queue never share Revents storage: the previous shared-array implementation
// was a genuine Go data race when two writers parked in poll(2) at once (the
// kernel writing Revents while another goroutine zeroed it). Level-triggered
// events kept it from deadlocking, but it was still a race.
// Queue never share Revents storage.
//
// Poll(2) is looped over EINTR. err is checked before the Revents bits are
// trusted, since a failed poll may leave them bogus. Returns os.ErrClosed when
// shutdown was signaled (POLLIN on shutdownFd) or either fd reported a problem
// condition (POLLHUP|POLLNVAL|POLLERR).
// Returns os.ErrClosed when shutdown was signaled (POLLIN on shutdownFd)
// or either fd reported a problem condition (POLLHUP|POLLNVAL|POLLERR).
func blockOn(fd, shutdownFd int32, events int16) error {
const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR
pfds := [2]unix.PollFd{