dandri/livekit
1
0 Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-10-23 08:42:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Proper RTCP report past mute. (#2588)

Browse Source 
- When audio is muted, server injects silence frames which moves the
  time stamp forward and adjusts offset. That cannot be used against
  publisher side sender report. Use a pinned version.
- Ignore small changes to propagation delay even while checking for
  sharp increase. That is spamming a lot for small changes, i.e.
  existing delta is 100 micro seconds or so and the new one is 300 micro
  seconds. Also rename to `longTerm` from `smoothed` as it is a slow
  varying long term estimate of propagation delay delta. And slow down
  that adaptation more.
This commit is contained in:
Raja Subramanian authored and GitHub committed 2024-03-19 11:59:24 +05:30
1 parent c30c1be689
commit 03ada9ba76
3 files changed
+47 -40
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pkg/sfu/buffer/rtpstats_receiver.go
+38 -36
View File
@@ -48,10 +48,10 @@ const (
// To account for path changes mid-stream, if the delta of the propagation delay is consistently higher, reset.
// Reset at whichever of the below happens later.
//
// A smoothed version of delta of propagation delay is maintained and delta propagation delay exceeding
// a factor of the smoothed version is considered a sharp increase. That will trigger the start of the
// A long term version of delta of propagation delay is maintained and delta propagation delay exceeding
// a factor of the long term version is considered a sharp increase. That will trigger the start of the
// path change condition and if it persists, propagation delay will be reset.
cPropagationDelayDeltaAdaptationFactor = float64(0.1)
cPropagationDelayDeltaAdaptationFactor = float64(0.05)
cPropagationDelayDeltaHighResetNumReports = 3
cPropagationDelayDeltaHighResetWait = 10 * time.Second
)
@@ -80,7 +80,7 @@ type RTPStatsReceiver struct {
history *protoutils.Bitmap[uint64]
propagationDelay time.Duration
smoothedDeltaPropagationDelay time.Duration
longTermDeltaPropagationDelay time.Duration
propagationDelayDeltaHighCount int
propagationDelayDeltaHighStartTime time.Time
@@ -361,7 +361,7 @@ func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData)
return []interface{}{
"propagationDelay", r.propagationDelay.String(),
"receivedPropagationDelay", propagationDelay.String(),
"smoothedDeltaPropagationDelay", r.smoothedDeltaPropagationDelay.String(),
"longTermDeltaPropagationDelay", r.longTermDeltaPropagationDelay.String(),
"receivedDeltaPropagationDelay", deltaPropagationDelay.String(),
"deltaHighCount", r.propagationDelayDeltaHighCount,
"sinceDeltaHighStart", time.Since(r.propagationDelayDeltaHighStartTime).String(),
@@ -372,7 +372,7 @@ func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData)
}
initPropagationDelay := func(pd time.Duration) {
r.propagationDelay = pd
r.smoothedDeltaPropagationDelay = 0
r.longTermDeltaPropagationDelay = 0
r.propagationDelayDeltaHighCount = 0
r.propagationDelayDeltaHighStartTime = time.Time{}
}
@@ -385,39 +385,41 @@ func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData)
r.logger.Debugw("initializing propagation delay", getPropagationFields()...)
} else {
deltaPropagationDelay = propagationDelay - r.propagationDelay
if r.smoothedDeltaPropagationDelay != 0 && deltaPropagationDelay > 0 && deltaPropagationDelay > r.smoothedDeltaPropagationDelay*time.Duration(cPropagationDelayDeltaThresholdMaxFactor) {
r.logger.Debugw("sharp increase in propagation delay, skipping", getPropagationFields()...) // TODO-REMOVE
r.propagationDelayDeltaHighCount++
if r.propagationDelayDeltaHighStartTime.IsZero() {
r.propagationDelayDeltaHighStartTime = time.Now()
}
if r.propagationDelayDeltaHighCount >= cPropagationDelayDeltaHighResetNumReports && time.Since(r.propagationDelayDeltaHighStartTime) >= cPropagationDelayDeltaHighResetWait {
r.logger.Debugw("re-initializing propagation delay", append(getPropagationFields(), "newPropagationDelay", propagationDelay.String())...)
initPropagationDelay(propagationDelay)
}
} else {
if r.smoothedDeltaPropagationDelay == 0 {
r.smoothedDeltaPropagationDelay = deltaPropagationDelay
} else {
r.smoothedDeltaPropagationDelay += time.Duration(cPropagationDelayDeltaAdaptationFactor * float64(deltaPropagationDelay-r.smoothedDeltaPropagationDelay))
}
r.propagationDelayDeltaHighCount = 0
r.propagationDelayDeltaHighStartTime = time.Time{}
if deltaPropagationDelay.Abs() > cPropagationDelayDeltaThresholdMin {
factor := cPropagationDelayFallFactor
if propagationDelay > r.propagationDelay {
factor = cPropagationDelayRiseFactor
if deltaPropagationDelay.Abs() > cPropagationDelayDeltaThresholdMin { // ignore small changes
if r.longTermDeltaPropagationDelay != 0 && deltaPropagationDelay > 0 && deltaPropagationDelay > r.longTermDeltaPropagationDelay*time.Duration(cPropagationDelayDeltaThresholdMaxFactor) {
r.logger.Debugw("sharp increase in propagation delay, skipping", getPropagationFields()...) // TODO-REMOVE
r.propagationDelayDeltaHighCount++
if r.propagationDelayDeltaHighStartTime.IsZero() {
r.propagationDelayDeltaHighStartTime = time.Now()
}
if r.propagationDelayDeltaHighCount >= cPropagationDelayDeltaHighResetNumReports && time.Since(r.propagationDelayDeltaHighStartTime) >= cPropagationDelayDeltaHighResetWait {
r.logger.Debugw("re-initializing propagation delay", append(getPropagationFields(), "newPropagationDelay", propagationDelay.String())...)
initPropagationDelay(propagationDelay)
}
} else {
r.propagationDelayDeltaHighCount = 0
r.propagationDelayDeltaHighStartTime = time.Time{}
if deltaPropagationDelay.Abs() > cPropagationDelayDeltaThresholdMin {
factor := cPropagationDelayFallFactor
if propagationDelay > r.propagationDelay {
factor = cPropagationDelayRiseFactor
}
fields := append(
getPropagationFields(),
"adjustedPropagationDelay", r.propagationDelay+time.Duration(factor*float64(propagationDelay-r.propagationDelay)),
) // TODO-REMOVE
r.logger.Debugw("adapting propagation delay", fields...) // TODO-REMOVE
r.propagationDelay += time.Duration(factor * float64(propagationDelay-r.propagationDelay))
}
fields := append(
getPropagationFields(),
"adjustedPropagationDelay", r.propagationDelay+time.Duration(factor*float64(propagationDelay-r.propagationDelay)),
) // TODO-REMOVE
r.logger.Debugw("adapting propagation delay", fields...) // TODO-REMOVE
r.propagationDelay += time.Duration(factor * float64(propagationDelay-r.propagationDelay))
}
}
if r.longTermDeltaPropagationDelay == 0 {
r.longTermDeltaPropagationDelay = deltaPropagationDelay
} else {
r.longTermDeltaPropagationDelay += time.Duration(cPropagationDelayDeltaAdaptationFactor * float64(deltaPropagationDelay-r.longTermDeltaPropagationDelay))
}
}
// adjust receive time to estimated propagation delay
srDataCopy.At = ntpTime.Add(r.propagationDelay)
pkg/sfu/forwarder.go
+2 -2
View File
@@ -561,10 +561,10 @@ func (f *Forwarder) GetCurrentSpatialAndTSOffset() (int32, uint64) {
defer f.lock.RUnlock()
if f.kind == webrtc.RTPCodecTypeAudio {
return 0, f.rtpMunger.GetTSOffset()
return 0, f.rtpMunger.GetPinnedTSOffset()
}
return f.vls.GetCurrent().Spatial, f.rtpMunger.GetTSOffset()
return f.vls.GetCurrent().Spatial, f.rtpMunger.GetPinnedTSOffset()
}
func (f *Forwarder) isDeficientLocked() bool {
pkg/sfu/rtpmunger.go
+7 -2
View File
@@ -83,6 +83,7 @@ type RTPMunger struct {
extLastTS uint64
extSecondLastTS uint64
tsOffset uint64
pinnedTSOffset uint64
lastMarker bool
secondLastMarker bool
@@ -107,6 +108,7 @@ func (r *RTPMunger) DebugInfo() map[string]interface{} {
"ExtLastTS": r.extLastTS,
"ExtSecondLastTS": r.extSecondLastTS,
"TSOffset": r.tsOffset,
"PinnedTSOffset": r.pinnedTSOffset,
"LastMarker": r.lastMarker,
"SecondLastMarker": r.secondLastMarker,
}
@@ -123,8 +125,8 @@ func (r *RTPMunger) GetLast() RTPMungerState {
}
}
func (r *RTPMunger) GetTSOffset() uint64 {
return r.tsOffset
func (r *RTPMunger) GetPinnedTSOffset() uint64 {
return r.pinnedTSOffset
}
func (r *RTPMunger) SeedLast(state RTPMungerState) {
@@ -146,6 +148,8 @@ func (r *RTPMunger) SetLastSnTs(extPkt *buffer.ExtPacket) {
r.extLastTS = extPkt.ExtTimestamp
r.extSecondLastTS = extPkt.ExtTimestamp
r.tsOffset = 0
r.pinnedTSOffset = r.tsOffset
}
func (r *RTPMunger) UpdateSnTsOffsets(extPkt *buffer.ExtPacket, snAdjust uint64, tsAdjust uint64) {
@@ -155,6 +159,7 @@ func (r *RTPMunger) UpdateSnTsOffsets(extPkt *buffer.ExtPacket, snAdjust uint64,
r.updateSnOffset()
r.tsOffset = extPkt.ExtTimestamp - r.extLastTS - tsAdjust
r.pinnedTSOffset = r.tsOffset
}
func (r *RTPMunger) PacketDropped(extPkt *buffer.ExtPacket) {