From faff67162bf7721f23f6a26e83375ce7b3759f36 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Thu, 21 Dec 2023 09:56:54 +0530 Subject: [PATCH] Consolidate TrackInfo. (#2331) * Consolidate TrackInfo. TrackInfo was spread across a bit. Consolidating it. * TODO comments * test * update TrackInfo on SSRC change * further consolidation * log mimes only * update receivers on SSRC set * clone proto on return * feedback: break loop on mime match * prevent data race --- pkg/rtc/mediatrack.go | 91 ++-- pkg/rtc/mediatrack_test.go | 26 +- pkg/rtc/mediatrackreceiver.go | 401 ++++++++++-------- pkg/rtc/participant.go | 5 +- pkg/rtc/participant_sdp.go | 2 +- pkg/rtc/types/interfaces.go | 1 - .../typesfakes/fake_local_media_track.go | 66 --- pkg/rtc/types/typesfakes/fake_media_track.go | 66 --- pkg/rtc/wrappedreceiver.go | 6 + pkg/sfu/buffer/videolayerutils.go | 1 + pkg/sfu/receiver.go | 41 +- pkg/sfu/streamtrackermanager.go | 26 +- 12 files changed, 325 insertions(+), 407 deletions(-) diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 584c953c3..73796cb73 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -22,7 +22,6 @@ import ( "github.com/pion/rtcp" "github.com/pion/webrtc/v3" "go.uber.org/atomic" - "google.golang.org/protobuf/proto" "github.com/livekit/mediatransportutil/pkg/twcc" "github.com/livekit/protocol/livekit" @@ -52,7 +51,6 @@ type MediaTrack struct { } type MediaTrackParams struct { - TrackInfo *livekit.TrackInfo SignalCid string SdpCid string ParticipantID livekit.ParticipantID @@ -71,13 +69,12 @@ type MediaTrackParams struct { SimTracks map[uint32]SimulcastTrackInfo } -func NewMediaTrack(params MediaTrackParams) *MediaTrack { +func NewMediaTrack(params MediaTrackParams, ti *livekit.TrackInfo) *MediaTrack { t := &MediaTrack{ params: params, } t.MediaTrackReceiver = NewMediaTrackReceiver(MediaTrackReceiverParams{ - TrackInfo: params.TrackInfo, MediaTrack: t, IsRelayed: false, ParticipantID: params.ParticipantID, @@ -88,19 +85,9 @@ func NewMediaTrack(params MediaTrackParams) *MediaTrack { AudioConfig: params.AudioConfig, Telemetry: params.Telemetry, Logger: params.Logger, - }) - t.MediaTrackReceiver.OnVideoLayerUpdate(func(layers []*livekit.VideoLayer) { - t.params.Telemetry.TrackPublishedUpdate(context.Background(), t.PublisherID(), - &livekit.TrackInfo{ - Sid: string(t.ID()), - Type: livekit.TrackType_VIDEO, - Muted: t.IsMuted(), - Simulcast: t.IsSimulcast(), - Layers: layers, - }) - }) + }, ti) - if params.TrackInfo.Type == livekit.TrackType_AUDIO { + if ti.Type == livekit.TrackType_AUDIO { t.MediaLossProxy = NewMediaLossProxy(MediaLossProxyParams{ Logger: params.Logger, }) @@ -113,7 +100,7 @@ func NewMediaTrack(params MediaTrackParams) *MediaTrack { t.MediaTrackReceiver.OnMediaLossFeedback(t.MediaLossProxy.HandleMaxLossFeedback) } - if params.TrackInfo.Type == livekit.TrackType_VIDEO { + if ti.Type == livekit.TrackType_VIDEO { t.dynacastManager = NewDynacastManager(DynacastManagerParams{ DynacastPauseDelay: params.VideoConfig.DynacastPauseDelay, Logger: params.Logger, @@ -126,7 +113,7 @@ func NewMediaTrack(params MediaTrackParams) *MediaTrack { t.dynacastManager.NotifySubscriberMaxQuality( subscriberID, codec.MimeType, - buffer.SpatialLayerToVideoQuality(layer, t.params.TrackInfo), + buffer.SpatialLayerToVideoQuality(layer, t.MediaTrackReceiver.TrackInfo()), ) }, ) @@ -154,7 +141,7 @@ func (t *MediaTrack) OnSubscribedMaxQualityChange( for _, q := range maxSubscribedQualities { receiver := t.Receiver(q.CodecMime) if receiver != nil { - receiver.SetMaxExpectedSpatialLayer(buffer.VideoQualityToSpatialLayer(q.Quality, t.params.TrackInfo)) + receiver.SetMaxExpectedSpatialLayer(buffer.VideoQualityToSpatialLayer(q.Quality, t.MediaTrackReceiver.TrackInfo())) } } } @@ -177,8 +164,8 @@ func (t *MediaTrack) HasSdpCid(cid string) bool { return true } - info := t.params.TrackInfo - for _, c := range info.Codecs { + ti := t.MediaTrackReceiver.TrackInfoClone() + for _, c := range ti.Codecs { if c.Cid == cid { return true } @@ -187,24 +174,11 @@ func (t *MediaTrack) HasSdpCid(cid string) bool { } func (t *MediaTrack) ToProto() *livekit.TrackInfo { - info := t.MediaTrackReceiver.TrackInfo(true) - info.Muted = t.IsMuted() - info.Simulcast = t.IsSimulcast() - return info + return t.MediaTrackReceiver.TrackInfoClone() } -func (t *MediaTrack) SetPendingCodecSid(codecs []*livekit.SimulcastCodec) { - ti := proto.Clone(t.params.TrackInfo).(*livekit.TrackInfo) - for _, c := range codecs { - for _, origin := range ti.Codecs { - if strings.Contains(origin.MimeType, c.Codec) { - origin.Cid = c.Cid - break - } - } - } - t.params.TrackInfo = ti - t.MediaTrackReceiver.UpdateTrackInfo(ti) +func (t *MediaTrack) UpdateCodecCid(codecs []*livekit.SimulcastCodec) { + t.MediaTrackReceiver.UpdateCodecCid(codecs) } // AddReceiver adds a new RTP receiver to the track, returns true when receiver represents a new codec @@ -233,24 +207,25 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra } }) + ti := t.MediaTrackReceiver.TrackInfoClone() t.lock.Lock() mime := strings.ToLower(track.Codec().MimeType) - layer := buffer.RidToSpatialLayer(track.RID(), t.trackInfo) + layer := buffer.RidToSpatialLayer(track.RID(), ti) t.params.Logger.Debugw("AddReceiver", "mime", track.Codec().MimeType) wr := t.MediaTrackReceiver.Receiver(mime) if wr == nil { priority := -1 - for idx, c := range t.params.TrackInfo.Codecs { + for idx, c := range ti.Codecs { if strings.EqualFold(mime, c.MimeType) { priority = idx break } } - if len(t.params.TrackInfo.Codecs) == 0 { + if len(ti.Codecs) == 0 { priority = 0 } if priority < 0 { - t.params.Logger.Warnw("could not find codec for webrtc receiver", nil, "webrtcCodec", mime, "track", logger.Proto(t.params.TrackInfo)) + t.params.Logger.Warnw("could not find codec for webrtc receiver", nil, "webrtcCodec", mime, "track", logger.Proto(ti)) t.lock.Unlock() return false } @@ -258,7 +233,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra newWR := sfu.NewWebRTCReceiver( receiver, track, - t.params.TrackInfo, + ti, LoggerWithCodecMime(t.params.Logger, mime), twcc, t.params.VideoConfig.StreamTracker, @@ -277,16 +252,20 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra } } }) - newWR.OnStatsUpdate(func(_ *sfu.WebRTCReceiver, stat *livekit.AnalyticsStat) { - // LK-TODO: this needs to be receiver/mime aware - key := telemetry.StatsKeyForTrack(livekit.StreamType_UPSTREAM, t.PublisherID(), t.ID(), t.params.TrackInfo.Source, t.params.TrackInfo.Type) - t.params.Telemetry.TrackStats(key, stat) - }) + // SIMULCAST-CODEC-TODO: these need to be receiver/mime aware, setting it up only for primary now + if priority == 0 { + newWR.OnStatsUpdate(func(_ *sfu.WebRTCReceiver, stat *livekit.AnalyticsStat) { + key := telemetry.StatsKeyForTrack(livekit.StreamType_UPSTREAM, t.PublisherID(), t.ID(), ti.Source, ti.Type) + t.params.Telemetry.TrackStats(key, stat) + }) + + newWR.OnMaxLayerChange(t.onMaxLayerChange) + } if t.PrimaryReceiver() == nil { // primary codec published, set potential codecs - potentialCodecs := make([]webrtc.RTPCodecParameters, 0, len(t.params.TrackInfo.Codecs)) + potentialCodecs := make([]webrtc.RTPCodecParameters, 0, len(ti.Codecs)) parameters := receiver.GetParameters() - for _, c := range t.params.TrackInfo.Codecs { + for _, c := range ti.Codecs { for _, nc := range parameters.Codecs { if strings.EqualFold(nc.MimeType, c.MimeType) { potentialCodecs = append(potentialCodecs, nc) @@ -301,8 +280,6 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra } } - newWR.OnMaxLayerChange(t.onMaxLayerChange) - t.buffer = buff t.MediaTrackReceiver.SetupReceiver(newWR, priority, mid) @@ -367,17 +344,7 @@ func (t *MediaTrack) HasPendingCodec() bool { } func (t *MediaTrack) onMaxLayerChange(maxLayer int32) { - ti := &livekit.TrackInfo{ - Sid: t.trackInfo.Sid, - Type: t.trackInfo.Type, - } - - if layer, ok := t.MediaTrackReceiver.layerDimensions[livekit.VideoQuality(maxLayer)]; ok { - ti.Layers = []*livekit.VideoLayer{{Quality: livekit.VideoQuality(maxLayer), Width: layer.Width, Height: layer.Height}} - } else if maxLayer == -1 { - ti.Layers = []*livekit.VideoLayer{{Quality: livekit.VideoQuality_OFF}} - } - t.params.Telemetry.TrackPublishedUpdate(context.Background(), t.PublisherID(), ti) + t.MediaTrackReceiver.NotifyMaxLayerChange(maxLayer) } func (t *MediaTrack) Restart() { diff --git a/pkg/rtc/mediatrack_test.go b/pkg/rtc/mediatrack_test.go index 9a0587447..8d96bbae4 100644 --- a/pkg/rtc/mediatrack_test.go +++ b/pkg/rtc/mediatrack_test.go @@ -35,9 +35,7 @@ func TestTrackInfo(t *testing.T) { Muted: true, } - mt := NewMediaTrack(MediaTrackParams{ - TrackInfo: &ti, - }) + mt := NewMediaTrack(MediaTrackParams{}, &ti) outInfo := mt.ToProto() require.Equal(t, ti.Muted, outInfo.Muted) require.Equal(t, ti.Name, outInfo.Name) @@ -51,17 +49,17 @@ func TestTrackInfo(t *testing.T) { require.Equal(t, ti.Simulcast, outInfo.Simulcast) // make it simulcasted - mt.simulcasted.Store(true) + mt.SetSimulcast(true) require.True(t, mt.ToProto().Simulcast) } func TestGetQualityForDimension(t *testing.T) { t.Run("landscape source", func(t *testing.T) { - mt := NewMediaTrack(MediaTrackParams{TrackInfo: &livekit.TrackInfo{ + mt := NewMediaTrack(MediaTrackParams{}, &livekit.TrackInfo{ Type: livekit.TrackType_VIDEO, Width: 1080, Height: 720, - }}) + }) require.Equal(t, livekit.VideoQuality_LOW, mt.GetQualityForDimension(120, 120)) require.Equal(t, livekit.VideoQuality_LOW, mt.GetQualityForDimension(300, 200)) @@ -71,11 +69,11 @@ func TestGetQualityForDimension(t *testing.T) { }) t.Run("portrait source", func(t *testing.T) { - mt := NewMediaTrack(MediaTrackParams{TrackInfo: &livekit.TrackInfo{ + mt := NewMediaTrack(MediaTrackParams{}, &livekit.TrackInfo{ Type: livekit.TrackType_VIDEO, Width: 540, Height: 960, - }}) + }) require.Equal(t, livekit.VideoQuality_LOW, mt.GetQualityForDimension(200, 400)) require.Equal(t, livekit.VideoQuality_MEDIUM, mt.GetQualityForDimension(400, 400)) @@ -84,7 +82,7 @@ func TestGetQualityForDimension(t *testing.T) { }) t.Run("layers provided", func(t *testing.T) { - mt := NewMediaTrack(MediaTrackParams{TrackInfo: &livekit.TrackInfo{ + mt := NewMediaTrack(MediaTrackParams{}, &livekit.TrackInfo{ Type: livekit.TrackType_VIDEO, Width: 1080, Height: 720, @@ -105,7 +103,7 @@ func TestGetQualityForDimension(t *testing.T) { Height: 720, }, }, - }}) + }) require.Equal(t, livekit.VideoQuality_LOW, mt.GetQualityForDimension(120, 120)) require.Equal(t, livekit.VideoQuality_LOW, mt.GetQualityForDimension(300, 300)) @@ -114,7 +112,7 @@ func TestGetQualityForDimension(t *testing.T) { }) t.Run("highest layer with smallest dimensions", func(t *testing.T) { - mt := NewMediaTrack(MediaTrackParams{TrackInfo: &livekit.TrackInfo{ + mt := NewMediaTrack(MediaTrackParams{}, &livekit.TrackInfo{ Type: livekit.TrackType_VIDEO, Width: 1080, Height: 720, @@ -135,7 +133,7 @@ func TestGetQualityForDimension(t *testing.T) { Height: 720, }, }, - }}) + }) require.Equal(t, livekit.VideoQuality_LOW, mt.GetQualityForDimension(120, 120)) require.Equal(t, livekit.VideoQuality_LOW, mt.GetQualityForDimension(300, 300)) @@ -143,7 +141,7 @@ func TestGetQualityForDimension(t *testing.T) { require.Equal(t, livekit.VideoQuality_HIGH, mt.GetQualityForDimension(1000, 700)) require.Equal(t, livekit.VideoQuality_HIGH, mt.GetQualityForDimension(1200, 800)) - mt = NewMediaTrack(MediaTrackParams{TrackInfo: &livekit.TrackInfo{ + mt = NewMediaTrack(MediaTrackParams{}, &livekit.TrackInfo{ Type: livekit.TrackType_VIDEO, Width: 1080, Height: 720, @@ -164,7 +162,7 @@ func TestGetQualityForDimension(t *testing.T) { Height: 720, }, }, - }}) + }) require.Equal(t, livekit.VideoQuality_MEDIUM, mt.GetQualityForDimension(120, 120)) require.Equal(t, livekit.VideoQuality_MEDIUM, mt.GetQualityForDimension(300, 300)) diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index 358a9f615..b5fefa8a3 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -15,6 +15,7 @@ package rtc import ( + "context" "errors" "fmt" "sort" @@ -23,12 +24,10 @@ import ( "github.com/pion/rtcp" "github.com/pion/webrtc/v3" - "go.uber.org/atomic" "google.golang.org/protobuf/proto" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" - "github.com/livekit/protocol/utils" "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/rtc/types" @@ -74,8 +73,7 @@ func (m mediaTrackReceiverState) String() string { type simulcastReceiver struct { sfu.TrackReceiver - priority int - layerSSRCs [livekit.VideoQuality_HIGH + 1]uint32 + priority int } func (r *simulcastReceiver) Priority() int { @@ -83,7 +81,6 @@ func (r *simulcastReceiver) Priority() int { } type MediaTrackReceiverParams struct { - TrackInfo *livekit.TrackInfo MediaTrack types.MediaTrack IsRelayed bool ParticipantID livekit.ParticipantID @@ -97,32 +94,26 @@ type MediaTrackReceiverParams struct { } type MediaTrackReceiver struct { - params MediaTrackReceiverParams - muted atomic.Bool - simulcasted atomic.Bool + params MediaTrackReceiverParams lock sync.RWMutex receivers []*simulcastReceiver - receiversShadow []*simulcastReceiver trackInfo *livekit.TrackInfo - layerDimensions map[livekit.VideoQuality]*livekit.VideoLayer potentialCodecs []webrtc.RTPCodecParameters state mediaTrackReceiverState onSetupReceiver func(mime string) onMediaLossFeedback func(dt *sfu.DownTrack, report *rtcp.ReceiverReport) - onVideoLayerUpdate func(layers []*livekit.VideoLayer) onClose []func() *MediaTrackSubscriptions } -func NewMediaTrackReceiver(params MediaTrackReceiverParams) *MediaTrackReceiver { +func NewMediaTrackReceiver(params MediaTrackReceiverParams, ti *livekit.TrackInfo) *MediaTrackReceiver { t := &MediaTrackReceiver{ - params: params, - trackInfo: proto.Clone(params.TrackInfo).(*livekit.TrackInfo), - layerDimensions: make(map[livekit.VideoQuality]*livekit.VideoLayer), - state: mediaTrackReceiverStateOpen, + params: params, + trackInfo: proto.Clone(ti).(*livekit.TrackInfo), + state: mediaTrackReceiverStateOpen, } t.MediaTrackSubscriptions = NewMediaTrackSubscriptions(MediaTrackSubscriptionsParams{ @@ -138,22 +129,18 @@ func NewMediaTrackReceiver(params MediaTrackReceiverParams) *MediaTrackReceiver if t.trackInfo.Muted { t.SetMuted(true) } - - if t.trackInfo != nil && t.Kind() == livekit.TrackType_VIDEO { - t.UpdateVideoLayers(t.trackInfo.Layers) - // LK-TODO: maybe use this or simulcast flag in TrackInfo to set simulcasted here - } - return t } func (t *MediaTrackReceiver) Restart() { t.lock.Lock() - receivers := t.receiversShadow + receivers := t.receivers + ti := t.trackInfo t.lock.Unlock() + hq := buffer.VideoQualityToSpatialLayer(livekit.VideoQuality_HIGH, ti) for _, receiver := range receivers { - receiver.SetMaxExpectedSpatialLayer(buffer.VideoQualityToSpatialLayer(livekit.VideoQuality_HIGH, t.params.TrackInfo)) + receiver.SetMaxExpectedSpatialLayer(hq) } } @@ -210,10 +197,18 @@ func (t *MediaTrackReceiver) SetupReceiver(receiver sfu.TrackReceiver, priority } } - t.shadowReceiversLocked() - + var receiverCodecs []string + for _, r := range t.receivers { + receiverCodecs = append(receiverCodecs, r.Codec().MimeType) + } + t.params.Logger.Debugw( + "setup receiver", + "mime", receiver.Codec().MimeType, + "priority", priority, + "receivers", receiverCodecs, + "mid", mid, + ) onSetupReceiver := t.onSetupReceiver - t.params.Logger.Debugw("setup receiver", "mime", receiver.Codec().MimeType, "priority", priority, "receivers", t.receiversShadow, "mid", mid) t.lock.Unlock() if onSetupReceiver != nil { @@ -255,32 +250,9 @@ func (t *MediaTrackReceiver) SetPotentialCodecs(codecs []webrtc.RTPCodecParamete sort.Slice(t.receivers, func(i, j int) bool { return t.receivers[i].Priority() < t.receivers[j].Priority() }) - t.shadowReceiversLocked() t.lock.Unlock() } -func (t *MediaTrackReceiver) shadowReceiversLocked() { - t.receiversShadow = make([]*simulcastReceiver, len(t.receivers)) - copy(t.receiversShadow, t.receivers) -} - -func (t *MediaTrackReceiver) SetLayerSsrc(mime string, rid string, ssrc uint32) { - t.lock.Lock() - defer t.lock.Unlock() - - layer := buffer.RidToSpatialLayer(rid, t.params.TrackInfo) - if layer == buffer.InvalidLayerSpatial { - // non-simulcast case will not have `rid` - layer = 0 - } - for _, receiver := range t.receiversShadow { - if strings.EqualFold(receiver.Codec().MimeType, mime) && int(layer) < len(receiver.layerSSRCs) { - receiver.layerSSRCs[layer] = ssrc - return - } - } -} - func (t *MediaTrackReceiver) ClearReceiver(mime string, willBeResumed bool) { t.params.Logger.Debugw("clearing receiver", "mime", mime) t.lock.Lock() @@ -292,7 +264,6 @@ func (t *MediaTrackReceiver) ClearReceiver(mime string, willBeResumed bool) { } } - t.shadowReceiversLocked() t.lock.Unlock() t.removeAllSubscribersForMime(mime, willBeResumed) @@ -306,8 +277,7 @@ func (t *MediaTrackReceiver) ClearAllReceivers(willBeResumed bool) { mimes = append(mimes, receiver.Codec().MimeType) } - t.receivers = t.receivers[:0] - t.receiversShadow = nil + t.receivers = nil t.lock.Unlock() for _, mime := range mimes { @@ -319,10 +289,6 @@ func (t *MediaTrackReceiver) OnMediaLossFeedback(f func(dt *sfu.DownTrack, rr *r t.onMediaLossFeedback = f } -func (t *MediaTrackReceiver) OnVideoLayerUpdate(f func(layers []*livekit.VideoLayer)) { - t.onVideoLayerUpdate = f -} - func (t *MediaTrackReceiver) IsOpen() bool { t.lock.RLock() defer t.lock.RUnlock() @@ -353,7 +319,7 @@ func (t *MediaTrackReceiver) TryClose() bool { return true } - for _, receiver := range t.receiversShadow { + for _, receiver := range t.receivers { if dr, _ := receiver.TrackReceiver.(*DummyReceiver); dr != nil && dr.Receiver() != nil { t.lock.RUnlock() return false @@ -422,11 +388,17 @@ func (t *MediaTrackReceiver) PublisherVersion() uint32 { } func (t *MediaTrackReceiver) IsSimulcast() bool { - return t.simulcasted.Load() + t.lock.RLock() + defer t.lock.RUnlock() + + return t.trackInfo.Simulcast } func (t *MediaTrackReceiver) SetSimulcast(simulcast bool) { - t.simulcasted.Store(simulcast) + t.lock.Lock() + defer t.lock.Unlock() + + t.trackInfo.Simulcast = simulcast } func (t *MediaTrackReceiver) Name() string { @@ -437,15 +409,17 @@ func (t *MediaTrackReceiver) Name() string { } func (t *MediaTrackReceiver) IsMuted() bool { - return t.muted.Load() + t.lock.RLock() + defer t.lock.RUnlock() + + return t.trackInfo.Muted } func (t *MediaTrackReceiver) SetMuted(muted bool) { - t.muted.Store(muted) - - t.lock.RLock() - receivers := t.receiversShadow - t.lock.RUnlock() + t.lock.Lock() + t.trackInfo.Muted = muted + receivers := t.receivers + t.lock.Unlock() for _, receiver := range receivers { receiver.SetUpTrackPaused(muted) } @@ -471,7 +445,7 @@ func (t *MediaTrackReceiver) AddSubscriber(sub types.LocalParticipant) (types.Su return nil, ErrNotOpen } - receivers := t.receiversShadow + receivers := t.receivers potentialCodecs := make([]webrtc.RTPCodecParameters, len(t.potentialCodecs)) copy(potentialCodecs, t.potentialCodecs) t.lock.RUnlock() @@ -553,134 +527,219 @@ func (t *MediaTrackReceiver) RevokeDisallowedSubscribers(allowedSubscriberIdenti return revokedSubscriberIdentities } -func (t *MediaTrackReceiver) UpdateTrackInfo(ti *livekit.TrackInfo) { - clonedInfo := proto.Clone(ti).(*livekit.TrackInfo) +func (t *MediaTrackReceiver) updateTrackInfoOfReceivers() { + t.lock.RLock() + receivers := t.receivers + ti := t.trackInfo + t.lock.RUnlock() + + for _, r := range receivers { + r.UpdateTrackInfo(ti) + } +} + +func (t *MediaTrackReceiver) SetLayerSsrc(mime string, rid string, ssrc uint32) { t.lock.Lock() - originInfo := t.trackInfo - for _, ci := range clonedInfo.Codecs { - for _, originCi := range originInfo.Codecs { - if strings.EqualFold(ci.MimeType, originCi.MimeType) && originCi.Mid != "" { - ci.Mid = originCi.Mid + layer := buffer.RidToSpatialLayer(rid, t.trackInfo) + if layer == buffer.InvalidLayerSpatial { + // non-simulcast case will not have `rid` + layer = 0 + } + quality := buffer.SpatialLayerToVideoQuality(layer, t.trackInfo) + // set video layer ssrc info + for _, ci := range t.trackInfo.Codecs { + if !strings.EqualFold(ci.MimeType, mime) { + continue + } + + // if origin layer has ssrc, don't override it + var matchingLayer *livekit.VideoLayer + ssrcFound := false + for _, l := range ci.Layers { + if l.Quality == quality { + matchingLayer = l + if l.Ssrc != 0 { + ssrcFound = true + } + break + } + } + if !ssrcFound && matchingLayer != nil { + matchingLayer.Ssrc = ssrc + } + break + } + + // for client don't use simulcast codecs (old client version or single codec) + if len(t.trackInfo.Codecs) == 0 { + // if origin layer has ssrc, don't override it + var matchingLayer *livekit.VideoLayer + ssrcFound := false + for _, l := range t.trackInfo.Layers { + if l.Quality == quality { + matchingLayer = l + if l.Ssrc != 0 { + ssrcFound = true + } + break + } + } + if !ssrcFound && matchingLayer != nil { + matchingLayer.Ssrc = ssrc + } + } + t.lock.Unlock() + + t.updateTrackInfoOfReceivers() +} + +func (t *MediaTrackReceiver) UpdateCodecCid(codecs []*livekit.SimulcastCodec) { + t.lock.Lock() + for _, c := range codecs { + for _, origin := range t.trackInfo.Codecs { + if strings.Contains(origin.MimeType, c.Codec) { + origin.Cid = c.Cid break } } } - t.trackInfo = clonedInfo t.lock.Unlock() - if ti != nil && t.Kind() == livekit.TrackType_VIDEO { - t.UpdateVideoLayers(ti.Layers) - } + t.updateTrackInfoOfReceivers() } -func (t *MediaTrackReceiver) TrackInfo(generateLayer bool) *livekit.TrackInfo { - t.lock.RLock() - defer t.lock.RUnlock() +func (t *MediaTrackReceiver) UpdateTrackInfo(ti *livekit.TrackInfo) { + clonedInfo := proto.Clone(ti).(*livekit.TrackInfo) - ti := proto.Clone(t.trackInfo).(*livekit.TrackInfo) - if !generateLayer { - return ti - } - - layers := t.getVideoLayersLocked() - - // set video layer ssrc info - for i, ci := range ti.Codecs { - for _, receiver := range t.receiversShadow { - if receiver.priority == i { - originLayers := ci.Layers - ci.Layers = []*livekit.VideoLayer{} - for layerIdx, layer := range layers { - ci.Layers = append(ci.Layers, proto.Clone(layer).(*livekit.VideoLayer)) - - // if origin layer has ssrc, don't override it - ssrcFound := false - for _, l := range originLayers { - if l.Quality == ci.Layers[layerIdx].Quality { - if l.Ssrc != 0 { - ci.Layers[layerIdx].Ssrc = l.Ssrc - ssrcFound = true - } - break - } - } - if !ssrcFound && int(layer.Quality) < len(receiver.layerSSRCs) { - ci.Layers[layerIdx].Ssrc = receiver.layerSSRCs[layer.Quality] - } - } - - if i == 0 { - ti.Layers = ci.Layers - } - break + t.lock.Lock() + // patch Mid and SSRC of codecs/layers by keeping original if available + for i, ci := range clonedInfo.Codecs { + for _, originCi := range t.trackInfo.Codecs { + if !strings.EqualFold(ci.MimeType, originCi.MimeType) { + continue } + + if originCi.Mid != "" { + ci.Mid = originCi.Mid + } + + for _, layer := range ci.Layers { + for _, originLayer := range originCi.Layers { + if layer.Quality == originLayer.Quality { + if originLayer.Ssrc != 0 { + layer.Ssrc = originLayer.Ssrc + } + break + } + } + } + break + } + + if i == 0 { + clonedInfo.Layers = ci.Layers } } // for client don't use simulcast codecs (old client version or single codec) - if len(ti.Codecs) == 0 && len(t.receiversShadow) > 0 { - receiver := t.receiversShadow[0] - originLayers := ti.Layers - ti.Layers = []*livekit.VideoLayer{} - for layerIdx, layer := range layers { - ti.Layers = append(ti.Layers, proto.Clone(layer).(*livekit.VideoLayer)) - - // if origin layer has ssrc, don't override it - ssrcFound := false - for _, l := range originLayers { - if l.Quality == ti.Layers[layerIdx].Quality { - if l.Ssrc != 0 { - ti.Layers[layerIdx].Ssrc = l.Ssrc - ssrcFound = true + if len(clonedInfo.Codecs) == 0 { + for _, layer := range clonedInfo.Layers { + for _, originLayer := range t.trackInfo.Layers { + if layer.Quality == originLayer.Quality { + if originLayer.Ssrc != 0 { + layer.Ssrc = originLayer.Ssrc } break } } - if !ssrcFound && int(layer.Quality) < len(receiver.layerSSRCs) { - ti.Layers[layerIdx].Ssrc = receiver.layerSSRCs[layer.Quality] - } } } - return ti -} + t.trackInfo = clonedInfo + t.lock.Unlock() -func (t *MediaTrackReceiver) Version() utils.TimedVersion { - t.lock.RLock() - defer t.lock.RUnlock() - - return utils.TimedVersionFromProto(t.trackInfo.Version) + t.updateTrackInfoOfReceivers() } func (t *MediaTrackReceiver) UpdateVideoLayers(layers []*livekit.VideoLayer) { t.lock.Lock() - for _, layer := range layers { - t.layerDimensions[layer.Quality] = layer + // set video layer ssrc info + for i, ci := range t.trackInfo.Codecs { + originLayers := ci.Layers + ci.Layers = []*livekit.VideoLayer{} + for layerIdx, layer := range layers { + ci.Layers = append(ci.Layers, proto.Clone(layer).(*livekit.VideoLayer)) + for _, l := range originLayers { + if l.Quality == ci.Layers[layerIdx].Quality { + if l.Ssrc != 0 { + ci.Layers[layerIdx].Ssrc = l.Ssrc + } + break + } + } + } + + if i == 0 { + t.trackInfo.Layers = ci.Layers + } + } + + // for client don't use simulcast codecs (old client version or single codec) + if len(t.trackInfo.Codecs) == 0 { + originLayers := t.trackInfo.Layers + t.trackInfo.Layers = []*livekit.VideoLayer{} + for layerIdx, layer := range layers { + t.trackInfo.Layers = append(t.trackInfo.Layers, proto.Clone(layer).(*livekit.VideoLayer)) + for _, l := range originLayers { + if l.Quality == t.trackInfo.Layers[layerIdx].Quality { + if l.Ssrc != 0 { + t.trackInfo.Layers[layerIdx].Ssrc = l.Ssrc + } + break + } + } + } } t.lock.Unlock() + t.updateTrackInfoOfReceivers() t.MediaTrackSubscriptions.UpdateVideoLayers() - if t.onVideoLayerUpdate != nil { - t.onVideoLayerUpdate(layers) - } - - // TODO: this might need to trigger a participant update for clients to pick up dimension change } -func (t *MediaTrackReceiver) GetVideoLayers() []*livekit.VideoLayer { +func (t *MediaTrackReceiver) TrackInfo() *livekit.TrackInfo { t.lock.RLock() defer t.lock.RUnlock() - return t.getVideoLayersLocked() + return t.trackInfo } -func (t *MediaTrackReceiver) getVideoLayersLocked() []*livekit.VideoLayer { - layers := make([]*livekit.VideoLayer, 0) - for _, layer := range t.layerDimensions { - layers = append(layers, proto.Clone(layer).(*livekit.VideoLayer)) - } +func (t *MediaTrackReceiver) TrackInfoClone() *livekit.TrackInfo { + t.lock.RLock() + defer t.lock.RUnlock() - return layers + return proto.Clone(t.trackInfo).(*livekit.TrackInfo) +} + +func (t *MediaTrackReceiver) NotifyMaxLayerChange(maxLayer int32) { + t.lock.RLock() + quality := buffer.SpatialLayerToVideoQuality(maxLayer, t.trackInfo) + ti := &livekit.TrackInfo{ + Sid: t.trackInfo.Sid, + Type: t.trackInfo.Type, + Layers: []*livekit.VideoLayer{{Quality: quality}}, + } + if quality != livekit.VideoQuality_OFF { + for _, layer := range t.trackInfo.Layers { + if layer.Quality == quality { + ti.Layers[0].Width = layer.Width + ti.Layers[0].Height = layer.Height + break + } + } + } + t.lock.RUnlock() + + t.params.Telemetry.TrackPublishedUpdate(context.Background(), t.PublisherID(), ti) } // GetQualityForDimension finds the closest quality to use for desired dimensions @@ -708,7 +767,7 @@ func (t *MediaTrackReceiver) GetQualityForDimension(width, height uint32) liveki // default sizes representing qualities low - high layerSizes := []uint32{180, 360, origSize} var providedSizes []uint32 - for _, layer := range t.layerDimensions { + for _, layer := range t.trackInfo.Layers { providedSizes = append(providedSizes, layer.Height) } if len(providedSizes) > 0 { @@ -757,13 +816,13 @@ func (t *MediaTrackReceiver) DebugInfo() map[string]interface{} { info := map[string]interface{}{ "ID": t.ID(), "Kind": t.Kind().String(), - "PubMuted": t.muted.Load(), + "PubMuted": t.IsMuted(), } info["DownTracks"] = t.MediaTrackSubscriptions.DebugInfo() t.lock.RLock() - receivers := t.receiversShadow + receivers := t.receivers t.lock.RUnlock() for _, receiver := range receivers { info[receiver.Codec().MimeType] = receiver.DebugInfo() @@ -776,20 +835,20 @@ func (t *MediaTrackReceiver) PrimaryReceiver() sfu.TrackReceiver { t.lock.RLock() defer t.lock.RUnlock() - if len(t.receiversShadow) == 0 { + if len(t.receivers) == 0 { return nil } - if dr, ok := t.receiversShadow[0].TrackReceiver.(*DummyReceiver); ok { + if dr, ok := t.receivers[0].TrackReceiver.(*DummyReceiver); ok { return dr.Receiver() } - return t.receiversShadow[0].TrackReceiver + return t.receivers[0].TrackReceiver } func (t *MediaTrackReceiver) Receiver(mime string) sfu.TrackReceiver { t.lock.RLock() defer t.lock.RUnlock() - for _, r := range t.receiversShadow { + for _, r := range t.receivers { if strings.EqualFold(r.Codec().MimeType, mime) { if dr, ok := r.TrackReceiver.(*DummyReceiver); ok { return dr.Receiver() @@ -804,17 +863,17 @@ func (t *MediaTrackReceiver) Receivers() []sfu.TrackReceiver { t.lock.RLock() defer t.lock.RUnlock() - receivers := make([]sfu.TrackReceiver, 0, len(t.receiversShadow)) - for _, r := range t.receiversShadow { + receivers := make([]sfu.TrackReceiver, 0, len(t.receivers)) + for _, r := range t.receivers { receivers = append(receivers, r.TrackReceiver) } return receivers } func (t *MediaTrackReceiver) SetRTT(rtt uint32) { - t.lock.Lock() - receivers := t.receiversShadow - t.lock.Unlock() + t.lock.RLock() + receivers := t.receivers + t.lock.RUnlock() for _, r := range receivers { if wr, ok := r.TrackReceiver.(*sfu.WebRTCReceiver); ok { @@ -847,9 +906,9 @@ func (t *MediaTrackReceiver) IsEncrypted() bool { } func (t *MediaTrackReceiver) GetTrackStats() *livekit.RTPStats { - t.lock.Lock() - receivers := t.receiversShadow - t.lock.Unlock() + t.lock.RLock() + receivers := t.receivers + t.lock.RUnlock() stats := make([]*livekit.RTPStats, 0, len(receivers)) for _, receiver := range receivers { diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index badc6b954..6f07849b8 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -1622,7 +1622,7 @@ func (p *ParticipantImpl) addPendingTrackLocked(req *livekit.AddTrackRequest) *l return nil } - track.(*MediaTrack).SetPendingCodecSid(req.SimulcastCodecs) + track.(*MediaTrack).UpdateCodecCid(req.SimulcastCodecs) ti := track.ToProto() return ti } @@ -1907,7 +1907,6 @@ func (p *ParticipantImpl) addMigrateMutedTrack(cid string, ti *livekit.TrackInfo func (p *ParticipantImpl) addMediaTrack(signalCid string, sdpCid string, ti *livekit.TrackInfo) *MediaTrack { mt := NewMediaTrack(MediaTrackParams{ - TrackInfo: proto.Clone(ti).(*livekit.TrackInfo), SignalCid: signalCid, SdpCid: sdpCid, ParticipantID: p.params.SID, @@ -1923,7 +1922,7 @@ func (p *ParticipantImpl) addMediaTrack(signalCid string, sdpCid string, ti *liv SubscriberConfig: p.params.Config.Subscriber, PLIThrottleConfig: p.params.PLIThrottleConfig, SimTracks: p.params.SimTracks, - }) + }, ti) mt.OnSubscribedMaxQualityChange(p.onSubscribedMaxQualityChange) diff --git a/pkg/rtc/participant_sdp.go b/pkg/rtc/participant_sdp.go index 51e98c478..ed47d2da9 100644 --- a/pkg/rtc/participant_sdp.go +++ b/pkg/rtc/participant_sdp.go @@ -242,7 +242,7 @@ func (p *ParticipantImpl) configurePublisherAnswer(answer webrtc.SessionDescript _, ti, _ = p.getPendingTrack(streamID, livekit.TrackType_AUDIO) p.pendingTracksLock.RUnlock() } else { - ti = track.TrackInfo(false) + ti = track.ToProto() } break } diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index bcea1f58b..05f7d5a2d 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -446,7 +446,6 @@ type MediaTrack interface { UpdateTrackInfo(ti *livekit.TrackInfo) ToProto() *livekit.TrackInfo - Version() utils.TimedVersion PublisherID() livekit.ParticipantID PublisherIdentity() livekit.ParticipantIdentity diff --git a/pkg/rtc/types/typesfakes/fake_local_media_track.go b/pkg/rtc/types/typesfakes/fake_local_media_track.go index 555467591..be86e74c4 100644 --- a/pkg/rtc/types/typesfakes/fake_local_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_local_media_track.go @@ -7,7 +7,6 @@ import ( "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/sfu" "github.com/livekit/protocol/livekit" - "github.com/livekit/protocol/utils" ) type FakeLocalMediaTrack struct { @@ -343,16 +342,6 @@ type FakeLocalMediaTrack struct { updateVideoLayersArgsForCall []struct { arg1 []*livekit.VideoLayer } - VersionStub func() utils.TimedVersion - versionMutex sync.RWMutex - versionArgsForCall []struct { - } - versionReturns struct { - result1 utils.TimedVersion - } - versionReturnsOnCall map[int]struct { - result1 utils.TimedVersion - } invocations map[string][][]interface{} invocationsMutex sync.RWMutex } @@ -2157,59 +2146,6 @@ func (fake *FakeLocalMediaTrack) UpdateVideoLayersArgsForCall(i int) []*livekit. return argsForCall.arg1 } -func (fake *FakeLocalMediaTrack) Version() utils.TimedVersion { - fake.versionMutex.Lock() - ret, specificReturn := fake.versionReturnsOnCall[len(fake.versionArgsForCall)] - fake.versionArgsForCall = append(fake.versionArgsForCall, struct { - }{}) - stub := fake.VersionStub - fakeReturns := fake.versionReturns - fake.recordInvocation("Version", []interface{}{}) - fake.versionMutex.Unlock() - if stub != nil { - return stub() - } - if specificReturn { - return ret.result1 - } - return fakeReturns.result1 -} - -func (fake *FakeLocalMediaTrack) VersionCallCount() int { - fake.versionMutex.RLock() - defer fake.versionMutex.RUnlock() - return len(fake.versionArgsForCall) -} - -func (fake *FakeLocalMediaTrack) VersionCalls(stub func() utils.TimedVersion) { - fake.versionMutex.Lock() - defer fake.versionMutex.Unlock() - fake.VersionStub = stub -} - -func (fake *FakeLocalMediaTrack) VersionReturns(result1 utils.TimedVersion) { - fake.versionMutex.Lock() - defer fake.versionMutex.Unlock() - fake.VersionStub = nil - fake.versionReturns = struct { - result1 utils.TimedVersion - }{result1} -} - -func (fake *FakeLocalMediaTrack) VersionReturnsOnCall(i int, result1 utils.TimedVersion) { - fake.versionMutex.Lock() - defer fake.versionMutex.Unlock() - fake.VersionStub = nil - if fake.versionReturnsOnCall == nil { - fake.versionReturnsOnCall = make(map[int]struct { - result1 utils.TimedVersion - }) - } - fake.versionReturnsOnCall[i] = struct { - result1 utils.TimedVersion - }{result1} -} - func (fake *FakeLocalMediaTrack) Invocations() map[string][][]interface{} { fake.invocationsMutex.RLock() defer fake.invocationsMutex.RUnlock() @@ -2287,8 +2223,6 @@ func (fake *FakeLocalMediaTrack) Invocations() map[string][][]interface{} { defer fake.updateTrackInfoMutex.RUnlock() fake.updateVideoLayersMutex.RLock() defer fake.updateVideoLayersMutex.RUnlock() - fake.versionMutex.RLock() - defer fake.versionMutex.RUnlock() copiedInvocations := map[string][][]interface{}{} for key, value := range fake.invocations { copiedInvocations[key] = value diff --git a/pkg/rtc/types/typesfakes/fake_media_track.go b/pkg/rtc/types/typesfakes/fake_media_track.go index 1f88d07a5..d4bdfc17e 100644 --- a/pkg/rtc/types/typesfakes/fake_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_media_track.go @@ -7,7 +7,6 @@ import ( "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/sfu" "github.com/livekit/protocol/livekit" - "github.com/livekit/protocol/utils" ) type FakeMediaTrack struct { @@ -279,16 +278,6 @@ type FakeMediaTrack struct { updateVideoLayersArgsForCall []struct { arg1 []*livekit.VideoLayer } - VersionStub func() utils.TimedVersion - versionMutex sync.RWMutex - versionArgsForCall []struct { - } - versionReturns struct { - result1 utils.TimedVersion - } - versionReturnsOnCall map[int]struct { - result1 utils.TimedVersion - } invocations map[string][][]interface{} invocationsMutex sync.RWMutex } @@ -1743,59 +1732,6 @@ func (fake *FakeMediaTrack) UpdateVideoLayersArgsForCall(i int) []*livekit.Video return argsForCall.arg1 } -func (fake *FakeMediaTrack) Version() utils.TimedVersion { - fake.versionMutex.Lock() - ret, specificReturn := fake.versionReturnsOnCall[len(fake.versionArgsForCall)] - fake.versionArgsForCall = append(fake.versionArgsForCall, struct { - }{}) - stub := fake.VersionStub - fakeReturns := fake.versionReturns - fake.recordInvocation("Version", []interface{}{}) - fake.versionMutex.Unlock() - if stub != nil { - return stub() - } - if specificReturn { - return ret.result1 - } - return fakeReturns.result1 -} - -func (fake *FakeMediaTrack) VersionCallCount() int { - fake.versionMutex.RLock() - defer fake.versionMutex.RUnlock() - return len(fake.versionArgsForCall) -} - -func (fake *FakeMediaTrack) VersionCalls(stub func() utils.TimedVersion) { - fake.versionMutex.Lock() - defer fake.versionMutex.Unlock() - fake.VersionStub = stub -} - -func (fake *FakeMediaTrack) VersionReturns(result1 utils.TimedVersion) { - fake.versionMutex.Lock() - defer fake.versionMutex.Unlock() - fake.VersionStub = nil - fake.versionReturns = struct { - result1 utils.TimedVersion - }{result1} -} - -func (fake *FakeMediaTrack) VersionReturnsOnCall(i int, result1 utils.TimedVersion) { - fake.versionMutex.Lock() - defer fake.versionMutex.Unlock() - fake.VersionStub = nil - if fake.versionReturnsOnCall == nil { - fake.versionReturnsOnCall = make(map[int]struct { - result1 utils.TimedVersion - }) - } - fake.versionReturnsOnCall[i] = struct { - result1 utils.TimedVersion - }{result1} -} - func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} { fake.invocationsMutex.RLock() defer fake.invocationsMutex.RUnlock() @@ -1857,8 +1793,6 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} { defer fake.updateTrackInfoMutex.RUnlock() fake.updateVideoLayersMutex.RLock() defer fake.updateVideoLayersMutex.RUnlock() - fake.versionMutex.RLock() - defer fake.versionMutex.RUnlock() copiedInvocations := map[string][][]interface{}{} for key, value := range fake.invocations { copiedInvocations[key] = value diff --git a/pkg/rtc/wrappedreceiver.go b/pkg/rtc/wrappedreceiver.go index 8df479ce7..a887ef832 100644 --- a/pkg/rtc/wrappedreceiver.go +++ b/pkg/rtc/wrappedreceiver.go @@ -294,6 +294,12 @@ func (d *DummyReceiver) TrackInfo() *livekit.TrackInfo { return nil } +func (d *DummyReceiver) UpdateTrackInfo(ti *livekit.TrackInfo) { + if r, ok := d.receiver.Load().(sfu.TrackReceiver); ok { + r.UpdateTrackInfo(ti) + } +} + func (d *DummyReceiver) IsClosed() bool { if r, ok := d.receiver.Load().(sfu.TrackReceiver); ok { return r.IsClosed() diff --git a/pkg/sfu/buffer/videolayerutils.go b/pkg/sfu/buffer/videolayerutils.go index 895af9f18..f099c2e7c 100644 --- a/pkg/sfu/buffer/videolayerutils.go +++ b/pkg/sfu/buffer/videolayerutils.go @@ -25,6 +25,7 @@ const ( FullResolution = "f" ) +// SIMULCAST-CODEC-TODO: these need to be codec mime aware if and when each codec suppports different layers func LayerPresenceFromTrackInfo(trackInfo *livekit.TrackInfo) *[livekit.VideoQuality_HIGH + 1]bool { if trackInfo == nil || len(trackInfo.Layers) == 0 { return nil diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 9afff8e98..e555b99f9 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -24,6 +24,7 @@ import ( "github.com/pion/rtcp" "github.com/pion/webrtc/v3" "go.uber.org/atomic" + "google.golang.org/protobuf/proto" "github.com/livekit/mediatransportutil/pkg/bucket" "github.com/livekit/mediatransportutil/pkg/twcc" @@ -71,6 +72,7 @@ type TrackReceiver interface { DebugInfo() map[string]interface{} TrackInfo() *livekit.TrackInfo + UpdateTrackInfo(ti *livekit.TrackInfo) // Get primary receiver if this receiver represents a RED codec; otherwise it will return itself GetPrimaryReceiverForRed() TrackReceiver @@ -104,7 +106,7 @@ type WebRTCReceiver struct { closeOnce sync.Once closed atomic.Bool useTrackers bool - trackInfo *livekit.TrackInfo + trackInfo atomic.Pointer[livekit.TrackInfo] rtcpCh chan []rtcp.Packet @@ -200,21 +202,21 @@ func NewWebRTCReceiver( opts ...ReceiverOpts, ) *WebRTCReceiver { w := &WebRTCReceiver{ - logger: logger, - receiver: receiver, - trackID: livekit.TrackID(track.ID()), - streamID: track.StreamID(), - codec: track.Codec(), - kind: track.Kind(), - twcc: twcc, - trackInfo: trackInfo, - isSVC: IsSvcCodec(track.Codec().MimeType), - isRED: IsRedCodec(track.Codec().MimeType), + logger: logger, + receiver: receiver, + trackID: livekit.TrackID(track.ID()), + streamID: track.StreamID(), + codec: track.Codec(), + kind: track.Kind(), + twcc: twcc, + isSVC: IsSvcCodec(track.Codec().MimeType), + isRED: IsRedCodec(track.Codec().MimeType), } for _, opt := range opts { w = opt(w) } + w.trackInfo.Store(proto.Clone(trackInfo).(*livekit.TrackInfo)) w.downTrackSpreader = NewDownTrackSpreader(DownTrackSpreaderParams{ Threshold: w.lbThreshold, @@ -232,7 +234,7 @@ func NewWebRTCReceiver( w.onStatsUpdate(w, stat) } }) - w.connectionStats.Start(w.trackInfo) + w.connectionStats.Start(trackInfo) w.streamTrackerManager = NewStreamTrackerManager(logger, trackInfo, w.isSVC, w.codec.ClockRate, trackersConfig) w.streamTrackerManager.SetListener(w) @@ -250,7 +252,12 @@ func NewWebRTCReceiver( } func (w *WebRTCReceiver) TrackInfo() *livekit.TrackInfo { - return w.trackInfo + return w.trackInfo.Load() +} + +func (w *WebRTCReceiver) UpdateTrackInfo(ti *livekit.TrackInfo) { + w.trackInfo.Store(proto.Clone(ti).(*livekit.TrackInfo)) + w.streamTrackerManager.UpdateTrackInfo(ti) } func (w *WebRTCReceiver) OnStatsUpdate(fn func(w *WebRTCReceiver, stat *livekit.AnalyticsStat)) { @@ -328,7 +335,7 @@ func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buff layer := int32(0) if w.Kind() == webrtc.RTPCodecTypeVideo && !w.isSVC { - layer = buffer.RidToSpatialLayer(track.RID(), w.trackInfo) + layer = buffer.RidToSpatialLayer(track.RID(), w.trackInfo.Load()) } buff.SetLogger(w.logger.WithValues("layer", layer)) buff.SetTWCC(w.twcc) @@ -700,9 +707,13 @@ func (w *WebRTCReceiver) closeTracks() { } func (w *WebRTCReceiver) DebugInfo() map[string]interface{} { + isSimulcast := !w.isSVC + if ti := w.trackInfo.Load(); ti != nil { + isSimulcast = isSimulcast && len(ti.Layers) > 1 + } info := map[string]interface{}{ "SVC": w.isSVC, - "Simulcast": !w.isSVC && len(w.trackInfo.Layers) > 1, + "Simulcast": isSimulcast, } w.upTrackMu.RLock() diff --git a/pkg/sfu/streamtrackermanager.go b/pkg/sfu/streamtrackermanager.go index c9a22bff9..a934b9e93 100644 --- a/pkg/sfu/streamtrackermanager.go +++ b/pkg/sfu/streamtrackermanager.go @@ -22,6 +22,8 @@ import ( "time" "github.com/frostbyte73/core" + "go.uber.org/atomic" + "google.golang.org/protobuf/proto" "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/sfu/buffer" @@ -57,7 +59,7 @@ type endsSenderReport struct { type StreamTrackerManager struct { logger logger.Logger - trackInfo *livekit.TrackInfo + trackInfo atomic.Pointer[livekit.TrackInfo] isSVC bool clockRate uint32 @@ -92,15 +94,15 @@ func NewStreamTrackerManager( ) *StreamTrackerManager { s := &StreamTrackerManager{ logger: logger, - trackInfo: trackInfo, isSVC: isSVC, maxPublishedLayer: buffer.InvalidLayerSpatial, maxTemporalLayerSeen: buffer.InvalidLayerTemporal, clockRate: clockRate, closed: core.NewFuse(), } + s.trackInfo.Store(proto.Clone(trackInfo).(*livekit.TrackInfo)) - switch s.trackInfo.Source { + switch trackInfo.Source { case livekit.TrackSource_SCREEN_SHARE: s.trackerConfig = trackersConfig.Screenshare case livekit.TrackSource_CAMERA: @@ -111,7 +113,7 @@ func NewStreamTrackerManager( s.maxExpectedLayerFromTrackInfo() - if s.trackInfo.Type == livekit.TrackType_VIDEO { + if trackInfo.Type == livekit.TrackType_VIDEO { go s.bitrateReporter() } return s @@ -316,6 +318,11 @@ func (s *StreamTrackerManager) IsPaused() bool { return s.paused } +func (s *StreamTrackerManager) UpdateTrackInfo(ti *livekit.TrackInfo) { + s.trackInfo.Store(proto.Clone(ti).(*livekit.TrackInfo)) + s.maxExpectedLayerFromTrackInfo() +} + func (s *StreamTrackerManager) SetMaxExpectedSpatialLayer(layer int32) int32 { s.lock.Lock() prev := s.maxExpectedLayer @@ -540,10 +547,13 @@ func (s *StreamTrackerManager) removeAvailableLayer(layer int32) { func (s *StreamTrackerManager) maxExpectedLayerFromTrackInfo() { s.maxExpectedLayer = buffer.InvalidLayerSpatial - for _, layer := range s.trackInfo.Layers { - spatialLayer := buffer.VideoQualityToSpatialLayer(layer.Quality, s.trackInfo) - if spatialLayer > s.maxExpectedLayer { - s.maxExpectedLayer = spatialLayer + ti := s.trackInfo.Load() + if ti != nil { + for _, layer := range ti.Layers { + spatialLayer := buffer.VideoQualityToSpatialLayer(layer.Quality, ti) + if spatialLayer > s.maxExpectedLayer { + s.maxExpectedLayer = spatialLayer + } } } }