From 15f565510c70dd0fdc47f325c2c6ec88f5ced709 Mon Sep 17 00:00:00 2001 From: Anunay Maheshwari Date: Wed, 26 Mar 2025 16:22:32 +0530 Subject: [PATCH] fix(video): determine svc/simulcast from SDP for advanced codecs (#3549) * fix(video): determine svc/simulcast from SDP for advanced codecs * fix(explicit-svc): cleanup * fix(explicit-svc): remove from list on close/remove * fix(explicit-svc): reorder VLS selection, cleanup * fix(explicit-svc): todo comments for temporal layer selector * fix(explicit-svc): remove from simulcastTrackIds even if client does not support unpublish --- pkg/rtc/mediatrack.go | 10 +- pkg/rtc/mediatrackreceiver.go | 10 +- pkg/rtc/mediatracksubscriptions.go | 3 +- pkg/rtc/participant.go | 37 ++++ pkg/rtc/subscriptionmanager_test.go | 5 +- pkg/rtc/types/interfaces.go | 1 + .../typesfakes/fake_local_media_track.go | 65 +++++++ pkg/service/servicefakes/fake_sipstore.go | 162 ------------------ pkg/service/wire_gen.go | 10 +- pkg/sfu/downtrack.go | 2 + pkg/sfu/forwarder.go | 65 ++++--- pkg/sfu/forwarder_test.go | 2 +- pkg/sfu/mime/mimetype.go | 4 - pkg/sfu/receiver.go | 8 +- 14 files changed, 168 insertions(+), 216 deletions(-) diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 4be5466c3..8d8eacd81 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -181,6 +181,10 @@ func (t *MediaTrack) SignalCid() string { return t.params.SignalCid } +func (t *MediaTrack) SdpCid() string { + return t.params.SdpCid +} + func (t *MediaTrack) HasSdpCid(cid string) bool { if t.params.SdpCid == cid { return true @@ -395,12 +399,6 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe return false } - // LK-TODO: can remove this completely when VideoLayers protocol becomes the default as it has info from client or if we decide to use TrackInfo.Simulcast - if t.numUpTracks.Inc() > 1 || track.RID() != "" { - // cannot only rely on numUpTracks since we fire metadata events immediately after the first layer - t.SetSimulcast(true) - } - var bitrates int if len(ti.Layers) > int(layer) { bitrates = int(ti.Layers[layer].GetBitrate()) diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index 2eebea050..88b45abc5 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -498,9 +498,11 @@ func (t *MediaTrackReceiver) SetSimulcast(simulcast bool) { t.lock.Lock() defer t.lock.Unlock() - trackInfo := t.TrackInfoClone() - trackInfo.Simulcast = simulcast - t.trackInfo.Store(trackInfo) + if t.IsSimulcast() != simulcast { + trackInfo := t.TrackInfoClone() + trackInfo.Simulcast = simulcast + t.trackInfo.Store(trackInfo) + } } func (t *MediaTrackReceiver) Name() string { @@ -593,7 +595,7 @@ func (t *MediaTrackReceiver) AddSubscriber(sub types.LocalParticipant) (types.Su Logger: tLogger, DisableRed: t.TrackInfo().GetDisableRed() || !t.params.AudioConfig.ActiveREDEncoding, }) - subTrack, err := t.MediaTrackSubscriptions.AddSubscriber(sub, wr) + subTrack, err := t.MediaTrackSubscriptions.AddSubscriber(sub, wr, t.IsSimulcast()) // media track could have been closed while adding subscription remove := false diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index 63d0f175c..c7af3d77c 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -92,7 +92,7 @@ func (t *MediaTrackSubscriptions) IsSubscriber(subID livekit.ParticipantID) bool } // AddSubscriber subscribes sub to current mediaTrack -func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr *WrappedReceiver) (types.SubscribedTrack, error) { +func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr *WrappedReceiver, isReceiverSimulcast bool) (types.SubscribedTrack, error) { trackID := t.params.MediaTrack.ID() subscriberID := sub.ID() @@ -146,6 +146,7 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr * RTCPWriter: sub.WriteSubscriberRTCP, DisableSenderReportPassThrough: sub.GetDisableSenderReportPassThrough(), SupportsCodecChange: sub.SupportsCodecChange(), + IsReceiverSimulcast: isReceiverSimulcast, }) if err != nil { return nil, err diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 1e0d09d11..9cdc3aa03 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -37,6 +37,7 @@ import ( "github.com/livekit/protocol/auth" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" + sdpHelper "github.com/livekit/protocol/sdp" "github.com/livekit/protocol/utils" "github.com/livekit/protocol/utils/guid" @@ -206,6 +207,8 @@ type ParticipantImpl struct { pendingPublishingTracks map[livekit.TrackID]*pendingTrackInfo pendingRemoteTracks []*pendingRemoteTrack + simulcastTrackIds sync.Map + // supported codecs enabledPublishCodecs []*livekit.Codec enabledSubscribeCodecs []*livekit.Codec @@ -870,10 +873,29 @@ func (p *ParticipantImpl) synthesizeAddTrackRequests(offer webrtc.SessionDescrip return nil } +func (p *ParticipantImpl) findAndStoreSimulcastTrackIds(offer webrtc.SessionDescription) { + parsed, err := offer.Unmarshal() + if err != nil { + return + } + + for _, m := range parsed.MediaDescriptions { + if m.MediaName.Media != "video" { + continue + } + if sdpHelper.IsMediaDescriptionSimulcast(m) { + trackID := sdpHelper.GetTrackIDFromMediaDescription(m) + p.simulcastTrackIds.LoadOrStore(trackID, true) + } + } +} + // HandleOffer an offer from remote participant, used when clients make the initial connection func (p *ParticipantImpl) HandleOffer(offer webrtc.SessionDescription) error { p.pubLogger.Debugw("received offer", "transport", livekit.SignalTarget_PUBLISHER, "offer", offer) + p.findAndStoreSimulcastTrackIds(offer) + if p.params.UseOneShotSignallingMode { if err := p.synthesizeAddTrackRequests(offer); err != nil { return err @@ -1059,6 +1081,7 @@ func (p *ParticipantImpl) Close(sendLeave bool, reason types.ParticipantCloseRea p.pendingTracksLock.Lock() p.pendingTracks = make(map[string]*pendingTrackInfo) p.pendingPublishingTracks = make(map[livekit.TrackID]*pendingTrackInfo) + p.simulcastTrackIds.Clear() p.pendingTracksLock.Unlock() p.UpTrackManager.Close(isExpectedToResume) @@ -1737,6 +1760,11 @@ func (p *ParticipantImpl) onSubscriberOffer(offer webrtc.SessionDescription) err func (p *ParticipantImpl) removePublishedTrack(track types.MediaTrack) { p.RemovePublishedTrack(track, false, true) + + if lmt, ok := track.(types.LocalMediaTrack); ok { + p.simulcastTrackIds.Delete(lmt.SdpCid()) + } + if p.ProtocolVersion().SupportsUnpublish() { p.sendTrackUnpublished(track.ID()) } else { @@ -2455,6 +2483,7 @@ func (p *ParticipantImpl) mediaTrackReceived(track sfu.TrackRemote, rtpReceiver // only assign version on a fresh publish, i. e. avoid updating version in scenarios like migration ti.Version = p.params.VersionGenerator.Next().ToProto() } + mt = p.addMediaTrack(signalCid, track.ID(), ti) newTrack = true @@ -2468,6 +2497,12 @@ func (p *ParticipantImpl) mediaTrackReceived(track sfu.TrackRemote, rtpReceiver p.dirty.Store(true) } + isSimulcast, ok := p.simulcastTrackIds.Load(track.ID()) + if !ok { + isSimulcast = false + } + mt.SetSimulcast(isSimulcast.(bool)) + p.pendingTracksLock.Unlock() mt.AddReceiver(rtpReceiver, track, mid) @@ -2621,6 +2656,8 @@ func (p *ParticipantImpl) addMediaTrack(signalCid string, sdpCid string, ti *liv ) } + p.simulcastTrackIds.Delete(mt.SdpCid()) + p.pendingTracksLock.Lock() if pti := p.pendingTracks[signalCid]; pti != nil { p.sendTrackPublished(signalCid, pti.trackInfos[0]) diff --git a/pkg/rtc/subscriptionmanager_test.go b/pkg/rtc/subscriptionmanager_test.go index a8dfe7032..011192da0 100644 --- a/pkg/rtc/subscriptionmanager_test.go +++ b/pkg/rtc/subscriptionmanager_test.go @@ -242,10 +242,7 @@ func TestUnsubscribe(t *testing.T) { sm.lock.RLock() subLen := len(sm.subscriptions) sm.lock.RUnlock() - if subLen != 0 { - return false - } - return true + return subLen == 0 }, subSettleTimeout, subCheckInterval, "Track was not unsubscribed") // no traces should be left diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index d44bb329c..0c2ead4f2 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -543,6 +543,7 @@ type LocalMediaTrack interface { Restart() SignalCid() string + SdpCid() string HasSdpCid(cid string) bool GetConnectionScoreAndQuality() (float32, livekit.ConnectionQuality) diff --git a/pkg/rtc/types/typesfakes/fake_local_media_track.go b/pkg/rtc/types/typesfakes/fake_local_media_track.go index 877267f6c..eee0f33fb 100644 --- a/pkg/rtc/types/typesfakes/fake_local_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_local_media_track.go @@ -287,6 +287,16 @@ type FakeLocalMediaTrack struct { revokeDisallowedSubscribersReturnsOnCall map[int]struct { result1 []livekit.ParticipantIdentity } + SdpCidStub func() string + sdpCidMutex sync.RWMutex + sdpCidArgsForCall []struct { + } + sdpCidReturns struct { + result1 string + } + sdpCidReturnsOnCall map[int]struct { + result1 string + } SetMutedStub func(bool) setMutedMutex sync.RWMutex setMutedArgsForCall []struct { @@ -1835,6 +1845,59 @@ func (fake *FakeLocalMediaTrack) RevokeDisallowedSubscribersReturnsOnCall(i int, }{result1} } +func (fake *FakeLocalMediaTrack) SdpCid() string { + fake.sdpCidMutex.Lock() + ret, specificReturn := fake.sdpCidReturnsOnCall[len(fake.sdpCidArgsForCall)] + fake.sdpCidArgsForCall = append(fake.sdpCidArgsForCall, struct { + }{}) + stub := fake.SdpCidStub + fakeReturns := fake.sdpCidReturns + fake.recordInvocation("SdpCid", []interface{}{}) + fake.sdpCidMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalMediaTrack) SdpCidCallCount() int { + fake.sdpCidMutex.RLock() + defer fake.sdpCidMutex.RUnlock() + return len(fake.sdpCidArgsForCall) +} + +func (fake *FakeLocalMediaTrack) SdpCidCalls(stub func() string) { + fake.sdpCidMutex.Lock() + defer fake.sdpCidMutex.Unlock() + fake.SdpCidStub = stub +} + +func (fake *FakeLocalMediaTrack) SdpCidReturns(result1 string) { + fake.sdpCidMutex.Lock() + defer fake.sdpCidMutex.Unlock() + fake.SdpCidStub = nil + fake.sdpCidReturns = struct { + result1 string + }{result1} +} + +func (fake *FakeLocalMediaTrack) SdpCidReturnsOnCall(i int, result1 string) { + fake.sdpCidMutex.Lock() + defer fake.sdpCidMutex.Unlock() + fake.SdpCidStub = nil + if fake.sdpCidReturnsOnCall == nil { + fake.sdpCidReturnsOnCall = make(map[int]struct { + result1 string + }) + } + fake.sdpCidReturnsOnCall[i] = struct { + result1 string + }{result1} +} + func (fake *FakeLocalMediaTrack) SetMuted(arg1 bool) { fake.setMutedMutex.Lock() fake.setMutedArgsForCall = append(fake.setMutedArgsForCall, struct { @@ -2270,6 +2333,8 @@ func (fake *FakeLocalMediaTrack) Invocations() map[string][][]interface{} { defer fake.restartMutex.RUnlock() fake.revokeDisallowedSubscribersMutex.RLock() defer fake.revokeDisallowedSubscribersMutex.RUnlock() + fake.sdpCidMutex.RLock() + defer fake.sdpCidMutex.RUnlock() fake.setMutedMutex.RLock() defer fake.setMutedMutex.RUnlock() fake.setRTTMutex.RLock() diff --git a/pkg/service/servicefakes/fake_sipstore.go b/pkg/service/servicefakes/fake_sipstore.go index 055018fe0..bc914444c 100644 --- a/pkg/service/servicefakes/fake_sipstore.go +++ b/pkg/service/servicefakes/fake_sipstore.go @@ -146,34 +146,6 @@ type FakeSIPStore struct { result1 *livekit.SIPTrunkInfo result2 error } - SelectSIPDispatchRuleStub func(context.Context, string) ([]*livekit.SIPDispatchRuleInfo, error) - selectSIPDispatchRuleMutex sync.RWMutex - selectSIPDispatchRuleArgsForCall []struct { - arg1 context.Context - arg2 string - } - selectSIPDispatchRuleReturns struct { - result1 []*livekit.SIPDispatchRuleInfo - result2 error - } - selectSIPDispatchRuleReturnsOnCall map[int]struct { - result1 []*livekit.SIPDispatchRuleInfo - result2 error - } - SelectSIPInboundTrunkStub func(context.Context, string) ([]*livekit.SIPInboundTrunkInfo, error) - selectSIPInboundTrunkMutex sync.RWMutex - selectSIPInboundTrunkArgsForCall []struct { - arg1 context.Context - arg2 string - } - selectSIPInboundTrunkReturns struct { - result1 []*livekit.SIPInboundTrunkInfo - result2 error - } - selectSIPInboundTrunkReturnsOnCall map[int]struct { - result1 []*livekit.SIPInboundTrunkInfo - result2 error - } StoreSIPDispatchRuleStub func(context.Context, *livekit.SIPDispatchRuleInfo) error storeSIPDispatchRuleMutex sync.RWMutex storeSIPDispatchRuleArgsForCall []struct { @@ -870,136 +842,6 @@ func (fake *FakeSIPStore) LoadSIPTrunkReturnsOnCall(i int, result1 *livekit.SIPT }{result1, result2} } -func (fake *FakeSIPStore) SelectSIPDispatchRule(arg1 context.Context, arg2 string) ([]*livekit.SIPDispatchRuleInfo, error) { - fake.selectSIPDispatchRuleMutex.Lock() - ret, specificReturn := fake.selectSIPDispatchRuleReturnsOnCall[len(fake.selectSIPDispatchRuleArgsForCall)] - fake.selectSIPDispatchRuleArgsForCall = append(fake.selectSIPDispatchRuleArgsForCall, struct { - arg1 context.Context - arg2 string - }{arg1, arg2}) - stub := fake.SelectSIPDispatchRuleStub - fakeReturns := fake.selectSIPDispatchRuleReturns - fake.recordInvocation("SelectSIPDispatchRule", []interface{}{arg1, arg2}) - fake.selectSIPDispatchRuleMutex.Unlock() - if stub != nil { - return stub(arg1, arg2) - } - if specificReturn { - return ret.result1, ret.result2 - } - return fakeReturns.result1, fakeReturns.result2 -} - -func (fake *FakeSIPStore) SelectSIPDispatchRuleCallCount() int { - fake.selectSIPDispatchRuleMutex.RLock() - defer fake.selectSIPDispatchRuleMutex.RUnlock() - return len(fake.selectSIPDispatchRuleArgsForCall) -} - -func (fake *FakeSIPStore) SelectSIPDispatchRuleCalls(stub func(context.Context, string) ([]*livekit.SIPDispatchRuleInfo, error)) { - fake.selectSIPDispatchRuleMutex.Lock() - defer fake.selectSIPDispatchRuleMutex.Unlock() - fake.SelectSIPDispatchRuleStub = stub -} - -func (fake *FakeSIPStore) SelectSIPDispatchRuleArgsForCall(i int) (context.Context, string) { - fake.selectSIPDispatchRuleMutex.RLock() - defer fake.selectSIPDispatchRuleMutex.RUnlock() - argsForCall := fake.selectSIPDispatchRuleArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2 -} - -func (fake *FakeSIPStore) SelectSIPDispatchRuleReturns(result1 []*livekit.SIPDispatchRuleInfo, result2 error) { - fake.selectSIPDispatchRuleMutex.Lock() - defer fake.selectSIPDispatchRuleMutex.Unlock() - fake.SelectSIPDispatchRuleStub = nil - fake.selectSIPDispatchRuleReturns = struct { - result1 []*livekit.SIPDispatchRuleInfo - result2 error - }{result1, result2} -} - -func (fake *FakeSIPStore) SelectSIPDispatchRuleReturnsOnCall(i int, result1 []*livekit.SIPDispatchRuleInfo, result2 error) { - fake.selectSIPDispatchRuleMutex.Lock() - defer fake.selectSIPDispatchRuleMutex.Unlock() - fake.SelectSIPDispatchRuleStub = nil - if fake.selectSIPDispatchRuleReturnsOnCall == nil { - fake.selectSIPDispatchRuleReturnsOnCall = make(map[int]struct { - result1 []*livekit.SIPDispatchRuleInfo - result2 error - }) - } - fake.selectSIPDispatchRuleReturnsOnCall[i] = struct { - result1 []*livekit.SIPDispatchRuleInfo - result2 error - }{result1, result2} -} - -func (fake *FakeSIPStore) SelectSIPInboundTrunk(arg1 context.Context, arg2 string) ([]*livekit.SIPInboundTrunkInfo, error) { - fake.selectSIPInboundTrunkMutex.Lock() - ret, specificReturn := fake.selectSIPInboundTrunkReturnsOnCall[len(fake.selectSIPInboundTrunkArgsForCall)] - fake.selectSIPInboundTrunkArgsForCall = append(fake.selectSIPInboundTrunkArgsForCall, struct { - arg1 context.Context - arg2 string - }{arg1, arg2}) - stub := fake.SelectSIPInboundTrunkStub - fakeReturns := fake.selectSIPInboundTrunkReturns - fake.recordInvocation("SelectSIPInboundTrunk", []interface{}{arg1, arg2}) - fake.selectSIPInboundTrunkMutex.Unlock() - if stub != nil { - return stub(arg1, arg2) - } - if specificReturn { - return ret.result1, ret.result2 - } - return fakeReturns.result1, fakeReturns.result2 -} - -func (fake *FakeSIPStore) SelectSIPInboundTrunkCallCount() int { - fake.selectSIPInboundTrunkMutex.RLock() - defer fake.selectSIPInboundTrunkMutex.RUnlock() - return len(fake.selectSIPInboundTrunkArgsForCall) -} - -func (fake *FakeSIPStore) SelectSIPInboundTrunkCalls(stub func(context.Context, string) ([]*livekit.SIPInboundTrunkInfo, error)) { - fake.selectSIPInboundTrunkMutex.Lock() - defer fake.selectSIPInboundTrunkMutex.Unlock() - fake.SelectSIPInboundTrunkStub = stub -} - -func (fake *FakeSIPStore) SelectSIPInboundTrunkArgsForCall(i int) (context.Context, string) { - fake.selectSIPInboundTrunkMutex.RLock() - defer fake.selectSIPInboundTrunkMutex.RUnlock() - argsForCall := fake.selectSIPInboundTrunkArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2 -} - -func (fake *FakeSIPStore) SelectSIPInboundTrunkReturns(result1 []*livekit.SIPInboundTrunkInfo, result2 error) { - fake.selectSIPInboundTrunkMutex.Lock() - defer fake.selectSIPInboundTrunkMutex.Unlock() - fake.SelectSIPInboundTrunkStub = nil - fake.selectSIPInboundTrunkReturns = struct { - result1 []*livekit.SIPInboundTrunkInfo - result2 error - }{result1, result2} -} - -func (fake *FakeSIPStore) SelectSIPInboundTrunkReturnsOnCall(i int, result1 []*livekit.SIPInboundTrunkInfo, result2 error) { - fake.selectSIPInboundTrunkMutex.Lock() - defer fake.selectSIPInboundTrunkMutex.Unlock() - fake.SelectSIPInboundTrunkStub = nil - if fake.selectSIPInboundTrunkReturnsOnCall == nil { - fake.selectSIPInboundTrunkReturnsOnCall = make(map[int]struct { - result1 []*livekit.SIPInboundTrunkInfo - result2 error - }) - } - fake.selectSIPInboundTrunkReturnsOnCall[i] = struct { - result1 []*livekit.SIPInboundTrunkInfo - result2 error - }{result1, result2} -} - func (fake *FakeSIPStore) StoreSIPDispatchRule(arg1 context.Context, arg2 *livekit.SIPDispatchRuleInfo) error { fake.storeSIPDispatchRuleMutex.Lock() ret, specificReturn := fake.storeSIPDispatchRuleReturnsOnCall[len(fake.storeSIPDispatchRuleArgsForCall)] @@ -1271,10 +1113,6 @@ func (fake *FakeSIPStore) Invocations() map[string][][]interface{} { defer fake.loadSIPOutboundTrunkMutex.RUnlock() fake.loadSIPTrunkMutex.RLock() defer fake.loadSIPTrunkMutex.RUnlock() - fake.selectSIPDispatchRuleMutex.RLock() - defer fake.selectSIPDispatchRuleMutex.RUnlock() - fake.selectSIPInboundTrunkMutex.RLock() - defer fake.selectSIPInboundTrunkMutex.RUnlock() fake.storeSIPDispatchRuleMutex.RLock() defer fake.storeSIPDispatchRuleMutex.RUnlock() fake.storeSIPInboundTrunkMutex.RLock() diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index 3857da717..23075bc69 100644 --- a/pkg/service/wire_gen.go +++ b/pkg/service/wire_gen.go @@ -89,23 +89,23 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live } rtcEgressLauncher := NewEgressLauncher(egressClient, ioInfoService) topicFormatter := rpc.NewTopicFormatter() - roomClient, err := rpc.NewTypedRoomClient(clientParams) + v, err := rpc.NewTypedRoomClient(clientParams) if err != nil { return nil, err } - participantClient, err := rpc.NewTypedParticipantClient(clientParams) + v2, err := rpc.NewTypedParticipantClient(clientParams) if err != nil { return nil, err } - roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, roomClient, participantClient) + roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, v, v2) if err != nil { return nil, err } - agentDispatchInternalClient, err := rpc.NewTypedAgentDispatchInternalClient(clientParams) + v3, err := rpc.NewTypedAgentDispatchInternalClient(clientParams) if err != nil { return nil, err } - agentDispatchService := NewAgentDispatchService(agentDispatchInternalClient, topicFormatter, roomAllocator, router) + agentDispatchService := NewAgentDispatchService(v3, topicFormatter, roomAllocator, router) egressService := NewEgressService(egressClient, rtcEgressLauncher, objectStore, ioInfoService, roomService) ingressConfig := getIngressConfig(conf) ingressClient, err := rpc.NewIngressClient(clientParams) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index d2472db61..3c7d2d80e 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -240,6 +240,7 @@ type DowntrackParams struct { RTCPWriter func([]rtcp.Packet) error DisableSenderReportPassThrough bool SupportsCodecChange bool + IsReceiverSimulcast bool } // DownTrack implements TrackLocal, is the track used to write packets @@ -405,6 +406,7 @@ func NewDownTrack(params DowntrackParams) (*DownTrack, error) { d.params.Logger, false, d.rtpStats, + d.params.IsReceiverSimulcast, ) d.connectionStats = connectionquality.NewConnectionStats(connectionquality.ConnectionStatsParams{ diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 5251e7e90..672f0c7d5 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -223,6 +223,7 @@ type Forwarder struct { started bool preStartTime time.Time + isReceiverSimulcast bool extFirstTS uint64 lastSSRC uint32 lastReferencePayloadType int8 @@ -248,6 +249,7 @@ func NewForwarder( logger logger.Logger, skipReferenceTS bool, rtpStats *rtpstats.RTPStatsSender, + isReceiverSimulcast bool, ) *Forwarder { f := &Forwarder{ mime: mime.MimeTypeUnknown, @@ -255,6 +257,7 @@ func NewForwarder( logger: logger, skipReferenceTS: skipReferenceTS, rtpStats: rtpStats, + isReceiverSimulcast: isReceiverSimulcast, referenceLayerSpatial: buffer.InvalidLayerSpatial, lastAllocation: VideoAllocationDefault, lastReferencePayloadType: -1, @@ -340,40 +343,46 @@ func (f *Forwarder) DetermineCodec(codec webrtc.RTPCodecCapability, extensions [ } case mime.MimeTypeVP9: - // DD-TODO : we only enable dd layer selector for av1/vp9 now, in the future we can enable it for vp8 too - isDDAvailable := ddAvailable(extensions) - if isDDAvailable { - if f.vls != nil { - f.vls = videolayerselector.NewDependencyDescriptorFromOther(f.vls) - } else { - f.vls = videolayerselector.NewDependencyDescriptor(f.logger) - } - } else { - if f.vls != nil { - f.vls = videolayerselector.NewVP9FromOther(f.vls) - } else { - f.vls = videolayerselector.NewVP9(f.logger) - } - } - // SVC-TODO: Support for VP9 simulcast. When DD is not available, have to pick selector based on VP9 SVC or Simulcast - - case mime.MimeTypeAV1: - // DD-TODO : we only enable dd layer selector for av1/vp9 now, in the future we can enable it for vp8 too - isDDAvailable := ddAvailable(extensions) - if isDDAvailable { - if f.vls != nil { - f.vls = videolayerselector.NewDependencyDescriptorFromOther(f.vls) - } else { - f.vls = videolayerselector.NewDependencyDescriptor(f.logger) - } - } else { + if f.isReceiverSimulcast { if f.vls != nil { f.vls = videolayerselector.NewSimulcastFromOther(f.vls) } else { f.vls = videolayerselector.NewSimulcast(f.logger) } + // VP9-SIMULCAST-TODO: Add temporal layer selector for VP9 + } else { + isDDAvailable := ddAvailable(extensions) + if isDDAvailable { + if f.vls != nil { + f.vls = videolayerselector.NewDependencyDescriptorFromOther(f.vls) + } else { + f.vls = videolayerselector.NewDependencyDescriptor(f.logger) + } + } else { + if f.vls != nil { + f.vls = videolayerselector.NewVP9FromOther(f.vls) + } else { + f.vls = videolayerselector.NewVP9(f.logger) + } + } + } + + case mime.MimeTypeAV1: + isDDAvailable := ddAvailable(extensions) + if f.isReceiverSimulcast || !isDDAvailable { + // AV1-SIMULCAST-TODO: Add temporal layer selector for AV1 + if f.vls != nil { + f.vls = videolayerselector.NewSimulcastFromOther(f.vls) + } else { + f.vls = videolayerselector.NewSimulcast(f.logger) + } + } else { + if f.vls != nil { + f.vls = videolayerselector.NewDependencyDescriptorFromOther(f.vls) + } else { + f.vls = videolayerselector.NewDependencyDescriptor(f.logger) + } } - // SVC-TODO: Support for AV1 Simulcast } } diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 722c76769..2239ee690 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -32,7 +32,7 @@ func disable(f *Forwarder) { } func newForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType) *Forwarder { - f := NewForwarder(kind, logger.GetLogger(), true, nil) + f := NewForwarder(kind, logger.GetLogger(), true, nil, false) f.DetermineCodec(codec, nil) return f } diff --git a/pkg/sfu/mime/mimetype.go b/pkg/sfu/mime/mimetype.go index 118a7bad8..df24ab686 100644 --- a/pkg/sfu/mime/mimetype.go +++ b/pkg/sfu/mime/mimetype.go @@ -240,10 +240,6 @@ func IsMimeTypeVideo(mimeType MimeType) bool { return strings.HasPrefix(mimeType.String(), MimeTypePrefixVideo) } -// SVC-TODO: Have to use more conditions to differentiate between -// SVC-TODO: SVC and non-SVC (could be single layer or simulcast). -// SVC-TODO: May only need to differentiate between simulcast and non-simulcast -// SVC-TODO: i. e. may be possible to treat single layer as SVC to get proper/intended functionality. func IsMimeTypeSVC(mimeType MimeType) bool { switch mimeType { case MimeTypeAV1, MimeTypeVP9: diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 3f4fd9eff..4e3de0699 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -265,10 +265,16 @@ func NewWebRTCReceiver( codecState: ReceiverCodecStateNormal, kind: track.Kind(), onRTCP: onRTCP, - isSVC: mime.IsMimeTypeStringSVC(track.Codec().MimeType), isRED: mime.IsMimeTypeStringRED(track.Codec().MimeType), } + isSVC := false + isSimulcast := trackInfo.GetSimulcast() + if !isSimulcast { + isSVC = mime.IsMimeTypeStringSVC(track.Codec().MimeType) + } + w.isSVC = isSVC + for _, opt := range opts { w = opt(w) }