diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 8a862bd81..f37fb05fc 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -1059,7 +1059,7 @@ func (p *ParticipantImpl) setupSubscriptionManager() { Telemetry: p.params.Telemetry, OnTrackSubscribed: p.onTrackSubscribed, OnTrackUnsubscribed: p.onTrackUnsubscribed, - OnSubcriptionError: p.onSubscriptionError, + OnSubscriptionError: p.onSubscriptionError, }) } diff --git a/pkg/rtc/subscriptionmanager.go b/pkg/rtc/subscriptionmanager.go index acff6cbf2..43fee0d39 100644 --- a/pkg/rtc/subscriptionmanager.go +++ b/pkg/rtc/subscriptionmanager.go @@ -48,7 +48,7 @@ type SubscriptionManagerParams struct { TrackResolver types.MediaTrackResolver OnTrackSubscribed func(subTrack types.SubscribedTrack) OnTrackUnsubscribed func(subTrack types.SubscribedTrack) - OnSubcriptionError func(trackID livekit.TrackID) + OnSubscriptionError func(trackID livekit.TrackID) Telemetry telemetry.TelemetryService } @@ -58,21 +58,24 @@ type SubscriptionManager struct { lock sync.RWMutex subscriptions map[livekit.TrackID]*trackSubscription subscribedTo map[livekit.ParticipantID]map[livekit.TrackID]struct{} - reconcileCh chan livekit.TrackID - closeCh chan struct{} - doneCh chan struct{} + // keeps track of tracks that are already queued for reconcile to avoid duplicating reconcile requests + pendingReconcile map[livekit.TrackID]struct{} + reconcileCh chan livekit.TrackID + closeCh chan struct{} + doneCh chan struct{} onSubscribeStatusChanged func(publisherID livekit.ParticipantID, subscribed bool) } func NewSubscriptionManager(params SubscriptionManagerParams) *SubscriptionManager { m := &SubscriptionManager{ - params: params, - subscriptions: make(map[livekit.TrackID]*trackSubscription), - subscribedTo: make(map[livekit.ParticipantID]map[livekit.TrackID]struct{}), - reconcileCh: make(chan livekit.TrackID, 50), - closeCh: make(chan struct{}), - doneCh: make(chan struct{}), + params: params, + subscriptions: make(map[livekit.TrackID]*trackSubscription), + subscribedTo: make(map[livekit.ParticipantID]map[livekit.TrackID]struct{}), + pendingReconcile: make(map[livekit.TrackID]struct{}), + reconcileCh: make(chan livekit.TrackID, 50), + closeCh: make(chan struct{}), + doneCh: make(chan struct{}), } go m.reconcileWorker() @@ -307,7 +310,7 @@ func (m *SubscriptionManager) reconcileSubscription(s *trackSubscription) { "attempt", numAttempts, ) s.maybeRecordError(m.params.Telemetry, m.params.Participant.ID(), err, false) - m.params.OnSubcriptionError(s.trackID) + m.params.OnSubscriptionError(s.trackID) } else { s.logger.Debugw("failed to subscribe, retrying", "error", err, @@ -343,13 +346,20 @@ func (m *SubscriptionManager) reconcileSubscription(s *trackSubscription) { if s.durationSinceStart() > subscriptionTimeout { s.logger.Errorw("track not bound after timeout", nil) s.maybeRecordError(m.params.Telemetry, m.params.Participant.ID(), ErrTrackNotBound, false) - m.params.OnSubcriptionError(s.trackID) + m.params.OnSubscriptionError(s.trackID) } } } -// trigger an immediate reconcilation, when trackID is empty, will reconcile all subscriptions +// trigger an immediate reconciliation, when trackID is empty, will reconcile all subscriptions func (m *SubscriptionManager) queueReconcile(trackID livekit.TrackID) { + m.lock.Lock() + if _, ok := m.pendingReconcile[trackID]; ok { + // already reconciled + m.lock.Unlock() + return + } + m.lock.Unlock() select { case m.reconcileCh <- trackID: default: @@ -369,9 +379,10 @@ func (m *SubscriptionManager) reconcileWorker() { case <-reconcileTicker.C: m.reconcileSubscriptions() case trackID := <-m.reconcileCh: - m.lock.RLock() + m.lock.Lock() s := m.subscriptions[trackID] - m.lock.RUnlock() + delete(m.pendingReconcile, trackID) + m.lock.Unlock() if s != nil { m.reconcileSubscription(s) } else { diff --git a/pkg/rtc/subscriptionmanager_test.go b/pkg/rtc/subscriptionmanager_test.go index 496f501d6..983b73f45 100644 --- a/pkg/rtc/subscriptionmanager_test.go +++ b/pkg/rtc/subscriptionmanager_test.go @@ -54,7 +54,7 @@ func TestSubscribe(t *testing.T) { sm.params.OnTrackSubscribed = func(subTrack types.SubscribedTrack) { subCount.Add(1) } - sm.params.OnSubcriptionError = func(trackID livekit.TrackID) { + sm.params.OnSubscriptionError = func(trackID livekit.TrackID) { failed.Store(true) } numParticipantSubscribed := atomic.Int32{} @@ -115,7 +115,7 @@ func TestSubscribe(t *testing.T) { resolver := newTestResolver(false, true, "pub", "pubID") sm.params.TrackResolver = resolver.Resolve failed := atomic.Bool{} - sm.params.OnSubcriptionError = func(trackID livekit.TrackID) { + sm.params.OnSubscriptionError = func(trackID livekit.TrackID) { failed.Store(true) } @@ -156,7 +156,7 @@ func TestSubscribe(t *testing.T) { resolver := newTestResolver(true, true, "pub", "pubID") sm.params.TrackResolver = resolver.Resolve failed := atomic.Bool{} - sm.params.OnSubcriptionError = func(trackID livekit.TrackID) { + sm.params.OnSubscriptionError = func(trackID livekit.TrackID) { failed.Store(true) } @@ -278,6 +278,7 @@ func TestSubscribeStatusChanged(t *testing.T) { require.Equal(t, int32(1), numParticipantSubscribed.Load()) require.Equal(t, int32(0), numParticipantUnsubscribed.Load()) + require.True(t, sm.IsSubscribedTo("pubID")) // now unsubscribe track2, no event should be fired sm.UnsubscribeFromTrack("track2") @@ -292,6 +293,7 @@ func TestSubscribeStatusChanged(t *testing.T) { return !s1.needsUnsubscribe() }, subSettleTimeout, subCheckInterval, "track1 should be unsubscribed") require.Equal(t, int32(1), numParticipantUnsubscribed.Load()) + require.False(t, sm.IsSubscribedTo("pubID")) } // clients may send update subscribed settings prior to subscription events coming through @@ -334,7 +336,7 @@ func newTestSubscriptionManager(t *testing.T) *SubscriptionManager { Logger: logger.GetLogger(), OnTrackSubscribed: func(subTrack types.SubscribedTrack) {}, OnTrackUnsubscribed: func(subTrack types.SubscribedTrack) {}, - OnSubcriptionError: func(trackID livekit.TrackID) {}, + OnSubscriptionError: func(trackID livekit.TrackID) {}, TrackResolver: func(identity livekit.ParticipantIdentity, trackID livekit.TrackID) types.MediaResolverResult { return types.MediaResolverResult{} },