Repository: pion/interceptor (go)

Issue:
A question about arrival_group_accumulator.go
```
func (a *arrivalGroupAccumulator) run(in <-chan []cc.Acknowledgment, agWriter func(arrivalGroup)) {
	init := false
	group := arrivalGroup{}
	for acks := range in {
		for _, next := range acks {
			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)
					continue
				}

				if interArrivalTimePkt(group, next) <= a.interArrivalThreshold &&
					interGroupDelayVariationPkt(group, next) < a.interGroupDelayVariationTreshold {
					group.add(next)
					continue
				}
				// println("Arrival len", len(group.packets), group.packets[len(group.packets)-1].Departure.Sub(group.packets[0].Departure).Milliseconds())
				agWriter(group)
				group = arrivalGroup{}
				group.add(next)
			}
		}
	}
}
```
The above code is excerpted from lines 23 to 55 of arrival_group_accumulator.go. After each new pkt is added, the group's departure_time is updated to the new packet's departure_time. In other words, as long as the time difference between the departure_time of the newly added packet and the previous packet is less than the threshold, it will be added to the group. How can we ensure that the time difference between the first packet and the last packet is less than 5 ms (the threshold)?

@thatsnotright 

Pull request description:
Fixes bug in group accumulator
#### Description

See referenced issue for background

#### Reference issue
Fixes #274 


Patch:
diff --git a/pkg/gcc/arrival_group.go b/pkg/gcc/arrival_group.go
index e0300f1..5aa1d89 100644
--- a/pkg/gcc/arrival_group.go
+++ b/pkg/gcc/arrival_group.go
@@ -16,10 +16,17 @@ type arrivalGroup struct {
 	arrival   time.Time
 }
 
