Add support for TCP HyStart

Signed-off-by: Spike Curtis <spike@coder.com>

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

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",

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 {

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")
+	}
+}

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 {

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