Optimise SendSideBWE.WriteRTCP

- instead of passing each `cc.Acknowledgement` 1 by 1 in respective
  channel; pass them all at once to decrease the internal overhead of go
  channels

Author: adityaa30

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

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{}

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
 }
 

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

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)
 			}()
 

pkg/gcc/send_side_bwe_test.go

@@ -1,6 +1,7 @@
 package gcc
 
 import (
+	"fmt"
 	"math/rand"
 	"testing"
 
@@ -106,32 +107,39 @@ func TestSendSideBWE_ErrorOnWriteRTCPAtClosedState(t *testing.T) {
 }
 
 func BenchmarkSendSideBWE_WriteRTCP(b *testing.B) {
-	bwe, err := NewSendSideBWE(SendSideBWEPacer(NewNoOpPacer()))
-	require.NoError(b, err)
-	require.NotNil(b, bwe)
+	numSequencesPerTwccReport := []int{10, 100, 500, 1000}
 
-	r := twcc.NewRecorder(5000)
-	seq := uint16(0)
-	arrivalTime := int64(0)
+	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)
 
-	for i := 0; i < b.N; i++ {
-		seqs := rand.Intn(1000) + 500
-		for j := 0; j < seqs; j++ {
-			seq++
+			r := twcc.NewRecorder(5000)
+			seq := uint16(0)
+			arrivalTime := int64(0)
 
-			if rand.Intn(5) == 0 {
-				// skip this packet
-			}
+			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++
 
-			arrivalTime += int64(rtcp.TypeTCCDeltaScaleFactor * (rand.Intn(128) + 1))
-			r.Record(5000, seq, arrivalTime)
-		}
+					if rand.Intn(5) == 0 { //nolint:gosec,staticcheck
+						// skip this packet
+					}
 
-		rtcpPackets := r.BuildFeedbackPacket()
-		require.Equal(b, 1, len(rtcpPackets))
+					arrivalTime += int64(rtcp.TypeTCCDeltaScaleFactor * (rand.Intn(128) + 1)) //nolint:gosec
+					r.Record(5000, seq, arrivalTime)
+				}
 
-		require.NoError(b, bwe.WriteRTCP(rtcpPackets, nil))
-	}
+				rtcpPackets := r.BuildFeedbackPacket()
+				require.Equal(b, 1, len(rtcpPackets))
 
-	require.NoError(b, bwe.Close())
+				require.NoError(b, bwe.WriteRTCP(rtcpPackets, nil))
+			}
+
+			require.NoError(b, bwe.Close())
+		})
+	}
 }