Merge remote-tracking branch 'origin/master' into raja_min_packets

pkg/sfu/buffer/buffer_test.go

@@ -180,6 +180,7 @@ func TestNewBuffer(t *testing.T) {
 					Header: rtp.Header{
 						SequenceNumber: 65534,
 					},
+					Payload: []byte{1},
 				},
 				{
 					Header: rtp.Header{
@@ -201,8 +202,7 @@ func TestNewBuffer(t *testing.T) {
 			buff := NewBuffer(123, pool, pool)
 			buff.codecType = webrtc.RTPCodecTypeVideo
 			require.NotNil(t, buff)
-			buff.OnRtcpFeedback(func(_ []rtcp.Packet) {
-			})
+			buff.OnRtcpFeedback(func(_ []rtcp.Packet) {})
 			buff.Bind(webrtc.RTPParameters{
 				HeaderExtensions: nil,
 				Codecs:           []webrtc.RTPCodecParameters{vp8Codec},

pkg/sfu/buffer/rtpstats.go

@@ -377,6 +377,11 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa
 
 	first := false
 	if !r.initialized {
+		if payloadSize == 0 {
+			// do not start on a padding only packet
+			return
+		}
+
 		r.initialized = true
 
 		r.startTime = time.Now()
@@ -402,6 +407,14 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa
 				extStartSNOverridden: r.extStartSN,
 			}
 		}
+
+		r.logger.Debugw(
+			"rtp stream start",
+			"startTime", r.startTime.String(),
+			"firstTime", r.firstTime.String(),
+			"startSN", r.extStartSN,
+			"startTS", r.extStartTS,
+		)
 	}
 
 	if r.resyncOnNextPacket {
@@ -506,6 +519,12 @@ func (r *RTPStats) maybeAdjustStart(rtph *rtp.Header, pktSize uint64, hdrSize ui
 		return false
 	}
 
+	if payloadSize == 0 {
+		// do not start on a padding only packet
+		r.logger.Infow("adjusting start, skipping on padding only packet")
+		return true
+	}
+
 	r.packetsLost += uint32(uint16(r.extStartSN)-rtph.SequenceNumber) - 1
 	snBeforeAdjust := r.extStartSN
 	r.extStartSN = uint32(rtph.SequenceNumber)
@@ -529,7 +548,7 @@ func (r *RTPStats) maybeAdjustStart(rtph *rtp.Header, pktSize uint64, hdrSize ui
 		r.tsCycles++
 	}
 	r.logger.Infow(
-		"adjusting starting sequence number",
+		"adjusting start",
 		"snBefore", snBeforeAdjust,
 		"snAfter", r.extStartSN,
 		"snCyles", r.cycles,
@@ -817,8 +836,13 @@ func (r *RTPStats) maybeAdjustFirstPacketTime(ts uint32) {
 	if firstTime.Before(r.firstTime) {
 		r.logger.Debugw(
 			"adjusting first packet time",
+			"startTime", r.startTime.String(),
+			"nowTime", now.String(),
 			"before", r.firstTime.String(),
 			"after", firstTime.String(),
+			"adjustment", r.firstTime.Sub(firstTime),
+			"nowTS", ts,
+			"extStartTS", r.extStartTS,
 		)
 		if r.firstTime.Sub(firstTime) > firstPacketTimeAdjustThreshold {
 			r.logger.Infow("first packet time adjustment too big, ignoring",

pkg/sfu/downtrack.go

@@ -784,9 +784,8 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa
 			TransportWideExtID: uint8(d.transportWideExtID),
 			WriteStream:        d.writeStream,
 			Metadata: sendPacketMetadata{
-				isPadding:       true,
-				disableCounter:  true,
-				disableRTPStats: paddingOnMute,
+				isPadding:      true,
+				disableCounter: true,
 			},
 			OnSent: d.packetSent,
 		})
@@ -1294,10 +1293,8 @@ func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan
 					AbsSendTimeExtID:   uint8(d.absSendTimeExtID),
 					TransportWideExtID: uint8(d.transportWideExtID),
 					WriteStream:        d.writeStream,
-					Metadata: sendPacketMetadata{
-						isBlankFrame: true,
-					},
-					OnSent: d.packetSent,
+					Metadata:           sendPacketMetadata{},
+					OnSent:             d.packetSent,
 				})
 
 				// only the first frame will need frameEndNeeded to close out the
@@ -1791,11 +1788,8 @@ func (d *DownTrack) sendSilentFrameOnMuteForOpus() {
 				AbsSendTimeExtID:   uint8(d.absSendTimeExtID),
 				TransportWideExtID: uint8(d.transportWideExtID),
 				WriteStream:        d.writeStream,
-				Metadata: sendPacketMetadata{
-					isBlankFrame:    true,
-					disableRTPStats: true,
-				},
-				OnSent: d.packetSent,
+				Metadata:           sendPacketMetadata{},
+				OnSent:             d.packetSent,
 			})
 		}
 
@@ -1812,16 +1806,14 @@ func (d *DownTrack) HandleRTCPSenderReportData(_payloadType webrtc.PayloadType,
 }
 
 type sendPacketMetadata struct {
-	layer           int32
-	arrival         time.Time
-	isKeyFrame      bool
-	isRTX           bool
-	isPadding       bool
-	isBlankFrame    bool
-	disableCounter  bool
-	disableRTPStats bool
-	tp              *TranslationParams
-	pool            *[]byte
+	layer          int32
+	arrival        time.Time
+	isKeyFrame     bool
+	isRTX          bool
+	isPadding      bool
+	disableCounter bool
+	tp             *TranslationParams
+	pool           *[]byte
 }
 
 func (d *DownTrack) packetSent(md interface{}, hdr *rtp.Header, payloadSize int, sendTime time.Time, sendError error) {
@@ -1839,10 +1831,9 @@ func (d *DownTrack) packetSent(md interface{}, hdr *rtp.Header, payloadSize int,
 		return
 	}
 
-	headerSize := hdr.MarshalSize()
 	if !spmd.disableCounter {
 		// STREAM-ALLOCATOR-TODO: remove this stream allocator bytes counter once stream allocator changes fully to pull bytes counter
-		size := uint32(headerSize + payloadSize)
+		size := uint32(hdr.MarshalSize() + payloadSize)
 		d.streamAllocatorBytesCounter.Add(size)
 		if spmd.isRTX {
 			d.bytesRetransmitted.Add(size)
@@ -1851,16 +1842,15 @@ func (d *DownTrack) packetSent(md interface{}, hdr *rtp.Header, payloadSize int,
 		}
 	}
 
-	if !spmd.disableRTPStats {
-		packetTime := spmd.arrival
-		if packetTime.IsZero() {
-			packetTime = sendTime
-		}
-		if spmd.isPadding {
-			d.rtpStats.Update(hdr, 0, payloadSize, packetTime)
-		} else {
-			d.rtpStats.Update(hdr, payloadSize, 0, packetTime)
-		}
+	// update RTPStats
+	packetTime := spmd.arrival
+	if packetTime.IsZero() {
+		packetTime = sendTime
+	}
+	if spmd.isPadding {
+		d.rtpStats.Update(hdr, 0, payloadSize, packetTime)
+	} else {
+		d.rtpStats.Update(hdr, payloadSize, 0, packetTime)
 	}
 
 	if spmd.isKeyFrame {