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

overlay/batch: ship never-grown slots as plain writes

Browse Source 
A slot that stays single-segment is byte-identical to the packet it
was seeded from, but flushSlot re-emitted it via WriteGSO with a
seeded pseudo-sum, forcing the kernel to software-checksum up to
~1400B that arrived with a perfectly valid checksum. Keep the borrowed
seed packet on the slot (valid until Flush per the Commit contract)
and emit it through the plain DATA_VALID path when numSeg is still 1
at flush time. appendPayload and mergeSlots only touch hdrBuf once
numSeg >= 2, so the raw bytes are pristine whenever the fast path
fires. This is every non-coalesced TCP/UDP packet: request/response
flows, many-flow fan-in, and each run's leftover tail.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014ugV2edVqoz3tBvq9J6yWp
This commit is contained in:
JackDoan
2026-07-29 17:31:29 -05:00
parent 69cf816f80
commit 2190900107
4 changed files with 214 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files
overlay/batch/tcp_coalesce.go
+12 -3
View File
@@ -37,7 +37,11 @@ const tcpCoalesceHdrCap = 100
// The caller (listenOut) must keep those buffers alive until Flush. // The caller (listenOut) must keep those buffers alive until Flush.
type coalesceSlot struct { type coalesceSlot struct {
passthrough bool passthrough bool
rawPkt []byte // borrowed when passthrough // rawPkt is borrowed: the whole packet for passthrough 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 []byte
fk flowKey fk flowKey
hdrBuf [tcpCoalesceHdrCap]byte hdrBuf [tcpCoalesceHdrCap]byte
@@ -235,7 +239,12 @@ func (c *TCPCoalescer) Flush() error {
var first error var first error
for _, s := range c.slots { for _, s := range c.slots {
var err error var err error
if s.passthrough { if s.passthrough || 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 and mergeSlots only touch hdrBuf
// once numSeg >= 2, so rawPkt is still pristine here.
_, err = c.w.Write(s.rawPkt) _, err = c.w.Write(s.rawPkt)
} else { } else {
err = c.flushSlot(s) err = c.flushSlot(s)
@@ -268,7 +277,7 @@ func (c *TCPCoalescer) seed(pkt []byte, info parsedTCP) {
} }
s := c.take() s := c.take()
s.passthrough = false s.passthrough = false
s.rawPkt = nil s.rawPkt = pkt // kept for the numSeg==1 fast path in Flush
copy(s.hdrBuf[:], pkt[:info.hdrLen]) copy(s.hdrBuf[:], pkt[:info.hdrLen])
s.hdrLen = info.hdrLen s.hdrLen = info.hdrLen
s.ipHdrLen = info.ipHdrLen s.ipHdrLen = info.ipHdrLen
overlay/batch/tcp_coalesce_test.go
+142 -128
View File
@@ -217,17 +217,15 @@ func TestCoalescerSeedThenFlushAlone(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
// Single-segment flush goes through WriteGSO with GSO_NONE // A slot that never grew past one segment flushes as a plain Write of
// (virtio NEEDS_CSUM lets the kernel fill in the L4 csum). // the original packet bytes: the original (already valid) checksum
if len(w.gsoWrites) != 1 || len(w.writes) != 0 { // ships via the DATA_VALID path, so the kernel does no csum work.
// WriteGSO is reserved for slots that actually coalesced (>=2 segs).
if len(w.writes) != 1 || len(w.gsoWrites) != 0 {
t.Fatalf("single-seg flush: writes=%d gso=%d", len(w.writes), len(w.gsoWrites)) t.Fatalf("single-seg flush: writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
} }
g := w.gsoWrites[0] if !bytes.Equal(w.writes[0], pkt) {
if g.total() != 40+1000 { t.Errorf("plain write not byte-identical to committed packet: got %d bytes want %d", len(w.writes[0]), len(pkt))
t.Errorf("super total=%d want %d", g.total(), 40+1000)
}
if g.payLen() != 1000 {
t.Errorf("payLen=%d want 1000", g.payLen())
} }
} }
@@ -284,9 +282,10 @@ func TestCoalescerRejectsSeqGap(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
// Each packet flushes as its own single-segment WriteGSO. // Each packet stays a single-segment slot and flushes as its own plain
if len(w.gsoWrites) != 2 || len(w.writes) != 0 { // write of the original bytes.
t.Fatalf("seq gap: want 2 gso writes got writes=%d gso=%d", len(w.writes), len(w.gsoWrites)) if len(w.writes) != 2 || len(w.gsoWrites) != 0 {
t.Fatalf("seq gap: want 2 plain writes got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
} }
} }
@@ -297,8 +296,9 @@ func TestCoalescerRejectsFlagMismatch(t *testing.T) {
if err := c.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { if err := c.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil {
t.Fatal(err) t.Fatal(err)
} }
// SYN|ACK is non-admissible. Must flush matching flow's slot (gso) // SYN|ACK is non-admissible. Must flush the matching flow's slot —
// and then plain-write the SYN packet itself. // single-segment, so a plain write of the original bytes — and then
// plain-write the SYN packet itself.
syn := buildTCPv4(2200, tcpSyn|tcpAck, pay) syn := buildTCPv4(2200, tcpSyn|tcpAck, pay)
if err := c.Commit(syn); err != nil { if err := c.Commit(syn); err != nil {
t.Fatal(err) t.Fatal(err)
@@ -306,8 +306,11 @@ func TestCoalescerRejectsFlagMismatch(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if len(w.writes) != 1 || len(w.gsoWrites) != 1 { if len(w.writes) != 2 || len(w.gsoWrites) != 0 {
t.Fatalf("flag mismatch: want 1 plain + 1 gso, got writes=%d gso=%d", len(w.writes), len(w.gsoWrites)) t.Fatalf("flag mismatch: want 2 plain writes (flushed seed + SYN), got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
}
if !bytes.Equal(w.writes[1], syn) {
t.Errorf("second plain write should be the SYN packet, got %d bytes", len(w.writes[1]))
} }
} }
@@ -346,13 +349,14 @@ func TestCoalescerShortLastSegmentClosesChain(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
// Expect two gso writes: the first two packets coalesced, then the // Expect one gso write for the first two packets coalesced, then the
// third flushed alone (single-seg via GSO_NONE). // third — still single-segment — flushed as a plain write of the
if len(w.gsoWrites) != 2 { // original packet.
t.Fatalf("want 2 gso writes got %d", len(w.gsoWrites)) if len(w.gsoWrites) != 1 {
t.Fatalf("want 1 gso write got %d", len(w.gsoWrites))
} }
if len(w.writes) != 0 { if len(w.writes) != 1 {
t.Fatalf("want 0 plain writes got %d", len(w.writes)) t.Fatalf("want 1 plain write got %d", len(w.writes))
} }
if w.gsoWrites[0].gsoSize != 1200 { if w.gsoWrites[0].gsoSize != 1200 {
t.Errorf("gsoSize=%d want 1200", w.gsoWrites[0].gsoSize) t.Errorf("gsoSize=%d want 1200", w.gsoWrites[0].gsoSize)
@@ -360,6 +364,9 @@ func TestCoalescerShortLastSegmentClosesChain(t *testing.T) {
if got, want := w.gsoWrites[0].total(), 40+1200+500; got != want { if got, want := w.gsoWrites[0].total(), 40+1200+500; got != want {
t.Errorf("super len=%d want %d", got, want) t.Errorf("super len=%d want %d", got, want)
} }
if got, want := len(w.writes[0]), 40+1200; got != want {
t.Errorf("plain write len=%d want %d", got, want)
}
} }
func TestCoalescerPSHFinalizesChain(t *testing.T) { func TestCoalescerPSHFinalizesChain(t *testing.T) {
@@ -378,12 +385,13 @@ func TestCoalescerPSHFinalizesChain(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
// First two coalesce; the third seeds a fresh slot that flushes alone. // First two coalesce into one gso write; the third seeds a fresh slot
if len(w.gsoWrites) != 2 { // that stays single-segment and flushes as a plain write.
t.Fatalf("want 2 gso writes got %d", len(w.gsoWrites)) if len(w.gsoWrites) != 1 {
t.Fatalf("want 1 gso write got %d", len(w.gsoWrites))
} }
if len(w.writes) != 0 { if len(w.writes) != 1 {
t.Fatalf("want 0 plain writes got %d", len(w.writes)) t.Fatalf("want 1 plain write got %d", len(w.writes))
} }
} }
@@ -436,9 +444,10 @@ func TestCoalescerRejectsDifferentFlow(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
// Two independent flows, each flushes its own single-segment WriteGSO. // Two independent flows, each stays single-segment and flushes as its
if len(w.gsoWrites) != 2 || len(w.writes) != 0 { // own plain write of the original bytes.
t.Fatalf("diff flow: want 2 gso writes got writes=%d gso=%d", len(w.writes), len(w.gsoWrites)) if len(w.writes) != 2 || len(w.gsoWrites) != 0 {
t.Fatalf("diff flow: want 2 plain writes got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
} }
} }
@@ -544,12 +553,15 @@ func TestCoalescerMultipleFlowsInSameBatch(t *testing.T) {
// coalesced events both queued, Flush emits them in Add order rather than // coalesced events both queued, Flush emits them in Add order rather than
// writing passthrough packets synchronously. // writing passthrough packets synchronously.
func TestCoalescerPreservesArrivalOrder(t *testing.T) { func TestCoalescerPreservesArrivalOrder(t *testing.T) {
w := &orderedFakeWriter{gsoEnabled: true} w := &fakeTunWriter{gsoEnabled: true}
c := newTestTCPCoalescer(t, w) c := newTestTCPCoalescer(t, w)
// Sequence: coalesceable TCP, ICMP (passthrough), coalesceable TCP on // Sequence: coalesceable TCP, ICMP (passthrough), coalesceable TCP on
// a different flow. Expected emit order: gso(X), plain(ICMP), gso(Y). // a different flow. Both TCP slots stay single-segment, so all three
// emit as plain writes; the packet order (X, ICMP, Y) is asserted by
// byte content since the kinds no longer distinguish them.
pay := make([]byte, 1200) pay := make([]byte, 1200)
if err := c.Commit(buildTCPv4Ports(1000, 2000, 100, tcpAck, pay)); err != nil { tcpX := buildTCPv4Ports(1000, 2000, 100, tcpAck, pay)
if err := c.Commit(tcpX); err != nil {
t.Fatal(err) t.Fatal(err)
} }
icmp := make([]byte, 28) icmp := make([]byte, 28)
@@ -561,40 +573,25 @@ func TestCoalescerPreservesArrivalOrder(t *testing.T) {
if err := c.Commit(icmp); err != nil { if err := c.Commit(icmp); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if err := c.Commit(buildTCPv4Ports(3000, 2000, 500, tcpAck, pay)); err != nil { tcpY := buildTCPv4Ports(3000, 2000, 500, tcpAck, pay)
if err := c.Commit(tcpY); err != nil {
t.Fatal(err) t.Fatal(err)
} }
// Nothing should have hit the writer synchronously. // Nothing should have hit the writer synchronously.
if len(w.events) != 0 { if len(w.order) != 0 {
t.Fatalf("Add emitted events synchronously: %v", w.events) t.Fatalf("Add emitted events synchronously: %v", w.order)
} }
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if got, want := w.events, []string{"gso", "plain", "gso"}; !stringSliceEq(got, want) { if got, want := w.order, []string{"write", "write", "write"}; !stringSliceEq(got, want) {
t.Fatalf("flush order=%v want %v", got, want) t.Fatalf("flush order=%v want %v", got, want)
} }
} for i, want := range [][]byte{tcpX, icmp, tcpY} {
if !bytes.Equal(w.writes[i], want) {
// orderedFakeWriter records only the sequence of call types so tests can t.Fatalf("write %d out of arrival order: got %d bytes, want %d bytes", i, len(w.writes[i]), len(want))
// assert arrival order without inspecting bytes. }
type orderedFakeWriter struct { }
gsoEnabled bool
events []string
}
func (w *orderedFakeWriter) Write(p []byte) (int, error) {
w.events = append(w.events, "plain")
return len(p), nil
}
func (w *orderedFakeWriter) WriteGSO(hdr []byte, transportHdr []byte, pays [][]byte, _ tio.GSOProto) error {
w.events = append(w.events, "gso")
return nil
}
func (w *orderedFakeWriter) Capabilities() tio.Capabilities {
return tio.Capabilities{TSO: w.gsoEnabled, USO: w.gsoEnabled}
} }
func stringSliceEq(a, b []string) bool { func stringSliceEq(a, b []string) bool {
@@ -761,17 +758,14 @@ func TestCoalescerCwrSealsFlow(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if len(w.writes) != 1 { // All three emissions are plain writes: the seed before CWR and the
t.Fatalf("want 1 plain write (CWR), got %d", len(w.writes)) // fresh seed after both stay single-segment, and the CWR packet itself
// is passthrough. Order: seed, CWR, reseed.
if len(w.writes) != 3 || len(w.gsoWrites) != 0 {
t.Fatalf("want 3 plain writes (seed, CWR, reseed), got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
} }
// Two GSO writes: the first seed before CWR, and a fresh seed after. if flags := w.writes[1][20+13]; flags&tcpCwr == 0 {
if len(w.gsoWrites) != 2 { t.Errorf("middle write flags=0x%02x want CWR (passthrough in arrival order)", flags)
t.Fatalf("want 2 gso writes, got %d", len(w.gsoWrites))
}
for i, g := range w.gsoWrites {
if len(g.pays) != 1 {
t.Errorf("gso %d pay count=%d want 1", i, len(g.pays))
}
} }
} }
@@ -791,22 +785,25 @@ func TestCoalescerEceMismatchReseeds(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if len(w.gsoWrites) != 2 { // Each seed stays single-segment and flushes as its own plain write.
t.Fatalf("want 2 separate seeds, got %d gso writes", len(w.gsoWrites)) if len(w.writes) != 2 || len(w.gsoWrites) != 0 {
t.Fatalf("want 2 separate plain writes, got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
} }
for i, g := range w.gsoWrites { if flags := w.writes[0][20+13]; flags&tcpEce == 0 {
if len(g.pays) != 1 { t.Errorf("first write lost ECE: flags=0x%02x", flags)
t.Errorf("gso %d pay count=%d want 1", i, len(g.pays)) }
} if flags := w.writes[1][20+13]; flags&tcpEce != 0 {
t.Errorf("second write gained ECE: flags=0x%02x", flags)
} }
} }
// TestCoalescerDifferingECNReseeds confirms that segments with differing IP // TestCoalescerDifferingECNReseeds confirms that segments with differing IP
// ECN codepoints do NOT coalesce: headersMatch compares the full ToS byte, // ECN codepoints do NOT coalesce: headersMatch compares the full ToS byte,
// matching kernel GRO. Two ECT(0) segments merge; a CE stamp mid-run seals // matching kernel GRO. Two ECT(0) segments merge into a superpacket; a CE
// the ECT(0) chain and starts a fresh superpacket that keeps CE; a trailing // stamp mid-run seals the ECT(0) chain and reseeds, and the trailing ECT(0)
// ECT(0) starts yet another. Each superpacket keeps its own codepoint — // reseeds again — those reseeds stay single-segment and ship as plain
// ORing the marks (the old buggy behavior) would have fabricated a false CE // writes of the original packets, each keeping its own codepoint. ORing
// the marks (the old buggy behavior) would have fabricated a false CE
// across the whole burst. // across the whole burst.
func TestCoalescerDifferingECNReseeds(t *testing.T) { func TestCoalescerDifferingECNReseeds(t *testing.T) {
w := &fakeTunWriter{gsoEnabled: true} w := &fakeTunWriter{gsoEnabled: true}
@@ -828,31 +825,34 @@ func TestCoalescerDifferingECNReseeds(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if len(w.gsoWrites) != 3 { // gso: the two ECT(0) segments merged; then plain CE alone; then plain
t.Fatalf("want 3 superpackets (ECN split), got %d (plain=%d)", len(w.gsoWrites), len(w.writes))
}
// gso[0]: the two ECT(0) segments merged; gso[1]: CE alone; gso[2]:
// trailing ECT(0) alone. Emitted in seq order. // trailing ECT(0) alone. Emitted in seq order.
type want struct { if len(w.gsoWrites) != 1 || len(w.writes) != 2 {
pays int t.Fatalf("want 1 gso (ECT0 pair) + 2 plain (ECN split), got gso=%d plain=%d", len(w.gsoWrites), len(w.writes))
ecn byte
} }
wants := []want{{2, ecnECT0}, {1, ecnCE}, {1, ecnECT0}} if got, want := w.order, []string{"gso", "write", "write"}; !stringSliceEq(got, want) {
for i, wnt := range wants { t.Fatalf("emission order=%v want %v", got, want)
g := w.gsoWrites[i] }
if len(g.pays) != wnt.pays { g := w.gsoWrites[0]
t.Errorf("gso %d pay count=%d want %d", i, len(g.pays), wnt.pays) if len(g.pays) != 2 {
} t.Errorf("gso pay count=%d want 2", len(g.pays))
if got := g.hdr[1] & 0x03; got != wnt.ecn { }
t.Errorf("gso %d ECN=0x%02x want 0x%02x", i, got, wnt.ecn) if got := g.hdr[1] & 0x03; got != ecnECT0 {
t.Errorf("gso ECN=0x%02x want 0x%02x", got, ecnECT0)
}
wantECN := []byte{ecnCE, ecnECT0}
for i, wnt := range wantECN {
if got := w.writes[i][1] & 0x03; got != wnt {
t.Errorf("plain %d ECN=0x%02x want 0x%02x", i, got, wnt)
} }
} }
} }
// TestCoalescerECT0ThenECT1NoCE is the core regression for the ECN merge // TestCoalescerECT0ThenECT1NoCE is the core regression for the ECN merge
// bug: ORing ECT(0)=0b10 with ECT(1)=0b01 fabricates CE=0b11. The two // bug: ORing ECT(0)=0b10 with ECT(1)=0b01 fabricates CE=0b11. The two
// segments must land in separate superpackets, each preserving its own // segments must land in separate emissions — both stay single-segment, so
// codepoint, and neither may end up CE-marked. // each ships as a plain write of its original bytes, preserving its own
// codepoint — and neither may end up CE-marked.
func TestCoalescerECT0ThenECT1NoCE(t *testing.T) { func TestCoalescerECT0ThenECT1NoCE(t *testing.T) {
w := &fakeTunWriter{gsoEnabled: true} w := &fakeTunWriter{gsoEnabled: true}
c := newTestTCPCoalescer(t, w) c := newTestTCPCoalescer(t, w)
@@ -866,16 +866,16 @@ func TestCoalescerECT0ThenECT1NoCE(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if len(w.gsoWrites) != 2 { if len(w.writes) != 2 || len(w.gsoWrites) != 0 {
t.Fatalf("want 2 separate superpackets (ECT0 vs ECT1), got %d", len(w.gsoWrites)) t.Fatalf("want 2 separate plain writes (ECT0 vs ECT1), got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
} }
wantECN := []byte{ecnECT0, ecnECT1} wantECN := []byte{ecnECT0, ecnECT1}
for i, g := range w.gsoWrites { for i, p := range w.writes {
if got := g.hdr[1] & 0x03; got != wantECN[i] { if got := p[1] & 0x03; got != wantECN[i] {
t.Errorf("gso %d ECN=0x%02x want 0x%02x", i, got, wantECN[i]) t.Errorf("write %d ECN=0x%02x want 0x%02x", i, got, wantECN[i])
} }
if got := g.hdr[1] & 0x03; got == ecnCE { if got := p[1] & 0x03; got == ecnCE {
t.Errorf("gso %d fabricated CE from ECT merge", i) t.Errorf("write %d fabricated CE from ECT merge", i)
} }
} }
} }
@@ -899,8 +899,9 @@ func TestCoalescerDscpMismatchReseeds(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if len(w.gsoWrites) != 2 { // Both seeds stay single-segment → two plain writes, no gso.
t.Fatalf("want 2 separate seeds (different DSCP), got %d", len(w.gsoWrites)) if len(w.writes) != 2 || len(w.gsoWrites) != 0 {
t.Fatalf("want 2 separate plain writes (different DSCP), got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
} }
} }
@@ -1043,8 +1044,10 @@ func TestCoalescerSortKeepsPSHBoundary(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if len(w.gsoWrites) != 2 { // The PSH-sealed pair is a real superpacket; the fresh seed stays
t.Fatalf("want 2 gso writes (PSH-sealed and fresh seed), got %d", len(w.gsoWrites)) // single-segment and flushes as a plain write.
if len(w.gsoWrites) != 1 || len(w.writes) != 1 {
t.Fatalf("want 1 gso (PSH-sealed pair) + 1 plain (fresh seed), got gso=%d plain=%d", len(w.gsoWrites), len(w.writes))
} }
} }
@@ -1075,11 +1078,19 @@ func TestCoalescerSortKeepsPassthroughBarrier(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
// We expect: gso(merged 1000+3400 ranges sorted but not contiguous so 2 // All four packets emit as plain writes: 1000 and 3400 are separate
// gso writes), plain(SYN), gso(2200 alone). The pre-barrier sort should // single-segment slots (not contiguous, so the post-sort merge can't
// land 1000 before 3400, and the post-barrier 2200 stays after the SYN. // fold them), the SYN is passthrough, and the post-barrier 2200 stays
if len(w.writes) != 1 { // a single-segment slot after the SYN. The pre-barrier sort must land
t.Fatalf("want 1 plain SYN passthrough, got %d", len(w.writes)) // 1000 before 3400, and 2200 must never move before the SYN.
if len(w.writes) != 4 || len(w.gsoWrites) != 0 {
t.Fatalf("want 4 plain writes, got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
}
wantSeqs := []uint32{1000, 3400, 9999, 2200}
for i, want := range wantSeqs {
if seq := binary.BigEndian.Uint32(w.writes[i][24:28]); seq != want {
t.Errorf("write %d seq=%d want %d", i, seq, want)
}
} }
} }
@@ -1106,23 +1117,25 @@ func TestCoalescerIPv6DifferingECNReseeds(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if len(w.gsoWrites) != 3 { // Like the v4 test: the ECT(0) pair merges into one superpacket; the CE
t.Fatalf("want 3 superpackets (ECN split), got %d", len(w.gsoWrites)) // and trailing ECT(0) reseeds stay single-segment and ship as plain
// writes, in seq order.
if len(w.gsoWrites) != 1 || len(w.writes) != 2 {
t.Fatalf("want 1 gso (ECT0 pair) + 2 plain (ECN split), got gso=%d plain=%d", len(w.gsoWrites), len(w.writes))
} }
// Byte 1 high nibble holds TC[3:0]; ECN is the low 2 bits of that nibble, // Byte 1 high nibble holds TC[3:0]; ECN is the low 2 bits of that nibble,
// which appears in byte 1 mask 0x30 (>>4 to read the codepoint value). // which appears in byte 1 mask 0x30 (>>4 to read the codepoint value).
type want struct { g := w.gsoWrites[0]
pays int if len(g.pays) != 2 {
ecn byte t.Errorf("gso pay count=%d want 2", len(g.pays))
} }
wants := []want{{2, ecnECT0}, {1, ecnCE}, {1, ecnECT0}} if got := (g.hdr[1] >> 4) & 0x03; got != ecnECT0 {
for i, wnt := range wants { t.Errorf("gso v6 ECN=0x%02x want 0x%02x", got, ecnECT0)
g := w.gsoWrites[i] }
if len(g.pays) != wnt.pays { wantECN := []byte{ecnCE, ecnECT0}
t.Errorf("gso %d pay count=%d want %d", i, len(g.pays), wnt.pays) for i, wnt := range wantECN {
} if got := (w.writes[i][1] >> 4) & 0x03; got != wnt {
if got := (g.hdr[1] >> 4) & 0x03; got != wnt.ecn { t.Errorf("plain %d v6 ECN=0x%02x want 0x%02x", i, got, wnt)
t.Errorf("gso %d v6 ECN=0x%02x want 0x%02x", i, got, wnt.ecn)
} }
} }
} }
@@ -1303,7 +1316,8 @@ func TestCoalescerNonAtomicSequentialIDsCoalesce(t *testing.T) {
// TestCoalescerNonAtomicIDGapDoesNotCoalesce: with DF clear and an ID jump // TestCoalescerNonAtomicIDGapDoesNotCoalesce: with DF clear and an ID jump
// mid-flow, neither the append path nor the flush-time merge may combine // mid-flow, neither the append path nor the flush-time merge may combine
// the segments — TSO would re-stamp seed+n and rewrite the second // the segments — TSO would re-stamp seed+n and rewrite the second
// packet's ID, which is meaningful on non-atomic datagrams. // packet's ID, which is meaningful on non-atomic datagrams. Each stays a
// single-segment slot and flushes as a plain write with its original ID.
func TestCoalescerNonAtomicIDGapDoesNotCoalesce(t *testing.T) { func TestCoalescerNonAtomicIDGapDoesNotCoalesce(t *testing.T) {
w := &fakeTunWriter{gsoEnabled: true} w := &fakeTunWriter{gsoEnabled: true}
c := newTestTCPCoalescer(t, w) c := newTestTCPCoalescer(t, w)
@@ -1323,11 +1337,11 @@ func TestCoalescerNonAtomicIDGapDoesNotCoalesce(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if len(w.gsoWrites) != 2 { if len(w.writes) != 2 || len(w.gsoWrites) != 0 {
t.Fatalf("ID gap on DF=0 must not coalesce (append or merge): gso=%d", len(w.gsoWrites)) t.Fatalf("ID gap on DF=0 must not coalesce (append or merge): writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
} }
for i, want := range []uint16{700, 900} { for i, want := range []uint16{700, 900} {
if id := binary.BigEndian.Uint16(w.gsoWrites[i].hdr[4:6]); id != want { if id := binary.BigEndian.Uint16(w.writes[i][4:6]); id != want {
t.Errorf("write %d: ID=%d want %d (must be preserved)", i, id, want) t.Errorf("write %d: ID=%d want %d (must be preserved)", i, id, want)
} }
} }
overlay/batch/udp_coalesce.go
+10 -3
View File
@@ -25,7 +25,11 @@ const udpCoalesceHdrCap = 64
// udpSlot is one entry in the UDPCoalescer's ordered event queue. // udpSlot is one entry in the UDPCoalescer's ordered event queue.
type udpSlot struct { type udpSlot struct {
passthrough bool passthrough bool
rawPkt []byte // borrowed when passthrough // rawPkt is borrowed: the whole packet for passthrough 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 []byte
fk flowKey fk flowKey
hdrBuf [udpCoalesceHdrCap]byte hdrBuf [udpCoalesceHdrCap]byte
@@ -170,7 +174,10 @@ func (c *UDPCoalescer) Flush() error {
var first error var first error
for _, s := range c.slots { for _, s := range c.slots {
var err error var err error
if s.passthrough { if s.passthrough || s.numSeg == 1 {
// A slot that never grew 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.
_, err = c.w.Write(s.rawPkt) _, err = c.w.Write(s.rawPkt)
} else { } else {
err = c.flushSlot(s) err = c.flushSlot(s)
@@ -201,7 +208,7 @@ func (c *UDPCoalescer) seed(pkt []byte, info parsedUDP) {
} }
s := c.take() s := c.take()
s.passthrough = false s.passthrough = false
s.rawPkt = nil s.rawPkt = pkt // kept for the numSeg==1 fast path in Flush
copy(s.hdrBuf[:], pkt[:info.hdrLen]) copy(s.hdrBuf[:], pkt[:info.hdrLen])
s.hdrLen = info.hdrLen s.hdrLen = info.hdrLen
s.ipHdrLen = info.ipHdrLen s.ipHdrLen = info.ipHdrLen
overlay/batch/udp_coalesce_test.go
+50 -29
View File
@@ -1,6 +1,7 @@
package batch package batch
import ( import (
"bytes"
"encoding/binary" "encoding/binary"
"io" "io"
"testing" "testing"
@@ -112,11 +113,16 @@ func TestUDPCoalescerSeedThenFlushAlone(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
// Single-segment flush goes through WriteGSO; the writer infers GSO_NONE // A slot that never grew past one datagram flushes as a plain Write of
// from len(pays)==1 and the kernel fills in the UDP csum (NEEDS_CSUM). // the original packet bytes: the original (already valid) checksum
if len(w.gsoWrites) != 1 || len(w.writes) != 0 { // ships via the DATA_VALID path, so the kernel does no csum work.
// WriteGSO is reserved for slots that actually coalesced (>=2 segs).
if len(w.writes) != 1 || len(w.gsoWrites) != 0 {
t.Fatalf("single-seg flush: writes=%d gso=%d", len(w.writes), len(w.gsoWrites)) t.Fatalf("single-seg flush: writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
} }
if !bytes.Equal(w.writes[0], pkt) {
t.Errorf("plain write not byte-identical to committed packet: got %d bytes want %d", len(w.writes[0]), len(pkt))
}
} }
func TestUDPCoalescerCoalescesEqualSized(t *testing.T) { func TestUDPCoalescerCoalescesEqualSized(t *testing.T) {
@@ -180,14 +186,16 @@ func TestUDPCoalescerShortLastSegmentSeals(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if len(w.gsoWrites) != 2 { // The sealed 3-datagram chain is a real superpacket; the re-seed stays
t.Fatalf("want 2 gso writes (sealed + new seed), got %d", len(w.gsoWrites)) // single-segment and flushes as a plain write of the original packet.
if len(w.gsoWrites) != 1 || len(w.writes) != 1 {
t.Fatalf("want 1 gso (sealed) + 1 plain (new seed), got gso=%d plain=%d", len(w.gsoWrites), len(w.writes))
} }
if len(w.gsoWrites[0].pays) != 3 { if len(w.gsoWrites[0].pays) != 3 {
t.Errorf("first super: want 3 pays, got %d", len(w.gsoWrites[0].pays)) t.Errorf("super: want 3 pays, got %d", len(w.gsoWrites[0].pays))
} }
if len(w.gsoWrites[1].pays) != 1 { if got, want := len(w.writes[0]), 20+8+1200; got != want {
t.Errorf("second super: want 1 pay (re-seed), got %d", len(w.gsoWrites[1].pays)) t.Errorf("re-seed plain write len=%d want %d", got, want)
} }
} }
@@ -204,8 +212,14 @@ func TestUDPCoalescerLargerThanSeedReseeds(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if len(w.gsoWrites) != 2 { // Both seeds stay single-segment → two plain writes in arrival order.
t.Fatalf("want 2 separate seeds, got %d", len(w.gsoWrites)) if len(w.writes) != 2 || len(w.gsoWrites) != 0 {
t.Fatalf("want 2 separate plain writes, got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
}
for i, want := range []int{20 + 8 + 800, 20 + 8 + 1200} {
if len(w.writes[i]) != want {
t.Errorf("write %d len=%d want %d", i, len(w.writes[i]), want)
}
} }
} }
@@ -268,8 +282,9 @@ func TestUDPCoalescerCapsAtMaxSegs(t *testing.T) {
// Differing IP ECN codepoints must not coalesce: udpHeadersMatch compares // Differing IP ECN codepoints must not coalesce: udpHeadersMatch compares
// the full ToS byte (matching kernel GRO). A CE-marked datagram mid-run // the full ToS byte (matching kernel GRO). A CE-marked datagram mid-run
// seals the Not-ECT chain and seeds a fresh superpacket that keeps CE; the // seals the Not-ECT chain and reseeds; the trailing Not-ECT datagram
// trailing Not-ECT datagram seeds another. // reseeds again. All three stay single-segment, so each ships as a plain
// write of its original bytes, keeping its own codepoint.
func TestUDPCoalescerDifferingECNReseeds(t *testing.T) { func TestUDPCoalescerDifferingECNReseeds(t *testing.T) {
w := &fakeTunWriter{gsoEnabled: true} w := &fakeTunWriter{gsoEnabled: true}
c := newTestUDPCoalescer(t, w) c := newTestUDPCoalescer(t, w)
@@ -286,16 +301,13 @@ func TestUDPCoalescerDifferingECNReseeds(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if len(w.gsoWrites) != 3 { if len(w.writes) != 3 || len(w.gsoWrites) != 0 {
t.Fatalf("want 3 separate seeds (differing ECN), got %d (plain=%d)", len(w.gsoWrites), len(w.writes)) t.Fatalf("want 3 separate plain writes (differing ECN), got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
} }
wantECN := []byte{0x00, 0x03, 0x00} wantECN := []byte{0x00, 0x03, 0x00}
for i, g := range w.gsoWrites { for i, p := range w.writes {
if len(g.pays) != 1 { if got := p[1] & 0x03; got != wantECN[i] {
t.Errorf("gso %d pay count=%d want 1", i, len(g.pays)) t.Errorf("write %d ECN=%#x want %#x", i, got, wantECN[i])
}
if got := g.hdr[1] & 0x03; got != wantECN[i] {
t.Errorf("gso %d ECN=%#x want %#x", i, got, wantECN[i])
} }
} }
} }
@@ -353,8 +365,9 @@ func TestUDPCoalescerDSCPMismatchReseeds(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if len(w.gsoWrites) != 2 { // Both seeds stay single-segment → two plain writes, no gso.
t.Fatalf("want 2 separate seeds (different DSCP), got %d", len(w.gsoWrites)) if len(w.writes) != 2 || len(w.gsoWrites) != 0 {
t.Fatalf("want 2 separate plain writes (different DSCP), got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
} }
} }
@@ -437,9 +450,16 @@ func TestUDPCoalescerZeroLengthMidFlowSealsAndPreservesOrder(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
// The empty datagram sealed the first slot, so the trailing full packet // The empty datagram sealed the first slot, so the trailing full packet
// can't join it: two single-segment superpackets bracket one plain write. // can't join it. All three emit as plain writes (the two full datagrams
if len(w.gsoWrites) != 2 || len(w.writes) != 1 { // stayed single-segment; the empty one is passthrough) in per-flow
t.Fatalf("want 2 gso writes + 1 plain, got gso=%d plain=%d", len(w.gsoWrites), len(w.writes)) // arrival order: full, empty, full.
if len(w.writes) != 3 || len(w.gsoWrites) != 0 {
t.Fatalf("want 3 plain writes, got gso=%d plain=%d", len(w.gsoWrites), len(w.writes))
}
for i, want := range []int{20 + 8 + 800, 20 + 8, 20 + 8 + 800} {
if len(w.writes[i]) != want {
t.Errorf("write %d len=%d want %d (order full, empty, full)", i, len(w.writes[i]), want)
}
} }
} }
@@ -484,7 +504,8 @@ func TestUDPCoalescerNonAtomicSequentialIDsCoalesce(t *testing.T) {
} }
// TestUDPCoalescerNonAtomicIDGapReseeds: an ID jump on a DF=0 flow breaks // TestUDPCoalescerNonAtomicIDGapReseeds: an ID jump on a DF=0 flow breaks
// the chain; each datagram must keep its own (meaningful) ID. // the chain; each datagram stays a single-segment slot and flushes as a
// plain write that keeps its own (meaningful) ID.
func TestUDPCoalescerNonAtomicIDGapReseeds(t *testing.T) { func TestUDPCoalescerNonAtomicIDGapReseeds(t *testing.T) {
w := &fakeTunWriter{gsoEnabled: true} w := &fakeTunWriter{gsoEnabled: true}
c := newTestUDPCoalescer(t, w) c := newTestUDPCoalescer(t, w)
@@ -504,11 +525,11 @@ func TestUDPCoalescerNonAtomicIDGapReseeds(t *testing.T) {
if err := c.Flush(); err != nil { if err := c.Flush(); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if len(w.gsoWrites) != 2 { if len(w.writes) != 2 || len(w.gsoWrites) != 0 {
t.Fatalf("ID gap on DF=0 must reseed: gso=%d", len(w.gsoWrites)) t.Fatalf("ID gap on DF=0 must reseed: writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
} }
for i, want := range []uint16{40, 50} { for i, want := range []uint16{40, 50} {
if id := binary.BigEndian.Uint16(w.gsoWrites[i].hdr[4:6]); id != want { if id := binary.BigEndian.Uint16(w.writes[i][4:6]); id != want {
t.Errorf("write %d: ID=%d want %d (must be preserved)", i, id, want) t.Errorf("write %d: ID=%d want %d (must be preserved)", i, id, want)
} }
} }