From fdde44d926139c4c7fc27538eb40f35fe6a76139 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 5 Jun 2024 13:02:33 +0530 Subject: [PATCH] Attach time stamp offset to publisher sender report. (#2757) There are cases where the RTP time stamp does not increment acros mute/unmute. Seems to happen fairly consistently with React Native clients. Something like the following happens - Track is progressing - Mute at `t = x`, assume RTP time stamp at the point is `y` and RTP clock rate is `z`. - Through mute, more RTCP sender reports come in from publisher and the RTP time stamp in those reports are progressing at expected rate of `z` RTP clock ticks per second. - Forwarding path uses those sender reports from publisher to build the sender report for subscribers. - Unmute happens at `t = x + a` seconds. - Ideally packets coming in after that, should have a time stamp of `y + (a * z)`, but they tend to have something a little bit more than `y`. - RTCP sender reports also have a time stamp that goes back. SFU ignores these. - Mean while the forwarding path has adjusted to the new RTP time stamp base and it has calculated a TS offset (from publisher -> subscriber). Effectively, that offset comes out close to `(a * z)`, i. e. jump corresponding to the mute interval. - When it is time to send a RTCP sender report to subscriber, the old sender report from publisher is used (as intervening ones from publisher were rejected because time stamp is moving backwards). The problem is that the old report is used with new offset. So, it looks like time stamp jumped ahead by `a` seconds. Address it by storing time stamp offset at the time of receiving the publisher side sender report. And use that while sending subscriber side sender report. There are very edge cases where this can get mismatched, but should be rare. Hopefully, this should prevent unnecessary jumps in time stamp in RTCP sender report to subscribers. --- pkg/sfu/forwarder.go | 59 ++++++++++++++++++++++++++------------------ 1 file changed, 35 insertions(+), 24 deletions(-) diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index bc0e7f31c..87c6cc345 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -216,8 +216,9 @@ func (f ForwarderState) String() string { // ------------------------------------------------------------------- type refInfo struct { - senderReport *buffer.RTCPSenderReportData - tsOffset uint64 + senderReport *buffer.RTCPSenderReportData + tsOffset uint64 + isTSOffsetValid bool } type Forwarder struct { @@ -567,16 +568,39 @@ func (f *Forwarder) GetMaxSubscribedSpatial() int32 { return layer } +func (f *Forwarder) getReferenceLayer() (int32, int32) { + if f.lastSSRC == 0 { + return buffer.InvalidLayerSpatial, buffer.InvalidLayerSpatial + } + + if f.kind == webrtc.RTPCodecTypeAudio { + return 0, 0 + } + + currentLayerSpatial := f.vls.GetCurrent().Spatial + if currentLayerSpatial < 0 || currentLayerSpatial > buffer.DefaultMaxLayerSpatial { + return buffer.InvalidLayerSpatial, buffer.InvalidLayerSpatial + } + + if f.refIsSVC { + return 0, currentLayerSpatial + } + + return currentLayerSpatial, currentLayerSpatial +} + func (f *Forwarder) SetRefSenderReport(isSVC bool, layer int32, srData *buffer.RTCPSenderReportData) { f.lock.Lock() defer f.lock.Unlock() f.refIsSVC = isSVC - if isSVC { - layer = 0 - } + refLayer, _ := f.getReferenceLayer() if layer >= 0 && int(layer) < len(f.refInfos) { - f.refInfos[layer].senderReport = srData + f.refInfos[layer] = refInfo{srData, 0, false} + if layer == refLayer { + f.refInfos[layer].tsOffset = f.rtpMunger.GetTSOffset() + f.refInfos[layer].isTSOffsetValid = true + } } } @@ -611,7 +635,7 @@ func (f *Forwarder) clearRefSenderReportsLocked() { // By clearing sender report on (re)start of a stream, subscribers will wait for a fresh report // after unmute to send sender report. for layer := int32(0); layer < buffer.DefaultMaxLayerSpatial+1; layer++ { - f.refInfos[layer] = refInfo{nil, 0} + f.refInfos[layer] = refInfo{nil, 0, false} } } @@ -619,23 +643,12 @@ func (f *Forwarder) GetSenderReportParams() (int32, uint64, *buffer.RTCPSenderRe f.lock.RLock() defer f.lock.RUnlock() - if f.kind == webrtc.RTPCodecTypeAudio { - return 0, f.refInfos[0].tsOffset, f.refInfos[0].senderReport + refLayer, currentLayerSpatial := f.getReferenceLayer() + if refLayer == buffer.InvalidLayerSpatial || !f.refInfos[refLayer].isTSOffsetValid { + return buffer.InvalidLayerSpatial, 0, nil } - currentLayerSpatial := f.vls.GetCurrent().Spatial - if currentLayerSpatial < 0 || currentLayerSpatial > buffer.DefaultMaxLayerSpatial { - return currentLayerSpatial, 0, nil - } - - refSenderReport := f.refInfos[currentLayerSpatial].senderReport - tsOffset := f.refInfos[currentLayerSpatial].tsOffset - if f.refIsSVC { - refSenderReport = f.refInfos[0].senderReport - tsOffset = f.refInfos[0].tsOffset - } - - return currentLayerSpatial, tsOffset, refSenderReport + return currentLayerSpatial, f.refInfos[refLayer].tsOffset, f.refInfos[refLayer].senderReport } func (f *Forwarder) isDeficientLocked() bool { @@ -1588,7 +1601,6 @@ func (f *Forwarder) processSourceSwitch(extPkt *buffer.ExtPacket, layer int32) e f.codecMunger.SetLast(extPkt) f.clearRefSenderReportsLocked() - f.refInfos[layer].tsOffset = f.rtpMunger.GetTSOffset() f.logger.Debugw( "starting forwarding", @@ -1827,7 +1839,6 @@ func (f *Forwarder) processSourceSwitch(extPkt *buffer.ExtPacket, layer int32) e } f.rtpMunger.UpdateSnTsOffsets(extPkt, 1, extNextTS-extLastTS) - f.refInfos[layer].tsOffset = f.rtpMunger.GetTSOffset() f.codecMunger.UpdateOffsets(extPkt) return nil }