From 15078eb9f4814d5837fea317b8dea15924bb1427 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Thu, 4 May 2023 13:00:57 +0530 Subject: [PATCH] Keep track of expected RTP time stamp and control drift. (#1681) * Keep track of expected RTP time stamp and control drift. - Use monotonic clock in RTCP Sender Report and packet times - Keep the time stamp close to expected time stamp on layer/SSRC switches * clean up * fix test compile * more test compile failures --- pkg/sfu/buffer/buffer.go | 28 +++--- pkg/sfu/buffer/rtpstats.go | 149 +++++++++++++++++++++----------- pkg/sfu/buffer/rtpstats_test.go | 14 +-- pkg/sfu/downtrack.go | 25 ++++-- pkg/sfu/forwarder.go | 90 +++++++++++++++---- pkg/sfu/forwarder_test.go | 2 +- pkg/sfu/streamtrackermanager.go | 5 +- pkg/sfu/testutils/data.go | 4 +- 8 files changed, 218 insertions(+), 99 deletions(-) diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 8b9a7094d..0faea1fe8 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -27,17 +27,17 @@ import ( ) const ( - ReportDelta = 1e9 + ReportDelta = time.Second ) type pendingPacket struct { - arrivalTime int64 + arrivalTime time.Time packet []byte } type ExtPacket struct { VideoLayer - Arrival int64 + Arrival time.Time Packet *rtp.Packet Payload interface{} KeyFrame bool @@ -58,7 +58,7 @@ type Buffer struct { closeOnce sync.Once mediaSSRC uint32 clockRate uint32 - lastReport int64 + lastReport time.Time twccExt uint8 audioLevelExt uint8 bound bool @@ -163,7 +163,7 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili b.deltaStatsSnapshotId = b.rtpStats.NewSnapshotId() b.clockRate = codec.ClockRate - b.lastReport = time.Now().UnixNano() + b.lastReport = time.Now() b.mime = strings.ToLower(codec.MimeType) for _, ext := range params.HeaderExtensions { @@ -260,12 +260,12 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) { copy(packet, pkt) b.pPackets = append(b.pPackets, pendingPacket{ packet: packet, - arrivalTime: time.Now().UnixNano(), + arrivalTime: time.Now(), }) return } - b.calc(pkt, time.Now().UnixNano()) + b.calc(pkt, time.Now()) return } @@ -392,7 +392,7 @@ func (b *Buffer) SetRTT(rtt uint32) { } } -func (b *Buffer) calc(pkt []byte, arrivalTime int64) { +func (b *Buffer) calc(pkt []byte, arrivalTime time.Time) { pktBuf, err := b.bucket.AddPacket(pkt) if err != nil { // @@ -486,7 +486,7 @@ func (b *Buffer) doFpsCalc(ep *ExtPacket) { } } -func (b *Buffer) updateStreamState(p *rtp.Packet, arrivalTime int64) { +func (b *Buffer) updateStreamState(p *rtp.Packet, arrivalTime time.Time) { flowState := b.rtpStats.Update(&p.Header, len(p.Payload), int(p.PaddingSize), arrivalTime) if b.nacker != nil { @@ -500,12 +500,12 @@ func (b *Buffer) updateStreamState(p *rtp.Packet, arrivalTime int64) { } } -func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime int64) { +func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime time.Time) { // submit to TWCC even if it is a padding only packet. Clients use padding only packets as probes // for bandwidth estimation if b.twcc != nil && b.twccExt != 0 { if ext := p.GetExtension(b.twccExt); ext != nil { - b.twcc.Push(binary.BigEndian.Uint16(ext[0:2]), arrivalTime, p.Marker) + b.twcc.Push(binary.BigEndian.Uint16(ext[0:2]), arrivalTime.UnixNano(), p.Marker) } } @@ -530,7 +530,7 @@ func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime int64) { } } -func (b *Buffer) getExtPacket(rtpPacket *rtp.Packet, arrivalTime int64) *ExtPacket { +func (b *Buffer) getExtPacket(rtpPacket *rtp.Packet, arrivalTime time.Time) *ExtPacket { ep := &ExtPacket{ Packet: rtpPacket, Arrival: arrivalTime, @@ -615,8 +615,8 @@ func (b *Buffer) doNACKs() { } } -func (b *Buffer) doReports(arrivalTime int64) { - timeDiff := arrivalTime - b.lastReport +func (b *Buffer) doReports(arrivalTime time.Time) { + timeDiff := arrivalTime.Sub(b.lastReport) if timeDiff < ReportDelta { return } diff --git a/pkg/sfu/buffer/rtpstats.go b/pkg/sfu/buffer/rtpstats.go index c7b8575f9..b2cca9f5d 100644 --- a/pkg/sfu/buffer/rtpstats.go +++ b/pkg/sfu/buffer/rtpstats.go @@ -1,6 +1,7 @@ package buffer import ( + "errors" "fmt" "math" "sync" @@ -123,11 +124,15 @@ type RTPStats struct { lastRRTime time.Time lastRR rtcp.ReceptionReport - highestTS uint32 - tsCycles uint32 - highestTime int64 + extStartTS uint64 + highestTS uint32 + tsCycles uint32 - lastTransit uint32 + firstTime time.Time + highestTime time.Time + + lastTransit uint32 + lastJitterRTP uint32 bytes uint64 headerBytes uint64 @@ -180,6 +185,8 @@ type RTPStats struct { firstSenderReportRTP uint32 firstFeedSenderReportNTP mediatransportutil.NtpTime firstFeedSenderReportRTP uint32 + lastSRTime time.Time + lastSRNTP mediatransportutil.NtpTime nextSnapshotId uint32 snapshots map[uint32]*Snapshot @@ -206,7 +213,7 @@ func (r *RTPStats) Seed(from *RTPStats) { r.resyncOnNextPacket = from.resyncOnNextPacket r.startTime = from.startTime - // do not clone endTime as a non-zero endTime indiacates an ended object + // do not clone endTime as a non-zero endTime indicates an ended object r.extStartSN = from.extStartSN r.highestSN = from.highestSN @@ -216,11 +223,15 @@ func (r *RTPStats) Seed(from *RTPStats) { r.lastRRTime = from.lastRRTime r.lastRR = from.lastRR + r.extStartTS = from.extStartTS r.highestTS = from.highestTS r.tsCycles = from.tsCycles + + r.firstTime = from.firstTime r.highestTime = from.highestTime r.lastTransit = from.lastTransit + r.lastJitterRTP = from.lastJitterRTP r.bytes = from.bytes r.headerBytes = from.headerBytes @@ -280,6 +291,8 @@ func (r *RTPStats) Seed(from *RTPStats) { r.firstSenderReportRTP = from.firstSenderReportRTP r.firstFeedSenderReportNTP = from.firstFeedSenderReportNTP r.firstFeedSenderReportRTP = from.firstFeedSenderReportRTP + r.lastSRTime = from.lastSRTime + r.lastSRNTP = from.lastSRNTP r.nextSnapshotId = from.nextSnapshotId for id, ss := range from.snapshots { @@ -324,7 +337,7 @@ func (r *RTPStats) IsActive() bool { return r.initialized && r.endTime.IsZero() } -func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, packetTime int64) (flowState RTPFlowState) { +func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, packetTime time.Time) (flowState RTPFlowState) { r.lock.Lock() defer r.lock.Unlock() @@ -338,14 +351,17 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa r.startTime = time.Now() - r.highestSN = rtph.SequenceNumber - 1 - r.highestTS = rtph.Timestamp - r.highestTime = packetTime - r.extStartSN = uint32(rtph.SequenceNumber) + r.highestSN = rtph.SequenceNumber - 1 r.cycles = 0 + + r.extStartTS = uint64(rtph.Timestamp) + r.highestTS = rtph.Timestamp r.tsCycles = 0 + r.firstTime = packetTime + r.highestTime = packetTime + first = true // initialize snapshots if any @@ -378,7 +394,7 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa } // adjust start to account for out-of-order packets before a cycle completes - if !r.maybeAdjustStartSN(rtph, packetTime, pktSize, hdrSize, payloadSize) { + if !r.maybeAdjustStartSN(rtph, pktSize, hdrSize, payloadSize) { if !r.isSnInfoLost(rtph.SequenceNumber) { r.bytesDuplicate += pktSize r.headerBytesDuplicate += hdrSize @@ -447,7 +463,7 @@ func (r *RTPStats) ResyncOnNextPacket() { r.resyncOnNextPacket = true } -func (r *RTPStats) maybeAdjustStartSN(rtph *rtp.Header, packetTime int64, pktSize uint64, hdrSize uint64, payloadSize int) bool { +func (r *RTPStats) maybeAdjustStartSN(rtph *rtp.Header, pktSize uint64, hdrSize uint64, payloadSize int) bool { if (r.getExtHighestSN() - r.extStartSN + 1) >= (NumSequenceNumbers / 2) { return false } @@ -501,7 +517,7 @@ func (r *RTPStats) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt uint32 return } - rtt, err := mediatransportutil.GetRttMsFromReceiverReportOnly(&rr) + rtt, err := mediatransportutil.GetRttMs(&rr, r.lastSRNTP, r.lastSRTime) if err == nil { isRttChanged = rtt != r.rtt } else { @@ -795,6 +811,32 @@ func (r *RTPStats) GetRtcpSenderReportDataExt() *RTCPSenderReportDataExt { } } +func (r *RTPStats) GetExpectedRTPTimestamp(at time.Time) (uint32, error) { + r.lock.RLock() + defer r.lock.RUnlock() + + if !r.initialized { + return 0, errors.New("uninitilaized") + } + + timeDiff := at.Sub(r.firstTime) + rtpDiff := timeDiff.Nanoseconds() * int64(r.params.ClockRate) / 1e9 + expectedExtRTP := r.extStartTS + uint64(rtpDiff) + r.logger.Debugw( + "expected RTP timestamp", + "firstTime", r.firstTime.String(), + "checkAt", at.String(), + "timeDiff", timeDiff, + "firstRTP", r.extStartTS, + "rtpDiff", rtpDiff, + "expectedExtRTP", expectedExtRTP, + "expectedRTP", uint32(expectedExtRTP), + "highestTS", r.highestTS, + "highestTime", r.highestTime.String(), + ) + return uint32(expectedExtRTP), nil +} + func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, srDataExt *RTCPSenderReportDataExt) *rtcp.SenderReport { r.lock.Lock() defer r.lock.Unlock() @@ -808,28 +850,27 @@ func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, srDataExt *RTCPSenderReportD return nil } - // NTP timestamp in sender report from publisher side could have a different base, - // i. e. although it should be wall clock at time of send, have observed instances of older timer. - // It is not possible to accurately calculate current time in the NTP time base of the publisher side. - // So, using a smoothed version of one way delay for use in sender reports. - now := time.Now() + // construct current time based on monotonic clock + timeSinceFirst := time.Since(r.firstTime) + now := r.firstTime.Add(timeSinceFirst) nowNTP := mediatransportutil.ToNtpTime(now) - nowRTP := r.highestTS - isUsingSmoothed := true - smoothedLocalTimeOfLatestSenderReportNTP := srDataExt.SenderReportData.NTPTimestamp.Time().Add(srDataExt.SmoothedOWD) - if smoothedLocalTimeOfLatestSenderReportNTP.After(now) { - isUsingSmoothed = false - r.logger.Debugw("smoothed time of NTP is ahead", - "now", now, - "smoothed", smoothedLocalTimeOfLatestSenderReportNTP, - "diff", smoothedLocalTimeOfLatestSenderReportNTP.Sub(now), + expectedExtRTP := r.extStartTS + uint64(timeSinceFirst.Nanoseconds()*int64(r.params.ClockRate)/1e9) + if getExtTS(r.highestTS, r.tsCycles) > expectedExtRTP || now.Before(r.highestTime) { + r.logger.Debugw( + "anachronous sender report", + "firstTime", r.firstTime.String(), + "currentTime", now.String(), + "timSinceFirst", timeSinceFirst, + "extStartTS", r.extStartTS, + "highestExtRTP", getExtTS(r.highestTS, r.tsCycles), + "expectedExtRTP", expectedExtRTP, ) - nowRTP += uint32(now.Sub(time.Unix(0, r.highestTime)).Milliseconds() * int64(r.params.ClockRate) / 1000) - } else { - nowRTP = srDataExt.SenderReportData.RTPTimestamp + uint32(now.Sub(smoothedLocalTimeOfLatestSenderReportNTP).Milliseconds()*int64(r.params.ClockRate)/1000) } + timeSinceHighest := time.Since(r.highestTime) + nowRTP := r.highestTS + uint32(timeSinceHighest.Nanoseconds()*int64(r.params.ClockRate)/1e9) + // TODO-REMOVE-AFTER-DEBUG if r.firstSenderReportNTP == 0 { r.firstSenderReportNTP = nowNTP @@ -838,57 +879,50 @@ func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, srDataExt *RTCPSenderReportD r.firstFeedSenderReportNTP = srDataExt.SenderReportData.NTPTimestamp r.firstFeedSenderReportRTP = srDataExt.SenderReportData.RTPTimestamp } else { - highestTime := time.Unix(0, r.highestTime) ntpTime := nowNTP.Time() - ntpDiffLocal := ntpTime.Sub(highestTime) + ntpDiffLocal := ntpTime.Sub(r.highestTime) rtpDiffLocal := int32(nowRTP - r.highestTS) rtpOffsetLocal := int32(nowRTP - r.highestTS - uint32(ntpDiffLocal.Nanoseconds()*int64(r.params.ClockRate)/1e9)) - ntpDiffSmoothed := ntpTime.Sub(smoothedLocalTimeOfLatestSenderReportNTP) - rtpDiffSmoothed := int32(nowRTP - srDataExt.SenderReportData.RTPTimestamp) - rtpOffsetSmoothed := int32(nowRTP - srDataExt.SenderReportData.RTPTimestamp - uint32(ntpDiffSmoothed.Nanoseconds()*int64(r.params.ClockRate)/1e9)) - timeSinceFirst := nowNTP.Time().Sub(r.firstSenderReportNTP.Time()) rtpDiffSinceFirst := getExtTS(nowRTP, r.tsCycles) - getExtTS(r.firstSenderReportRTP, 0) drift := int64(uint64(timeSinceFirst.Nanoseconds()*int64(r.params.ClockRate)/1e9) - rtpDiffSinceFirst) - driftTime := (float64(drift) * 1000) / float64(r.params.ClockRate) + driftMs := (float64(drift) * 1000) / float64(r.params.ClockRate) feedTimeSinceFirst := srDataExt.SenderReportData.NTPTimestamp.Time().Sub(r.firstFeedSenderReportNTP.Time()) // using tsCycles for extending feed time stamp too feedRtpDiffSinceFirst := getExtTS(srDataExt.SenderReportData.RTPTimestamp, r.tsCycles) - getExtTS(r.firstFeedSenderReportRTP, 0) feedDrift := int64(uint64(feedTimeSinceFirst.Nanoseconds()*int64(r.params.ClockRate)/1e9) - feedRtpDiffSinceFirst) - feedDriftTime := (float64(feedDrift) * 1000) / float64(r.params.ClockRate) + feedDriftMs := (float64(feedDrift) * 1000) / float64(r.params.ClockRate) r.logger.Debugw( "sending sender report", "highestTS", r.highestTS, - "highestTime", highestTime, - "smoothedTime", smoothedLocalTimeOfLatestSenderReportNTP, + "highestTime", r.highestTime.String(), "reportTS", nowRTP, - "reportTime", ntpTime, + "reportTime", ntpTime.String(), "rtpDiffLocal", rtpDiffLocal, "ntpDiffLocal", ntpDiffLocal, "rtpOffsetLocal", rtpOffsetLocal, - "rtpDiffSmoothed", rtpDiffSmoothed, - "ntpDiffSmoothed", ntpDiffSmoothed, - "rtpOffsetSmoothed", rtpOffsetSmoothed, "timeSinceFirst", timeSinceFirst, "rtpDiffSinceFirst", rtpDiffSinceFirst, "drift", drift, - "driftTime(ms)", driftTime, - "smoothed", isUsingSmoothed, + "driftMs", driftMs, "feedRTP", srDataExt.SenderReportData.RTPTimestamp, - "feedNTP", srDataExt.SenderReportData.NTPTimestamp.Time(), - "feedArrival", srDataExt.SenderReportData.ArrivalTime, + "feedNTP", srDataExt.SenderReportData.NTPTimestamp.Time().String(), + "feedArrival", srDataExt.SenderReportData.ArrivalTime.String(), "smoothedOWD", srDataExt.SmoothedOWD, "feedTimeSinceFirst", feedTimeSinceFirst, "feedRtpDiffSinceFirst", feedRtpDiffSinceFirst, "feedDrift", feedDrift, - "feedDriftTime(ms)", feedDriftTime, + "feedDriftMs", feedDriftMs, ) } + r.lastSRTime = now + r.lastSRNTP = nowNTP + return &rtcp.SenderReport{ SSRC: ssrc, NTPTime: uint64(nowNTP), @@ -1400,8 +1434,20 @@ func (r *RTPStats) getIntervalStats(startInclusive uint16, endExclusive uint16) return } -func (r *RTPStats) updateJitter(rtph *rtp.Header, packetTime int64) { - packetTimeRTP := uint32(packetTime / 1e6 * int64(r.params.ClockRate/1e3)) +func (r *RTPStats) updateJitter(rtph *rtp.Header, packetTime time.Time) { + // Do not update jitter on multiple packets of same frame. + // All packets of a frame have the same time stamp. + // NOTE: This does not protect against using more than one packet of the same frame + // if packets arrive out-of-order. For example, + // p1f1 -> p1f2 -> p2f1 + // In this case, p2f1 (packet 2, frame 1) will still be used in jitter calculation + // although it is the second packet of a frame because of out-of-order receival. + if r.lastJitterRTP == rtph.Timestamp { + return + } + + timeSinceFirst := packetTime.Sub(r.firstTime) + packetTimeRTP := uint32(timeSinceFirst.Nanoseconds() * int64(r.params.ClockRate) / 1e9) transit := packetTimeRTP - rtph.Timestamp if r.lastTransit != 0 { @@ -1422,6 +1468,7 @@ func (r *RTPStats) updateJitter(rtph *rtp.Header, packetTime int64) { } r.lastTransit = transit + r.lastJitterRTP = rtph.Timestamp } func (r *RTPStats) updateGapHistogram(gap int) { diff --git a/pkg/sfu/buffer/rtpstats_test.go b/pkg/sfu/buffer/rtpstats_test.go index b1be40650..70c852e7b 100644 --- a/pkg/sfu/buffer/rtpstats_test.go +++ b/pkg/sfu/buffer/rtpstats_test.go @@ -43,7 +43,7 @@ func TestRTPStats(t *testing.T) { timestamp += uint32(now.Sub(lastFrameTime).Seconds() * float64(clockRate)) for i := 0; i < packetsPerFrame; i++ { packet := getPacket(sequenceNumber, timestamp, packetSize) - r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano()) + r.Update(&packet.Header, len(packet.Payload), 0, time.Now()) if (sequenceNumber % 100) == 0 { jump := uint16(rand.Float64() * 120.0) sequenceNumber += jump @@ -70,7 +70,7 @@ func TestRTPStats_Update(t *testing.T) { sequenceNumber := uint16(rand.Float64() * float64(1<<16)) timestamp := uint32(rand.Float64() * float64(1<<32)) packet := getPacket(sequenceNumber, timestamp, 1000) - flowState := r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano()) + flowState := r.Update(&packet.Header, len(packet.Payload), 0, time.Now()) require.False(t, flowState.HasLoss) require.True(t, r.initialized) require.Equal(t, sequenceNumber, r.highestSN) @@ -80,14 +80,14 @@ func TestRTPStats_Update(t *testing.T) { sequenceNumber++ timestamp += 3000 packet = getPacket(sequenceNumber, timestamp, 1000) - flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano()) + flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now()) require.False(t, flowState.HasLoss) require.Equal(t, sequenceNumber, r.highestSN) require.Equal(t, timestamp, r.highestTS) // out-of-order packet = getPacket(sequenceNumber-10, timestamp-30000, 1000) - flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano()) + flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now()) require.False(t, flowState.HasLoss) require.Equal(t, sequenceNumber, r.highestSN) require.Equal(t, timestamp, r.highestTS) @@ -96,7 +96,7 @@ func TestRTPStats_Update(t *testing.T) { // duplicate packet = getPacket(sequenceNumber-10, timestamp-30000, 1000) - flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano()) + flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now()) require.False(t, flowState.HasLoss) require.Equal(t, sequenceNumber, r.highestSN) require.Equal(t, timestamp, r.highestTS) @@ -107,7 +107,7 @@ func TestRTPStats_Update(t *testing.T) { sequenceNumber += 10 timestamp += 30000 packet = getPacket(sequenceNumber, timestamp, 1000) - flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano()) + flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now()) require.True(t, flowState.HasLoss) require.Equal(t, sequenceNumber-9, flowState.LossStartInclusive) require.Equal(t, sequenceNumber, flowState.LossEndExclusive) @@ -115,7 +115,7 @@ func TestRTPStats_Update(t *testing.T) { // out-of-order should decrement number of lost packets packet = getPacket(sequenceNumber-15, timestamp-45000, 1000) - flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano()) + flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now()) require.False(t, flowState.HasLoss) require.Equal(t, sequenceNumber, r.highestSN) require.Equal(t, timestamp, r.highestTS) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 6ad785500..0e6638b0e 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -276,7 +276,12 @@ func NewDownTrack( kind: kind, codec: codecs[0].RTPCodecCapability, } - d.forwarder = NewForwarder(d.kind, d.logger, d.receiver.GetReferenceLayerRTPTimestamp) + d.forwarder = NewForwarder( + d.kind, + d.logger, + d.receiver.GetReferenceLayerRTPTimestamp, + d.getExpectedRTPTimestamp, + ) d.forwarder.OnParkedLayerExpired(func() { if sal := d.getStreamAllocatorListener(); sal != nil { sal.OnSubscriptionChanged(d) @@ -638,7 +643,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { } } - d.rtpStats.Update(hdr, len(payload), 0, time.Now().UnixNano()) + d.rtpStats.Update(hdr, len(payload), 0, extPkt.Arrival) return nil } @@ -717,7 +722,7 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool) int { } if !paddingOnMute { - d.rtpStats.Update(&hdr, 0, len(payload), time.Now().UnixNano()) + d.rtpStats.Update(&hdr, 0, len(payload), time.Now()) } // @@ -1214,7 +1219,7 @@ func (d *DownTrack) writeOpusBlankFrame(hdr *rtp.Header, frameEndNeeded bool) (i _, err := d.writeStream.WriteRTP(hdr, payload) if err == nil { - d.rtpStats.Update(hdr, len(payload), 0, time.Now().UnixNano()) + d.rtpStats.Update(hdr, len(payload), 0, time.Now()) } return hdr.MarshalSize() + len(payload), err } @@ -1233,7 +1238,7 @@ func (d *DownTrack) writeOpusRedBlankFrame(hdr *rtp.Header, frameEndNeeded bool) _, err := d.writeStream.WriteRTP(hdr, payload) if err == nil { - d.rtpStats.Update(hdr, len(payload), 0, time.Now().UnixNano()) + d.rtpStats.Update(hdr, len(payload), 0, time.Now()) } return hdr.MarshalSize() + len(payload), err } @@ -1254,7 +1259,7 @@ func (d *DownTrack) writeVP8BlankFrame(hdr *rtp.Header, frameEndNeeded bool) (in _, err = d.writeStream.WriteRTP(hdr, payload) if err == nil { - d.rtpStats.Update(hdr, len(payload), 0, time.Now().UnixNano()) + d.rtpStats.Update(hdr, len(payload), 0, time.Now()) } return hdr.MarshalSize() + len(payload), err } @@ -1276,7 +1281,7 @@ func (d *DownTrack) writeH264BlankFrame(hdr *rtp.Header, frameEndNeeded bool) (i payload := buf[:offset] _, err := d.writeStream.WriteRTP(hdr, payload) if err == nil { - d.rtpStats.Update(hdr, len(payload), 0, time.Now().UnixNano()) + d.rtpStats.Update(hdr, len(payload), 0, time.Now()) } return hdr.MarshalSize() + offset, err } @@ -1499,7 +1504,7 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) { d.streamAllocatorBytesCounter.Add(uint32(pkt.Header.MarshalSize() + len(payload))) d.bytesRetransmitted.Add(uint32(pkt.Header.MarshalSize() + len(payload))) - d.rtpStats.Update(&pkt.Header, len(payload), 0, time.Now().UnixNano()) + d.rtpStats.Update(&pkt.Header, len(payload), 0, time.Now()) } } @@ -1624,6 +1629,10 @@ func (d *DownTrack) DebugInfo() map[string]interface{} { } } +func (d *DownTrack) getExpectedRTPTimestamp(at time.Time) (uint32, error) { + return d.rtpStats.GetExpectedRTPTimestamp(at) +} + func (d *DownTrack) GetConnectionScoreAndQuality() (float32, livekit.ConnectionQuality) { return d.connectionStats.GetScoreAndQuality() } diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 4c1ba5028..e6da06859 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -158,6 +158,7 @@ type Forwarder struct { kind webrtc.RTPCodecType logger logger.Logger getReferenceLayerRTPTimestamp func(ts uint32, layer int32, referenceLayer int32) (uint32, error) + getExpectedRTPTimestamp func(at time.Time) (uint32, error) muted bool pubMuted bool @@ -185,11 +186,13 @@ func NewForwarder( kind webrtc.RTPCodecType, logger logger.Logger, getReferenceLayerRTPTimestamp func(ts uint32, layer int32, referenceLayer int32) (uint32, error), + getExpectedRTPTimestamp func(at time.Time) (uint32, error), ) *Forwarder { f := &Forwarder{ kind: kind, logger: logger, getReferenceLayerRTPTimestamp: getReferenceLayerRTPTimestamp, + getExpectedRTPTimestamp: getExpectedRTPTimestamp, referenceLayerSpatial: buffer.InvalidLayerSpatial, lastAllocation: VideoAllocationDefault, rtpMunger: NewRTPMunger(logger), @@ -1446,26 +1449,46 @@ func (f *Forwarder) getTranslationParamsCommon(extPkt *buffer.ExtPacket, layer i // Compute how much time passed between the old RTP extPkt // and the current packet, and fix timestamp on source change - td := uint32(1) + // + // There are three time stamps to consider here + // 1. lastTS -> time stamp of last sent packet + // 2. refTS -> time stamp of this packet (after munging) calculated using feed's RTCP sender report + // 3. expectedTS -> time stamp of this packet (after munging) calculated using this stream's RTCP sender report + // Ideally, refTS and expectedTS should be very close and lastTS should be before both of those. + // But, cases like muting/unmuting, clock vagaries make them not satisfy those conditions always. + // + // There are 6 orderings to consider (considering only inequalities). Resolve them using following rules + // 1. Timestamp has to move forward + // 2. Keep next time stamp close to expected + lastTS := f.rtpMunger.GetLast().LastTS + refTS := lastTS + expectedTS := lastTS + switchingAt := time.Now() if f.getReferenceLayerRTPTimestamp != nil { - refTS, err := f.getReferenceLayerRTPTimestamp(extPkt.Packet.Timestamp, layer, f.referenceLayerSpatial) + ts, err := f.getReferenceLayerRTPTimestamp(extPkt.Packet.Timestamp, layer, f.referenceLayerSpatial) if err == nil { - last := f.rtpMunger.GetLast() - td = refTS - last.LastTS - if td == 0 || td > (1<<31) { - f.logger.Debugw("reference timestamp out-of-order, using default", "lastTS", last.LastTS, "refTS", refTS, "td", int32(td)) - td = 1 - } else if td > uint32(0.5*float32(f.codec.ClockRate)) { - // log jumps greater than 0.5 seconds - f.logger.Debugw("reference timestamp too far ahead", "lastTS", last.LastTS, "refTS", refTS, "td", td) - } - f.logger.Debugw("reference timestamp on switch", "lastTS", last.LastTS, "refTS", refTS, "td", int32(td), "switchingAt", time.Now()) - } else { - f.logger.Debugw("reference timestamp get error, using default", "error", err) + refTS = ts } } + if f.getExpectedRTPTimestamp != nil { + ts, err := f.getExpectedRTPTimestamp(switchingAt) + if err == nil { + expectedTS = ts + } + } + nextTS, explain := getNextTimestamp(lastTS, refTS, expectedTS) + f.logger.Debugw( + "next timestamp on switch", + "switchingAt", switchingAt.String(), + "lastTS", lastTS, + "refTS", refTS, + "expectedTS", expectedTS, + "nextTS", nextTS, + "jump", nextTS-lastTS, + "explanation", explain, + ) - f.rtpMunger.UpdateSnTsOffsets(extPkt, 1, td) + f.rtpMunger.UpdateSnTsOffsets(extPkt, 1, nextTS-lastTS) f.codecMunger.UpdateOffsets(extPkt) } @@ -1734,3 +1757,40 @@ done: return float64(distance) / float64(maxSeenLayer.Temporal+1) } + +func getNextTimestamp(lastTS uint32, refTS uint32, expectedTS uint32) (uint32, string) { + isInOrder := func(val1, val2 uint32) bool { + diff := val1 - val2 + return diff != 0 && diff < (1<<31) + } + + rl := isInOrder(refTS, lastTS) + el := isInOrder(expectedTS, lastTS) + er := isInOrder(expectedTS, refTS) + + nextTS := lastTS + 1 + explain := "l = r = e" + + switch { + case rl && el && er: // lastTS < refTS < expectedTS + nextTS = uint32(float64(refTS) + 0.95*float64(expectedTS-refTS)) + explain = fmt.Sprintf("l < r < e, %d, %d", refTS-lastTS, expectedTS-refTS) + case rl && el && !er: // lastTS < expectedTS < refTS + nextTS = uint32(float64(expectedTS) + 0.5*float64(refTS-expectedTS)) + explain = fmt.Sprintf("l < e < r, %d, %d", expectedTS-lastTS, refTS-expectedTS) + case !rl && el && er: // refTS < lastTS < expectedTS + nextTS = uint32(float64(lastTS) + 0.5*float64(expectedTS-lastTS)) + explain = fmt.Sprintf("r < l < e, %d, %d", lastTS-refTS, expectedTS-lastTS) + case !rl && !el && er: // refTS < expectedTS < lastTS + nextTS = lastTS + 1 + explain = fmt.Sprintf("r < e < l, %d, %d", expectedTS-refTS, lastTS-expectedTS) + case rl && !el && !er: // expectedTS < lastTS < refTS + nextTS = uint32(float64(lastTS) + 0.5*float64(refTS-lastTS)) + explain = fmt.Sprintf("e < l < r, %d, %d", lastTS-expectedTS, refTS-lastTS) + case !rl && !el && !er: // expectedTS < refTS < lastTS + nextTS = lastTS + 1 + explain = fmt.Sprintf("e < r < l, %d, %d", refTS-expectedTS, lastTS-refTS) + } + + return nextTS, explain +} diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index ac3166f70..c99374ded 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -18,7 +18,7 @@ func disable(f *Forwarder) { } func newForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType) *Forwarder { - f := NewForwarder(kind, logger.GetLogger(), nil) + f := NewForwarder(kind, logger.GetLogger(), nil, nil) f.DetermineCodec(codec, nil) return f } diff --git a/pkg/sfu/streamtrackermanager.go b/pkg/sfu/streamtrackermanager.go index e4598e675..8e324e6a3 100644 --- a/pkg/sfu/streamtrackermanager.go +++ b/pkg/sfu/streamtrackermanager.go @@ -529,8 +529,9 @@ func (s *StreamTrackerManager) GetReferenceLayerRTPTimestamp(ts uint32, layer in // NOTE: It is possible that reference layer has stopped (due to dynacast/adaptive streaming OR publisher // constraints). It should be okay even if the layer has stopped for a long time when using modulo arithmetic for // RTP time stamp (uint32 arithmetic). - ntpDiff := float64(int64(srRef.SenderReportData.NTPTimestamp-srLayer.SenderReportData.NTPTimestamp)) / float64(1<<32) - normalizedTS := srLayer.SenderReportData.RTPTimestamp + uint32(ntpDiff*float64(s.clockRate)) + ntpDiff := srRef.SenderReportData.NTPTimestamp.Time().Sub(srLayer.SenderReportData.NTPTimestamp.Time()) + rtpDiff := ntpDiff.Nanoseconds() * int64(s.clockRate) / 1e9 + normalizedTS := srLayer.SenderReportData.RTPTimestamp + uint32(rtpDiff) // now that both RTP timestamps correspond to roughly the same NTP time, // the diff between them is the offset in RTP timestamp units between layer and referenceLayer. diff --git a/pkg/sfu/testutils/data.go b/pkg/sfu/testutils/data.go index 8b243b40f..2ab28767a 100644 --- a/pkg/sfu/testutils/data.go +++ b/pkg/sfu/testutils/data.go @@ -1,6 +1,8 @@ package testutils import ( + "time" + "github.com/pion/rtp" "github.com/pion/webrtc/v3" @@ -18,7 +20,7 @@ type TestExtPacketParams struct { SSRC uint32 PayloadSize int PaddingSize byte - ArrivalTime int64 + ArrivalTime time.Time VideoLayer buffer.VideoLayer }