From 610d68a4098cd99e63f2cd0f4295450d130e26e5 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Mon, 11 Mar 2024 12:25:07 +0530 Subject: [PATCH] Clean up using publisher side clock rate. (#2568) It is not used any more. --- pkg/rtc/wrappedreceiver.go | 7 ------- pkg/sfu/buffer/rtpstats_sender.go | 29 ++++----------------------- pkg/sfu/downtrack.go | 2 +- pkg/sfu/receiver.go | 5 ----- pkg/sfu/streamtrackermanager.go | 33 ------------------------------- 5 files changed, 5 insertions(+), 71 deletions(-) diff --git a/pkg/rtc/wrappedreceiver.go b/pkg/rtc/wrappedreceiver.go index 3abd3d584..e088c9eb3 100644 --- a/pkg/rtc/wrappedreceiver.go +++ b/pkg/rtc/wrappedreceiver.go @@ -317,13 +317,6 @@ func (d *DummyReceiver) GetRedReceiver() sfu.TrackReceiver { return d } -func (d *DummyReceiver) GetCalculatedClockRate(layer int32) uint32 { - if r, ok := d.receiver.Load().(sfu.TrackReceiver); ok { - return r.GetCalculatedClockRate(layer) - } - return 0 -} - func (d *DummyReceiver) GetReferenceLayerRTPTimestamp(ts uint32, layer int32, referenceLayer int32) (uint32, error) { if r, ok := d.receiver.Load().(sfu.TrackReceiver); ok { return r.GetReferenceLayerRTPTimestamp(ts, layer, referenceLayer) diff --git a/pkg/sfu/buffer/rtpstats_sender.go b/pkg/sfu/buffer/rtpstats_sender.go index 585d9cb5c..b78dc6b2a 100644 --- a/pkg/sfu/buffer/rtpstats_sender.go +++ b/pkg/sfu/buffer/rtpstats_sender.go @@ -647,7 +647,7 @@ func (r *RTPStatsSender) GetExpectedRTPTimestamp(at time.Time) (expectedTSExt ui return } -func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, calculatedClockRate uint32) *rtcp.SenderReport { +func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32) *rtcp.SenderReport { r.lock.Lock() defer r.lock.Unlock() @@ -662,24 +662,7 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, calculatedClockRate ui } now := r.firstTime.Add(timeSinceFirst) nowNTP := mediatransportutil.ToNtpTime(now) - nowRTPExtUsingTime := r.extStartTS + uint64(timeSinceFirst.Nanoseconds()*int64(r.params.ClockRate)/1e9) - nowRTPExt := nowRTPExtUsingTime - - /* - // TODO: Bad reports or unpaced publishing contorts the calculated clock rate a lot resulting in - // subsscriber sender reports jumping around. Needs more thinking. - // - // It is possible that publisher is pacing at a slower rate. - // That would make `highestTS` to be lagging the RTP time stamp in the RTCP Sender Report from publisher. - // Check for that using calculated clock rate and use the later time stamp if applicable. - var nowRTPExtUsingRate uint64 - if calculatedClockRate != 0 { - nowRTPExtUsingRate = r.extStartTS + uint64(float64(calculatedClockRate)*timeSinceFirst.Seconds()) - if nowRTPExtUsingRate > nowRTPExt { - nowRTPExt = nowRTPExtUsingRate - } - } - */ + nowRTPExt := r.extStartTS + uint64(timeSinceFirst.Nanoseconds()*int64(r.params.ClockRate)/1e9) srData := &RTCPSenderReportData{ NTPTimestamp: nowNTP, @@ -707,9 +690,7 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, calculatedClockRate ui "timeSinceHighest", now.Sub(r.highestTime).String(), "firstTime", r.firstTime.String(), "timeSinceFirst", timeSinceFirst.String(), - "nowRTPExtUsingTime", nowRTPExtUsingTime, - "calculatedClockRate", calculatedClockRate, - // TODO "nowRTPExtUsingRate", nowRTPExtUsingRate, + "nowRTPExt", nowRTPExt, "timeSinceLastReport", timeSinceLastReport.String(), "rtpDiffSinceLastReport", rtpDiffSinceLastReport, "windowClockRate", windowClockRate, @@ -737,9 +718,7 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, calculatedClockRate ui "timeSinceHighest", now.Sub(r.highestTime).String(), "firstTime", r.firstTime.String(), "timeSinceFirst", timeSinceFirst.String(), - "nowRTPExtUsingTime", nowRTPExtUsingTime, - "calculatedClockRate", calculatedClockRate, - // TODO "nowRTPExtUsingRate", nowRTPExtUsingRate, + "nowRTPExt", nowRTPExt, ) return nil } diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 1ed09223c..1bc07b8ff 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -1307,7 +1307,7 @@ func (d *DownTrack) CreateSenderReport() *rtcp.SenderReport { if clockLayer == buffer.InvalidLayerSpatial { clockLayer = d.forwarder.GetReferenceLayerSpatial() } - return d.rtpStats.GetRtcpSenderReport(d.ssrc, d.params.Receiver.GetCalculatedClockRate(clockLayer)) + return d.rtpStats.GetRtcpSenderReport(d.ssrc) } func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan struct{} { diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 9407303ef..eec709dba 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -83,7 +83,6 @@ type TrackReceiver interface { GetTemporalLayerFpsForSpatial(layer int32) []float32 - GetCalculatedClockRate(layer int32) uint32 GetReferenceLayerRTPTimestamp(ts uint32, layer int32, referenceLayer int32) (uint32, error) GetRTCPSenderReportData(layer int32) (*buffer.RTCPSenderReportData, *buffer.RTCPSenderReportData) @@ -783,10 +782,6 @@ func (w *WebRTCReceiver) GetTemporalLayerFpsForSpatial(layer int32) []float32 { return b.GetTemporalLayerFpsForSpatial(layer) } -func (w *WebRTCReceiver) GetCalculatedClockRate(layer int32) uint32 { - return w.streamTrackerManager.GetCalculatedClockRate(layer) -} - func (w *WebRTCReceiver) GetReferenceLayerRTPTimestamp(ts uint32, layer int32, referenceLayer int32) (uint32, error) { return w.streamTrackerManager.GetReferenceLayerRTPTimestamp(ts, layer, referenceLayer) } diff --git a/pkg/sfu/streamtrackermanager.go b/pkg/sfu/streamtrackermanager.go index 8acb64518..2b80e7c25 100644 --- a/pkg/sfu/streamtrackermanager.go +++ b/pkg/sfu/streamtrackermanager.go @@ -35,8 +35,6 @@ import ( const ( senderReportThresholdSeconds = float64(60.0) - - minDurationForClockRateCalculation = 15 * time.Second ) // --------------------------------------------------- @@ -643,37 +641,6 @@ func (s *StreamTrackerManager) GetRTCPSenderReportData(layer int32) (*buffer.RTC return s.senderReports[layer].first, s.senderReports[layer].newest } -func (s *StreamTrackerManager) GetCalculatedClockRate(layer int32) uint32 { - s.senderReportMu.RLock() - defer s.senderReportMu.RUnlock() - - if layer < 0 || int(layer) >= len(s.senderReports) { - // invalid layer - return 0 - } - - srFirst := s.senderReports[layer].first - srNewest := s.senderReports[layer].newest - if srFirst == nil || srFirst.NTPTimestamp == 0 || srNewest == nil || srNewest.NTPTimestamp == 0 || srFirst.RTPTimestamp == srNewest.RTPTimestamp { - // sender reports invalid or same - return 0 - } - - if s.senderReports[layer].lastUpdated.IsZero() || time.Since(s.senderReports[layer].lastUpdated).Seconds() > senderReportThresholdSeconds { - // sender report updated too far back - return 0 - } - - tsf := srNewest.NTPTimestamp.Time().Sub(srFirst.NTPTimestamp.Time()) - if tsf < minDurationForClockRateCalculation { - // not enough time has elapsed to get a stable clock rate calculation - return 0 - } - - rdsf := srNewest.RTPTimestampExt - srFirst.RTPTimestampExt - return uint32(float64(rdsf) / tsf.Seconds()) -} - func (s *StreamTrackerManager) GetReferenceLayerRTPTimestamp(ts uint32, layer int32, referenceLayer int32) (uint32, error) { s.senderReportMu.RLock() defer s.senderReportMu.RUnlock()