+func newArrivalGroup(a cc.Acknowledgment) arrivalGroup {
+	return arrivalGroup{
+		packets:   []cc.Acknowledgment{a},
+		departure: a.Departure,
+		arrival:   a.Arrival,
+	}
+}
+
 func (g *arrivalGroup) add(a cc.Acknowledgment) {
 	g.packets = append(g.packets, a)
 	g.arrival = a.Arrival
-	g.departure = a.Departure
 }
 
 func (g arrivalGroup) String() string {
diff --git a/pkg/gcc/arrival_group_accumulator.go b/pkg/gcc/arrival_group_accumulator.go
index df170a1..c568ffd 100644
--- a/pkg/gcc/arrival_group_accumulator.go
+++ b/pkg/gcc/arrival_group_accumulator.go
@@ -29,7 +29,7 @@ func (a *arrivalGroupAccumulator) run(in <-chan []cc.Acknowledgment, agWriter fu
 	for acks := range in {
 		for _, next := range acks {
 			if !init {
-				group.add(next)
+				group = newArrivalGroup(next)
 				init = true
 				continue
 			}
@@ -38,11 +38,16 @@ func (a *arrivalGroupAccumulator) run(in <-chan []cc.Acknowledgment, agWriter fu
 				continue
 			}
 			if next.Departure.After(group.departure) {
+				// A sequence of packets which are sent within a burst_time interval
+				// constitute a group.
 				if interDepartureTimePkt(group, next) <= a.interDepartureThreshold {
 					group.add(next)
 					continue
 				}
 
+				// A Packet which has an inter-arrival time less than burst_time and
+				// an inter-group delay variation d(i) less than 0 is considered
+				// being part of the current group of packets.
 				if interArrivalTimePkt(group, next) <= a.interArrivalThreshold &&
 					interGroupDelayVariationPkt(group, next) < a.interGroupDelayVariationTreshold {
 					group.add(next)
@@ -50,24 +55,23 @@ func (a *arrivalGroupAccumulator) run(in <-chan []cc.Acknowledgment, agWriter fu
 				}
 
 				agWriter(group)
-				group = arrivalGroup{}
-				group.add(next)
+				group = newArrivalGroup(next)
 			}
 		}
 	}
 }
 
-func interArrivalTimePkt(a arrivalGroup, b cc.Acknowledgment) time.Duration {
-	return b.Arrival.Sub(a.arrival)
+func interArrivalTimePkt(group arrivalGroup, ack cc.Acknowledgment) time.Duration {
+	return ack.Arrival.Sub(group.arrival)
 }
 
-func interDepartureTimePkt(a arrivalGroup, b cc.Acknowledgment) time.Duration {
-	if len(a.packets) == 0 {
+func interDepartureTimePkt(group arrivalGroup, ack cc.Acknowledgment) time.Duration {
+	if len(group.packets) == 0 {
 		return 0
 	}
-	return b.Departure.Sub(a.packets[len(a.packets)-1].Departure)
+	return ack.Departure.Sub(group.departure)
 }
 
-func interGroupDelayVariationPkt(a arrivalGroup, b cc.Acknowledgment) time.Duration {
-	return b.Arrival.Sub(a.arrival) - b.Departure.Sub(a.departure)
+func interGroupDelayVariationPkt(group arrivalGroup, ack cc.Acknowledgment) time.Duration {
+	return ack.Arrival.Sub(group.arrival) - ack.Departure.Sub(group.departure)
 }


Test patch:
diff --git a/pkg/gcc/arrival_group_accumulator_test.go b/pkg/gcc/arrival_group_accumulator_test.go
index 7dc1a0c..25d87c0 100644
--- a/pkg/gcc/arrival_group_accumulator_test.go
+++ b/pkg/gcc/arrival_group_accumulator_test.go
@@ -70,7 +70,7 @@ func TestArrivalGroupAccumulator(t *testing.T) {
 					},
 				},
 				arrival:   time.Time{}.Add(20 * time.Millisecond),
-				departure: time.Time{}.Add(3 * time.Millisecond),
+				departure: time.Time{},
 			}},
 		},
 		{
@@ -102,7 +102,7 @@ func TestArrivalGroupAccumulator(t *testing.T) {
 						},
 					},
 					arrival:   time.Time{}.Add(20 * time.Millisecond),
-					departure: time.Time{}.Add(3 * time.Millisecond),
+					departure: time.Time{}.Add(0 * time.Millisecond),
 				},
 				{
 					packets: []cc.Acknowledgment{
@@ -156,6 +156,66 @@ func TestArrivalGroupAccumulator(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "newGroupBecauseOfInterDepartureTime",
+			log: []cc.Acknowledgment{
+				{
+					SequenceNumber: 0,
+					Departure:      time.Time{},
+					Arrival:        time.Time{}.Add(4 * time.Millisecond),
+				},
+				{
+					SequenceNumber: 1,
+					Departure:      time.Time{}.Add(3 * time.Millisecond),
+					Arrival:        time.Time{}.Add(4 * time.Millisecond),
+				},
+				{
+					SequenceNumber: 2,
+					Departure:      time.Time{}.Add(6 * time.Millisecond),
+					Arrival:        time.Time{}.Add(10 * time.Millisecond),
+				},
+				{
+					SequenceNumber: 3,
+					Departure:      time.Time{}.Add(9 * time.Millisecond),
+					Arrival:        time.Time{}.Add(10 * time.Millisecond),
+				},
+				triggerNewGroupElement,
+			},
+			exp: []arrivalGroup{
+				{
+					packets: []cc.Acknowledgment{
+						{
+							SequenceNumber: 0,
+							Departure:      time.Time{},
+							Arrival:        time.Time{}.Add(4 * time.Millisecond),
+						},
+						{
+							SequenceNumber: 1,
+							Departure:      time.Time{}.Add(3 * time.Millisecond),
+							Arrival:        time.Time{}.Add(4 * time.Millisecond),
+						},
+					},
+					departure: time.Time{},
+					arrival:   time.Time{}.Add(4 * time.Millisecond),
+				},
+				{
+					packets: []cc.Acknowledgment{
+						{
+							SequenceNumber: 2,
+							Departure:      time.Time{}.Add(6 * time.Millisecond),
+							Arrival:        time.Time{}.Add(10 * time.Millisecond),
+						},
+						{
+							SequenceNumber: 3,
+							Departure:      time.Time{}.Add(9 * time.Millisecond),
+							Arrival:        time.Time{}.Add(10 * time.Millisecond),
+						},
+					},
+					departure: time.Time{}.Add(6 * time.Millisecond),
+					arrival:   time.Time{}.Add(10 * time.Millisecond),
+				},
+			},
+		},
 	}
 
 	for _, tc := range cases {
diff --git a/pkg/gcc/arrival_group_test.go b/pkg/gcc/arrival_group_test.go
index 6857085..cf24f5a 100644
--- a/pkg/gcc/arrival_group_test.go
+++ b/pkg/gcc/arrival_group_test.go
@@ -71,7 +71,46 @@ func TestArrivalGroup(t *testing.T) {
 					Arrival:        time.Time{}.Add(time.Second),
 				}},
 				arrival:   time.Time{}.Add(time.Second),
-				departure: time.Time{}.Add(time.Second),
+				departure: time.Time{},
+			},
+		},
+		{
+			name: "departure time of group is the departure time of the first packet in the group",
+			acks: []cc.Acknowledgment{{
+				SequenceNumber: 0,
+				Size:           0,
+				Departure:      time.Time{}.Add(27 * time.Millisecond),
+				Arrival:        time.Time{},
+			}, {
+				SequenceNumber: 1,
+				Size:           1,
+				Departure:      time.Time{}.Add(32 * time.Millisecond),
+				Arrival:        time.Time{}.Add(37 * time.Millisecond),
+			}, {
+				SequenceNumber: 2,
+				Size:           2,
+				Departure:      time.Time{}.Add(50 * time.Millisecond),
+				Arrival:        time.Time{}.Add(56 * time.Millisecond),
+			}},
+			expected: arrivalGroup{
+				packets: []cc.Acknowledgment{{
+					SequenceNumber: 0,
+					Size:           0,
+					Departure:      time.Time{}.Add(27 * time.Millisecond),
+					Arrival:        time.Time{},
+				}, {
+					SequenceNumber: 1,
+					Size:           1,
+					Departure:      time.Time{}.Add(32 * time.Millisecond),
+					Arrival:        time.Time{}.Add(37 * time.Millisecond),
+				}, {
+					SequenceNumber: 2,
+					Size:           2,
+					Departure:      time.Time{}.Add(50 * time.Millisecond),
+					Arrival:        time.Time{}.Add(56 * time.Millisecond),
+				}},
+				arrival:   time.Time{}.Add(56 * time.Millisecond),
+				departure: time.Time{}.Add(27 * time.Millisecond),
 			},
 		},
 	}
@@ -80,8 +119,12 @@ func TestArrivalGroup(t *testing.T) {
 		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
 			ag := arrivalGroup{}
-			for _, ack := range tc.acks {
-				ag.add(ack)
+			for i, ack := range tc.acks {
+				if i == 0 {
+					ag = newArrivalGroup(ack)
+				} else {
+					ag.add(ack)
+				}
 			}
 			assert.Equal(t, tc.expected, ag)
 		})