From 7ad51f49f16fd567086f45a00ceea52c1c5491ee Mon Sep 17 00:00:00 2001 From: David Zhao Date: Sun, 29 May 2022 22:09:02 -0700 Subject: [PATCH] Fixed unclean DownTrack close when removed before bound. (#736) * Fixed unclean DownTrack close when removed before bound. When a DownTrack is closed before it had a chance to be bound to a transceiver, we'd skip close and leave it hanging. This is unlikely in normal operations. However, it can be seen with permissions and subscription APIs. * remove remaining peerID references --- pkg/rtc/mediatrack.go | 1 - pkg/rtc/mediatracksubscriptions.go | 5 + pkg/rtc/signalhandler.go | 3 +- pkg/rtc/uptrackmanager.go | 30 +++--- pkg/rtc/utils.go | 4 +- pkg/sfu/downtrack.go | 154 ++++++++++++++++------------- pkg/sfu/downtrackspreader.go | 10 +- pkg/sfu/receiver.go | 11 +-- 8 files changed, 120 insertions(+), 98 deletions(-) diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 848c3912c..d65935564 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -154,7 +154,6 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra newWR := sfu.NewWebRTCReceiver( receiver, track, - t.PublisherID(), t.params.TrackInfo, LoggerWithCodecMime(t.params.Logger, mime), twcc, diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index 63d3d2a5f..7d9df5177 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -144,6 +144,7 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr * codecs, wr, t.params.BufferFactory, + sub.Identity(), subscriberID, t.params.ReceiverConfig.PacketBufferSize, LoggerWithTrack(sub.GetLogger(), trackID), @@ -309,6 +310,10 @@ func (t *MediaTrackSubscriptions) RevokeDisallowedSubscribers(allowedSubscriberI } if !found { + t.params.Logger.Infow("revoking subscription", + "subscriber", subTrack.SubscriberIdentity(), + "subscriberID", subTrack.SubscriberID(), + ) go subTrack.DownTrack().Close() revokedSubscriberIdentities = append(revokedSubscriberIdentities, subTrack.SubscriberIdentity()) } diff --git a/pkg/rtc/signalhandler.go b/pkg/rtc/signalhandler.go index b3d0d698e..b5f095b1c 100644 --- a/pkg/rtc/signalhandler.go +++ b/pkg/rtc/signalhandler.go @@ -38,7 +38,8 @@ func HandleParticipantSignal(room types.Room, participant types.LocalParticipant participant.SetTrackMuted(livekit.TrackID(msg.Mute.Sid), msg.Mute.Muted, false) case *livekit.SignalRequest_Subscription: var err error - if participant.CanSubscribe() { + // always allow unsubscribe + if participant.CanSubscribe() || !msg.Subscription.Subscribe { updateErr := room.UpdateSubscriptions( participant, livekit.StringsAsTrackIDs(msg.Subscription.TrackSids), diff --git a/pkg/rtc/uptrackmanager.go b/pkg/rtc/uptrackmanager.go index 8be498a44..a6f7b9f00 100644 --- a/pkg/rtc/uptrackmanager.go +++ b/pkg/rtc/uptrackmanager.go @@ -198,21 +198,21 @@ func (u *UpTrackManager) UpdateSubscriptionPermission( u.lock.Lock() defer u.lock.Unlock() - // store as is for use when migrating - u.subscriptionPermission = subscriptionPermission if subscriptionPermission == nil { + // store as is for use when migrating + u.subscriptionPermission = subscriptionPermission // possible to get a nil when migrating return nil } if err := u.parseSubscriptionPermissions(subscriptionPermission, resolverBySid); err != nil { - // do not accept permissions if parse fails - u.subscriptionPermission = nil + // when failed, do not override previous permissions return err } + // store as is for use when migrating + u.subscriptionPermission = subscriptionPermission u.processPendingSubscriptions(resolverByIdentity) - u.maybeRevokeSubscriptions(resolverByIdentity) return nil @@ -322,19 +322,15 @@ func (u *UpTrackManager) parseSubscriptionPermissions( } // per participant permissions - u.subscriberPermissions = make(map[livekit.ParticipantIdentity]*livekit.TrackPermission) + subscriberPermissions := make(map[livekit.ParticipantIdentity]*livekit.TrackPermission) for _, trackPerms := range subscriptionPermission.TrackPermissions { subscriberIdentity := livekit.ParticipantIdentity(trackPerms.ParticipantIdentity) if subscriberIdentity == "" { if trackPerms.ParticipantSid == "" { - u.subscriberPermissions = nil return ErrSubscriptionPermissionNeedsId } - var sub types.LocalParticipant - if resolver != nil { - sub = resolver(livekit.ParticipantID(trackPerms.ParticipantSid)) - } + sub := resolver(livekit.ParticipantID(trackPerms.ParticipantSid)) if sub == nil { u.params.Logger.Warnw("could not find subscriber for permissions update", nil, "subscriberID", trackPerms.ParticipantSid) continue @@ -353,9 +349,11 @@ func (u *UpTrackManager) parseSubscriptionPermissions( } } - u.subscriberPermissions[subscriberIdentity] = trackPerms + subscriberPermissions[subscriberIdentity] = trackPerms } + u.subscriberPermissions = subscriberPermissions + return nil } @@ -437,11 +435,13 @@ func (u *UpTrackManager) maybeRemovePendingSubscription(trackID livekit.TrackID, } } +// creates subscriptions for tracks if permissions have been granted func (u *UpTrackManager) processPendingSubscriptions(resolver func(participantIdentity livekit.ParticipantIdentity) types.LocalParticipant) { updatedPendingSubscriptions := make(map[livekit.TrackID][]livekit.ParticipantIdentity) for trackID, pending := range u.pendingSubscriptions { track := u.getPublishedTrack(trackID) if track == nil { + // published track is gone continue } @@ -488,11 +488,9 @@ func (u *UpTrackManager) maybeRevokeSubscriptions(resolver func(participantIdent revokedSubscribers := track.RevokeDisallowedSubscribers(allowed) for _, subIdentity := range revokedSubscribers { - var sub types.LocalParticipant - if resolver != nil { - sub = resolver(subIdentity) - } + sub := resolver(subIdentity) if sub == nil { + // participant may have disconnected continue } diff --git a/pkg/rtc/utils.go b/pkg/rtc/utils.go index d0ec0a21e..d0a276eeb 100644 --- a/pkg/rtc/utils.go +++ b/pkg/rtc/utils.go @@ -35,12 +35,12 @@ func PackDataTrackLabel(participantID livekit.ParticipantID, trackID livekit.Tra return string(participantID) + trackIdSeparator + string(trackID) + trackIdSeparator + label } -func UnpackDataTrackLabel(packed string) (peerID livekit.ParticipantID, trackID livekit.TrackID, label string) { +func UnpackDataTrackLabel(packed string) (participantID livekit.ParticipantID, trackID livekit.TrackID, label string) { parts := strings.Split(packed, trackIdSeparator) if len(parts) != 3 { return "", livekit.TrackID(packed), "" } - peerID = livekit.ParticipantID(parts[0]) + participantID = livekit.ParticipantID(parts[0]) trackID = livekit.TrackID(parts[1]) label = parts[2] return diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index b89c553ca..eb7ac462b 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -3,7 +3,6 @@ package sfu import ( "encoding/binary" "errors" - "fmt" "io" "strings" "sync" @@ -30,9 +29,11 @@ type TrackSender interface { UpTrackBitrateAvailabilityChange() WriteRTP(p *buffer.ExtPacket, layer int32) error Close() + IsClosed() bool // ID is the globally unique identifier for this Track. ID() string - PeerID() livekit.ParticipantID + SubscriberID() livekit.ParticipantID + SubscriberIdentity() livekit.ParticipantIdentity } const ( @@ -57,7 +58,8 @@ var ( ErrNotVP8 = errors.New("not VP8") ErrOutOfOrderVP8PictureIdCacheMiss = errors.New("out-of-order VP8 picture id not found in cache") ErrFilteredVP8TemporalLayer = errors.New("filtered VP8 temporal layer") - ErrTrackAlreadyBind = errors.New("already bind") + ErrDownTrackAlreadyBound = errors.New("already bound") + ErrDownTrackClosed = errors.New("downtrack closed") ) var ( @@ -101,19 +103,25 @@ type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport) // DownTrack implements TrackLocal, is the track used to write packets // to SFU Subscriber, the track handle the packets for simple, simulcast // and SVC Publisher. +// A DownTrack has the following lifecycle +// - new +// - bound / unbound +// - closed +// once closed, a DownTrack cannot be re-used. type DownTrack struct { - logger logger.Logger - id livekit.TrackID - peerID livekit.ParticipantID - bound atomic.Bool - kind webrtc.RTPCodecType - mime string - ssrc uint32 - streamID string - maxTrack int - payloadType uint8 - sequencer *sequencer - bufferFactory *buffer.Factory + logger logger.Logger + id livekit.TrackID + subscriberIdentity livekit.ParticipantIdentity + subscriberID livekit.ParticipantID + bound atomic.Bool + kind webrtc.RTPCodecType + mime string + ssrc uint32 + streamID string + maxTrack int + payloadType uint8 + sequencer *sequencer + bufferFactory *buffer.Factory forwarder *Forwarder @@ -129,7 +137,7 @@ type DownTrack struct { onBind func() receiverReportListeners []ReceiverReportListener listenerLock sync.RWMutex - closeOnce sync.Once + isClosed atomic.Bool rtpStats *buffer.RTPStats @@ -186,7 +194,8 @@ func NewDownTrack( codecs []webrtc.RTPCodecCapability, r TrackReceiver, bf *buffer.Factory, - peerID livekit.ParticipantID, + subIdentity livekit.ParticipantIdentity, + subID livekit.ParticipantID, mt int, logger logger.Logger, ) (*DownTrack, error) { @@ -201,15 +210,16 @@ func NewDownTrack( } d := &DownTrack{ - logger: logger, - id: r.TrackID(), - peerID: peerID, - maxTrack: mt, - streamID: r.StreamID(), - bufferFactory: bf, - receiver: r, - upstreamCodecs: codecs, - kind: kind, + logger: logger, + id: r.TrackID(), + subscriberIdentity: subIdentity, + subscriberID: subID, + maxTrack: mt, + streamID: r.StreamID(), + bufferFactory: bf, + receiver: r, + upstreamCodecs: codecs, + kind: kind, } d.forwarder = NewForwarder(d.kind, d.logger) @@ -255,8 +265,11 @@ func NewDownTrack( // This asserts that the code requested is supported by the remote peer. // If so it sets up all the state (SSRC and PayloadType) to have a call func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, error) { + if d.IsClosed() { + return webrtc.RTPCodecParameters{}, ErrDownTrackClosed + } if d.bound.Load() { - return webrtc.RTPCodecParameters{}, ErrTrackAlreadyBind + return webrtc.RTPCodecParameters{}, ErrDownTrackAlreadyBound } var codec webrtc.RTPCodecParameters for _, c := range d.upstreamCodecs { @@ -316,7 +329,9 @@ func (d *DownTrack) Codec() webrtc.RTPCodecCapability { return d.codec } // StreamID is the group this track belongs too. This must be unique func (d *DownTrack) StreamID() string { return d.streamID } -func (d *DownTrack) PeerID() livekit.ParticipantID { return d.peerID } +func (d *DownTrack) SubscriberIdentity() livekit.ParticipantIdentity { return d.subscriberIdentity } + +func (d *DownTrack) SubscriberID() livekit.ParticipantID { return d.subscriberID } // Sets RTP header extensions for this track func (d *DownTrack) SetRTPHeaderExtensions(rtpHeaderExtensions []webrtc.RTPHeaderExtensionParameter) { @@ -349,7 +364,7 @@ func (d *DownTrack) Stop() error { if d.transceiver != nil { return d.transceiver.Stop() } - return fmt.Errorf("d.transceiver not exists") + return errors.New("downtrack transceiver does not exist") } func (d *DownTrack) SetTransceiver(transceiver *webrtc.RTPTransceiver) { @@ -613,6 +628,10 @@ func (d *DownTrack) Mute(muted bool) { } } +func (d *DownTrack) IsClosed() bool { + return d.isClosed.Load() +} + func (d *DownTrack) Close() { d.CloseWithFlush(true) } @@ -624,50 +643,52 @@ func (d *DownTrack) Close() { // 2. in case of session migration, participant migrate from other node, video track should // be resumed with same participant, set flush=false since we don't need to flush decoder. func (d *DownTrack) CloseWithFlush(flush bool) { - if !d.bound.Load() { + if d.isClosed.Swap(true) { + // already closed return } - if d.forwarder != nil { - d.forwarder.Mute(true) + + d.logger.Infow("close down track", "flushBlankFrame", flush) + if d.bound.Load() { + if d.forwarder != nil { + d.forwarder.Mute(true) + } + // write blank frames after disabling so that other frames do not interfere. + // Idea here is to send blank key frames to flush the decoder buffer at the remote end. + // Otherwise, with transceiver re-use last frame from previous stream is held in the + // display buffer and there could be a brief moment where the previous stream is displayed. + if flush { + doneFlushing := d.writeBlankFrameRTP(RTPBlankFramesCloseSeconds, d.blankFramesGeneration.Inc()) + + // wait a limited time to flush + timer := time.NewTimer(flushTimeout) + defer timer.Stop() + + select { + case <-doneFlushing: + case <-timer.C: + d.blankFramesGeneration.Inc() // in case flush is still running + } + } + + d.bound.Store(false) + d.logger.Debugw("closing sender", "kind", d.kind) + d.receiver.DeleteDownTrack(d.subscriberID) } - // write blank frames after disabling so that other frames do not interfere. - // Idea here is to send blank key frames to flush the decoder buffer at the remote end. - // Otherwise, with transceiver re-use last frame from previous stream is held in the - // display buffer and there could be a brief moment where the previous stream is displayed. - d.logger.Infow("close down track", "peerID", d.peerID, "trackID", d.id, "flushBlankFrame", flush) - if flush { - doneFlushing := d.writeBlankFrameRTP(RTPBlankFramesCloseSeconds, d.blankFramesGeneration.Inc()) + d.connectionStats.Close() + d.rtpStats.Stop() + d.logger.Debugw("rtp stats", "stats", d.rtpStats.ToString()) - // wait a limited time to flush - timer := time.NewTimer(flushTimeout) - defer timer.Stop() - - select { - case <-doneFlushing: - case <-timer.C: - d.blankFramesGeneration.Inc() // in case flush is still running - } + if d.onMaxLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo { + d.onMaxLayerChanged(d, InvalidLayerSpatial) } - d.closeOnce.Do(func() { - d.logger.Debugw("closing sender", "peerID", d.peerID, "trackID", d.id, "kind", d.kind) - d.receiver.DeleteDownTrack(d.peerID) + if d.onCloseHandler != nil { + d.onCloseHandler() + } - d.connectionStats.Close() - d.rtpStats.Stop() - d.logger.Debugw("rtp stats", "stats", d.rtpStats.ToString()) - - if d.onMaxLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo { - d.onMaxLayerChanged(d, InvalidLayerSpatial) - } - - if d.onCloseHandler != nil { - d.onCloseHandler() - } - - d.stopKeyFrameRequester() - }) + d.stopKeyFrameRequester() } func (d *DownTrack) SetMaxSpatialLayer(spatialLayer int32) { @@ -1349,7 +1370,8 @@ func (d *DownTrack) DebugInfo() map[string]interface{} { } return map[string]interface{}{ - "PeerID": d.peerID, + "SubscriberID": d.subscriberID, + "Subscriber": d.subscriberIdentity, "TrackID": d.id, "StreamID": d.streamID, "SSRC": d.ssrc, diff --git a/pkg/sfu/downtrackspreader.go b/pkg/sfu/downtrackspreader.go index d3e557727..7121c2df4 100644 --- a/pkg/sfu/downtrackspreader.go +++ b/pkg/sfu/downtrackspreader.go @@ -61,23 +61,23 @@ func (d *DownTrackSpreader) Store(ts TrackSender) { d.downTrackMu.Lock() defer d.downTrackMu.Unlock() - d.downTracks[ts.PeerID()] = ts + d.downTracks[ts.SubscriberID()] = ts d.shadowDownTracks() } -func (d *DownTrackSpreader) Free(peerID livekit.ParticipantID) { +func (d *DownTrackSpreader) Free(subscriberID livekit.ParticipantID) { d.downTrackMu.Lock() defer d.downTrackMu.Unlock() - delete(d.downTracks, peerID) + delete(d.downTracks, subscriberID) d.shadowDownTracks() } -func (d *DownTrackSpreader) HasDownTrack(peerID livekit.ParticipantID) bool { +func (d *DownTrackSpreader) HasDownTrack(subscriberID livekit.ParticipantID) bool { d.downTrackMu.RLock() defer d.downTrackMu.RUnlock() - _, ok := d.downTracks[peerID] + _, ok := d.downTracks[subscriberID] return ok } diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index e9dadd1a4..a8f081899 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -49,7 +49,7 @@ type TrackReceiver interface { SetMaxExpectedSpatialLayer(layer int32) AddDownTrack(track TrackSender) error - DeleteDownTrack(peerID livekit.ParticipantID) + DeleteDownTrack(participantID livekit.ParticipantID) DebugInfo() map[string]interface{} @@ -63,7 +63,6 @@ type WebRTCReceiver struct { pliThrottleConfig config.PLIThrottleConfig audioConfig config.AudioConfig - peerID livekit.ParticipantID trackID livekit.TrackID streamID string kind webrtc.RTPCodecType @@ -163,7 +162,6 @@ func WithLoadBalanceThreshold(downTracks int) ReceiverOpts { func NewWebRTCReceiver( receiver *webrtc.RTPReceiver, track *webrtc.TrackRemote, - pid livekit.ParticipantID, trackInfo *livekit.TrackInfo, logger logger.Logger, twcc *twcc.Responder, @@ -171,7 +169,6 @@ func NewWebRTCReceiver( ) *WebRTCReceiver { w := &WebRTCReceiver{ logger: logger, - peerID: pid, receiver: receiver, trackID: livekit.TrackID(track.ID()), streamID: track.StreamID(), @@ -381,7 +378,7 @@ func (w *WebRTCReceiver) AddDownTrack(track TrackSender) error { return ErrReceiverClosed } - if w.downTrackSpreader.HasDownTrack(track.PeerID()) { + if w.downTrackSpreader.HasDownTrack(track.SubscriberID()) { return ErrDownTrackAlreadyExist } @@ -423,12 +420,12 @@ func (w *WebRTCReceiver) OnCloseHandler(fn func()) { } // DeleteDownTrack removes a DownTrack from a Receiver -func (w *WebRTCReceiver) DeleteDownTrack(peerID livekit.ParticipantID) { +func (w *WebRTCReceiver) DeleteDownTrack(subscriberID livekit.ParticipantID) { if w.closed.Load() { return } - w.downTrackSpreader.Free(peerID) + w.downTrackSpreader.Free(subscriberID) } func (w *WebRTCReceiver) sendRTCP(packets []rtcp.Packet) {