Aggregate method for RTPDeltaInfo (#1562)

pkg/sfu/buffer/rtpstats.go

@@ -1492,3 +1492,101 @@ func AggregateRTPStats(statsList []*livekit.RTPStats) *livekit.RTPStats {
 		RttMax:               maxRtt,
 	}
 }
+
+func AggregateRTPDeltaInfo(deltaInfoList []*RTPDeltaInfo) *RTPDeltaInfo {
+	if len(deltaInfoList) == 0 {
+		return nil
+	}
+
+	startTime := time.Time{}
+	endTime := time.Time{}
+
+	packets := uint32(0)
+	bytes := uint64(0)
+	headerBytes := uint64(0)
+
+	packetsDuplicate := uint32(0)
+	bytesDuplicate := uint64(0)
+	headerBytesDuplicate := uint64(0)
+
+	packetsPadding := uint32(0)
+	bytesPadding := uint64(0)
+	headerBytesPadding := uint64(0)
+
+	packetsLost := uint32(0)
+	packetsMissing := uint32(0)
+
+	frames := uint32(0)
+
+	maxRtt := uint32(0)
+	maxJitter := float64(0)
+
+	nacks := uint32(0)
+	plis := uint32(0)
+	firs := uint32(0)
+
+	for _, deltaInfo := range deltaInfoList {
+		if startTime.IsZero() || startTime.After(deltaInfo.StartTime) {
+			startTime = deltaInfo.StartTime
+		}
+
+		endedAt := deltaInfo.StartTime.Add(deltaInfo.Duration)
+		if endTime.IsZero() || endTime.Before(endedAt) {
+			endTime = endedAt
+		}
+
+		packets += deltaInfo.Packets
+		bytes += deltaInfo.Bytes
+		headerBytes += deltaInfo.HeaderBytes
+
+		packetsDuplicate += deltaInfo.PacketsDuplicate
+		bytesDuplicate += deltaInfo.BytesDuplicate
+		headerBytesDuplicate += deltaInfo.HeaderBytesDuplicate
+
+		packetsPadding += deltaInfo.PacketsPadding
+		bytesPadding += deltaInfo.BytesPadding
+		headerBytesPadding += deltaInfo.HeaderBytesPadding
+
+		packetsLost += deltaInfo.PacketsLost
+		packetsMissing += deltaInfo.PacketsMissing
+
+		frames += deltaInfo.Frames
+
+		if deltaInfo.RttMax > maxRtt {
+			maxRtt = deltaInfo.RttMax
+		}
+
+		if deltaInfo.JitterMax > maxJitter {
+			maxJitter = deltaInfo.JitterMax
+		}
+
+		nacks += deltaInfo.Nacks
+		plis += deltaInfo.Plis
+		firs += deltaInfo.Firs
+	}
+	if startTime.IsZero() || endTime.IsZero() {
+		return nil
+	}
+
+	return &RTPDeltaInfo{
+		StartTime:            startTime,
+		Duration:             endTime.Sub(startTime),
+		Packets:              packets,
+		Bytes:                bytes,
+		HeaderBytes:          headerBytes,
+		PacketsDuplicate:     packetsDuplicate,
+		BytesDuplicate:       bytesDuplicate,
+		HeaderBytesDuplicate: headerBytesDuplicate,
+		PacketsPadding:       packetsPadding,
+		BytesPadding:         bytesPadding,
+		HeaderBytesPadding:   headerBytesPadding,
+		PacketsLost:          packetsLost,
+		PacketsMissing:       packetsMissing,
+		Frames:               frames,
+		RttMax:               maxRtt,
+		JitterMax:            maxJitter,
+		Nacks:                nacks,
+		Plis:                 plis,
+		Firs:                 firs,
+	}
+}

pkg/sfu/connectionquality/connectionstats.go

@@ -106,29 +106,22 @@ func (cs *ConnectionStats) ReceiverReportReceived(at time.Time) {
 }
 
 func (cs *ConnectionStats) updateScore(streams map[uint32]*buffer.StreamStatsWithLayers, at time.Time) float32 {
-	var endedAt time.Time
-	var stat windowStat
+	deltaInfoList := make([]*buffer.RTPDeltaInfo, 0, len(streams))
 	for _, s := range streams {
-		if stat.startedAt.IsZero() || stat.startedAt.After(s.RTPStats.StartTime) {
-			stat.startedAt = s.RTPStats.StartTime
-		}
-		streamEndedAt := s.RTPStats.StartTime.Add(s.RTPStats.Duration)
-		if endedAt.IsZero() || endedAt.Before(streamEndedAt) {
-			endedAt = streamEndedAt
-		}
-		stat.packetsExpected += s.RTPStats.Packets + s.RTPStats.PacketsPadding
-		stat.packetsLost += s.RTPStats.PacketsLost
-		stat.packetsMissing += s.RTPStats.PacketsMissing
-		if stat.rttMax < s.RTPStats.RttMax {
-			stat.rttMax = s.RTPStats.RttMax
-		}
-		if stat.jitterMax < s.RTPStats.JitterMax {
-			stat.jitterMax = s.RTPStats.JitterMax
-		}
-		stat.bytes += s.RTPStats.Bytes - s.RTPStats.HeaderBytes // only use media payload size
+		deltaInfoList = append(deltaInfoList, s.RTPStats)
 	}
-	if !stat.startedAt.IsZero() && !endedAt.IsZero() {
-		stat.duration = endedAt.Sub(stat.startedAt)
+	agg := buffer.AggregateRTPDeltaInfo(deltaInfoList)
+
+	var stat windowStat
+	if agg != nil {
+		stat.startedAt = agg.StartTime
+		stat.duration = agg.Duration
+		stat.packetsExpected = agg.Packets + agg.PacketsPadding
+		stat.packetsLost = agg.PacketsLost
+		stat.packetsMissing = agg.PacketsMissing
+		stat.bytes = agg.Bytes - agg.HeaderBytes // only use media payload size
+		stat.rttMax = agg.RttMax
+		stat.jitterMax = agg.JitterMax
 	}
 	cs.scorer.Update(&stat, at)
 

pkg/sfu/connectionquality/connectionstats_test.go

@@ -59,10 +59,18 @@ func TestConnectionQuality(t *testing.T) {
 				RTPStats: &buffer.RTPDeltaInfo{
 					StartTime:   now,
 					Duration:    duration,
-					Packets:     250,
+					Packets:     120,
 					PacketsLost: 30,
 				},
 			},
+			2: {
+				RTPStats: &buffer.RTPDeltaInfo{
+					StartTime:   now,
+					Duration:    duration,
+					Packets:     130,
+					PacketsLost: 0,
+				},
+			},
 		}
 		cs.updateScore(streams, now.Add(duration))
 		mos, quality = cs.GetScoreAndQuality()