dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-29 07:39:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Reduce locking in media track + telemetry listener on move participant. (#4790)

Browse Source 
* 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
This commit is contained in:
Raja Subramanian
2026-08-25 02:45:40 +05:30
committed by GitHub
parent 5c7713b336
commit b0710f7a6a
9 changed files with 134 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/datatrack.go
+1 -1
View File
@@ -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{
pkg/rtc/mediatrack.go
+44 -86
View File
@@ -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()
}
pkg/rtc/mediatrackreceiver.go
+5 -5
View File
@@ -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
pkg/rtc/participant.go
+52 -35
View File
@@ -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)
pkg/rtc/participant_data_track.go
+1 -1
View File
@@ -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,
),
},
pkg/rtc/signalanddatastats.go
+9 -5
View File
@@ -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)
pkg/rtc/subscriptionmanager.go
+9 -10
View File
@@ -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(),
pkg/rtc/subscriptionmanager_test.go
+8 -5
View File
@@ -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,
})
pkg/rtc/types/interfaces.go
+5 -4
View File
@@ -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 {