Repository: pion/interceptor
Pull request #146 (2022-07-16 15:41:24)

Clean up delay based bwe and use less goroutines
Reduces the number of goroutines that are spawned for a process that always runs sequentially anyway. Additionally, cleans up and moves some code to more appropriate files.

codecov[bot]: # [Codecov](https://codecov.io/gh/pion/interceptor/pull/146?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) Report
> Merging [#146](https://codecov.io/gh/pion/interceptor/pull/146?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) (0120bf6) into [master](https://codecov.io/gh/pion/interceptor/commit/b0b785b9dfd341ca692c5b3e89e97d7d1a863051?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) (b0b785b) will **decrease** coverage by `0.25%`.
> The diff coverage is `89.91%`.

> :exclamation: Current head 0120bf6 differs from pull request most recent head f7029ad. Consider uploading reports for the commit f7029ad to get more accurate results

```diff
@@            Coverage Diff             @@
##           master     #146      +/-   ##
==========================================
- Coverage   79.19%   78.94%   -0.26%     
==========================================
  Files          51       51              
  Lines        2519     2503      -16     
==========================================
- Hits         1995     1976      -19     
- Misses        425      427       +2     
- Partials       99      100       +1     
```

| Flag | Coverage Δ | |
|---|---|---|
| go | `78.94% <89.91%> (-0.26%)` | :arrow_down: |
| wasm | `?` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/pion/interceptor/pull/146?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | Coverage Δ | |
|---|---|---|
| [pkg/gcc/rate\_controller.go](https://codecov.io/gh/pion/interceptor/pull/146/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2djYy9yYXRlX2NvbnRyb2xsZXIuZ28=) | `58.77% <74.57%> (-1.23%)` | :arrow_down: |
| [pkg/gcc/rtt\_estimator.go](https://codecov.io/gh/pion/interceptor/pull/146/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2djYy9ydHRfZXN0aW1hdG9yLmdv) | `87.50% <83.33%> (-1.79%)` | :arrow_down: |
| [pkg/gcc/arrival\_group\_accumulator.go](https://codecov.io/gh/pion/interceptor/pull/146/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2djYy9hcnJpdmFsX2dyb3VwX2FjY3VtdWxhdG9yLmdv) | `91.89% <89.28%> (-8.11%)` | :arrow_down: |
| [pkg/gcc/rate\_calculator.go](https://codecov.io/gh/pion/interceptor/pull/146/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2djYy9yYXRlX2NhbGN1bGF0b3IuZ28=) | `91.89% <90.62%> (-0.80%)` | :arrow_down: |
| [pkg/gcc/delay\_based\_bwe.go](https://codecov.io/gh/pion/interceptor/pull/146/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2djYy9kZWxheV9iYXNlZF9id2UuZ28=) | `100.00% <100.00%> (+4.54%)` | :arrow_up: |
| [pkg/gcc/overuse\_detector.go](https://codecov.io/gh/pion/interceptor/pull/146/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2djYy9vdmVydXNlX2RldGVjdG9yLmdv) | `100.00% <100.00%> (ø)` | |
| [pkg/gcc/slope\_estimator.go](https://codecov.io/gh/pion/interceptor/pull/146/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2djYy9zbG9wZV9lc3RpbWF0b3IuZ28=) | `100.00% <100.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pion/interceptor/pull/146?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/pion/interceptor/pull/146?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion). Last update [b0b785b...f7029ad](https://codecov.io/gh/pion/interceptor/pull/146?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).


Patch:
diff --git a/pkg/gcc/arrival_group_accumulator.go b/pkg/gcc/arrival_group_accumulator.go
index ece28cc..24a650b 100644
--- a/pkg/gcc/arrival_group_accumulator.go
+++ b/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)
 }
diff --git a/pkg/gcc/delay_based_bwe.go b/pkg/gcc/delay_based_bwe.go
index 7b848cb..61c1274 100644
--- a/pkg/gcc/delay_based_bwe.go
+++ b/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,16 +21,6 @@ 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 {
@@ -43,6 +28,8 @@ type delayController struct {
 	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
 }
diff --git a/pkg/gcc/overuse_detector.go b/pkg/gcc/overuse_detector.go
index 467b641..40fcdb3 100644
--- a/pkg/gcc/overuse_detector.go
+++ b/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,
+	})
 }
diff --git a/pkg/gcc/rate_calculator.go b/pkg/gcc/rate_calculator.go
index 56dbe51..838c05e 100644
--- a/pkg/gcc/rate_calculator.go
+++ b/pkg/gcc/rate_calculator.go
@@ -16,49 +16,44 @@ func newRateCalculator(window time.Duration) *rateCalculator {
 	}
 }
 
-func (c *rateCalculator) run(in <-chan cc.Acknowledgment) <-chan int {
-	out := make(chan int)
-	go func() {
-		var history []cc.Acknowledgment
-		init := false
-		sum := 0
-		for next := range in {
-			if next.Arrival.IsZero() {
-				// Ignore packet if it didn't arrive
-				continue
-			}
-			history = append(history, next)
-			sum += next.Size
+func (c *rateCalculator) run(in <-chan cc.Acknowledgment, onRateUpdate func(int)) {
+	var history []cc.Acknowledgment
+	init := false
+	sum := 0
+	for next := range in {
+		if next.Arrival.IsZero() {
+			// Ignore packet if it didn't arrive
+			continue
+		}
+		history = append(history, next)
+		sum += next.Size
 
-			if !init {
-				init = true
-				// Don't know any timeframe here, only arrival of last packet,
-				// which is by definition in the window that ends with the last
-				// arrival time
-				out <- next.Size * 8
-				continue
-			}
+		if !init {
+			init = true
+			// Don't know any timeframe here, only arrival of last packet,
+			// which is by definition in the window that ends with the last
+			// arrival time
+			onRateUpdate(next.Size * 8)
+			continue
+		}
 
-			del := 0
-			for _, ack := range history {
-				deadline := next.Arrival.Add(-c.window)
-				if !ack.Arrival.Before(deadline) {
-					break
-				}
-				del++
-				sum -= ack.Size
+		del := 0
+		for _, ack := range history {
+			deadline := next.Arrival.Add(-c.window)
+			if !ack.Arrival.Before(deadline) {
+				break
 			}
-			history = history[del:]
-			if len(history) == 0 {
-				out <- 0
-				continue
-			}
-			dt := next.Arrival.Sub(history[0].Arrival)
-			bits := 8 * sum
-			rate := int(float64(bits) / dt.Seconds())
-			out <- rate
+			del++
+			sum -= ack.Size
+		}
+		history = history[del:]
+		if len(history) == 0 {
+			onRateUpdate(0)
+			continue
 		}
-		close(out)
-	}()
-	return out
+		dt := next.Arrival.Sub(history[0].Arrival)
+		bits := 8 * sum
+		rate := int(float64(bits) / dt.Seconds())
+		onRateUpdate(rate)
+	}
 }
diff --git a/pkg/gcc/rate_controller.go b/pkg/gcc/rate_controller.go
index 9eaab67..ed75fda 100644
--- a/pkg/gcc/rate_controller.go
+++ b/pkg/gcc/rate_controller.go
@@ -2,11 +2,17 @@ package gcc
 
 import (
 	"math"
+	"sync"
 	"time"
 
 	"github.com/pion/logging"
 )
 
+const (
+	decreaseEMAAlpha = 0.95
+	beta             = 0.85
+)
+
 type rateController struct {
 	log                  logging.LeveledLogger
 	now                  now
@@ -14,6 +20,11 @@ type rateController struct {
 	minBitrate           int
 	maxBitrate           int
 
+	dsWriter func(DelayStats)
+
+	lock               sync.Mutex
+	init               bool
+	delayStats         DelayStats
 	target             int
 	lastUpdate         time.Time
 	lastState          state
@@ -39,13 +50,16 @@ func (a *exponentialMovingAverage) update(value float64) {
 	}
 }
 
-func newRateController(now now, initialTargetBitrate, minBitrate, maxBitrate int) *rateController {
+func newRateController(now now, initialTargetBitrate, minBitrate, maxBitrate int, dsw func(DelayStats)) *rateController {
 	return &rateController{
 		log:                  logging.NewDefaultLoggerFactory().NewLogger("gcc_rate_controller"),
 		now:                  now,
 		initialTargetBitrate: initialTargetBitrate,
 		minBitrate:           minBitrate,
 		maxBitrate:           maxBitrate,
+		dsWriter:             dsw,
+		init:                 false,
+		delayStats:           DelayStats{},
 		target:               initialTargetBitrate,
 		lastUpdate:           time.Time{},
 		lastState:            stateIncrease,
@@ -55,68 +69,71 @@ func newRateController(now now, initialTargetBitrate, minBitrate, maxBitrate int
 	}
 }
 
-func (c *rateController) run(in <-chan DelayStats, receivedRate <-chan int, rtt <-chan time.Duration) chan DelayStats {
-	out := make(chan DelayStats)
-	go func() {
-		c.lastUpdate = c.now()
-
-		defer func() {
-			close(out)
-		}()
-
-		var latestStats DelayStats
-		init := false
-
-		for {
-			select {
-			case c.latestReceivedRate = <-receivedRate:
-			case c.latestRTT = <-rtt:
-			case nextStats, ok := <-in:
-				if !ok {
-					return
-				}
-				if !init {
-					init = true
-					latestStats = nextStats
-					latestStats.State = stateIncrease
-					continue
-				}
-				latestStats = nextStats
-				latestStats.State = latestStats.State.transition(nextStats.Usage)
-
-				now := time.Now()
-				switch latestStats.State {
-				case stateHold:
-				case stateIncrease:
-					c.target = clampInt(c.increase(now), c.minBitrate, c.maxBitrate)
-					out <- DelayStats{
-						Measurement:      latestStats.Measurement,
-						Estimate:         latestStats.Estimate,
-						Threshold:        latestStats.Threshold,
-						lastReceiveDelta: latestStats.lastReceiveDelta,
-						Usage:            latestStats.Usage,
-						State:            latestStats.State,
-						TargetBitrate:    c.target,
-						RTT:              c.latestRTT,
-					}
-
-				case stateDecrease:
-					c.target = clampInt(c.decrease(), c.minBitrate, c.maxBitrate)
-					out <- DelayStats{
-						Measurement:      latestStats.Measurement,
-						Estimate:         latestStats.Estimate,
-						Threshold:        latestStats.Threshold,
-						lastReceiveDelta: latestStats.lastReceiveDelta,
-						Usage:            latestStats.Usage,
-						State:            latestStats.State,
-						TargetBitrate:    c.target,
-						RTT:              c.latestRTT,
-					}
-				}
-			}
+func (c *rateController) onReceivedRate(rate int) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	c.latestReceivedRate = rate
+}
+
+func (c *rateController) onRTT(rtt time.Duration) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	c.latestRTT = rtt
+}
+
+func (c *rateController) onDelayStats(ds DelayStats) {
+	now := time.Now()
+
+	if !c.init {
+		c.delayStats = ds
+		c.delayStats.State = stateIncrease
+		c.init = true
+		return
+	}
+	c.delayStats = ds
+	c.delayStats.State = c.delayStats.State.transition(ds.Usage)
+
+	if c.delayStats.State == stateHold {
+		return
+	}
+
+	var next DelayStats
+
+	c.lock.Lock()
+
+	switch c.delayStats.State {
+	case stateHold:
+		// should never occur due to check above, but makes the linter happy
+	case stateIncrease:
+		c.target = clampInt(c.increase(now), c.minBitrate, c.maxBitrate)
+		next = DelayStats{
+			Measurement:      c.delayStats.Measurement,
+			Estimate:         c.delayStats.Estimate,
+			Threshold:        c.delayStats.Threshold,
+			lastReceiveDelta: c.delayStats.lastReceiveDelta,
+			Usage:            c.delayStats.Usage,
+			State:            c.delayStats.State,
+			TargetBitrate:    c.target,
+			RTT:              c.latestRTT,
+		}
+
+	case stateDecrease:
+		c.target = clampInt(c.decrease(), c.minBitrate, c.maxBitrate)
+		next = DelayStats{
+			Measurement:      c.delayStats.Measurement,
+			Estimate:         c.delayStats.Estimate,
+			Threshold:        c.delayStats.Threshold,
+			lastReceiveDelta: c.delayStats.lastReceiveDelta,
+			Usage:            c.delayStats.Usage,
+			State:            c.delayStats.State,
+			TargetBitrate:    c.target,
+			RTT:              c.latestRTT,
 		}
-	}()
-	return out
+	}
+
+	c.lock.Unlock()
+
+	c.dsWriter(next)
 }
 
 func (c *rateController) increase(now time.Time) int {
diff --git a/pkg/gcc/rtt_estimator.go b/pkg/gcc/rtt_estimator.go
index b2275e0..4f9386f 100644
--- a/pkg/gcc/rtt_estimator.go
+++ b/pkg/gcc/rtt_estimator.go
@@ -17,33 +17,28 @@ func newRTTEstimator() *rttEstimator {
 	}
 }
 
-func (e *rttEstimator) run(in <-chan []cc.Acknowledgment) <-chan time.Duration {
-	out := make(chan time.Duration)
-	go func() {
-		history := []time.Duration{}
-		for acks := range in {
-			if len(acks) == 0 {
-				continue
-			}
-			minRTT := time.Duration(math.MaxInt64)
-			for _, ack := range acks {
-				if ack.RTT < minRTT {
-					minRTT = ack.RTT
-				}
-			}
-			history = append(history, minRTT)
-			if len(history) >= e.samples {
-				history = history[len(history)-e.samples:]
+func (e *rttEstimator) run(in <-chan []cc.Acknowledgment, onRTTUpdate func(time.Duration)) {
+	history := []time.Duration{}
+	for acks := range in {
+		if len(acks) == 0 {
+			continue
+		}
+		minRTT := time.Duration(math.MaxInt64)
+		for _, ack := range acks {
+			if ack.RTT < minRTT {
+				minRTT = ack.RTT
 			}
+		}
+		history = append(history, minRTT)
+		if len(history) >= e.samples {
+			history = history[len(history)-e.samples:]
+		}
 
-			sum := time.Duration(0)
-			for _, rtt := range history {
-				sum += rtt
-			}
-			rtt := float64(sum.Milliseconds()) / float64(len(history)) * float64(time.Millisecond)
-			out <- time.Duration(rtt)
+		sum := time.Duration(0)
+		for _, rtt := range history {
+			sum += rtt
 		}
-		close(out)
-	}()
-	return out
+		rtt := float64(sum.Milliseconds()) / float64(len(history)) * float64(time.Millisecond)
+		onRTTUpdate(time.Duration(rtt))
+	}
 }
diff --git a/pkg/gcc/slope_estimator.go b/pkg/gcc/slope_estimator.go
index 4ac6c59..cd417b8 100644
--- a/pkg/gcc/slope_estimator.go
+++ b/pkg/gcc/slope_estimator.go
@@ -1,41 +1,52 @@
 package gcc
 
+import "time"
+
+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 slopeEstimator struct {
 	estimator
+	init             bool
+	group            arrivalGroup
+	delayStatsWriter func(DelayStats)
 }
 
-func newSlopeEstimator(e estimator) *slopeEstimator {
+func newSlopeEstimator(e estimator, dsw func(DelayStats)) *slopeEstimator {
 	return &slopeEstimator{
-		estimator: e,
+		estimator:        e,
+		delayStatsWriter: dsw,
+	}
+}
+
+func (e *slopeEstimator) onArrivalGroup(ag arrivalGroup) {
+	if !e.init {
+		e.group = ag
+		e.init = true
+		return
 	}
+	measurement := interGroupDelayVariation(e.group, ag)
+	delta := ag.arrival.Sub(e.group.arrival)
+	e.group = ag
+	e.delayStatsWriter(DelayStats{
+		Measurement:      measurement,
+		Estimate:         e.updateEstimate(measurement),
+		Threshold:        0,
+		lastReceiveDelta: delta,
+		Usage:            0,
+		State:            0,
+		TargetBitrate:    0,
+		RTT:              0,
+	})
 }
 
-func (e *slopeEstimator) run(in <-chan arrivalGroup) <-chan DelayStats {
-	out := make(chan DelayStats)
-	go func() {
-		init := false
-		var last arrivalGroup
-		for next := range in {
-			if !init {
-				last = next
-				init = true
-				continue
-			}
-			measurement := interGroupDelayVariation(last, next)
-			delta := next.arrival.Sub(last.arrival)
-			last = next
-			out <- DelayStats{
-				Measurement:      measurement,
-				Estimate:         e.updateEstimate(measurement),
-				Threshold:        0,
-				lastReceiveDelta: delta,
-				Usage:            0,
-				State:            0,
-				TargetBitrate:    0,
-				RTT:              0,
-			}
-		}
-		close(out)
-	}()
-	return out
+func interGroupDelayVariation(a, b arrivalGroup) time.Duration {
+	return b.arrival.Sub(a.arrival) - b.departure.Sub(a.departure)
 }


Test patch:
diff --git a/pkg/gcc/arrival_group_accumulator_test.go b/pkg/gcc/arrival_group_accumulator_test.go
index b0d45ee..f177357 100644
--- a/pkg/gcc/arrival_group_accumulator_test.go
+++ b/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
diff --git a/pkg/gcc/overuse_detector_test.go b/pkg/gcc/overuse_detector_test.go
index dca1737..b395cf3 100644
--- a/pkg/gcc/overuse_detector_test.go
+++ b/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 {
diff --git a/pkg/gcc/rate_calculator_test.go b/pkg/gcc/rate_calculator_test.go
index e08ff01..860bc6c 100644
--- a/pkg/gcc/rate_calculator_test.go
+++ b/pkg/gcc/rate_calculator_test.go
@@ -80,11 +80,16 @@ func TestRateCalculator(t *testing.T) {
 	for _, tc := range cases {
 		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
-			rc := rateCalculator{
-				window: 500 * time.Millisecond,
-			}
+			rc := newRateCalculator(500 * time.Millisecond)
 			in := make(chan cc.Acknowledgment)
-			out := rc.run(in)
+			out := make(chan int)
+			onRateUpdate := func(rate int) {
+				out <- rate
+			}
+			go func() {
+				defer close(out)
+				rc.run(in, onRateUpdate)
+			}()
 			go func() {
 				for _, ack := range tc.acks {
 					in <- ack
diff --git a/pkg/gcc/rate_controller_test.go b/pkg/gcc/rate_controller_test.go
index 6225893..b1c6d37 100644
--- a/pkg/gcc/rate_controller_test.go
+++ b/pkg/gcc/rate_controller_test.go
@@ -43,16 +43,17 @@ func TestRateControllerRun(t *testing.T) {
 	for _, tc := range cases {
 		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
-			dc := newRateController(mockNoFn, 100_000, 1_000, 50_000_000)
+			out := make(chan DelayStats)
+			dc := newRateController(mockNoFn, 100_000, 1_000, 50_000_000, func(ds DelayStats) {
+				out <- ds
+			})
 			in := make(chan DelayStats)
-			receivedRate := make(chan int)
-			rtt := make(chan time.Duration)
-			out := dc.run(in, receivedRate, rtt)
-			receivedRate <- 100_000
-			rtt <- 300 * time.Millisecond
+			dc.onReceivedRate(100_000)
+			dc.onRTT(300 * time.Millisecond)
 			go func() {
+				defer close(out)
 				for _, state := range tc.usage {
-					in <- DelayStats{
+					dc.onDelayStats(DelayStats{
 						Measurement:   0,
 						Estimate:      0,
 						Threshold:     0,
@@ -60,9 +61,8 @@ func TestRateControllerRun(t *testing.T) {
 						State:         0,
 						TargetBitrate: 0,
 						RTT:           0,
-					}
+					})
 				}
-				time.Sleep(2 * time.Second)
 				close(in)
 			}()
 			received := []DelayStats{}
diff --git a/pkg/gcc/rtt_estimator_test.go b/pkg/gcc/rtt_estimator_test.go
index 3fd792d..61bc9d9 100644
--- a/pkg/gcc/rtt_estimator_test.go
+++ b/pkg/gcc/rtt_estimator_test.go
@@ -57,7 +57,13 @@ func TestRTTEstimator(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			re := newRTTEstimator()
 			in := make(chan []cc.Acknowledgment)
-			out := re.run(in)
+			out := make(chan time.Duration)
+			go func() {
+				defer close(out)
+				re.run(in, func(d time.Duration) {
+					out <- d
+				})
+			}()
 			go func() {
 				for _, acks := range tc.ackLists {
 					in <- acks
diff --git a/pkg/gcc/slope_estimator_test.go b/pkg/gcc/slope_estimator_test.go
index a92bf62..8130521 100644
--- a/pkg/gcc/slope_estimator_test.go
+++ b/pkg/gcc/slope_estimator_test.go
@@ -91,15 +91,16 @@ func TestSlopeEstimator(t *testing.T) {
 	for _, tc := range cases {
 		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
-			se := newSlopeEstimator(estimatorFunc(identity))
-			in := make(chan arrivalGroup)
-			out := se.run(in)
+			out := make(chan DelayStats)
+			se := newSlopeEstimator(estimatorFunc(identity), func(ds DelayStats) {
+				out <- ds
+			})
 			input := []time.Duration{}
 			go func() {
+				defer close(out)
 				for _, ag := range tc.ags {
-					in <- ag
+					se.onArrivalGroup(ag)
 				}
-				close(in)
 			}()
 			received := []DelayStats{}
 			for d := range out {