diff --git a/pkg/tcpip/stack/tcp.go b/pkg/tcpip/stack/tcp.go index 3393e6b29..6597ea224 100644 --- a/pkg/tcpip/stack/tcp.go +++ b/pkg/tcpip/stack/tcp.go @@ -85,6 +85,26 @@ type TCPCubicState struct { // WEst is the window computed by CUBIC at time // TimeSinceLastCongestion+RTT i.e WC(TimeSinceLastCongestion+RTT). WEst float64 + + // EndSeq is the sequence number that, when cumulatively ACK'd, ends the + // HyStart round + EndSeq seqnum.Value + + // CurrRTT is the minimum round-trip time from the current round + CurrRTT time.Duration + + // LastRTT is the minimum round-trip time from the previous round + LastRTT time.Duration + + // SampleCount is the number of samples from the current round + SampleCount uint + + // LastAck is the time we received the most recent ACK (or start of round if + // more recent). + LastAck tcpip.MonotonicTime + + // RoundStart is the time we started the most recent HyStart round + RoundStart tcpip.MonotonicTime } // TCPRACKState is used to hold a copy of the internal RACK state when the diff --git a/pkg/tcpip/transport/tcp/BUILD b/pkg/tcpip/transport/tcp/BUILD index e7d508ff2..cd7fcd8f9 100644 --- a/pkg/tcpip/transport/tcp/BUILD +++ b/pkg/tcpip/transport/tcp/BUILD @@ -99,6 +99,7 @@ go_test( name = "tcp_test", size = "small", srcs = [ + "cubic_test.go", "main_test.go", "segment_test.go", "timer_test.go", diff --git a/pkg/tcpip/transport/tcp/cubic.go b/pkg/tcpip/transport/tcp/cubic.go index ff33a71bb..4378f5e10 100644 --- a/pkg/tcpip/transport/tcp/cubic.go +++ b/pkg/tcpip/transport/tcp/cubic.go @@ -18,9 +18,53 @@ import ( "math" "time" + "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/stack" ) +// effectivelyInfinity is an initialization value used for round-trip times +// that are then set using minDuration. It is equal to approximately 100 +// years: large enough that it will always be greater than a real TCP +// round-trip time, and small enough that it fits in time.Duration. +const effectivelyInfinity = 876000 * time.Hour + +// c.f. RFC 9406 Section 4.3. RTT = round-trip time. +const ( + // The delay increase sensitivity is determined by minRTTThresh and + // maxRTTThresh. Smaller values of minRTTThresh may cause spurious exits + // from slow start. Larger values of maxRTTThresh may result in slow start + // not exiting until loss is encountered for connections on large RTT paths. + minRTTThresh = 4 * time.Millisecond + maxRTTThresh = 16 * time.Millisecond + + // minRTTDivisor is a fraction of RTT to compute the delay threshold. A + // smaller value would mean a larger threshold and thus less sensitivity to + // delay increase, and vice versa. + minRTTDivisor = 8 + + // nRTTSample is the minimum number of RTT samples in the round before + // considering whether to exit the round due to increased RTT. + nRTTSample = 8 + + // ackDelta is the maximum time between ACKs for them to be considered part + // of the same ACK Train during HyStart + ackDelta = 2 * time.Millisecond +) + +func minDuration(a, b time.Duration) time.Duration { + if a < b { + return a + } + return b +} + +func maxDuration(a, b time.Duration) time.Duration { + if a < b { + return b + } + return a +} + // cubicState stores the variables related to TCP CUBIC congestion // control algorithm state. // @@ -39,11 +83,19 @@ type cubicState struct { // newCubicCC returns a partially initialized cubic state with the constants // beta and c set and t set to current time. func newCubicCC(s *sender) *cubicState { + now := s.ep.stack.Clock().NowMonotonic() return &cubicState{ TCPCubicState: stack.TCPCubicState{ - T: s.ep.stack.Clock().NowMonotonic(), + T: now, Beta: 0.7, C: 0.4, + // by this point, the sender has initialized it's initial sequence + // number. + EndSeq: s.SndNxt, + LastRTT: effectivelyInfinity, + CurrRTT: effectivelyInfinity, + LastAck: now, + RoundStart: now, }, s: s, } @@ -66,6 +118,56 @@ func (c *cubicState) enterCongestionAvoidance() { } } +// updateHyStart tracks packet round-trip time (rtt) to find a safe threshold +// to exit slow start without triggering packet loss. It updates the SSThresh +// and sets FoundThresh when it does. +func (c *cubicState) updateHyStart(rtt time.Duration) { + if rtt < 0 { + // negative indicates unknown + return + } + now := c.s.ep.stack.Clock().NowMonotonic() + if c.EndSeq.LessThan(c.s.SndUna) { + c.beginHyStartRound(now) + } + // ACK train + if now.Sub(c.LastAck) < ackDelta && // ensures acks are part of the same "train" + c.LastRTT < effectivelyInfinity { + c.LastAck = now + thresh := c.LastRTT / 2 + if now.Sub(c.RoundStart) > thresh { + c.s.Ssthresh = c.s.SndCwnd + } + } + + // Delay increase + c.CurrRTT = minDuration(c.CurrRTT, rtt) + c.SampleCount++ + + if c.SampleCount >= nRTTSample && + c.LastRTT < effectivelyInfinity { + // i.e. LastRTT/minRTTDivisor, but clamped to minRTTThresh & maxRTTThresh + thresh := maxDuration( + minRTTThresh, + minDuration(maxRTTThresh, c.LastRTT/minRTTDivisor), + ) + if c.CurrRTT >= (c.LastRTT + thresh) { + // Triggered HyStart safe exit threshold + c.s.Ssthresh = c.s.SndCwnd + } + } +} + +// resetHyStartRound begins a new HyStart round +func (c *cubicState) beginHyStartRound(now tcpip.MonotonicTime) { + c.EndSeq = c.s.SndNxt + c.SampleCount = 0 + c.LastRTT = c.CurrRTT + c.CurrRTT = effectivelyInfinity + c.LastAck = now + c.RoundStart = now +} + // updateSlowStart will update the congestion window as per the slow-start // algorithm used by NewReno. If after adjusting the congestion window we cross // the ssThresh then it will return the number of packets that must be consumed @@ -92,7 +194,10 @@ func (c *cubicState) updateSlowStart(packetsAcked int) int { // Update updates cubic's internal state variables. It must be called on every // ACK received. // Refer: https://tools.ietf.org/html/rfc8312#section-4 -func (c *cubicState) Update(packetsAcked int) { +func (c *cubicState) Update(packetsAcked int, rtt time.Duration) { + if c.s.Ssthresh == InitialSsthresh && c.s.SndCwnd < c.s.Ssthresh { + c.updateHyStart(rtt) + } if c.s.SndCwnd < c.s.Ssthresh { packetsAcked = c.updateSlowStart(packetsAcked) if packetsAcked == 0 { diff --git a/pkg/tcpip/transport/tcp/cubic_test.go b/pkg/tcpip/transport/tcp/cubic_test.go new file mode 100644 index 000000000..7150fa116 --- /dev/null +++ b/pkg/tcpip/transport/tcp/cubic_test.go @@ -0,0 +1,283 @@ +// Copyright 2024 The gVisor Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package tcp + +import ( + "testing" + "time" + + "gvisor.dev/gvisor/pkg/tcpip" + "gvisor.dev/gvisor/pkg/tcpip/faketime" + "gvisor.dev/gvisor/pkg/tcpip/seqnum" + "gvisor.dev/gvisor/pkg/tcpip/stack" +) + +// TestHyStartAckTrain_OK tests that HyStart triggers early exit from slow start +// if ACKs come in the same rounde for longer than RTT/2 +func TestHyStartAckTrain_OK(t *testing.T) { + fClock := faketime.NewManualClock() + stackOpts := stack.Options{ + TransportProtocols: []stack.TransportProtocolFactory{NewProtocol}, + Clock: fClock, + } + s := stack.New(stackOpts) + ep := &Endpoint{ + stack: s, + cc: tcpip.CongestionControlOption("cubic"), + } + iss := seqnum.Value(0) + snd := &sender{ + ep: ep, + TCPSenderState: stack.TCPSenderState{ + SndUna: iss + 1, + SndNxt: iss + 1, + Ssthresh: InitialSsthresh, + }, + } + uut := newCubicCC(snd) + snd.cc = uut + + if uut.LastRTT != effectivelyInfinity { + t.Fatal() + } + if uut.CurrRTT != effectivelyInfinity { + t.Fatal() + } + + d0 := 4 * time.Millisecond + uut.updateHyStart(d0) + if uut.CurrRTT != d0 { + t.Fatal() + } + if snd.Ssthresh != InitialSsthresh { + t.Fatal("HyStart should not be triggered") + } + + // move SndNext and SndUna to advance to a new round. + snd.SndNxt = snd.SndNxt.Add(2000) + snd.SndUna = snd.SndUna.Add(1000) + fClock.Advance(d0) + r1ExpectedStart := fClock.NowMonotonic() + + d1 := 5 * time.Millisecond + uut.updateHyStart(d1) + if uut.LastRTT != d0 { + t.Fatal() + } + if uut.CurrRTT != d1 { + t.Fatal() + } + if uut.RoundStart != r1ExpectedStart { + t.Fatal() + } + + // Still in round after RTT/2 (2ms) triggers HyStart. Note that HyStart + // will ignore ACKs spaced more than 2ms apart, so we send one per ms 3 + // times + fClock.Advance(time.Millisecond) + uut.updateHyStart(d1) + if snd.Ssthresh != InitialSsthresh { + t.Fatal("HyStart should not be triggered") + } + if uut.LastAck != fClock.NowMonotonic() { + t.Fatal() + } + fClock.Advance(time.Millisecond) + uut.updateHyStart(d1) + if snd.Ssthresh != InitialSsthresh { + t.Fatal("HyStart should not be triggered") + } + if uut.LastAck != fClock.NowMonotonic() { + t.Fatal() + } + + // 3 ms---triggers HyStart setting Ssthresh + fClock.Advance(time.Millisecond) + uut.updateHyStart(d1) + if snd.Ssthresh == InitialSsthresh { + t.Fatal("HyStart SHOULD be triggered") + } +} + +// TestHyStartAckTrain_TooSpread tests that ACKs that are more than 2ms apart +// are ignored for purposes of triggering HyStart via the ACK train mechanism. +func TestHyStartAckTrain_TooSpread(t *testing.T) { + fClock := faketime.NewManualClock() + stackOpts := stack.Options{ + TransportProtocols: []stack.TransportProtocolFactory{NewProtocol}, + Clock: fClock, + } + s := stack.New(stackOpts) + ep := &Endpoint{ + stack: s, + cc: tcpip.CongestionControlOption("cubic"), + } + iss := seqnum.Value(0) + snd := &sender{ + ep: ep, + TCPSenderState: stack.TCPSenderState{ + SndUna: iss + 1, + SndNxt: iss + 1, + Ssthresh: InitialSsthresh, + }, + } + uut := newCubicCC(snd) + snd.cc = uut + + if uut.LastRTT != effectivelyInfinity { + t.Fatal() + } + if uut.CurrRTT != effectivelyInfinity { + t.Fatal() + } + d0 := 4 * time.Millisecond + uut.updateHyStart(d0) + if uut.CurrRTT != d0 { + t.Fatal() + } + if snd.Ssthresh != InitialSsthresh { + t.Fatal("HyStart should not be triggered") + } + + // move SndNext and SndUna to advance to a new round. + snd.SndNxt = snd.SndNxt.Add(2000) + snd.SndUna = snd.SndUna.Add(1000) + fClock.Advance(d0) + r1ExpectedStart := fClock.NowMonotonic() + + d1 := 5 * time.Millisecond + uut.updateHyStart(d1) + if uut.LastRTT != d0 { + t.Fatal() + } + if uut.CurrRTT != d1 { + t.Fatal() + } + if uut.RoundStart != r1ExpectedStart { + t.Fatal() + } + + // HyStart will ignore ACKs spaced more than 2ms apart + fClock.Advance(3 * time.Millisecond) + uut.updateHyStart(d1) + if snd.Ssthresh != InitialSsthresh { + t.Fatal("HyStart should not be triggered") + } + if uut.LastAck != r1ExpectedStart { + t.Fatal("Should ignore ACK 3ms later") + } +} + +// TestHyStartDelay_OK tests that HyStart triggers early exit from slow start +// if RTT exceeds previous round by at least minRTTThresh +func TestHyStartDelay_OK(t *testing.T) { + fClock := faketime.NewManualClock() + stackOpts := stack.Options{ + TransportProtocols: []stack.TransportProtocolFactory{NewProtocol}, + Clock: fClock, + } + s := stack.New(stackOpts) + ep := &Endpoint{ + stack: s, + cc: tcpip.CongestionControlOption("cubic"), + } + iss := seqnum.Value(0) + snd := &sender{ + ep: ep, + TCPSenderState: stack.TCPSenderState{ + SndUna: iss + 1, + SndNxt: iss + 1, + Ssthresh: InitialSsthresh, + }, + } + uut := newCubicCC(snd) + snd.cc = uut + + d0 := 4 * time.Millisecond + uut.updateHyStart(d0) + + // move SndNext and SndUna to advance to a new round. + snd.SndNxt = snd.SndNxt.Add(2000) + snd.SndUna = snd.SndUna.Add(1000) + fClock.Advance(d0) + + d1 := d0 + minRTTThresh + + // Delay detection requires at least nRTTSample measurements + for i := uint(1); i < nRTTSample; i++ { + uut.updateHyStart(d1) + if uut.SampleCount != i { + t.Fatal() + } + } + if snd.Ssthresh != InitialSsthresh { + t.Fatal("triggered with fewer than nRTTSample measurements") + } + uut.updateHyStart(d1) + if snd.Ssthresh == InitialSsthresh { + t.Fatal("didn't trigger SS exit") + } +} + +// TestHyStartDelay_BelowThresh tests that HyStart doesn't trigger early exit +// from slow start if at least one RTT measurement is below threshold +func TestHyStartDelay_BelowThresh(t *testing.T) { + fClock := faketime.NewManualClock() + stackOpts := stack.Options{ + TransportProtocols: []stack.TransportProtocolFactory{NewProtocol}, + Clock: fClock, + } + s := stack.New(stackOpts) + ep := &Endpoint{ + stack: s, + cc: tcpip.CongestionControlOption("cubic"), + } + iss := seqnum.Value(0) + snd := &sender{ + ep: ep, + TCPSenderState: stack.TCPSenderState{ + SndUna: iss + 1, + SndNxt: iss + 1, + Ssthresh: InitialSsthresh, + }, + } + uut := newCubicCC(snd) + snd.cc = uut + + d0 := 4 * time.Millisecond + uut.updateHyStart(d0) + + // move SndNext and SndUna to advance to a new round. + snd.SndNxt = snd.SndNxt.Add(2000) + snd.SndUna = snd.SndUna.Add(1000) + fClock.Advance(d0) + + d1 := d0 + minRTTThresh + + // Delay detection requires at least nRTTSample measurements + for i := uint(1); i < nRTTSample; i++ { + uut.updateHyStart(d1) + if uut.SampleCount != i { + t.Fatal() + } + } + if snd.Ssthresh != InitialSsthresh { + t.Fatal("triggered with fewer than nRTTSample measurements") + } + uut.updateHyStart(d1 - time.Millisecond) + if snd.Ssthresh != InitialSsthresh { + t.Fatal("triggered with a measurement under threshold") + } +} diff --git a/pkg/tcpip/transport/tcp/reno.go b/pkg/tcpip/transport/tcp/reno.go index 063552c7f..2d1b011db 100644 --- a/pkg/tcpip/transport/tcp/reno.go +++ b/pkg/tcpip/transport/tcp/reno.go @@ -14,6 +14,10 @@ package tcp +import ( + "time" +) + // renoState stores the variables related to TCP New Reno congestion // control algorithm. // @@ -69,7 +73,7 @@ func (r *renoState) reduceSlowStartThreshold() { // Update updates the congestion state based on the number of packets that // were acknowledged. // Update implements congestionControl.Update. -func (r *renoState) Update(packetsAcked int) { +func (r *renoState) Update(packetsAcked int, _ time.Duration) { if r.s.SndCwnd < r.s.Ssthresh { packetsAcked = r.updateSlowStart(packetsAcked) if packetsAcked == 0 { diff --git a/pkg/tcpip/transport/tcp/snd.go b/pkg/tcpip/transport/tcp/snd.go index f8d699fe4..5f2493904 100644 --- a/pkg/tcpip/transport/tcp/snd.go +++ b/pkg/tcpip/transport/tcp/snd.go @@ -49,6 +49,16 @@ const ( // before timing out the connection. // Linux default TCP_RETR2, net.ipv4.tcp_retries2. MaxRetries = 15 + + // InitialSsthresh is the the maximum int value, which depends on the + // platform. + InitialSsthresh = math.MaxInt + + // unknownRTT is used to indicate to congestion control algorithms that we + // were unable to measure the round-trip time when processing ACKs. + // Algorithms (such as HyStart) that use the round-trip time should ignore + // such Updates. + unknownRTT = time.Duration(-1) ) // congestionControl is an interface that must be implemented by any supported @@ -64,8 +74,9 @@ type congestionControl interface { // Update is invoked when processing inbound acks. It's passed the // number of packet's that were acked by the most recent cumulative - // acknowledgement. - Update(packetsAcked int) + // acknowledgement. rtt is the round-trip time, or is set to unknownRTT + // (above) to indicate the time is unknown. + Update(packetsAcked int, rtt time.Duration) // PostRecovery is invoked when the sender is exiting a fast retransmit/ // recovery phase. This provides congestion control algorithms a way @@ -252,9 +263,8 @@ func newSender(ep *Endpoint, iss, irs seqnum.Value, sndWnd seqnum.Size, mss uint // their initial values. func (s *sender) initCongestionControl(congestionControlName tcpip.CongestionControlOption) congestionControl { s.SndCwnd = InitialCwnd - // Set sndSsthresh to the maximum int value, which depends on the - // platform. - s.Ssthresh = int(^uint(0) >> 1) + // Set sndSsthresh to + s.Ssthresh = InitialSsthresh switch congestionControlName { case ccCubic: @@ -1411,9 +1421,12 @@ func (s *sender) inRecovery() bool { // +checklocks:s.ep.mu // +checklocksalias:s.rc.snd.ep.mu=s.ep.mu func (s *sender) handleRcvdSegment(rcvdSeg *segment) { + bestRTT := unknownRTT + // Check if we can extract an RTT measurement from this ack. if !rcvdSeg.parsedOptions.TS && s.RTTMeasureSeqNum.LessThan(rcvdSeg.ackNumber) { - s.updateRTO(s.ep.stack.Clock().NowMonotonic().Sub(s.RTTMeasureTime)) + bestRTT = s.ep.stack.Clock().NowMonotonic().Sub(s.RTTMeasureTime) + s.updateRTO(bestRTT) s.RTTMeasureSeqNum = s.SndNxt } @@ -1515,7 +1528,14 @@ func (s *sender) handleRcvdSegment(rcvdSeg *segment) { // some new data, i.e., only if it advances the left edge of // the send window. if s.ep.SendTSOk && rcvdSeg.parsedOptions.TSEcr != 0 { - s.updateRTO(s.ep.elapsed(s.ep.stack.Clock().NowMonotonic(), rcvdSeg.parsedOptions.TSEcr)) + tsRTT := s.ep.elapsed(s.ep.stack.Clock().NowMonotonic(), rcvdSeg.parsedOptions.TSEcr) + s.updateRTO(tsRTT) + // Following Linux, prefer RTT computed from ACKs to TSEcr because, + // "broken middle-boxes or peers may corrupt TS-ECR fields" + // https://github.com/torvalds/linux/blob/39cd87c4eb2b893354f3b850f916353f2658ae6f/net/ipv4/tcp_input.c#L3141C1-L3144C24 + if bestRTT == unknownRTT { + bestRTT = tsRTT + } } if s.shouldSchedulePTO() { @@ -1584,7 +1604,7 @@ func (s *sender) handleRcvdSegment(rcvdSeg *segment) { // If we are not in fast recovery then update the congestion // window based on the number of acknowledged packets. if !s.FastRecovery.Active { - s.cc.Update(originalOutstanding - s.Outstanding) + s.cc.Update(originalOutstanding-s.Outstanding, bestRTT) if s.FastRecovery.Last.LessThan(s.SndUna) { s.state = tcpip.Open // Update RACK when we are exiting fast or RTO