Improve dynamic simulcast, properly cleanup after pendingTracks (#84)

pkg/rtc/mediatrack.go

@@ -3,6 +3,7 @@ package rtc
 import (
 	"errors"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/livekit/protocol/utils"
@@ -12,7 +13,6 @@ import (
 	"github.com/pion/rtcp"
 	"github.com/pion/webrtc/v3"
 	"github.com/pion/webrtc/v3/pkg/rtcerr"
-	"github.com/thoas/go-funk"
 
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/logger"
@@ -37,6 +37,7 @@ type MediaTrack struct {
 	kind        livekit.TrackType
 	codec       webrtc.RTPCodecParameters
 	muted       utils.AtomicFlag
+	numUpTracks uint32
 	simulcasted bool
 
 	// channel to send RTCP packets to the source
@@ -98,25 +99,17 @@ func (t *MediaTrack) IsMuted() bool {
 func (t *MediaTrack) SetMuted(muted bool) {
 	t.muted.TrySet(muted)
 
-	// mute all of the subscribedtracks
 	t.lock.RLock()
+	if t.receiver != nil {
+		t.receiver.SetUpTrackPaused(muted)
+	}
+	// mute all of the subscribedtracks
 	for _, st := range t.subscribedTracks {
 		st.SetPublisherMuted(muted)
 	}
 	t.lock.RUnlock()
 }
 
-func (t *MediaTrack) SetSimulcastLayers(layers []livekit.VideoQuality) {
-	t.lock.RLock()
-	defer t.lock.RUnlock()
-	if t.receiver != nil {
-		layers16 := funk.Map(layers, func(l livekit.VideoQuality) uint16 {
-			return uint16(spatialLayerForQuality(l))
-		}).([]uint16)
-		t.receiver.SetAvailableLayers(layers16)
-	}
-}
-
 func (t *MediaTrack) OnClose(f func()) {
 	t.onClose = f
 }
@@ -239,6 +232,10 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error {
 	return nil
 }
 
+func (t *MediaTrack) NumUpTracks() uint32 {
+	return atomic.LoadUint32(&t.numUpTracks)
+}
+
 // AddReceiver adds a new RTP receiver to the track
 func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.TrackRemote, twcc *twcc.Responder) {
 	t.lock.Lock()
@@ -286,7 +283,8 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
 	if t.receiver == nil {
 		t.receiver = sfu.NewWebRTCReceiver(receiver, track, t.params.ParticipantID,
 			sfu.WithPliThrottle(0),
-			sfu.WithLoadBalanceThreshold(20))
+			sfu.WithLoadBalanceThreshold(20),
+			sfu.WithStreamTrackers())
 		t.receiver.SetRTCPCh(t.params.RTCPChan)
 		t.receiver.OnCloseHandler(func() {
 			t.lock.Lock()
@@ -304,6 +302,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
 	t.receiver.AddUpTrack(track, buff, t.shouldStartWithBestQuality())
 	// when RID is set, track is simulcasted
 	t.simulcasted = track.RID() != ""
+	atomic.AddUint32(&t.numUpTracks, 1)
 
 	buff.Bind(receiver.GetParameters(), buffer.Options{
 		MaxBitRate: t.params.ReceiverConfig.maxBitrate,

pkg/rtc/participant.go

@@ -731,15 +731,13 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w
 		return
 	}
 
-	// delete pending track if it's not simulcasting
-	// TODO: we should delete it after adding three tracks
-	ti := p.getPendingTrack(track.ID(), ToProtoTrackKind(track.Kind()), track.RID() == "")
-	if ti == nil {
-		return
-	}
-
 	// use existing mediatrack to handle simulcast
 	p.lock.Lock()
+	ti := p.getPendingTrack(track.ID(), ToProtoTrackKind(track.Kind()), false)
+	if ti == nil {
+		p.lock.Unlock()
+		return
+	}
 	ptrack := p.publishedTracks[ti.Sid]
 
 	var mt *MediaTrack
@@ -772,6 +770,11 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w
 		})
 	}
 	mt.AddReceiver(rtpReceiver, track, p.twcc)
+
+	// cleanup pendingTracks
+	if !mt.simulcasted || mt.NumUpTracks() == 3 {
+		_ = p.getPendingTrack(track.ID(), ToProtoTrackKind(track.Kind()), true)
+	}
 	p.lock.Unlock()
 
 	if newTrack {
@@ -800,8 +803,6 @@ func (p *ParticipantImpl) onDataChannel(dc *webrtc.DataChannel) {
 }
 
 func (p *ParticipantImpl) getPendingTrack(clientId string, kind livekit.TrackType, deleteAfter bool) *livekit.TrackInfo {
-	p.lock.Lock()
-	defer p.lock.Unlock()
 	ti := p.pendingTracks[clientId]
 
 	// then find the first one that matches type. with MediaStreamTrack, it's possible for the client id to

pkg/rtc/types/interfaces.go

@@ -93,7 +93,6 @@ type PublishedTrack interface {
 	Name() string
 	IsMuted() bool
 	SetMuted(muted bool)
-	SetSimulcastLayers(layers []livekit.VideoQuality)
 	AddSubscriber(participant Participant) error
 	RemoveSubscriber(participantId string)
 	IsSubscriber(subId string) bool

pkg/service/roommanager.go

@@ -476,16 +476,7 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici
 			case *livekit.SignalRequest_Leave:
 				_ = participant.Close()
 			case *livekit.SignalRequest_Simulcast:
-				for _, track := range participant.GetPublishedTracks() {
-					if track.ID() == msg.Simulcast.TrackSid {
-						logger.Debugw("updating simulcast layers",
-							"participant", participant.Identity(),
-							"pID", participant.ID(),
-							"track", track.ID(),
-							"layers", msg.Simulcast.Layers)
-						track.SetSimulcastLayers(msg.Simulcast.Layers)
-					}
-				}
+				// deprecated
 			}
 		}
 	}