diff --git a/overlay/batch/tcp_coalesce.go b/overlay/batch/tcp_coalesce.go index 2c44c3aa..9926a4a0 100644 --- a/overlay/batch/tcp_coalesce.go +++ b/overlay/batch/tcp_coalesce.go @@ -48,9 +48,11 @@ type coalesceSlot struct { numSeg int totalPay int nextSeq uint32 - // psh closes the chain: set when the last-accepted segment had PSH or - // was sub-gsoSize. No further appends after that. - psh bool + // sealed marks the chain permanently closed: the last-accepted segment had PSH or was sub-gsoSize, + // so no append or flush-time merge may follow. + // Distinct from mere eviction out of openSlots (e.g. on seq mismatch), + // which leaves sealed=false so reorderForFlush can still merge the slot. + sealed bool payIovs [][]byte } @@ -213,7 +215,7 @@ func (c *TCPCoalescer) commitParsed(pkt []byte, info parsedTCP) error { if open != nil { if c.canAppend(open, pkt, info) { c.appendPayload(open, pkt, info) - if open.psh { + if open.sealed { delete(c.openSlots, info.fk) c.lastSlot = nil } else { @@ -221,7 +223,8 @@ func (c *TCPCoalescer) commitParsed(pkt []byte, info parsedTCP) error { } return nil } - // Can't extend — seal it and fall through to seed a fresh slot. + // Can't extend: evict it from openSlots and fall through to seed a fresh slot. + // The slot stays unsealed so reorderForFlush may still merge it. delete(c.openSlots, info.fk) if c.lastSlot == open { c.lastSlot = nil @@ -279,10 +282,10 @@ func (c *TCPCoalescer) seed(pkt []byte, info parsedTCP) { s.numSeg = 1 s.totalPay = info.payLen s.nextSeq = info.seq + uint32(info.payLen) - s.psh = info.flags&tcpFlagPsh != 0 + s.sealed = info.flags&tcpFlagPsh != 0 s.payIovs = append(s.payIovs[:0], pkt[info.hdrLen:info.hdrLen+info.payLen]) c.slots = append(c.slots, s) - if !s.psh { + if !s.sealed { c.openSlots[info.fk] = s c.lastSlot = s } else if last := c.lastSlot; last != nil && last.fk == info.fk { @@ -296,7 +299,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. func (c *TCPCoalescer) canAppend(s *coalesceSlot, pkt []byte, info parsedTCP) bool { - if s.psh { + if s.sealed { return false } if info.hdrLen != s.hdrLen { @@ -337,7 +340,7 @@ func (c *TCPCoalescer) appendPayload(s *coalesceSlot, pkt []byte, info parsedTCP s.hdrBuf[s.ipHdrLen+13] |= tcpFlagPsh } if info.payLen < s.gsoSize || info.flags&tcpFlagPsh != 0 { - s.psh = true + s.sealed = true } } @@ -358,7 +361,7 @@ func (c *TCPCoalescer) release(s *coalesceSlot) { s.payIovs = s.payIovs[:0] s.numSeg = 0 s.totalPay = 0 - s.psh = false + s.sealed = false c.pool = append(c.pool, s) } @@ -597,11 +600,11 @@ func flowKeyCompare(a, b flowKey) int { // → ipHeadersMatch, which compares the full DSCP/ECN byte, so two slots with // differing ECN marks stay separate superpackets, each keeping its own mark. // -// Note: a slot sealed by reorder (canAppend returned false on seq -// mismatch) keeps psh=false, so this restriction does not block the -// reorder-fix merge — only legitimate PSH-set seals. +// Note: a slot evicted on reorder (canAppend returned false on seq mismatch) +// stays sealed=false, so this restriction does not block the reorder-fix merge, +// only chains ended by PSH or a short tail. func canMergeSlots(prev, s *coalesceSlot) bool { - if prev.psh { + if prev.sealed { return false } if prev.fk != s.fk { @@ -634,18 +637,16 @@ func canMergeSlots(prev, s *coalesceSlot) bool { } // mergeSlots folds src into dst in place: payIovs concatenated, counters -// and totals updated, PSH OR'd into the seed header so the push signal is -// not lost. The seed header's seq, gsoSize, and fk are unchanged. Caller -// is responsible for releasing src (it's no longer in c.slots after this call). +// and totals updated. The seed header's seq, gsoSize, and fk are unchanged. +// The caller must release src (it's no longer in c.slots after this call). func mergeSlots(dst, src *coalesceSlot) { dst.payIovs = append(dst.payIovs, src.payIovs...) dst.numSeg += src.numSeg dst.totalPay += src.totalPay dst.nextSeq = src.nextSeq - if src.psh { - dst.psh = true - dst.hdrBuf[dst.ipHdrLen+13] |= tcpFlagPsh - } + dst.sealed = src.sealed // dst is open — canMergeSlots rejects a sealed prev + // carry PSH through + dst.hdrBuf[dst.ipHdrLen+13] |= src.hdrBuf[src.ipHdrLen+13] & tcpFlagPsh } // ipv4HdrChecksum computes the IPv4 header checksum over hdr (which must diff --git a/overlay/batch/tcp_coalesce_test.go b/overlay/batch/tcp_coalesce_test.go index 35c41766..47e02007 100644 --- a/overlay/batch/tcp_coalesce_test.go +++ b/overlay/batch/tcp_coalesce_test.go @@ -1115,3 +1115,80 @@ func TestSortRunZeroAllocs(t *testing.T) { t.Fatalf("sortRun allocates %v times per run; want 0", allocs) } } + +// TestCoalescerMergeShortTailDoesNotFabricatePSH: a slot sealed by a +// sub-gsoSize tail segment has psh=true in the chain-closed sense but no +// PSH flag on any of its packets. When reorderForFlush folds it into the +// preceding slot, the merged header must not grow a PSH the sender never +// sent — mergeSlots must copy the wire flag from the source header, not +// synthesize it from the seal bool. +func TestCoalescerMergeShortTailDoesNotFabricatePSH(t *testing.T) { + w := &fakeTunWriter{gsoEnabled: true} + c := NewTCPCoalescer(w, test.NewLogger()) + pay := make([]byte, 1200) + short := make([]byte, 600) + // Arrival: seq 3400 (full), 4600 (short, seals the slot), then the + // reordered front of the window: 1000, 2200. Flush sorts the two slots + // into [1000..3400) + [3400..5200) and merges them. + if err := c.Commit(buildTCPv4(3400, tcpAck, pay)); err != nil { + t.Fatal(err) + } + if err := c.Commit(buildTCPv4(4600, tcpAck, short)); err != nil { + t.Fatal(err) + } + if err := c.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { + t.Fatal(err) + } + if err := c.Commit(buildTCPv4(2200, tcpAck, pay)); err != nil { + t.Fatal(err) + } + if err := c.Flush(); err != nil { + t.Fatal(err) + } + if len(w.gsoWrites) != 1 { + t.Fatalf("want 1 merged gso write got %d (plain=%d)", len(w.gsoWrites), len(w.writes)) + } + g := w.gsoWrites[0] + if len(g.pays) != 4 { + t.Fatalf("merged segs=%d want 4", len(g.pays)) + } + const ipHdrLen = 20 + if flags := g.hdr[ipHdrLen+13]; flags&tcpPsh != 0 { + t.Errorf("merged header flags=%#x: PSH fabricated by short-tail merge", flags) + } +} + +// TestCoalescerMergePreservesRealPSH is the positive companion: when the +// source slot's tail really carried PSH, the merged header must keep it. +func TestCoalescerMergePreservesRealPSH(t *testing.T) { + w := &fakeTunWriter{gsoEnabled: true} + c := NewTCPCoalescer(w, test.NewLogger()) + pay := make([]byte, 1200) + short := make([]byte, 600) + if err := c.Commit(buildTCPv4(3400, tcpAck, pay)); err != nil { + t.Fatal(err) + } + if err := c.Commit(buildTCPv4(4600, tcpAckPsh, short)); err != nil { + t.Fatal(err) + } + if err := c.Commit(buildTCPv4(1000, tcpAck, pay)); err != nil { + t.Fatal(err) + } + if err := c.Commit(buildTCPv4(2200, tcpAck, pay)); err != nil { + t.Fatal(err) + } + if err := c.Flush(); err != nil { + t.Fatal(err) + } + if len(w.gsoWrites) != 1 { + t.Fatalf("want 1 merged gso write got %d (plain=%d)", len(w.gsoWrites), len(w.writes)) + } + g := w.gsoWrites[0] + if len(g.pays) != 4 { + t.Fatalf("merged segs=%d want 4", len(g.pays)) + } + const ipHdrLen = 20 + if flags := g.hdr[ipHdrLen+13]; flags&tcpPsh == 0 { + t.Errorf("merged header flags=%#x: real PSH lost in merge", flags) + } +}