From b0710f7a6a351bb87fe82b252aa443b8da1ee500 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 25 Aug 2026 02:45:40 +0530 Subject: [PATCH] Reduce locking in media track + telemetry listener on move participant. (#4790) * Reduce locking in media track + telemetry listener on move participant. Telemetry listener was not getting it from new room on room move. * update comment * fix signal bytes reporting * room aware telemetry listener in media track * data track telemetry listener --- pkg/rtc/datatrack.go | 2 +- pkg/rtc/mediatrack.go | 130 ++++++++++------------------ pkg/rtc/mediatrackreceiver.go | 10 +-- pkg/rtc/participant.go | 87 +++++++++++-------- pkg/rtc/participant_data_track.go | 2 +- pkg/rtc/signalanddatastats.go | 14 +-- pkg/rtc/subscriptionmanager.go | 19 ++-- pkg/rtc/subscriptionmanager_test.go | 13 +-- pkg/rtc/types/interfaces.go | 9 +- 9 files changed, 134 insertions(+), 152 deletions(-) diff --git a/pkg/rtc/datatrack.go b/pkg/rtc/datatrack.go index fd22f6258..2f3629b47 100644 --- a/pkg/rtc/datatrack.go +++ b/pkg/rtc/datatrack.go @@ -127,7 +127,7 @@ func (d *DataTrack) AddSubscriber(sub types.LocalParticipant) (types.DataDownTra sub.ID(), sub.Kind(), sub.KindDetails(), - sub.GetTelemetryListener(), + sub.GetTelemetryListener, sub.GetReporter(), ) dataDownTrack, err := NewDataDownTrack(DataDownTrackParams{ diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 2f48b0e58..3c167e0f2 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -46,7 +46,7 @@ var _ types.LocalMediaTrack = (*MediaTrack)(nil) // Implements MediaTrack and PublishedTrack interface type MediaTrack struct { params MediaTrackParams - buffer *buffer.Buffer + buffer atomic.Pointer[buffer.Buffer] everSubscribed atomic.Bool *MediaTrackReceiver @@ -54,26 +54,20 @@ type MediaTrack struct { dynacastManager dynacast.DynacastManager - lock sync.RWMutex - migrated bool - published bool + // serialises receiver creation in AddReceiver, i. e. the check-then-create + // of a receiver for a mime type. Held across receiver set up + addReceiverLock sync.Mutex + + regressionTargetCodecReceived atomic.Bool + + migrated atomic.Bool + published atomic.Bool rttFromXR atomic.Bool - backupCodecPolicy livekit.BackupCodecPolicy - regressionTargetCodec mime.MimeType - regressionTargetCodecReceived bool - - onSubscribedMaxQualityChange func( - trackID livekit.TrackID, - trackInfo *livekit.TrackInfo, - subscribedQualities []*livekit.SubscribedCodec, - maxSubscribedQualities []types.SubscribedCodecQuality, - ) error - onSubscribedAudioCodecChange func( - trackID livekit.TrackID, - codecs []*livekit.SubscribedAudioCodec, - ) error + backupCodecPolicy livekit.BackupCodecPolicy + // immutable after construction + regressionTargetCodec mime.MimeType } type MediaTrackParams struct { @@ -89,7 +83,7 @@ type MediaTrackParams struct { PLIThrottleConfig sfu.PLIThrottleConfig AudioConfig sfu.AudioConfig VideoConfig config.VideoConfig - TelemetryListener types.ParticipantTelemetryListener + TelemetryListener func() types.ParticipantTelemetryListener Logger logger.Logger Reporter roomobs.TrackReporter SimTracks map[uint32]interceptor.SimulcastTrackInfo @@ -101,6 +95,16 @@ type MediaTrackParams struct { EnableRTPStreamRestartDetection bool UpdateTrackInfoByVideoSizeChange bool ForceBackupCodecPolicySimulcast bool + OnSubscribedMaxQualityChange func( + trackID livekit.TrackID, + trackInfo *livekit.TrackInfo, + subscribedQualities []*livekit.SubscribedCodec, + maxSubscribedQualities []types.SubscribedCodecQuality, + ) error + OnSubscribedAudioCodecChange func( + trackID livekit.TrackID, + codecs []*livekit.SubscribedAudioCodec, + ) error } func NewMediaTrack(params MediaTrackParams, ti *livekit.TrackInfo) *MediaTrack { @@ -138,8 +142,8 @@ func NewMediaTrack(params MediaTrackParams, ti *livekit.TrackInfo) *MediaTrack { Logger: params.Logger, }) t.MediaLossProxy.OnMediaLossUpdate(func(fractionalLoss uint8) { - if t.buffer != nil { - t.buffer.SetLastFractionLostReport(fractionalLoss) + if buff := t.buffer.Load(); buff != nil { + buff.SetLastFractionLostReport(fractionalLoss) } }) t.MediaTrackReceiver.OnMediaLossFeedback(t.MediaLossProxy.HandleMaxLossFeedback) @@ -200,30 +204,6 @@ func NewMediaTrack(params MediaTrackParams, ti *livekit.TrackInfo) *MediaTrack { return t } -func (t *MediaTrack) OnSubscribedMaxQualityChange( - f func( - trackID livekit.TrackID, - trackInfo *livekit.TrackInfo, - subscribedQualities []*livekit.SubscribedCodec, - maxSubscribedQualities []types.SubscribedCodecQuality, - ) error, -) { - t.lock.Lock() - t.onSubscribedMaxQualityChange = f - t.lock.Unlock() -} - -func (t *MediaTrack) OnSubscribedAudioCodecChange( - f func( - trackID livekit.TrackID, - codecs []*livekit.SubscribedAudioCodec, - ) error, -) { - t.lock.Lock() - t.onSubscribedAudioCodecChange = f - t.lock.Unlock() -} - func (t *MediaTrack) NotifySubscriberNodeMaxQuality(nodeID livekit.NodeID, qualities []types.SubscribedCodecQuality) { if t.dynacastManager != nil { t.dynacastManager.NotifySubscriberNodeMaxQuality(nodeID, qualities) @@ -352,7 +332,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe var regressCodec bool enableRegression := t.enableRegression() - t.lock.Lock() + t.addReceiverLock.Lock() mimeType := mime.NormalizeMimeType(track.Codec().MimeType) layer := buffer.GetSpatialLayerForRid(mimeType, track.RID(), ti) if layer < 0 { @@ -364,7 +344,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe "codec", track.Codec(), "trackInfo", logger.Proto(ti), ) - t.lock.Unlock() + t.addReceiverLock.Unlock() return newCodec, false } @@ -408,7 +388,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe "mime", mimeType, "track", logger.Proto(ti), ) - t.lock.Unlock() + t.addReceiverLock.Unlock() return newCodec, false } @@ -451,11 +431,8 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe } newWR.OnStatsUpdate(func(_ *sfu.WebRTCReceiver, stat *livekit.AnalyticsStat) { // send for only one codec, either primary (priority == 0) OR regressed codec - t.lock.RLock() - regressionTargetCodecReceived := t.regressionTargetCodecReceived - t.lock.RUnlock() - if priority == 0 || regressionTargetCodecReceived { - t.params.TelemetryListener.OnTrackStats(statsKey, stat) + if priority == 0 || t.regressionTargetCodecReceived.Load() { + t.params.TelemetryListener().OnTrackStats(statsKey, stat) if cs, ok := telemetry.CondenseStat(stat); ok { t.params.Reporter.Tx(func(tx roomobs.TrackTx) { @@ -480,10 +457,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe newWR.OnMaxLayerChange(func(mimeType mime.MimeType, maxLayer int32) { // send for only one codec, either primary (priority == 0) OR regressed codec - t.lock.RLock() - regressionTargetCodecReceived := t.regressionTargetCodecReceived - t.lock.RUnlock() - if priority == 0 || regressionTargetCodecReceived { + if priority == 0 || t.regressionTargetCodecReceived.Load() { t.MediaTrackReceiver.NotifyMaxLayerChange(mimeType, maxLayer) } }) @@ -508,7 +482,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe } } - t.buffer = buff + t.buffer.Store(buff) t.MediaTrackReceiver.SetupReceiver(newWR, priority, mid) @@ -537,13 +511,13 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe if newCodec && enableRegression { if mimeType == t.regressionTargetCodec { t.params.Logger.Infow("regression target codec received", "codec", mimeType) - t.regressionTargetCodecReceived = true + t.regressionTargetCodecReceived.Store(true) regressCodec = true - } else if t.regressionTargetCodecReceived { + } else if t.regressionTargetCodecReceived.Load() { regressCodec = true } } - t.lock.Unlock() + t.addReceiverLock.Unlock() if err := wr.(*sfu.WebRTCReceiver).AddUpTrack(track, buff); err != nil { t.params.Logger.Warnw( @@ -600,7 +574,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe }) buff.OnFinalRtpStats(func(stats *livekit.RTPStats) { - t.params.TelemetryListener.OnTrackPublishRTPStats( + t.params.TelemetryListener().OnTrackPublishRTPStats( t.params.ParticipantID(), t.ID(), mimeType, @@ -681,12 +655,8 @@ func (t *MediaTrack) OnDynacastSubscribedMaxQualityChange( subscribedQualities []*livekit.SubscribedCodec, maxSubscribedQualities []types.SubscribedCodecQuality, ) { - t.lock.RLock() - onSubscribedMaxQualityChange := t.onSubscribedMaxQualityChange - t.lock.RUnlock() - - if onSubscribedMaxQualityChange != nil && !t.IsMuted() { - _ = onSubscribedMaxQualityChange( + if t.params.OnSubscribedMaxQualityChange != nil && !t.IsMuted() { + _ = t.params.OnSubscribedMaxQualityChange( t.ID(), t.ToProto(), subscribedQualities, @@ -709,35 +679,23 @@ func (t *MediaTrack) OnDynacastSubscribedMaxQualityChange( } func (t *MediaTrack) OnDynacastSubscribedAudioCodecChange(codecs []*livekit.SubscribedAudioCodec) { - t.lock.RLock() - onSubscribedAudioCodecChange := t.onSubscribedAudioCodecChange - t.lock.RUnlock() - - if onSubscribedAudioCodecChange != nil { - _ = onSubscribedAudioCodecChange(t.ID(), codecs) + if t.params.OnSubscribedAudioCodecChange != nil { + _ = t.params.OnSubscribedAudioCodecChange(t.ID(), codecs) } } func (t *MediaTrack) SetMigrated(migrated bool) { - t.lock.Lock() - t.migrated = migrated - t.lock.Unlock() + t.migrated.Store(migrated) } func (t *MediaTrack) Migrated() bool { - t.lock.RLock() - defer t.lock.RUnlock() - return t.migrated + return t.migrated.Load() } func (t *MediaTrack) SetPublished(published bool) { - t.lock.Lock() - t.published = published - t.lock.Unlock() + t.published.Store(published) } func (t *MediaTrack) Published() bool { - t.lock.RLock() - defer t.lock.RUnlock() - return t.published + return t.published.Load() } diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index cffde83b5..ad478d0d8 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -125,7 +125,7 @@ type MediaTrackReceiverParams struct { ReceiverConfig ReceiverConfig SubscriberConfig DirectionConfig AudioConfig sfu.AudioConfig - TelemetryListener types.ParticipantTelemetryListener + TelemetryListener func() types.ParticipantTelemetryListener Logger logger.Logger RegressionTargetCodec mime.MimeType PreferVideoSizeFromMedia bool @@ -949,7 +949,7 @@ func (t *MediaTrackReceiver) UpdateAudioTrack(update *livekit.UpdateLocalAudioTr t.updateTrackInfoOfReceivers() - t.params.TelemetryListener.OnTrackPublishedUpdate(t.PublisherID(), clonedInfo) + t.params.TelemetryListener().OnTrackPublishedUpdate(t.PublisherID(), clonedInfo) t.params.Logger.Debugw("updated audio track", "before", logger.Proto(trackInfo), "after", logger.Proto(clonedInfo)) } @@ -973,7 +973,7 @@ func (t *MediaTrackReceiver) UpdateVideoTrack(update *livekit.UpdateLocalVideoTr t.updateTrackInfoOfReceivers() - t.params.TelemetryListener.OnTrackPublishedUpdate(t.PublisherID(), clonedInfo) + t.params.TelemetryListener().OnTrackPublishedUpdate(t.PublisherID(), clonedInfo) t.params.Logger.Debugw("updated video track", "before", logger.Proto(trackInfo), "after", logger.Proto(clonedInfo)) } @@ -1019,7 +1019,7 @@ func (t *MediaTrackReceiver) UpdateVideoSize(mimeType mime.MimeType, sizes []cod t.updateTrackInfoOfReceivers() - t.params.TelemetryListener.OnTrackPublishedUpdate(t.PublisherID(), clonedInfo) + t.params.TelemetryListener().OnTrackPublishedUpdate(t.PublisherID(), clonedInfo) t.params.Logger.Debugw("updated video sizes", "before", logger.Proto(trackInfo), "after", logger.Proto(clonedInfo)) } @@ -1050,7 +1050,7 @@ func (t *MediaTrackReceiver) NotifyMaxLayerChange(mimeType mime.MimeType, maxLay } } - t.params.TelemetryListener.OnTrackPublishedUpdate(t.PublisherID(), ti) + t.params.TelemetryListener().OnTrackPublishedUpdate(t.PublisherID(), ti) } // GetQualityForDimension finds the closest quality to use for desired dimensions diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 3fd569ba3..7a6215f18 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -251,6 +251,7 @@ type ParticipantImpl struct { params ParticipantParams participantListener atomic.Pointer[types.LocalParticipantListener] + telemetryListener atomic.Pointer[types.ParticipantTelemetryListener] participantHelper atomic.Value // types.LocalParticipantHelper id atomic.Value // types.ParticipantID @@ -398,17 +399,8 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { p.setupSignalling() p.id.Store(params.SID) - p.dataChannelStats = NewBytesTrackStats( - p.params.Country, - BytesTrackIDForParticipantID(BytesTrackTypeData, p.ID()), - p.ID(), - params.Grants.GetParticipantKind(), - params.Grants.GetKindDetails(), - params.TelemetryListener, - params.Reporter, - ) - p.reliableDataInfo.lastPubReliableSeq.Store(params.LastPubReliableSeq) p.setListener(params.ParticipantListener) + p.setTelemetryListener(params.TelemetryListener) p.participantHelper.Store(params.ParticipantHelper) if !params.DisableSupervisor { p.supervisor = supervisor.NewParticipantSupervisor(supervisor.ParticipantSupervisorParams{Logger: params.Logger}) @@ -417,6 +409,17 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { p.version.Store(params.InitialVersion) p.timedVersion.Update(params.VersionGenerator.Next()) + p.dataChannelStats = NewBytesTrackStats( + p.params.Country, + BytesTrackIDForParticipantID(BytesTrackTypeData, p.ID()), + p.ID(), + params.Grants.GetParticipantKind(), + params.Grants.GetKindDetails(), + p.GetTelemetryListener, + params.Reporter, + ) + p.reliableDataInfo.lastPubReliableSeq.Store(params.LastPubReliableSeq) + p.migrateState.Store(types.MigrateStateInit) p.state.Store(livekit.ParticipantInfo_JOINING) @@ -500,6 +503,25 @@ func (p *ParticipantImpl) ClearParticipantListener() { p.setListener(nil) } +func (p *ParticipantImpl) setTelemetryListener(listener types.ParticipantTelemetryListener) { + if listener == nil { + p.telemetryListener.Store(nil) + return + } + p.telemetryListener.Store(&listener) +} + +func (p *ParticipantImpl) GetTelemetryListener() types.ParticipantTelemetryListener { + if l := p.telemetryListener.Load(); l != nil { + return *l + } + return &types.NullParticipantTelemetryListener{} +} + +func (p *ParticipantImpl) ClearTelemetryListener() { + p.setTelemetryListener(nil) +} + func (p *ParticipantImpl) GetCountry() string { return p.params.Country } @@ -986,14 +1008,6 @@ func (p *ParticipantImpl) TelemetryGuard() *telemetry.ReferenceGuard { return p.telemetryGuard } -func (p *ParticipantImpl) GetTelemetryListener() types.ParticipantTelemetryListener { - if p.params.TelemetryListener == nil { - return &types.NullParticipantTelemetryListener{} - } - - return p.params.TelemetryListener -} - func (p *ParticipantImpl) AddOnClose(key string, callback func(types.LocalParticipant)) { if p.isClosed.Load() { if callback != nil { @@ -1409,7 +1423,7 @@ func (p *ParticipantImpl) SetMigrateInfo( // for migrating in tracks, there is no AddTrack, so record a synthetic publish request for _, t := range mediaTracks { - p.params.TelemetryListener.OnTrackPublishRequested(p.ID(), p.Identity(), t.GetTrack(), false) + p.GetTelemetryListener().OnTrackPublishRequested(p.ID(), p.Identity(), t.GetTrack(), false) } for _, t := range dataTracks { @@ -1425,7 +1439,7 @@ func (p *ParticipantImpl) SetMigrateInfo( p.ID(), p.Kind(), p.KindDetails(), - p.params.TelemetryListener, + p.GetTelemetryListener, p.params.Reporter, ), }, @@ -2177,7 +2191,6 @@ func (p *ParticipantImpl) setupSubscriptionManager() { DataTrackResolver: func(lp types.LocalParticipant, ti livekit.TrackID) types.DataResolverResult { return p.helper().ResolveDataTrack(lp, ti) }, - TelemetryListener: p.params.TelemetryListener, OnTrackSubscribed: p.onTrackSubscribed, OnTrackUnsubscribed: p.onTrackUnsubscribed, OnSubscriptionError: p.onSubscriptionError, @@ -2864,7 +2877,7 @@ func (p *ParticipantImpl) onSubscribedMaxQualityChange( break } } - p.params.TelemetryListener.OnTrackMaxSubscribedVideoQuality( + p.GetTelemetryListener().OnTrackMaxSubscribedVideoQuality( p.ID(), ti, maxSubscribedQuality.CodecMime, @@ -3167,7 +3180,7 @@ func (p *ParticipantImpl) addPendingTrack(req *livekit.AddTrackRequest) *livekit } p.pendingTracksLock.Unlock() - p.params.TelemetryListener.OnTrackPublishRequested(p.ID(), p.Identity(), utils.CloneProto(ti), true) + p.GetTelemetryListener().OnTrackPublishRequested(p.ID(), p.Identity(), utils.CloneProto(ti), true) return nil } @@ -3192,7 +3205,7 @@ func (p *ParticipantImpl) addPendingTrack(req *livekit.AddTrackRequest) *livekit } p.pendingTracksLock.Unlock() - p.params.TelemetryListener.OnTrackPublishRequested(p.ID(), p.Identity(), utils.CloneProto(ti), true) + p.GetTelemetryListener().OnTrackPublishRequested(p.ID(), p.Identity(), utils.CloneProto(ti), true) return ti } @@ -3258,9 +3271,9 @@ func (p *ParticipantImpl) setTrackMuted(mute *livekit.MuteTrackRequest, fromAdmi if trackInfo != nil && changed { if mute.Muted { - p.params.TelemetryListener.OnTrackMuted(p.ID(), trackInfo) + p.GetTelemetryListener().OnTrackMuted(p.ID(), trackInfo) } else { - p.params.TelemetryListener.OnTrackUnmuted(p.ID(), trackInfo) + p.GetTelemetryListener().OnTrackUnmuted(p.ID(), trackInfo) } } @@ -3499,7 +3512,7 @@ func (p *ParticipantImpl) addMediaTrack(signalCid string, ti *livekit.TrackInfo) ReceiverConfig: p.params.Config.Receiver, AudioConfig: p.params.AudioConfig, VideoConfig: p.params.VideoConfig, - TelemetryListener: p.params.TelemetryListener, + TelemetryListener: p.GetTelemetryListener, Logger: LoggerWithTrack(p.pubLogger, livekit.TrackID(ti.Sid), false), Reporter: p.params.Reporter.WithTrack(ti.Sid), SubscriberConfig: p.params.Config.Subscriber, @@ -3515,11 +3528,10 @@ func (p *ParticipantImpl) addMediaTrack(signalCid string, ti *livekit.TrackInfo) EnableRTPStreamRestartDetection: p.params.EnableRTPStreamRestartDetection, UpdateTrackInfoByVideoSizeChange: p.params.UseOneShotSignallingMode, ForceBackupCodecPolicySimulcast: p.params.ForceBackupCodecPolicySimulcast, + OnSubscribedMaxQualityChange: p.onSubscribedMaxQualityChange, + OnSubscribedAudioCodecChange: p.onSubscribedAudioCodecChange, }, ti) - mt.OnSubscribedMaxQualityChange(p.onSubscribedMaxQualityChange) - mt.OnSubscribedAudioCodecChange(p.onSubscribedAudioCodecChange) - // add to published and clean up pending if p.supervisor != nil { p.supervisor.SetPublishedTrack(livekit.TrackID(ti.Sid), mt) @@ -3551,7 +3563,7 @@ func (p *ParticipantImpl) addMediaTrack(signalCid string, ti *livekit.TrackInfo) p.supervisor.ClearPublishedTrack(trackID, mt) } - p.params.TelemetryListener.OnTrackUnpublished( + p.GetTelemetryListener().OnTrackUnpublished( p.ID(), p.Identity(), mt.ToProto(), @@ -3587,7 +3599,7 @@ func (p *ParticipantImpl) handleTrackPublished(track types.MediaTrack, isMigrate if !isSynthetic { // send webhook after callbacks are complete, persistence and state handling happens // in `onTrackPublished` cb - p.params.TelemetryListener.OnTrackPublished( + p.GetTelemetryListener().OnTrackPublished( p.ID(), p.Identity(), track.ToProto(), @@ -4265,7 +4277,7 @@ func (p *ParticipantImpl) MoveToRoom(params types.MoveToRoomParams) { track.(types.LocalMediaTrack).ClearSubscriberNodes() trackInfo := track.ToProto() - p.params.TelemetryListener.OnTrackUnpublished( + p.GetTelemetryListener().OnTrackUnpublished( p.ID(), p.Identity(), trackInfo, @@ -4274,6 +4286,9 @@ func (p *ParticipantImpl) MoveToRoom(params types.MoveToRoomParams) { ) } + p.params.Reporter.ReportEndTime(time.Now()) + p.SubscriptionManager.ClearAllSubscriptions() + // fire onClose callback for original room p.lock.Lock() onClose := p.onClose @@ -4289,13 +4304,15 @@ func (p *ParticipantImpl) MoveToRoom(params types.MoveToRoomParams) { p.telemetryGuard = &telemetry.ReferenceGuard{} p.lock.Unlock() - p.params.Reporter.ReportEndTime(time.Now()) p.params.LoggerResolver.Reset() p.params.ReporterResolver.Reset() + p.setListener(params.Listener) + p.setTelemetryListener(params.TelemetryListener) p.participantHelper.Store(params.Helper) - p.SubscriptionManager.ClearAllSubscriptions() + p.id.Store(params.ParticipantID) + grants := p.grants.Load().Clone() grants.Video.Room = string(params.RoomName) p.grants.Store(grants) diff --git a/pkg/rtc/participant_data_track.go b/pkg/rtc/participant_data_track.go index df9854744..eff47b82b 100644 --- a/pkg/rtc/participant_data_track.go +++ b/pkg/rtc/participant_data_track.go @@ -121,7 +121,7 @@ func (p *ParticipantImpl) HandlePublishDataTrackRequest(req *livekit.PublishData p.ID(), p.Kind(), p.KindDetails(), - p.params.TelemetryListener, + p.GetTelemetryListener, p.params.Reporter, ), }, diff --git a/pkg/rtc/signalanddatastats.go b/pkg/rtc/signalanddatastats.go index fe83bd9f4..fd7227d6e 100644 --- a/pkg/rtc/signalanddatastats.go +++ b/pkg/rtc/signalanddatastats.go @@ -60,7 +60,7 @@ type BytesTrackStats struct { sendMessages, recvMessages atomic.Uint32 totalSendBytes, totalRecvBytes atomic.Uint64 totalSendMessages, totalRecvMessages atomic.Uint32 - telemetryListener types.ParticipantTelemetryListener + telemetryListener func() types.ParticipantTelemetryListener reporter roomobs.TrackReporter done core.Fuse } @@ -71,7 +71,7 @@ func NewBytesTrackStats( pID livekit.ParticipantID, kind livekit.ParticipantInfo_Kind, kindDetails []livekit.ParticipantInfo_KindDetail, - telemetryListener types.ParticipantTelemetryListener, + telemetryListener func() types.ParticipantTelemetryListener, participantReporter roomobs.ParticipantSessionReporter, ) *BytesTrackStats { s := &BytesTrackStats{ @@ -134,7 +134,7 @@ func (s *BytesTrackStats) Stop() { func (s *BytesTrackStats) report() { if recv := s.recv.Swap(0); recv > 0 { packets := s.recvMessages.Swap(0) - s.telemetryListener.OnTrackStats( + s.telemetryListener().OnTrackStats( telemetry.StatsKeyForData(s.country, livekit.StreamType_UPSTREAM, s.pID, s.trackID), &livekit.AnalyticsStat{ Streams: []*livekit.AnalyticsStream{ @@ -149,7 +149,7 @@ func (s *BytesTrackStats) report() { if send := s.send.Swap(0); send > 0 { packets := s.sendMessages.Swap(0) - s.telemetryListener.OnTrackStats( + s.telemetryListener().OnTrackStats( telemetry.StatsKeyForData(s.country, livekit.StreamType_DOWNSTREAM, s.pID, s.trackID), &livekit.AnalyticsStat{ Streams: []*livekit.AnalyticsStream{ @@ -217,7 +217,7 @@ func NewBytesSignalStats( trackResolver: trackReporterResolver, } b.BytesTrackStats = BytesTrackStats{ - telemetryListener: b, + telemetryListener: b.telemetryListener, reporter: trackReporter, } return b @@ -295,6 +295,10 @@ func (s *BytesSignalStats) worker() { close(s.stopped) } +func (s *BytesSignalStats) telemetryListener() types.ParticipantTelemetryListener { + return s +} + func (s *BytesSignalStats) OnTrackStats(key telemetry.StatsKey, stat *livekit.AnalyticsStat) { stat.RoomId, stat.RoomName = s.ri.Sid, s.ri.Name s.telemetry.TrackStats(livekit.RoomID(s.ri.Sid), livekit.RoomName(s.ri.Name), key, stat) diff --git a/pkg/rtc/subscriptionmanager.go b/pkg/rtc/subscriptionmanager.go index 7e06a8f68..3b5ddd849 100644 --- a/pkg/rtc/subscriptionmanager.go +++ b/pkg/rtc/subscriptionmanager.go @@ -57,7 +57,6 @@ type SubscriptionManagerParams struct { OnTrackSubscribed func(subTrack types.SubscribedTrack) OnTrackUnsubscribed func(subTrack types.SubscribedTrack) OnSubscriptionError func(trackID livekit.TrackID, fatal bool, err error) - TelemetryListener types.ParticipantTelemetryListener SubscriptionLimitVideo, SubscriptionLimitAudio int32 @@ -503,7 +502,7 @@ func (m *SubscriptionManager) reconcileSubscription(s *mediaTrackSubscription) { numAttempts := s.getNumAttempts() if numAttempts == 0 { - m.params.TelemetryListener.OnTrackSubscribeRequested( + m.params.Participant.GetTelemetryListener().OnTrackSubscribeRequested( s.subscriberID, &livekit.TrackInfo{ Sid: string(s.trackID), @@ -523,14 +522,14 @@ func (m *SubscriptionManager) reconcileSubscription(s *mediaTrackSubscription) { // - ErrSubscriptionLimitExceeded: the participant have reached the limit of subscriptions, wait for the other subscription to be unsubscribed // We'll still log an event to reflect this in telemetry since it's been too long if s.durationSinceStart() > subscriptionTimeout { - s.maybeRecordError(m.params.TelemetryListener, err, true) + s.maybeRecordError(m.params.Participant.GetTelemetryListener(), err, true) } case ErrTrackNotFound: // source track was never published or closed // if after timeout we'd unsubscribe from it. // this is the *only* case we'd change desired state if s.durationSinceStart() > notFoundTimeout { - s.maybeRecordError(m.params.TelemetryListener, err, true) + s.maybeRecordError(m.params.Participant.GetTelemetryListener(), err, true) s.logger.Infow("unsubscribing from track after notFoundTimeout", "error", err) s.setDesired(false) m.queueReconcile(s.trackID) @@ -543,7 +542,7 @@ func (m *SubscriptionManager) reconcileSubscription(s *mediaTrackSubscription) { "failed to subscribe, triggering error handler", err, "attempt", s.getNumAttempts(), ) - s.maybeRecordError(m.params.TelemetryListener, err, false) + s.maybeRecordError(m.params.Participant.GetTelemetryListener(), err, false) m.params.OnSubscriptionError(s.trackID, true, err) } else { s.logger.Debugw( @@ -582,7 +581,7 @@ func (m *SubscriptionManager) reconcileSubscription(s *mediaTrackSubscription) { wait := min(time.Since(activeAt), s.durationSinceStart()) if wait > subscriptionTimeout { s.logger.Warnw("track not bound after timeout", nil) - s.maybeRecordError(m.params.TelemetryListener, ErrTrackNotBound, false) + s.maybeRecordError(m.params.Participant.GetTelemetryListener(), ErrTrackNotBound, false) m.params.OnSubscriptionError(s.trackID, true, ErrTrackNotBound) } } @@ -873,13 +872,13 @@ func (m *SubscriptionManager) addSubscriber(sub *mediaTrackSubscription, track t subTrack.AddOnBind(func(err error) { if err != nil { sub.logger.Infow("failed to bind track", "err", err) - sub.maybeRecordError(m.params.TelemetryListener, err, true) + sub.maybeRecordError(m.params.Participant.GetTelemetryListener(), err, true) m.UnsubscribeFromTrack(trackID) m.params.OnSubscriptionError(trackID, false, err) return } sub.setBound() - sub.maybeRecordSuccess(m.params.TelemetryListener) + sub.maybeRecordSuccess(m.params.Participant.GetTelemetryListener()) }) sub.setSubscribedTrack(subTrack) @@ -1003,7 +1002,7 @@ func (m *SubscriptionManager) handleSubscribedTrackClose(s *mediaTrackSubscripti // * the participant isn't closing // * it's not a migration if wasBound { - m.params.TelemetryListener.OnTrackUnsubscribed( + m.params.Participant.GetTelemetryListener().OnTrackUnsubscribed( s.subscriberID, &livekit.TrackInfo{Sid: string(s.trackID), Type: subTrack.MediaTrack().Kind()}, !isExpectedToResume, @@ -1013,7 +1012,7 @@ func (m *SubscriptionManager) handleSubscribedTrackClose(s *mediaTrackSubscripti if dt != nil { stats := dt.GetTrackStats() if stats != nil { - m.params.TelemetryListener.OnTrackSubscribeRTPStats( + m.params.Participant.GetTelemetryListener().OnTrackSubscribeRTPStats( s.subscriberID, s.trackID, dt.Mime(), diff --git a/pkg/rtc/subscriptionmanager_test.go b/pkg/rtc/subscriptionmanager_test.go index 6c4618927..fbbfc0428 100644 --- a/pkg/rtc/subscriptionmanager_test.go +++ b/pkg/rtc/subscriptionmanager_test.go @@ -82,7 +82,7 @@ func TestSubscribe(t *testing.T) { require.Equal(t, "pubID", string(sm.GetSubscribedParticipants()[0])) // ensure telemetry events are sent - tl := sm.params.TelemetryListener.(*typesfakes.FakeParticipantTelemetryListener) + tl := sm.params.Participant.GetTelemetryListener().(*typesfakes.FakeParticipantTelemetryListener) require.Equal(t, 1, tl.OnTrackSubscribeRequestedCallCount()) // ensure bound @@ -141,7 +141,7 @@ func TestSubscribe(t *testing.T) { require.Len(t, sm.GetSubscribedTracks(), 0) // trackSubscribed telemetry not sent - tl := sm.params.TelemetryListener.(*typesfakes.FakeParticipantTelemetryListener) + tl := sm.params.Participant.GetTelemetryListener().(*typesfakes.FakeParticipantTelemetryListener) require.Equal(t, 1, tl.OnTrackSubscribeRequestedCallCount()) require.Equal(t, 0, tl.OnTrackSubscribedCallCount()) @@ -250,7 +250,7 @@ func TestUnsubscribe(t *testing.T) { require.Len(t, sm.GetSubscribedTracks(), 0) require.False(t, res.TrackChangedNotifier.HasObservers()) - tl := sm.params.TelemetryListener.(*typesfakes.FakeParticipantTelemetryListener) + tl := sm.params.Participant.GetTelemetryListener().(*typesfakes.FakeParticipantTelemetryListener) require.Equal(t, 1, tl.OnTrackUnsubscribedCallCount()) } @@ -390,7 +390,7 @@ func TestSubscriptionLimits(t *testing.T) { require.Equal(t, "pubID", string(sm.GetSubscribedParticipants()[0])) // ensure telemetry events are sent - tl := sm.params.TelemetryListener.(*typesfakes.FakeParticipantTelemetryListener) + tl := sm.params.Participant.GetTelemetryListener().(*typesfakes.FakeParticipantTelemetryListener) require.Equal(t, 1, tl.OnTrackSubscribeRequestedCallCount()) // ensure bound @@ -538,6 +538,10 @@ func newTestSubscriptionManagerWithParams(params testSubscriptionParams) *Subscr p.IDReturns("subID") p.IdentityReturns("sub") p.KindReturns(livekit.ParticipantInfo_STANDARD) + + tl := &typesfakes.FakeParticipantTelemetryListener{} + p.GetTelemetryListenerReturns(tl) + return NewSubscriptionManager(SubscriptionManagerParams{ Participant: p, Logger: logger.GetLogger(), @@ -547,7 +551,6 @@ func newTestSubscriptionManagerWithParams(params testSubscriptionParams) *Subscr TrackResolver: func(sub types.LocalParticipant, trackID livekit.TrackID) types.MediaResolverResult { return types.MediaResolverResult{} }, - TelemetryListener: &typesfakes.FakeParticipantTelemetryListener{}, SubscriptionLimitAudio: params.SubscriptionLimitAudio, SubscriptionLimitVideo: params.SubscriptionLimitVideo, }) diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 9c3a23db3..a67b19ffd 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -370,10 +370,11 @@ type AddTrackParams struct { } type MoveToRoomParams struct { - RoomName livekit.RoomName - ParticipantID livekit.ParticipantID - Listener LocalParticipantListener - Helper LocalParticipantHelper + RoomName livekit.RoomName + ParticipantID livekit.ParticipantID + Listener LocalParticipantListener + TelemetryListener ParticipantTelemetryListener + Helper LocalParticipantHelper } type DataMessageCache struct {