diff --git a/pkg/tcpip/transport/tcp/snd.go b/pkg/tcpip/transport/tcp/snd.go index 056c681d744..df43eec436c 100644 --- a/pkg/tcpip/transport/tcp/snd.go +++ b/pkg/tcpip/transport/tcp/snd.go @@ -184,6 +184,17 @@ type sender struct { // RFC3522 Section 3.2. retransmitTS uint32 + // sackedOutsideRecovery is the number of packets that were newly + // SACKed while no loss recovery was in progress. SetPipe() only + // accounts SACKed segments into Outstanding during loss recovery, and + // the cumulative-ACK path skips previously-SACKed segments assuming + // SetPipe() accounted for them; packets counted here are the ones + // that assumption misses, and their count is removed from Outstanding + // when they are cumulatively ACKed. Reset whenever Outstanding is + // recomputed or reset wholesale (recovery entry, RTO, full window + // drain). See gvisor.dev/issue/14092. + sackedOutsideRecovery int + // startCork start corking the segments. startCork bool @@ -631,6 +642,7 @@ func (s *sender) retransmitTimerExpired() tcpip.Error { // We'll keep on transmitting (or retransmitting) as we get acks for // the data we transmit. s.Outstanding = 0 + s.sackedOutsideRecovery = 0 // Expunge all SACK information as per https://tools.ietf.org/html/rfc6675#section-5.1 // @@ -1137,6 +1149,8 @@ func (s *sender) enterRecovery() { // the 3 duplicate ACKs and are now not in flight. s.SndCwnd = s.Ssthresh + 3 s.SackedOut = 0 + // SetPipe() owns all SACK accounting for the duration of recovery. + s.sackedOutsideRecovery = 0 s.DupAckCount = 0 s.FastRecovery.First = s.SndUna s.FastRecovery.Last = s.SndNxt - 1 @@ -1556,8 +1570,27 @@ func (s *sender) handleRcvdSegment(rcvdSeg *segment) { // which have start/end before sndUna and are used to // indicate spurious retransmissions. if rcvdSeg.ackNumber.LessThan(sb.Start) && s.SndUna.LessThan(sb.Start) && sb.End.LessThanEq(s.SndNxt) && !s.ep.scoreboard.IsSACKED(sb) { + // Track segments that this block newly covers while + // no loss recovery is in progress: SetPipe() will + // not account for them (it is a no-op outside + // recovery), so the cumulative-ACK path must, or + // their packet count leaks into Outstanding + // permanently. See gvisor.dev/issue/14092. + var newlyCovered []*segment + if !s.FastRecovery.Active { + for seg := s.writeList.Front(); seg != nil && seg.sequenceNumber.LessThan(sb.End); seg = seg.Next() { + if seg.xmitCount != 0 && !s.ep.scoreboard.IsSACKED(seg.sackBlock()) { + newlyCovered = append(newlyCovered, seg) + } + } + } s.ep.scoreboard.Insert(sb) rcvdSeg.hasNewSACKInfo = true + for _, seg := range newlyCovered { + if s.ep.scoreboard.IsSACKED(seg.sackBlock()) { + s.sackedOutsideRecovery += s.pCount(seg, s.MaxPayloadSize) + } + } } } @@ -1694,11 +1727,21 @@ func (s *sender) handleRcvdSegment(rcvdSeg *segment) { // If SACK is enabled then only reduce outstanding if // the segment was not previously SACKED as these have - // already been accounted for in SetPipe(). + // already been accounted for in SetPipe() (during loss + // recovery) or in sackedOutsideRecovery below (outside + // of it). if !s.ep.SACKPermitted || !s.ep.scoreboard.IsSACKED(seg.sackBlock()) { s.Outstanding -= s.pCount(seg, s.MaxPayloadSize) } else { s.SackedOut -= s.pCount(seg, s.MaxPayloadSize) + // SetPipe() only accounts for SACKed segments + // during loss recovery; segments SACKed outside + // of it are recorded in sackedOutsideRecovery + // and must be removed from Outstanding here. + if n := min(s.pCount(seg, s.MaxPayloadSize), s.sackedOutsideRecovery); n > 0 { + s.Outstanding -= n + s.sackedOutsideRecovery -= n + } } seg.DecRef() ackLeft -= datalen @@ -1744,6 +1787,7 @@ func (s *sender) handleRcvdSegment(rcvdSeg *segment) { // RFC 6298 Rule 5.3 if s.SndUna == s.SndNxt { s.Outstanding = 0 + s.sackedOutsideRecovery = 0 // Reset firstRetransmittedSegXmitTime to the zero value. s.firstRetransmittedSegXmitTime = tcpip.MonotonicTime{} s.resendTimer.disable() diff --git a/pkg/tcpip/transport/tcp/test/e2e/tcp_sack_test.go b/pkg/tcpip/transport/tcp/test/e2e/tcp_sack_test.go index 6202a2aacc4..1081efb5169 100644 --- a/pkg/tcpip/transport/tcp/test/e2e/tcp_sack_test.go +++ b/pkg/tcpip/transport/tcp/test/e2e/tcp_sack_test.go @@ -947,6 +947,56 @@ func TestNoSpuriousRecoveryWithDSACK(t *testing.T) { verifySpuriousRecoveryMetric(t, c, 0 /* numSpuriousRecovery */, 0 /* numSpuriousRTO */) } +// TestSACKedThenAckedSegmentLeavesOutstanding is a regression test for +// gvisor.dev/issue/14092: a segment SACKed outside of loss recovery and then +// cumulatively ACKed must be removed from Outstanding. SetPipe() only +// accounts for SACKed segments during recovery, so before the fix the +// segment's packet count leaked into Outstanding permanently, throttling +// long-lived connections on reordering-but-lossless paths. +func TestSACKedThenAckedSegmentLeavesOutstanding(t *testing.T) { + for _, enableRACK := range []bool{true, false} { + t.Run(fmt.Sprintf("enableRACK: %v", enableRACK), func(t *testing.T) { + const numPackets = 4 + var c *context.Context + outstanding := make(chan int, 1) + probe := func(state *tcp.TCPEndpointState) { + // Match the cumulative ACK of the first two + // segments; the other two are still in flight. + if state.Sender.SndUna == c.IRS.Add(1+2*maxPayload) { + select { + case outstanding <- state.Sender.Outstanding: + default: + } + } + } + c = context.NewWithProbe(t, uint32(mtu), probe) + defer c.Cleanup() + + e2e.SendAndReceiveWithSACK(t, c, maxPayload, numPackets, enableRACK) + + seq := seqnum.Value(context.TestInitialSequenceNumber).Add(1) + // SACK the second segment. A single duplicate ACK does + // not trigger loss recovery: this is benign reordering, + // not loss, so SetPipe() never runs to account for the + // SACKed segment. + c.SendAckWithSACK(seq, 0, []header.SACKBlock{ + {Start: c.IRS.Add(1 + maxPayload), End: c.IRS.Add(1 + 2*maxPayload)}, + }) + // Cumulatively ACK the first two segments, covering the + // SACKed one. + c.SendAck(seq, 2*maxPayload) + + if got, want := <-outstanding, 2; got != want { + t.Errorf("got Sender.Outstanding = %d after a cumulative ACK covering a SACKed segment, want %d", got, want) + } + + // ACK the remaining data so the endpoint quiesces for + // the leak check. + c.SendAck(seq, numPackets*maxPayload) + }) + } +} + func TestMain(m *testing.M) { refs.SetLeakMode(refs.LeaksPanic) code := m.Run()