Clean up delay based bwe and use less goroutines

Reduces the number of goroutines that is spawned for a process that
always runs sequentially anyway. Additionally cleans up and moves some
code to more appropriate files.

Author: mengelbart

pkg/gcc/arrival_group_accumulator.go

@@ -20,39 +20,49 @@ func newArrivalGroupAccumulator() *arrivalGroupAccumulator {
 	}
 }
 
-func (a *arrivalGroupAccumulator) run(in <-chan cc.Acknowledgment) <-chan arrivalGroup {
-	out := make(chan arrivalGroup)
-	go func() {
-		init := false
-		group := arrivalGroup{}
-		for next := range in {
-			if !init {
+func (a *arrivalGroupAccumulator) run(in <-chan cc.Acknowledgment, agWriter func(arrivalGroup)) {
+	init := false
+	group := arrivalGroup{}
+	for next := range in {
+		if !init {
+			group.add(next)
+			init = true
+			continue
+		}
+		if next.Arrival.Before(group.arrival) {
+			// ignore out of order arrivals
+			continue
+		}
+		if next.Departure.After(group.departure) {
+			if interDepartureTimePkt(group, next) <= a.interDepartureThreshold {
 				group.add(next)
-				init = true
-				continue
-			}
-			if next.Arrival.Before(group.arrival) {
-				// ignore out of order arrivals
 				continue
 			}
-			if next.Departure.After(group.departure) {
-				if interDepartureTimePkt(group, next) <= a.interDepartureThreshold {
-					group.add(next)
-					continue
-				}
-
-				if interArrivalTimePkt(group, next) <= a.interArrivalThreshold &&
-					interGroupDelayVariationPkt(group, next) < a.interGroupDelayVariationTreshold {
-					group.add(next)
-					continue
-				}
-
-				out <- group
-				group = arrivalGroup{}
+
+			if interArrivalTimePkt(group, next) <= a.interArrivalThreshold &&
+				interGroupDelayVariationPkt(group, next) < a.interGroupDelayVariationTreshold {
 				group.add(next)
+				continue
 			}
+
+			agWriter(group)
+			group = arrivalGroup{}
+			group.add(next)
 		}
-		close(out)
-	}()
-	return out
+	}
+}
+
+func interArrivalTimePkt(a arrivalGroup, b cc.Acknowledgment) time.Duration {
+	return b.Arrival.Sub(a.arrival)
+}
+
+func interDepartureTimePkt(a arrivalGroup, b cc.Acknowledgment) time.Duration {
+	if len(a.packets) == 0 {
+		return 0
+	}
+	return b.Departure.Sub(a.packets[len(a.packets)-1].Departure)
+}
+
+func interGroupDelayVariationPkt(a arrivalGroup, b cc.Acknowledgment) time.Duration {
+	return b.Arrival.Sub(a.arrival) - b.Departure.Sub(a.departure)
 }

pkg/gcc/arrival_group_accumulator_test.go

@@ -160,7 +160,13 @@ func TestArrivalGroupAccumulator(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			aga := newArrivalGroupAccumulator()
 			in := make(chan cc.Acknowledgment)
-			out := aga.run(in)
+			out := make(chan arrivalGroup)
+			go func() {
+				defer close(out)
+				aga.run(in, func(ag arrivalGroup) {
+					out <- ag
+				})
+			}()
 			go func() {
 				for _, as := range tc.log {
 					in <- as

pkg/gcc/delay_based_bwe.go

@@ -7,11 +7,6 @@ import (
 	"github.com/pion/interceptor/internal/cc"
 )
 
-const (
-	decreaseEMAAlpha = 0.95
-	beta             = 0.85
-)
-
 // DelayStats contains some internal statistics of the delay based congestion
 // controller
 type DelayStats struct {
@@ -26,23 +21,15 @@ type DelayStats struct {
 	RTT           time.Duration
 }
 
-type estimator interface {
-	updateEstimate(measurement time.Duration) time.Duration
-}
-
-type estimatorFunc func(time.Duration) time.Duration
-
-func (f estimatorFunc) updateEstimate(d time.Duration) time.Duration {
-	return f(d)
-}
-
 type now func() time.Time
 
 type delayController struct {
 	ackPipe     chan<- cc.Acknowledgment
 	ackRatePipe chan<- cc.Acknowledgment
 	ackRTTPipe  chan<- []cc.Acknowledgment
 
+	*arrivalGroupAccumulator
+
 	onUpdateCallback func(DelayStats)
 
 	wg sync.WaitGroup
@@ -61,28 +48,39 @@ func newDelayController(c delayControllerConfig) *delayController {
 	ackRTTPipe := make(chan []cc.Acknowledgment)
 
 	delayController := &delayController{
-		ackPipe:     ackPipe,
-		ackRatePipe: ackRatePipe,
-		ackRTTPipe:  ackRTTPipe,
-		wg:          sync.WaitGroup{},
+		ackPipe:                 ackPipe,
+		ackRatePipe:             ackRatePipe,
+		ackRTTPipe:              ackRTTPipe,
+		arrivalGroupAccumulator: nil,
+		onUpdateCallback:        nil,
+		wg:                      sync.WaitGroup{},
 	}
 
+	rateController := newRateController(c.nowFn, c.initialBitrate, c.minBitrate, c.maxBitrate, func(ds DelayStats) {
+		if delayController.onUpdateCallback != nil {
+			delayController.onUpdateCallback(ds)
+		}
+	})
+	overuseDetector := newOveruseDetector(newAdaptiveThreshold(), 10*time.Millisecond, rateController.onDelayStats)
+	slopeEstimator := newSlopeEstimator(newKalman(), overuseDetector.onDelayStats)
+	arrivalGroupAccumulator := newArrivalGroupAccumulator()
+
 	rc := newRateCalculator(500 * time.Millisecond)
 	re := newRTTEstimator()
 
-	reaceivedRate := rc.run(ackRatePipe)
-	rtt := re.run(ackRTTPipe)
-
-	arrivalGroupAccumulator := newArrivalGroupAccumulator()
-	slopeEstimator := newSlopeEstimator(newKalman())
-	overuseDetector := newOveruseDetector(newAdaptiveThreshold(), 10*time.Millisecond)
-	rateController := newRateController(c.nowFn, c.initialBitrate, c.minBitrate, c.maxBitrate)
-
-	arrival := arrivalGroupAccumulator.run(ackPipe)
-	estimate := slopeEstimator.run(arrival)
-	state := overuseDetector.run(estimate)
-	delayStats := rateController.run(state, reaceivedRate, rtt)
-	delayController.loop(delayStats)
+	delayController.wg.Add(3)
+	go func() {
+		defer delayController.wg.Done()
+		arrivalGroupAccumulator.run(ackPipe, slopeEstimator.onArrivalGroup)
+	}()
+	go func() {
+		defer delayController.wg.Done()
+		rc.run(ackRatePipe, rateController.onReceivedRate)
+	}()
+	go func() {
+		defer delayController.wg.Done()
+		re.run(ackRTTPipe, rateController.onRTT)
+	}()
 
 	return delayController
 }
@@ -99,42 +97,12 @@ func (d *delayController) updateDelayEstimate(acks []cc.Acknowledgment) {
 	d.ackRTTPipe <- acks
 }
 
-func (d *delayController) loop(in chan DelayStats) {
-	d.wg.Add(1)
-	defer d.wg.Done()
-
-	go func() {
-		for next := range in {
-			if d.onUpdateCallback != nil {
-				d.onUpdateCallback(next)
-			}
-		}
-	}()
-}
-
 func (d *delayController) Close() error {
+	defer d.wg.Wait()
+
 	close(d.ackPipe)
 	close(d.ackRTTPipe)
 	close(d.ackRatePipe)
-	d.wg.Wait()
-	return nil
-}
-
-func interArrivalTimePkt(a arrivalGroup, b cc.Acknowledgment) time.Duration {
-	return b.Arrival.Sub(a.arrival)
-}
 
-func interDepartureTimePkt(a arrivalGroup, b cc.Acknowledgment) time.Duration {
-	if len(a.packets) == 0 {
-		return 0
-	}
-	return b.Departure.Sub(a.packets[len(a.packets)-1].Departure)
-}
-
-func interGroupDelayVariationPkt(a arrivalGroup, b cc.Acknowledgment) time.Duration {
-	return b.Arrival.Sub(a.arrival) - b.Departure.Sub(a.departure)
-}
-
-func interGroupDelayVariation(a, b arrivalGroup) time.Duration {
-	return b.arrival.Sub(a.arrival) - b.departure.Sub(a.departure)
+	return nil
 }

pkg/gcc/overuse_detector.go

@@ -11,69 +11,69 @@ type threshold interface {
 type overuseDetector struct {
 	threshold   threshold
 	overuseTime time.Duration
+
+	dsWriter func(DelayStats)
+
+	lastEstimate       time.Duration
+	lastUpdate         time.Time
+	increasingDuration time.Duration
+	increasingCounter  int
 }
 
-func newOveruseDetector(thresh threshold, overuseTime time.Duration) *overuseDetector {
+func newOveruseDetector(thresh threshold, overuseTime time.Duration, dsw func(DelayStats)) *overuseDetector {
 	return &overuseDetector{
-		threshold:   thresh,
-		overuseTime: overuseTime,
+		threshold:          thresh,
+		overuseTime:        overuseTime,
+		dsWriter:           dsw,
+		lastEstimate:       0,
+		lastUpdate:         time.Now(),
+		increasingDuration: 0,
+		increasingCounter:  0,
 	}
 }
 
-func (d *overuseDetector) run(in <-chan DelayStats) <-chan DelayStats {
-	out := make(chan DelayStats)
-	go func() {
-		lastEstimate := 0 * time.Millisecond
-		lastUpdate := time.Now()
-		var increasingDuration time.Duration
-		var increasingCounter int
-
-		for next := range in {
-			now := time.Now()
-			delta := now.Sub(lastUpdate)
-			lastUpdate = now
+func (d *overuseDetector) onDelayStats(ds DelayStats) {
+	now := time.Now()
+	delta := now.Sub(d.lastUpdate)
+	d.lastUpdate = now
 
-			thresholdUse, estimate, currentThreshold := d.threshold.compare(next.Estimate, next.lastReceiveDelta)
+	thresholdUse, estimate, currentThreshold := d.threshold.compare(ds.Estimate, ds.lastReceiveDelta)
 
-			use := usageNormal
-			if thresholdUse == usageOver {
-				if increasingDuration == 0 {
-					increasingDuration = delta / 2
-				} else {
-					increasingDuration += delta
-				}
-				increasingCounter++
-				if increasingDuration > d.overuseTime && increasingCounter > 1 {
-					if estimate > lastEstimate {
-						use = usageOver
-					}
-				}
-			}
-			if thresholdUse == usageUnder {
-				increasingCounter = 0
-				increasingDuration = 0
-				use = usageUnder
+	use := usageNormal
+	if thresholdUse == usageOver {
+		if d.increasingDuration == 0 {
+			d.increasingDuration = delta / 2
+		} else {
+			d.increasingDuration += delta
+		}
+		d.increasingCounter++
+		if d.increasingDuration > d.overuseTime && d.increasingCounter > 1 {
+			if estimate > d.lastEstimate {
+				use = usageOver
 			}
+		}
+	}
+	if thresholdUse == usageUnder {
+		d.increasingCounter = 0
+		d.increasingDuration = 0
+		use = usageUnder
+	}
 
-			if thresholdUse == usageNormal {
-				increasingDuration = 0
-				increasingCounter = 0
-				use = usageNormal
-			}
-			lastEstimate = estimate
+	if thresholdUse == usageNormal {
+		d.increasingDuration = 0
+		d.increasingCounter = 0
+		use = usageNormal
+	}
+	d.lastEstimate = estimate
 
-			out <- DelayStats{
-				Measurement:      next.Measurement,
-				Estimate:         estimate,
-				Threshold:        currentThreshold,
-				lastReceiveDelta: delta,
-				Usage:            use,
-				State:            0,
-				TargetBitrate:    0,
-				RTT:              0,
-			}
-		}
-		close(out)
-	}()
-	return out
+	d.dsWriter(DelayStats{
+		Measurement:      ds.Measurement,
+		Estimate:         estimate,
+		Threshold:        currentThreshold,
+		lastReceiveDelta: delta,
+		Usage:            use,
+		State:            0,
+		TargetBitrate:    0,
+		RTT:              0,
+	})
 }

pkg/gcc/overuse_detector_test.go

@@ -86,15 +86,17 @@ func TestOveruseDetectorWithoutDelay(t *testing.T) {
 	for _, tc := range cases {
 		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
-			od := newOveruseDetector(tc.thresh, tc.delay)
-			in := make(chan DelayStats)
-			out := od.run(in)
+			out := make(chan DelayStats)
+			dsw := func(ds DelayStats) {
+				out <- ds
+			}
+			od := newOveruseDetector(tc.thresh, tc.delay, dsw)
 			go func() {
+				defer close(out)
 				for _, e := range tc.estimates {
-					in <- e
+					od.onDelayStats(e)
 					time.Sleep(tc.delay)
 				}
-				close(in)
 			}()
 			received := []usage{}
 			for s := range out {