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

@@ -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())

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

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

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])

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

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)

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()

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()

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)

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{

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
 	}
 }
 

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
 }

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:

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)
 	}