dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-07-18 01:46:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make sender report pass through an option. (#2861)

Browse Source 
Enabled by default.

Also, tweak the long term propagation delay a bit. The first propagation
delay itself was too high and the long term initialized with a high
value. Prevent that and also ensure large negtaives do not have an
effect by using a lower bound of 0. Lower bound of 0 is okay as the main
purpose is to track sustained high positive values.
This commit is contained in:
Raja Subramanian
2024-07-15 11:27:31 +05:30
committed by GitHub
parent a995c71f84
commit faa66d1138
8 changed files with 144 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/mediatracksubscriptions.go
+13 -12
View File
@@ -129,18 +129,19 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr *
}
downTrack, err := sfu.NewDownTrack(sfu.DowntrackParams{
Codecs: codecs,
Source: t.params.MediaTrack.Source(),
Receiver: wr,
BufferFactory: sub.GetBufferFactory(),
SubID: subscriberID,
StreamID: streamID,
MaxTrack: maxTrack,
PlayoutDelayLimit: sub.GetPlayoutDelayConfig(),
Pacer: sub.GetPacer(),
Trailer: trailer,
Logger: LoggerWithTrack(sub.GetLogger().WithComponent(sutils.ComponentSub), trackID, t.params.IsRelayed),
RTCPWriter: sub.WriteSubscriberRTCP,
Codecs: codecs,
Source: t.params.MediaTrack.Source(),
Receiver: wr,
BufferFactory: sub.GetBufferFactory(),
SubID: subscriberID,
StreamID: streamID,
MaxTrack: maxTrack,
PlayoutDelayLimit: sub.GetPlayoutDelayConfig(),
Pacer: sub.GetPacer(),
Trailer: trailer,
Logger: LoggerWithTrack(sub.GetLogger().WithComponent(sutils.ComponentSub), trackID, t.params.IsRelayed),
RTCPWriter: sub.WriteSubscriberRTCP,
DisableSenderReportPassThrough: sub.GetDisableSenderReportPassThrough(),
})
if err != nil {
return nil, err
pkg/rtc/participant.go
+36 -31
View File
@@ -112,37 +112,38 @@ type ParticipantParams struct {
PLIThrottleConfig config.PLIThrottleConfig
CongestionControlConfig config.CongestionControlConfig
// codecs that are enabled for this room
PublishEnabledCodecs []*livekit.Codec
SubscribeEnabledCodecs []*livekit.Codec
Logger logger.Logger
SimTracks map[uint32]SimulcastTrackInfo
Grants *auth.ClaimGrants
InitialVersion uint32
ClientConf *livekit.ClientConfiguration
ClientInfo ClientInfo
Region string
Migration bool
AdaptiveStream bool
AllowTCPFallback bool
TCPFallbackRTTThreshold int
AllowUDPUnstableFallback bool
TURNSEnabled bool
GetParticipantInfo func(pID livekit.ParticipantID) *livekit.ParticipantInfo
GetRegionSettings func(ip string) *livekit.RegionSettings
DisableSupervisor bool
ReconnectOnPublicationError bool
ReconnectOnSubscriptionError bool
ReconnectOnDataChannelError bool
DataChannelMaxBufferedAmount uint64
VersionGenerator utils.TimedVersionGenerator
TrackResolver types.MediaTrackResolver
DisableDynacast bool
SubscriberAllowPause bool
SubscriptionLimitAudio int32
SubscriptionLimitVideo int32
PlayoutDelay *livekit.PlayoutDelay
SyncStreams bool
ForwardStats *sfu.ForwardStats
PublishEnabledCodecs []*livekit.Codec
SubscribeEnabledCodecs []*livekit.Codec
Logger logger.Logger
SimTracks map[uint32]SimulcastTrackInfo
Grants *auth.ClaimGrants
InitialVersion uint32
ClientConf *livekit.ClientConfiguration
ClientInfo ClientInfo
Region string
Migration bool
AdaptiveStream bool
AllowTCPFallback bool
TCPFallbackRTTThreshold int
AllowUDPUnstableFallback bool
TURNSEnabled bool
GetParticipantInfo func(pID livekit.ParticipantID) *livekit.ParticipantInfo
GetRegionSettings func(ip string) *livekit.RegionSettings
DisableSupervisor bool
ReconnectOnPublicationError bool
ReconnectOnSubscriptionError bool
ReconnectOnDataChannelError bool
DataChannelMaxBufferedAmount uint64
VersionGenerator utils.TimedVersionGenerator
TrackResolver types.MediaTrackResolver
DisableDynacast bool
SubscriberAllowPause bool
SubscriptionLimitAudio int32
SubscriptionLimitVideo int32
PlayoutDelay *livekit.PlayoutDelay
SyncStreams bool
ForwardStats *sfu.ForwardStats
DisableSenderReportPassThrough bool
}
type ParticipantImpl struct {
@@ -320,6 +321,10 @@ func (p *ParticipantImpl) GetPacer() pacer.Pacer {
return p.TransportManager.GetSubscriberPacer()
}
func (p *ParticipantImpl) GetDisableSenderReportPassThrough() bool {
return p.params.DisableSenderReportPassThrough
}
func (p *ParticipantImpl) ID() livekit.ParticipantID {
return p.params.SID
}
pkg/rtc/types/interfaces.go
+2
View File
@@ -427,6 +427,8 @@ type LocalParticipant interface {
SetSubscriberChannelCapacity(channelCapacity int64)
GetPacer() pacer.Pacer
GetDisableSenderReportPassThrough() bool
}
// Room is a container of participants, and can provide room-level actions
pkg/rtc/types/typesfakes/fake_local_participant.go
+65
View File
@@ -266,6 +266,16 @@ type FakeLocalParticipant struct {
getConnectionQualityReturnsOnCall map[int]struct {
result1 *livekit.ConnectionQualityInfo
}
GetDisableSenderReportPassThroughStub func() bool
getDisableSenderReportPassThroughMutex sync.RWMutex
getDisableSenderReportPassThroughArgsForCall []struct {
}
getDisableSenderReportPassThroughReturns struct {
result1 bool
}
getDisableSenderReportPassThroughReturnsOnCall map[int]struct {
result1 bool
}
GetICEConnectionDetailsStub func() []*types.ICEConnectionDetails
getICEConnectionDetailsMutex sync.RWMutex
getICEConnectionDetailsArgsForCall []struct {
@@ -2306,6 +2316,59 @@ func (fake *FakeLocalParticipant) GetConnectionQualityReturnsOnCall(i int, resul
}{result1}
}
func (fake *FakeLocalParticipant) GetDisableSenderReportPassThrough() bool {
fake.getDisableSenderReportPassThroughMutex.Lock()
ret, specificReturn := fake.getDisableSenderReportPassThroughReturnsOnCall[len(fake.getDisableSenderReportPassThroughArgsForCall)]
fake.getDisableSenderReportPassThroughArgsForCall = append(fake.getDisableSenderReportPassThroughArgsForCall, struct {
}{})
stub := fake.GetDisableSenderReportPassThroughStub
fakeReturns := fake.getDisableSenderReportPassThroughReturns
fake.recordInvocation("GetDisableSenderReportPassThrough", []interface{}{})
fake.getDisableSenderReportPassThroughMutex.Unlock()
if stub != nil {
return stub()
}
if specificReturn {
return ret.result1
}
return fakeReturns.result1
}
func (fake *FakeLocalParticipant) GetDisableSenderReportPassThroughCallCount() int {
fake.getDisableSenderReportPassThroughMutex.RLock()
defer fake.getDisableSenderReportPassThroughMutex.RUnlock()
return len(fake.getDisableSenderReportPassThroughArgsForCall)
}
func (fake *FakeLocalParticipant) GetDisableSenderReportPassThroughCalls(stub func() bool) {
fake.getDisableSenderReportPassThroughMutex.Lock()
defer fake.getDisableSenderReportPassThroughMutex.Unlock()
fake.GetDisableSenderReportPassThroughStub = stub
}
func (fake *FakeLocalParticipant) GetDisableSenderReportPassThroughReturns(result1 bool) {
fake.getDisableSenderReportPassThroughMutex.Lock()
defer fake.getDisableSenderReportPassThroughMutex.Unlock()
fake.GetDisableSenderReportPassThroughStub = nil
fake.getDisableSenderReportPassThroughReturns = struct {
result1 bool
}{result1}
}
func (fake *FakeLocalParticipant) GetDisableSenderReportPassThroughReturnsOnCall(i int, result1 bool) {
fake.getDisableSenderReportPassThroughMutex.Lock()
defer fake.getDisableSenderReportPassThroughMutex.Unlock()
fake.GetDisableSenderReportPassThroughStub = nil
if fake.getDisableSenderReportPassThroughReturnsOnCall == nil {
fake.getDisableSenderReportPassThroughReturnsOnCall = make(map[int]struct {
result1 bool
})
}
fake.getDisableSenderReportPassThroughReturnsOnCall[i] = struct {
result1 bool
}{result1}
}
func (fake *FakeLocalParticipant) GetICEConnectionDetails() []*types.ICEConnectionDetails {
fake.getICEConnectionDetailsMutex.Lock()
ret, specificReturn := fake.getICEConnectionDetailsReturnsOnCall[len(fake.getICEConnectionDetailsArgsForCall)]
@@ -6646,6 +6709,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
defer fake.getClientInfoMutex.RUnlock()
fake.getConnectionQualityMutex.RLock()
defer fake.getConnectionQualityMutex.RUnlock()
fake.getDisableSenderReportPassThroughMutex.RLock()
defer fake.getDisableSenderReportPassThroughMutex.RUnlock()
fake.getICEConnectionDetailsMutex.RLock()
defer fake.getICEConnectionDetailsMutex.RUnlock()
fake.getLoggerMutex.RLock()
pkg/sfu/buffer/rtpstats_base.go
+2 -4
View File
@@ -37,8 +37,6 @@ const (
cFirstPacketTimeAdjustWindow = 2 * time.Minute
cFirstPacketTimeAdjustThreshold = 15 * 1e9
cPassthroughNTPTimestamp = true
cSequenceNumberLargeJumpThreshold = 100
)
@@ -122,8 +120,8 @@ type RTCPSenderReportData struct {
AtAdjusted time.Time
}
func (r *RTCPSenderReportData) PropagationDelay() time.Duration {
if cPassthroughNTPTimestamp {
func (r *RTCPSenderReportData) PropagationDelay(passThrough bool) time.Duration {
if passThrough {
return 0
}
pkg/sfu/buffer/rtpstats_receiver.go
+9 -7
View File
@@ -260,7 +260,7 @@ func (r *RTPStatsReceiver) Update(
r.largeJumpCount++
if (r.largeJumpCount-1)%100 == 0 {
r.logger.Warnw(
"large sequence number gap OR time reversed", nil,
"large sequence number gap", nil,
append(getLoggingFields(), "count", r.largeJumpCount)...,
)
}
@@ -492,7 +492,6 @@ func (r *RTPStatsReceiver) updatePropagationDelayAndRecordSenderReport(srData *R
deltaPropagationDelay = propagationDelay - r.propagationDelay
if deltaPropagationDelay > cPropagationDelayDeltaThresholdMin { // ignore small changes for path change consideration
if r.longTermDeltaPropagationDelay != 0 &&
deltaPropagationDelay > 0 &&
deltaPropagationDelay > r.longTermDeltaPropagationDelay*time.Duration(cPropagationDelayDeltaThresholdMaxFactor) {
r.logger.Debugw("sharp increase in propagation delay", getPropagationFields()...)
r.propagationDelayDeltaHighCount++
@@ -522,10 +521,10 @@ func (r *RTPStatsReceiver) updatePropagationDelayAndRecordSenderReport(srData *R
r.propagationDelay += time.Duration(factor * float64(propagationDelay-r.propagationDelay))
}
if r.longTermDeltaPropagationDelay == 0 {
r.longTermDeltaPropagationDelay = deltaPropagationDelay
} else {
if deltaPropagationDelay < cPropagationDelayDeltaLongTermAdaptationThreshold {
if deltaPropagationDelay < cPropagationDelayDeltaLongTermAdaptationThreshold {
if r.longTermDeltaPropagationDelay == 0 {
r.longTermDeltaPropagationDelay = deltaPropagationDelay
} else {
// do not adapt to large +ve spikes, can happen when channel is congested and reports are delivered very late
// if the spike is in fact a path change, it will persist and handled by path change detection above
sinceLastReport := srData.NTPTimestamp.Time().Sub(r.srNewest.NTPTimestamp.Time())
@@ -533,6 +532,9 @@ func (r *RTPStatsReceiver) updatePropagationDelayAndRecordSenderReport(srData *R
r.longTermDeltaPropagationDelay += time.Duration(adaptationFactor * float64(deltaPropagationDelay-r.longTermDeltaPropagationDelay))
}
}
if r.longTermDeltaPropagationDelay < 0 {
r.longTermDeltaPropagationDelay = 0
}
}
// adjust receive time to estimated propagation delay
srData.AtAdjusted = ntpTime.Add(r.propagationDelay)
@@ -563,8 +565,8 @@ func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData)
return false
}
r.checkRTPClockSkewForSenderReport(srDataExt)
r.updatePropagationDelayAndRecordSenderReport(srDataExt)
r.checkRTPClockSkewForSenderReport(srDataExt)
r.checkRTPClockSkewAgainstMediaPathForSenderReport(srDataExt)
if err, loggingFields := r.maybeAdjustFirstPacketTime(r.srNewest, 0, r.timestamp.GetExtendedStart()); err != nil {
pkg/sfu/buffer/rtpstats_sender.go
+2 -2
View File
@@ -606,7 +606,7 @@ func (r *RTPStatsSender) GetExpectedRTPTimestamp(at time.Time) (expectedTSExt ui
return
}
func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *RTCPSenderReportData, tsOffset uint64) *rtcp.SenderReport {
func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *RTCPSenderReportData, tsOffset uint64, passThrough bool) *rtcp.SenderReport {
r.lock.Lock()
defer r.lock.Unlock()
@@ -620,7 +620,7 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *RTCPS
nowNTP mediatransportutil.NtpTime
nowRTPExt uint64
)
if cPassthroughNTPTimestamp {
if passThrough {
nowNTP = publisherSRData.NTPTimestamp
nowRTPExt = publisherSRData.RTPTimestampExt - tsOffset
} else {
pkg/sfu/downtrack.go
+15 -14
View File
@@ -202,18 +202,19 @@ type DownTrackStreamAllocatorListener interface {
type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport)
type DowntrackParams struct {
Codecs []webrtc.RTPCodecParameters
Source livekit.TrackSource
Receiver TrackReceiver
BufferFactory *buffer.Factory
SubID livekit.ParticipantID
StreamID string
MaxTrack int
PlayoutDelayLimit *livekit.PlayoutDelay
Pacer pacer.Pacer
Logger logger.Logger
Trailer []byte
RTCPWriter func([]rtcp.Packet) error
Codecs []webrtc.RTPCodecParameters
Source livekit.TrackSource
Receiver TrackReceiver
BufferFactory *buffer.Factory
SubID livekit.ParticipantID
StreamID string
MaxTrack int
PlayoutDelayLimit *livekit.PlayoutDelay
Pacer pacer.Pacer
Logger logger.Logger
Trailer []byte
RTCPWriter func([]rtcp.Packet) error
DisableSenderReportPassThrough bool
}
// DownTrack implements TrackLocal, is the track used to write packets
@@ -813,7 +814,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
_, _, refSenderReport := d.forwarder.GetSenderReportParams()
if refSenderReport != nil {
actExtCopy := *extPkt.AbsCaptureTimeExt
if err = actExtCopy.Rewrite(refSenderReport.PropagationDelay()); err == nil {
if err = actExtCopy.Rewrite(refSenderReport.PropagationDelay(!d.params.DisableSenderReportPassThrough)); err == nil {
actBytes, err = actExtCopy.Marshal()
if err == nil {
extensions = append(
@@ -1404,7 +1405,7 @@ func (d *DownTrack) CreateSenderReport() *rtcp.SenderReport {
}
_, tsOffset, refSenderReport := d.forwarder.GetSenderReportParams()
return d.rtpStats.GetRtcpSenderReport(d.ssrc, refSenderReport, tsOffset)
return d.rtpStats.GetRtcpSenderReport(d.ssrc, refSenderReport, tsOffset, !d.params.DisableSenderReportPassThrough)
}
func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan struct{} {