Repository: pion/interceptor
Pull request #147 (2022-07-16 15:53:09)

Feat/improve gcc writertcp performance
Improves performance of `WriteRTCP` by moving the loop to the worker goroutines. This includes #142  (and solves #141) but depends on #146.

codecov[bot]: # [Codecov](https://codecov.io/gh/pion/interceptor/pull/147?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) Report
> Merging [#147](https://codecov.io/gh/pion/interceptor/pull/147?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) (f6e9b17) 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.57%`.
> The diff coverage is `88.34%`.

> :exclamation: Current head f6e9b17 differs from pull request most recent head 6894efd. Consider uploading reports for the commit 6894efd to get more accurate results

```diff
@@            Coverage Diff             @@
##           master     #147      +/-   ##
==========================================
- Coverage   79.19%   78.62%   -0.58%     
==========================================
  Files          51       51              
  Lines        2519     2503      -16     
==========================================
- Hits         1995     1968      -27     
- Misses        425      433       +8     
- Partials       99      102       +3     
```

| Flag | Coverage Δ | |
|---|---|---|
| go | `78.62% <88.34%> (-0.58%)` | :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/147?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | Coverage Δ | |
|---|---|---|
| [pkg/gcc/arrival\_group\_accumulator.go](https://codecov.io/gh/pion/interceptor/pull/147/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2djYy9hcnJpdmFsX2dyb3VwX2FjY3VtdWxhdG9yLmdv) | `78.94% <66.66%> (-21.06%)` | :arrow_down: |
| [pkg/gcc/rate\_controller.go](https://codecov.io/gh/pion/interceptor/pull/147/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/147/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/rate\_calculator.go](https://codecov.io/gh/pion/interceptor/pull/147/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2djYy9yYXRlX2NhbGN1bGF0b3IuZ28=) | `92.10% <88.88%> (-0.58%)` | :arrow_down: |
| [pkg/gcc/delay\_based\_bwe.go](https://codecov.io/gh/pion/interceptor/pull/147/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/147/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/147/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%> (ø)` | |
| [pkg/gcc/kalman.go](https://codecov.io/gh/pion/interceptor/pull/147/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2djYy9rYWxtYW4uZ28=) | `94.33% <0.00%> (-5.67%)` | :arrow_down: |
| ... and [2 more](https://codecov.io/gh/pion/interceptor/pull/147/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pion/interceptor/pull/147?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/147?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion). Last update [b0b785b...6894efd](https://codecov.io/gh/pion/interceptor/pull/147?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 24a650b..9d13b55 100644
--- a/pkg/gcc/arrival_group_accumulator.go
+++ b/pkg/gcc/arrival_group_accumulator.go
@@ -20,34 +20,36 @@ func newArrivalGroupAccumulator() *arrivalGroupAccumulator {
 	}
 }
 
-func (a *arrivalGroupAccumulator) run(in <-chan cc.Acknowledgment, agWriter func(arrivalGroup)) {
+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 {
+	for acks := range in {
+		for _, next := range acks {
+			if !init {
 				group.add(next)
+				init = true
 				continue
 			}
-
-			if interArrivalTimePkt(group, next) <= a.interArrivalThreshold &&
-				interGroupDelayVariationPkt(group, next) < a.interGroupDelayVariationTreshold {
-				group.add(next)
+			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
+				}
 
-			agWriter(group)
-			group = arrivalGroup{}
-			group.add(next)
+				if interArrivalTimePkt(group, next) <= a.interArrivalThreshold &&
+					interGroupDelayVariationPkt(group, next) < a.interGroupDelayVariationTreshold {
+					group.add(next)
+					continue
+				}
+
+				agWriter(group)
+				group = arrivalGroup{}
+				group.add(next)
+			}
 		}
 	}
 }
diff --git a/pkg/gcc/delay_based_bwe.go b/pkg/gcc/delay_based_bwe.go
index 61c1274..a3a7655 100644
--- a/pkg/gcc/delay_based_bwe.go
+++ b/pkg/gcc/delay_based_bwe.go
@@ -24,8 +24,8 @@ type DelayStats struct {
 type now func() time.Time
 
 type delayController struct {
-	ackPipe     chan<- cc.Acknowledgment
-	ackRatePipe chan<- cc.Acknowledgment
+	ackPipe     chan<- []cc.Acknowledgment
+	ackRatePipe chan<- []cc.Acknowledgment
 	ackRTTPipe  chan<- []cc.Acknowledgment
 
 	*arrivalGroupAccumulator
@@ -43,8 +43,8 @@ type delayControllerConfig struct {
 }
 
 func newDelayController(c delayControllerConfig) *delayController {
-	ackPipe := make(chan cc.Acknowledgment)
-	ackRatePipe := make(chan cc.Acknowledgment)
+	ackPipe := make(chan []cc.Acknowledgment)
+	ackRatePipe := make(chan []cc.Acknowledgment)
 	ackRTTPipe := make(chan []cc.Acknowledgment)
 
 	delayController := &delayController{
@@ -90,10 +90,8 @@ func (d *delayController) onUpdate(f func(DelayStats)) {
 }
 
 func (d *delayController) updateDelayEstimate(acks []cc.Acknowledgment) {
-	for _, ack := range acks {
-		d.ackPipe <- ack
-		d.ackRatePipe <- ack
-	}
+	d.ackPipe <- acks
+	d.ackRatePipe <- acks
 	d.ackRTTPipe <- acks
 }
 
diff --git a/pkg/gcc/rate_calculator.go b/pkg/gcc/rate_calculator.go
index 838c05e..cdf736d 100644
--- a/pkg/gcc/rate_calculator.go
+++ b/pkg/gcc/rate_calculator.go
@@ -16,44 +16,46 @@ func newRateCalculator(window time.Duration) *rateCalculator {
 	}
 }
 
-func (c *rateCalculator) run(in <-chan cc.Acknowledgment, onRateUpdate func(int)) {
+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
+	for acks := range in {
+		for _, next := range acks {
+			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
-			onRateUpdate(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 := 0
+			for _, ack := range history {
+				deadline := next.Arrival.Add(-c.window)
+				if !ack.Arrival.Before(deadline) {
+					break
+				}
+				del++
+				sum -= ack.Size
 			}
-			del++
-			sum -= ack.Size
-		}
-		history = history[del:]
-		if len(history) == 0 {
-			onRateUpdate(0)
-			continue
+			history = history[del:]
+			if len(history) == 0 {
+				onRateUpdate(0)
+				continue
+			}
+			dt := next.Arrival.Sub(history[0].Arrival)
+			bits := 8 * sum
+			rate := int(float64(bits) / dt.Seconds())
+			onRateUpdate(rate)
 		}
-		dt := next.Arrival.Sub(history[0].Arrival)
-		bits := 8 * sum
-		rate := int(float64(bits) / dt.Seconds())
-		onRateUpdate(rate)
 	}
 }


Test patch:
diff --git a/pkg/gcc/arrival_group_accumulator_test.go b/pkg/gcc/arrival_group_accumulator_test.go
index f177357..a5b867b 100644
--- a/pkg/gcc/arrival_group_accumulator_test.go
+++ b/pkg/gcc/arrival_group_accumulator_test.go
@@ -159,7 +159,7 @@ func TestArrivalGroupAccumulator(t *testing.T) {
 		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
 			aga := newArrivalGroupAccumulator()
-			in := make(chan cc.Acknowledgment)
+			in := make(chan []cc.Acknowledgment)
 			out := make(chan arrivalGroup)
 			go func() {
 				defer close(out)
@@ -168,9 +168,7 @@ func TestArrivalGroupAccumulator(t *testing.T) {
 				})
 			}()
 			go func() {
-				for _, as := range tc.log {
-					in <- as
-				}
+				in <- tc.log
 				close(in)
 			}()
 			received := []arrivalGroup{}
diff --git a/pkg/gcc/rate_calculator_test.go b/pkg/gcc/rate_calculator_test.go
index 860bc6c..190a709 100644
--- a/pkg/gcc/rate_calculator_test.go
+++ b/pkg/gcc/rate_calculator_test.go
@@ -81,7 +81,7 @@ func TestRateCalculator(t *testing.T) {
 		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
 			rc := newRateCalculator(500 * time.Millisecond)
-			in := make(chan cc.Acknowledgment)
+			in := make(chan []cc.Acknowledgment)
 			out := make(chan int)
 			onRateUpdate := func(rate int) {
 				out <- rate
@@ -91,9 +91,7 @@ func TestRateCalculator(t *testing.T) {
 				rc.run(in, onRateUpdate)
 			}()
 			go func() {
-				for _, ack := range tc.acks {
-					in <- ack
-				}
+				in <- tc.acks
 				close(in)
 			}()
 
diff --git a/pkg/gcc/send_side_bwe_test.go b/pkg/gcc/send_side_bwe_test.go
index 39ce24c..34ddfa0 100644
--- a/pkg/gcc/send_side_bwe_test.go
+++ b/pkg/gcc/send_side_bwe_test.go
@@ -1,6 +1,8 @@
 package gcc
 
 import (
+	"fmt"
+	"math/rand"
 	"testing"
 
 	"github.com/pion/interceptor"
@@ -103,3 +105,41 @@ func TestSendSideBWE_ErrorOnWriteRTCPAtClosedState(t *testing.T) {
 	require.ErrorIs(t, bwe.WriteRTCP(pkts, nil), ErrSendSideBWEClosed)
 	require.Equal(t, bwe.isClosed(), true)
 }
+
+func BenchmarkSendSideBWE_WriteRTCP(b *testing.B) {
+	numSequencesPerTwccReport := []int{10, 100, 500, 1000}
+
+	for _, count := range numSequencesPerTwccReport {
+		b.Run(fmt.Sprintf("num_sequences=%d", count), func(b *testing.B) {
+			bwe, err := NewSendSideBWE(SendSideBWEPacer(NewNoOpPacer()))
+			require.NoError(b, err)
+			require.NotNil(b, bwe)
+
+			r := twcc.NewRecorder(5000)
+			seq := uint16(0)
+			arrivalTime := int64(0)
+
+			for i := 0; i < b.N; i++ {
+				// nolint:gosec
+				seqs := rand.Intn(count/2) + count // [count, count * 1.5)
+				for j := 0; j < seqs; j++ {
+					seq++
+
+					if rand.Intn(5) == 0 { //nolint:gosec,staticcheck
+						// skip this packet
+					}
+
+					arrivalTime += int64(rtcp.TypeTCCDeltaScaleFactor * (rand.Intn(128) + 1)) //nolint:gosec
+					r.Record(5000, seq, arrivalTime)
+				}
+
+				rtcpPackets := r.BuildFeedbackPacket()
+				require.Equal(b, 1, len(rtcpPackets))
+
+				require.NoError(b, bwe.WriteRTCP(rtcpPackets, nil))
+			}
+
+			require.NoError(b, bwe.Close())
+		})
+	}
+}