Add delay rate controller

gcc/delay_rate_controller.go

@@ -0,0 +1,189 @@
+// SPDX-FileCopyrightText: 2025 The Pion community <https://pion.ly>
+// SPDX-License-Identifier: MIT
+
+package gcc
+
+import (
+	"math"
+	"time"
+
+	"github.com/pion/logging"
+)
+
+const (
+	defaultDecreaseFactor = 0.85
+
+	// minNumDeltas is the number of samples at which the trend is trusted in
+	// full. Below it the trend is scaled down, so that an estimate based on
+	// little evidence cannot reach the overuse threshold on its own.
+	minNumDeltas = 60
+)
+
+type delayRateController struct {
+	log              logging.LeveledLogger
+	decreaseFactor   float64
+	arrivalGroups    *arrivalGroupAccumulator
+	lastArrivalGroup arrivalGroup
+	trend            *trendlineEstimator
+	overuse          *overuseDetector
+	samples          int
+	usage            usage
+	usageUpdated     bool
+	state            state
+	lastDecreaseRate *ewma
+	lastUpdate       time.Time
+	targetRate       int
+	minTarget        int
+	maxTarget        int
+}
+
+func newDelayRateController(initialRate, minRate, maxRate int, logger logging.LeveledLogger) *delayRateController {
+	return &delayRateController{
+		log:              logger,
+		decreaseFactor:   defaultDecreaseFactor,
+		arrivalGroups:    newArrivalGroupAccumulator(),
+		lastArrivalGroup: []arrivalGroupItem{},
+		trend:            newTrendlineEstimator(),
+		overuse:          newOveruseDetector(),
+		usage:            0,
+		usageUpdated:     false,
+		samples:          0,
+		state:            0,
+		lastDecreaseRate: newEWMA(0.95),
+		targetRate:       initialRate,
+		minTarget:        minRate,
+		maxTarget:        maxRate,
+	}
+}
+
+func (c *delayRateController) onPacketAcked(sequenceNumber uint64, size int, departure, arrival time.Time) {
+	next := c.arrivalGroups.onPacketAcked(
+		sequenceNumber,
+		size,
+		departure,
+		arrival,
+	)
+	if next == nil {
+		return
+	}
+	if len(next) == 0 {
+		// ignore empty groups, should never occur
+		return
+	}
+	if len(c.lastArrivalGroup) == 0 {
+		c.lastArrivalGroup = next
+
+		return
+	}
+
+	groupDeparture := next[len(next)-1].Departure
+	groupArrival := next[len(next)-1].Arrival
+	lastGroupDeparture := c.lastArrivalGroup[len(c.lastArrivalGroup)-1].Departure
+	lastGroupArrival := c.lastArrivalGroup[len(c.lastArrivalGroup)-1].Arrival
+
+	interArrivalTime := groupArrival.Sub(lastGroupArrival)
+	interDepartureTime := groupDeparture.Sub(lastGroupDeparture)
+
+	interGroupDelay := interArrivalTime - interDepartureTime
+
+	trend := c.trend.update(groupArrival, interGroupDelay)
+	c.samples++
+
+	modifiedTrend := 0.0
+	if c.samples >= 2 {
+		modifiedTrend = float64(min(c.samples, minNumDeltas)) * trend
+	}
+	c.usage = c.overuse.update(groupArrival, modifiedTrend)
+	c.usageUpdated = true
+	c.lastArrivalGroup = next
+
+	c.log.Tracef(
+		"ts=%v.%06d, seq=%v, interArrivalTime=%v, interDepartureTime=%v, interGroupDelay=%v, estimate=%f, modifiedTrend=%f, threshold=%f, usage=%v, state=%v", // nolint
+		c.lastArrivalGroup[0].Departure.UTC().Format("2006/01/02 15:04:05"),
+		c.lastArrivalGroup[0].Departure.UTC().Nanosecond()/1e3,
+		next[0].SequenceNumber,
+		interArrivalTime.Microseconds(),
+		interDepartureTime.Microseconds(),
+		interGroupDelay.Microseconds(),
+		trend,
+		modifiedTrend,
+		c.overuse.delayThreshold,
+		int(c.usage),
+		int(c.state),
+	)
+}
+
+func (c *delayRateController) update(ts time.Time, deliveryRate int, rtt time.Duration) int {
+	// The usage is only updated when an arrival group completes. Without a new
+	// one there is no new delay signal, and running the state machine again
+	// would act on the previous one twice.
+	if !c.usageUpdated {
+		return c.clampedTarget()
+	}
+	c.usageUpdated = false
+
+	deliveredRate := float64(deliveryRate)
+	c.state = c.state.transition(c.usage)
+	window := ts.Sub(c.lastUpdate)
+	c.lastUpdate = ts
+
+	// Both the increase cap and the decrease are relative to the rate that was
+	// actually delivered. Without an estimate (too few acknowledgements in the
+	// delivery rate window) there is nothing to base a new target on, so keep
+	// the current target instead of collapsing to the minimum.
+	if deliveryRate <= 0 {
+		return c.clampedTarget()
+	}
+
+	if c.state == stateIncrease {
+		if c.canIncreaseMultiplicatively(deliveredRate) {
+			c.targetRate = max(c.targetRate, multiplicativeIncrease(c.targetRate, window))
+		} else {
+			c.targetRate = additiveIncrease(c.targetRate, rtt, window)
+		}
+		c.targetRate = min(c.targetRate, int(1.5*deliveredRate))
+	}
+	if c.state == stateDecrease {
+		c.lastDecreaseRate.update(deliveredRate)
+		c.targetRate = int(c.decreaseFactor * deliveredRate)
+	}
+
+	return c.clampedTarget()
+}
+
+// clampedTarget limits the target rate to the configured min and max rates.
+func (c *delayRateController) clampedTarget() int {
+	c.targetRate = max(c.targetRate, c.minTarget)
+	c.targetRate = min(c.targetRate, c.maxTarget)
+
+	return c.targetRate
+}
+
+func (c *delayRateController) canIncreaseMultiplicatively(deliveredRate float64) bool {
+	avg := c.lastDecreaseRate.avg()
+	if avg == 0 {
+		return true
+	}
+	stdDev := math.Sqrt(c.lastDecreaseRate.varr())
+	lower := avg - 3*stdDev
+	upper := avg + 3*stdDev
+
+	return deliveredRate < lower || deliveredRate > upper
+}
+
+func multiplicativeIncrease(rate int, window time.Duration) int {
+	exponent := min(window.Seconds(), 1.0)
+	eta := math.Pow(1.08, exponent)
+
+	return int(eta * float64(rate))
+}
+
+func additiveIncrease(rate int, rtt, window time.Duration) int {
+	responseTime := 100 + rtt.Milliseconds()
+	alpha := 0.5 * min(float64(window.Milliseconds())/float64(responseTime), 1.0)
+	bitsPerFrame := float64(rate) / 30.0
+	packetsPerFrame := math.Ceil(bitsPerFrame / (1200 * 8))
+	expectedPacketSizeBits := bitsPerFrame / packetsPerFrame
+
+	return rate + max(1000, int(alpha*float64(expectedPacketSizeBits)))
+}

