Add Stats to jitter buffer, avoid divide by 0 (#613)

* add PacketsDropped to jitter buffer, avoid divide by 0

* more stats

* add lock around PopSamples

Author: frostbyte73

egressclient.go

@@ -18,9 +18,10 @@ import (
 	"context"
 	"net/http"
 
+	"github.com/twitchtv/twirp"
+
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/utils/xtwirp"
-	"github.com/twitchtv/twirp"
 )
 
 type EgressClient struct {

pkg/jitter/buffer.go

@@ -29,8 +29,7 @@ type Buffer struct {
 	maxLate         uint32
 	clockRate       uint32
 	onPacketDropped func()
-	packetsDropped  int
-	packetsTotal    int
+	stats           *BufferStats
 	logger          logger.Logger
 
 	mu          sync.Mutex
@@ -45,6 +44,14 @@ type Buffer struct {
 	minTS         uint32
 }
 
+type BufferStats struct {
+	PacketsPushed  uint64
+	PaddingPushed  uint64
+	PacketsDropped uint64
+	PacketsPopped  uint64
+	SamplesPopped  uint64
+}
+
 type packet struct {
 	prev, next     *packet
 	start, end     bool
@@ -57,6 +64,7 @@ func NewBuffer(depacketizer rtp.Depacketizer, clockRate uint32, maxLatency time.
 		depacketizer: depacketizer,
 		maxLate:      uint32(float64(maxLatency) / float64(time.Second) * float64(clockRate)),
 		clockRate:    clockRate,
+		stats:        &BufferStats{},
 		logger:       logger.LogRLogger(logr.Discard()),
 	}
 	for _, opt := range opts {
@@ -78,7 +86,11 @@ func (b *Buffer) Push(pkt *rtp.Packet) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	b.packetsTotal++
+	b.stats.PacketsPushed++
+	if pkt.Padding {
+		b.stats.PaddingPushed++
+	}
+
 	var start, end, padding bool
 	if len(pkt.Payload) == 0 {
 		// drop padding packets from the beginning of the stream
@@ -121,7 +133,7 @@ func (b *Buffer) Push(pkt *rtp.Packet) {
 	} else if beforePrev && !outsidePrevRange {
 		// drop if packet comes before previously pushed packet
 		if !p.padding {
-			b.packetsDropped++
+			b.stats.PacketsDropped++
 			if b.onPacketDropped != nil {
 				b.onPacketDropped()
 			}
@@ -229,18 +241,38 @@ func (b *Buffer) Pop(force bool) []*rtp.Packet {
 }
 
 func (b *Buffer) PopSamples(force bool) [][]*rtp.Packet {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
 	if force {
 		return b.forcePopSamples()
 	} else {
 		return b.popSamples()
 	}
 }
 
+func (b *Buffer) Stats() *BufferStats {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	return &BufferStats{
+		PacketsPushed:  b.stats.PacketsPushed,
+		PacketsDropped: b.stats.PacketsDropped,
+		PaddingPushed:  b.stats.PaddingPushed,
+		PacketsPopped:  b.stats.PacketsPopped,
+		SamplesPopped:  b.stats.SamplesPopped,
+	}
+}
+
 func (b *Buffer) PacketLoss() float64 {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	return float64(b.packetsDropped) / float64(b.packetsTotal)
+	if b.stats.PacketsPushed == 0 {
+		return 0
+	}
+
+	return float64(b.stats.PacketsDropped) / float64(b.stats.PacketsPushed)
 }
 
 func (b *Buffer) forcePop() []*rtp.Packet {
@@ -249,7 +281,13 @@ func (b *Buffer) forcePop() []*rtp.Packet {
 	var next *packet
 	for c := b.head; c != nil; c = next {
 		next = c.next
-		packets = append(packets, c.packet)
+		if !c.padding {
+			packets = append(packets, c.packet)
+			b.stats.PacketsPopped++
+		}
+		if c.end {
+			b.stats.SamplesPopped++
+		}
 		b.free(c)
 	}
 
@@ -265,15 +303,24 @@ func (b *Buffer) forcePopSamples() [][]*rtp.Packet {
 	var next *packet
 	for c := b.head; c != nil; c = next {
 		next = c.next
+
 		if c.start && len(sample) > 0 {
+			b.stats.SamplesPopped++
 			packets = append(packets, sample)
 			sample = make([]*rtp.Packet, 0)
 		}
-		sample = append(sample, c.packet)
+
+		if !c.padding {
+			sample = append(sample, c.packet)
+			b.stats.PacketsPopped++
+		}
+
 		if c.end {
+			b.stats.SamplesPopped++
 			packets = append(packets, sample)
 			sample = make([]*rtp.Packet, 0)
 		}
+
 		b.free(c)
 	}
 
@@ -301,16 +348,22 @@ func (b *Buffer) pop() []*rtp.Packet {
 	var next *packet
 	for c := b.head; ; c = next {
 		next = c.next
-		if !c.padding {
-			packets = append(packets, c.packet)
-		}
 		if next != nil {
 			if outsideRange(next.packet.SequenceNumber, c.packet.SequenceNumber) {
 				// adjust minTS to account for sequence number reset
 				b.minTS += next.packet.Timestamp - c.packet.Timestamp - b.maxSampleSize
 			}
 			next.prev = nil
 		}
+
+		if !c.padding {
+			packets = append(packets, c.packet)
+			b.stats.PacketsPopped++
+		}
+		if c.end {
+			b.stats.SamplesPopped++
+		}
+
 		if c == end {
 			b.prevSN = c.packet.SequenceNumber
 			b.head = next
@@ -320,6 +373,7 @@ func (b *Buffer) pop() []*rtp.Packet {
 			b.free(c)
 			return packets
 		}
+
 		b.free(c)
 	}
 }
@@ -344,20 +398,24 @@ func (b *Buffer) popSamples() [][]*rtp.Packet {
 	var next *packet
 	for c := b.head; ; c = next {
 		next = c.next
-		if !c.padding {
-			sample = append(sample, c.packet)
-		}
 		if next != nil {
 			if outsideRange(next.packet.SequenceNumber, c.packet.SequenceNumber) {
 				// adjust minTS to account for sequence number reset
 				b.minTS += next.packet.Timestamp - c.packet.Timestamp - b.maxSampleSize
 			}
 			next.prev = nil
 		}
+
+		if !c.padding {
+			sample = append(sample, c.packet)
+			b.stats.PacketsPopped++
+		}
 		if c.end {
+			b.stats.SamplesPopped++
 			packets = append(packets, sample)
 			sample = make([]*rtp.Packet, 0)
 		}
+
 		if c == end {
 			b.prevSN = c.packet.SequenceNumber
 			b.head = next
@@ -367,6 +425,7 @@ func (b *Buffer) popSamples() [][]*rtp.Packet {
 			b.free(c)
 			return packets
 		}
+
 		b.free(c)
 	}
 }
@@ -408,13 +467,13 @@ func (b *Buffer) drop() {
 		// lost packets will now be too old even if we receive them
 		// on sequence number reset, skip callback because we don't know whether we lost any
 		if !b.head.reset {
-			b.packetsDropped++
 			dropped = true
+			b.stats.PacketsDropped++
 		}
 
 		for b.head != nil && !b.head.start && before32(b.head.packet.Timestamp-b.maxSampleSize, b.minTS) {
 			dropped = true
-			b.packetsDropped++
+			b.stats.PacketsDropped++
 			b.prevSN = b.head.packet.SequenceNumber - 1
 			b.dropHead()
 		}
@@ -435,7 +494,7 @@ func (b *Buffer) drop() {
 		count := 0
 		ts := c.packet.Timestamp
 		for {
-			b.packetsDropped++
+			b.stats.PacketsDropped++
 			count++
 			b.dropHead()
 			c = b.head