From c0c38f2e1dd3cc857797c904e9cb18915dd0d6fb Mon Sep 17 00:00:00 2001 From: David Zhao Date: Fri, 25 Jun 2021 12:57:07 -0700 Subject: [PATCH] Fixed deadlocks with updated simulcast logic --- go.mod | 2 +- go.sum | 4 ++-- magefile.go | 2 +- pkg/rtc/mediatrack.go | 22 ++++++++++++++-------- pkg/rtc/participant.go | 2 +- pkg/rtc/subscribedtrack.go | 2 +- version/version.go | 2 +- 7 files changed, 21 insertions(+), 15 deletions(-) diff --git a/go.mod b/go.mod index a24452456..345271afe 100644 --- a/go.mod +++ b/go.mod @@ -46,4 +46,4 @@ require ( replace github.com/pion/webrtc/v3 => github.com/livekit/pion-webrtc/v3 v3.0.30 -replace github.com/pion/ion-sfu => github.com/livekit/ion-sfu v1.10.7-0.20210624043550-63c2bdd50f84 +replace github.com/pion/ion-sfu => github.com/livekit/ion-sfu v1.10.7-0.20210625234943-c2fcf7981b1a diff --git a/go.sum b/go.sum index 28c5cbfda..153fd2e8e 100644 --- a/go.sum +++ b/go.sum @@ -230,8 +230,8 @@ github.com/lightstep/lightstep-tracer-common/golang/gogo v0.0.0-20190605223551-b github.com/lightstep/lightstep-tracer-go v0.18.1/go.mod h1:jlF1pusYV4pidLvZ+XD0UBX0ZE6WURAspgAczcDHrL4= github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY/mKrcvA= github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts= -github.com/livekit/ion-sfu v1.10.7-0.20210624043550-63c2bdd50f84 h1:mdRsMaWgYmZxcrH2jL/7SUVouXE4LSTGabky7YXEUKM= -github.com/livekit/ion-sfu v1.10.7-0.20210624043550-63c2bdd50f84/go.mod h1:Wx6b4qGUjvSo1kGl+/fHl0ZF48g2IJOjzUFg0yCo9qY= +github.com/livekit/ion-sfu v1.10.7-0.20210625234943-c2fcf7981b1a h1:umZJ3i6CUqO8hDsWVKJSdICIMkACKvVJ0ihXegFBBV0= +github.com/livekit/ion-sfu v1.10.7-0.20210625234943-c2fcf7981b1a/go.mod h1:Wx6b4qGUjvSo1kGl+/fHl0ZF48g2IJOjzUFg0yCo9qY= github.com/livekit/pion-webrtc/v3 v3.0.30 h1:2RomVfztLegWIePXcyPFJDiF5pbT64aIX4RtpXdDksU= github.com/livekit/pion-webrtc/v3 v3.0.30/go.mod h1:XFQeLYBf++bWWA0sJqh6zF1ouWluosxwTOMOoTZGaD0= github.com/livekit/protocol v0.5.4 h1:OFOOahFXGOusgd8X1Ucnl8H/sxa3DzwSrnmKNIUOu/0= diff --git a/magefile.go b/magefile.go index 80cb473fc..068ffb30c 100644 --- a/magefile.go +++ b/magefile.go @@ -176,7 +176,7 @@ func PublishDocker() error { } // don't publish snapshot versions as latest or minor version - if strings.HasSuffix(version.Version, "SNAPSHOT") { + if strings.Contains(version.Version, "SNAPSHOT") { return nil } diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index fd6c078c0..77e8de577 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -173,14 +173,15 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error { subTrack.SetPublisherMuted(t.IsMuted()) go t.sendDownTrackBindingReports(sub) }) + downTrack.OnCloseHandler(func() { - t.lock.Lock() - delete(t.subscribedTracks, sub.ID()) - t.lock.Unlock() - - t.params.Stats.SubSubscribedTrack(t.kind.String()) - go func() { + t.lock.Lock() + delete(t.subscribedTracks, sub.ID()) + t.lock.Unlock() + + t.params.Stats.SubSubscribedTrack(t.kind.String()) + // ignore if the subscribing sub is not connected if sub.SubscriberPC().ConnectionState() == webrtc.PeerConnectionStateClosed { return @@ -214,7 +215,7 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error { t.subscribedTracks[sub.ID()] = subTrack - t.receiver.AddDownTrack(downTrack, len(t.subscribedTracks) < 10) + t.receiver.AddDownTrack(downTrack, t.shouldStartWithBestQuality()) // since sub will lock, run it in a gorountine to avoid deadlocks go func() { sub.AddSubscribedTrack(t.params.ParticipantID, subTrack) @@ -285,7 +286,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra }) t.params.Stats.AddPublishedTrack(t.kind.String()) } - t.receiver.AddUpTrack(track, buff, false) + t.receiver.AddUpTrack(track, buff, t.shouldStartWithBestQuality()) // when RID is set, track is simulcasted t.simulcasted = track.RID() != "" @@ -327,6 +328,11 @@ func (t *MediaTrack) ToProto() *livekit.TrackInfo { } } +// this function assumes caller holds lock +func (t *MediaTrack) shouldStartWithBestQuality() bool { + return len(t.subscribedTracks) < 10 +} + // TODO: send for all downtracks from the source participant // https://tools.ietf.org/html/rfc7941 func (t *MediaTrack) sendDownTrackBindingReports(sub types.Participant) { diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index ff4bb2a93..8e1bba288 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -533,8 +533,8 @@ func (p *ParticipantImpl) SendDataPacket(dp *livekit.DataPacket) error { func (p *ParticipantImpl) SetTrackMuted(trackId string, muted bool) { p.lock.RLock() - defer p.lock.RUnlock() track := p.publishedTracks[trackId] + p.lock.RUnlock() if track == nil { logger.Warnw("could not locate track", nil, "track", trackId) return diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index e0d701260..9107bde33 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -62,7 +62,7 @@ func (t *SubscribedTrack) updateDownTrackMute() { t.dt.Mute(muted) } -func spatialLayerForQuality(quality livekit.VideoQuality) int64 { +func spatialLayerForQuality(quality livekit.VideoQuality) int32 { switch quality { case livekit.VideoQuality_LOW: return 0 diff --git a/version/version.go b/version/version.go index d3f84932f..92878fdb1 100644 --- a/version/version.go +++ b/version/version.go @@ -1,3 +1,3 @@ package version -const Version = "0.9.14-SNAPSHOT" +const Version = "0.9.14"