gcc/delay_rate_controller_test.go

@@ -0,0 +1,205 @@
+// SPDX-FileCopyrightText: 2025 The Pion community <https://pion.ly>
+// SPDX-License-Identifier: MIT
+
+package gcc
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDelayRateController(t *testing.T) {
+	t.Run("init", func(t *testing.T) {
+		controller := newDelayRateController(1_000_000, 500_000, 2_000_000, nil)
+		assert.Nil(t, controller.log)
+		assert.Equal(t, controller.decreaseFactor, defaultDecreaseFactor)
+		assert.NotNil(t, controller.arrivalGroups)
+		assert.NotNil(t, controller.lastArrivalGroup)
+		assert.NotNil(t, controller.trend)
+		assert.NotNil(t, controller.overuse)
+		assert.Equal(t, controller.samples, 0)
+		assert.Equal(t, controller.usage, usage(0))
+		assert.Equal(t, controller.state, state(0))
+		assert.NotNil(t, controller.lastDecreaseRate)
+		assert.Zero(t, controller.lastUpdate)
+		assert.Equal(t, controller.minTarget, 500_000)
+		assert.Equal(t, controller.maxTarget, 2_000_000)
+		assert.Equal(t, controller.targetRate, 1_000_000)
+	})
+
+	t.Run("canIncreaseMultiplicatively", func(t *testing.T) {
+		cases := []struct {
+			deliveredRate float64
+			decreaseRate  ewma
+			expected      bool
+		}{
+			{deliveredRate: 1000, decreaseRate: ewma{average: 0, variance: 0}, expected: true},
+			{deliveredRate: 1000, decreaseRate: ewma{average: 1500, variance: 100}, expected: true},
+			{deliveredRate: 1000, decreaseRate: ewma{average: 1020, variance: 100}, expected: false},
+			{deliveredRate: 1000, decreaseRate: ewma{average: 800, variance: 50}, expected: true},
+			{deliveredRate: 1000, decreaseRate: ewma{average: 995, variance: 100}, expected: false},
+		}
+
+		for i, c := range cases {
+			t.Run(fmt.Sprintf("%v", i), func(t *testing.T) {
+				controller := newDelayRateController(1000, 500, 2000, nil)
+				controller.lastDecreaseRate = &c.decreaseRate
+				assert.Equal(t, c.expected, controller.canIncreaseMultiplicatively(c.deliveredRate))
+			})
+		}
+	})
+
+	t.Run("update", func(t *testing.T) {
+		cases := []struct {
+			name         string
+			usage        usage
+			initialRate  int
+			minRate      int
+			maxRate      int
+			deliveryRate int
+			expected     int
+		}{
+			{
+				name:         "hold_on_underuse",
+				usage:        usageUnder,
+				initialRate:  1_000_000,
+				minRate:      500_000,
+				maxRate:      2_000_000,
+				deliveryRate: 1_000_000,
+				expected:     1_000_000,
+			},
+			{
+				name:         "no_delivery_rate_holds_target_on_increase",
+				usage:        usageNormal,
+				initialRate:  1_000_000,
+				minRate:      500_000,
+				maxRate:      2_000_000,
+				deliveryRate: 0,
+				expected:     1_000_000,
+			},
+			{
+				name:         "no_delivery_rate_holds_target_on_decrease",
+				usage:        usageOver,
+				initialRate:  1_000_000,
+				minRate:      500_000,
+				maxRate:      2_000_000,
+				deliveryRate: 0,
+				expected:     1_000_000,
+			},
+			{
+				name:         "multiplicative_increase",
+				usage:        usageNormal,
+				initialRate:  1_000_000,
+				minRate:      500_000,
+				maxRate:      2_000_000,
+				deliveryRate: 1_000_000,
+				expected:     1_080_000,
+			},
+			{
+				name:         "increase_capped_at_1.5x_delivered",
+				usage:        usageNormal,
+				initialRate:  1_000_000,
+				minRate:      500_000,
+				maxRate:      2_000_000,
+				deliveryRate: 600_000,
+				expected:     900_000,
+			},
+			{
+				name:         "increase_clamped_at_max",
+				usage:        usageNormal,
+				initialRate:  1_990_000,
+				minRate:      500_000,
+				maxRate:      2_000_000,
+				deliveryRate: 10_000_000,
+				expected:     2_000_000,
+			},
+			{
+				name:         "decrease_relative_to_delivered",
+				usage:        usageOver,
+				initialRate:  1_000_000,
+				minRate:      500_000,
+				maxRate:      2_000_000,
+				deliveryRate: 1_000_000,
+				expected:     850_000,
+			},
+			{
+				name:         "decrease_clamped_at_min",
+				usage:        usageOver,
+				initialRate:  1_000_000,
+				minRate:      500_000,
+				maxRate:      2_000_000,
+				deliveryRate: 200_000,
+				expected:     500_000,
+			},
+		}
+		for _, tc := range cases {
+			t.Run(tc.name, func(t *testing.T) {
+				controller := newDelayRateController(tc.initialRate, tc.minRate, tc.maxRate, nil)
+				controller.usage = tc.usage
+				controller.usageUpdated = true
+				now := time.Time{}.Add(time.Hour)
+				controller.lastUpdate = now.Add(-time.Second)
+
+				res := controller.update(now, tc.deliveryRate, 100*time.Millisecond)
+				assert.InDelta(t, tc.expected, res, 1)
+			})
+		}
+	})
+
+	t.Run("updateWithoutNewArrivalGroup", func(t *testing.T) {
+		controller := newDelayRateController(1_000_000, 500_000, 2_000_000, nil)
+		controller.usage = usageOver
+		controller.usageUpdated = true
+		now := time.Time{}.Add(time.Hour)
+		controller.lastUpdate = now.Add(-time.Second)
+
+		// The first update acts on the overuse.
+		assert.Equal(t, 850_000, controller.update(now, 1_000_000, 100*time.Millisecond))
+
+		// The sender follows the lowered target, so the delivery rate drops with
+		// it. No arrival group completed since, so there is no new delay signal
+		// and the same overuse must not decrease the target again.
+		now = now.Add(100 * time.Millisecond)
+		assert.Equal(t, 850_000, controller.update(now, 850_000, 100*time.Millisecond))
+		now = now.Add(100 * time.Millisecond)
+		assert.Equal(t, 850_000, controller.update(now, 722_500, 100*time.Millisecond))
+	})
+
+	t.Run("multiplicativeIncrease", func(t *testing.T) {
+		cases := []struct {
+			initialRate int
+			rate        int
+			window      time.Duration
+			expected    float64
+		}{
+			{initialRate: 1000, rate: 1000, window: 100 * time.Millisecond, expected: 1007},
+		}
+		for i, c := range cases {
+			t.Run(fmt.Sprintf("%v", i), func(t *testing.T) {
+				res := multiplicativeIncrease(c.rate, c.window)
+				assert.InDelta(t, res, c.expected, 1)
+			})
+		}
+	})
+
+	t.Run("additiveIncrease", func(t *testing.T) {
+		cases := []struct {
+			initialRate int
+			rate        int
+			window      time.Duration
+			expected    int
+		}{
+			{initialRate: 1000, rate: 1000, window: 100 * time.Millisecond, expected: 2000},
+			{initialRate: 1_000_000, rate: 1_500_000, window: 100 * time.Millisecond, expected: 1_500_000 + 2083},
+		}
+		for i, c := range cases {
+			t.Run(fmt.Sprintf("%v", i), func(t *testing.T) {
+				res := additiveIncrease(c.rate, 100*time.Millisecond, c.window)
+				assert.InDelta(t, res, c.expected, 1)
+			})
+		}
+	})
+}