From 89a312d259fac53b91e13f0fe0de637874e9cd97 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Mon, 12 Feb 2024 11:49:14 +0530 Subject: [PATCH] Ignore duplicate RID. (#2471) Firefox on Windows 10 seems to be producing simulcast tracks with duplicate RID. That causes a leak as only one buffer is processed. Ignore duplicate rid. NOTE: This is not perfect as the actual layer -> rid is indeterminable at addition time. It would require looking at packets to determine the video dimensions and match to rid/layer to figure out which one is correct and which one is duplicate. To simplify though, taking the first one and dropping later ones. This could mean the correct resolution is not streamed, but that should be okay. The leak is far more destructive. --- pkg/rtc/mediatrack.go | 12 +++++++++- pkg/rtc/participant.go | 10 +++++++++ pkg/sfu/buffer/buffer.go | 4 ++-- pkg/sfu/receiver.go | 48 +++++++++++++++++++++------------------- 4 files changed, 48 insertions(+), 26 deletions(-) diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 2dfbfb9bd..16f6c24f3 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -309,7 +309,17 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra } t.lock.Unlock() - wr.(*sfu.WebRTCReceiver).AddUpTrack(track, buff) + if err := wr.(*sfu.WebRTCReceiver).AddUpTrack(track, buff); err != nil { + t.params.Logger.Warnw( + "adding up track failed", err, + "rid", track.RID(), + "layer", layer, + "ssrc", track.SSRC(), + "newCodec", newCodec, + ) + buff.Close() + 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() != "" { diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 7fb7b3c8c..3a29f321a 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -1137,6 +1137,8 @@ func (p *ParticipantImpl) UpdateMediaRTT(rtt uint32) { } } +// ---------------------------------------------------------- + type AnyTransportHandler struct { transport.UnimplementedHandler p *ParticipantImpl @@ -1154,6 +1156,8 @@ func (h AnyTransportHandler) OnICECandidate(c *webrtc.ICECandidate, target livek return h.p.onICECandidate(c, target) } +// ---------------------------------------------------------- + type PublisherTransportHandler struct { AnyTransportHandler } @@ -1174,6 +1178,8 @@ func (h PublisherTransportHandler) OnDataPacket(kind livekit.DataPacket_Kind, da h.p.onDataMessage(kind, data) } +// ---------------------------------------------------------- + type SubscriberTransportHandler struct { AnyTransportHandler } @@ -1190,6 +1196,8 @@ func (h SubscriberTransportHandler) OnInitialConnected() { h.p.onSubscriberInitialConnected() } +// ---------------------------------------------------------- + type PrimaryTransportHandler struct { transport.Handler p *ParticipantImpl @@ -1204,6 +1212,8 @@ func (h PrimaryTransportHandler) OnFullyEstablished() { h.p.onPrimaryTransportFullyEstablished() } +// ---------------------------------------------------------- + func (p *ParticipantImpl) setupTransportManager() error { p.twcc = twcc.NewTransportWideCCResponder() p.twcc.OnFeedback(func(pkts []rtcp.Packet) { diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index d657b6f8a..1e607eaf0 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -147,7 +147,7 @@ func (b *Buffer) SetLogger(logger logger.Logger) { b.Lock() defer b.Unlock() - b.logger = logger.WithComponent(sutils.ComponentSFU) + b.logger = logger.WithComponent(sutils.ComponentSFU).WithValues("ssrc", b.mediaSSRC) if b.rtpStats != nil { b.rtpStats.SetLogger(b.logger) } @@ -584,7 +584,7 @@ func (b *Buffer) patchExtPacket(ep *ExtPacket, buf []byte) *ExtPacket { n, err := b.getPacket(buf, ep.Packet.SequenceNumber) if err != nil { packetNotFoundCount := b.packetNotFoundCount.Inc() - if packetNotFoundCount%20 == 0 { + if (packetNotFoundCount-1)%20 == 0 { b.logger.Warnw("could not get packet from bucket", err, "sn", ep.Packet.SequenceNumber, "headSN", b.bucket.HeadSequenceNumber(), "count", packetNotFoundCount) } return nil diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index bdb8853ea..0eba7f48d 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -42,6 +42,7 @@ var ( ErrReceiverClosed = errors.New("receiver closed") ErrDownTrackAlreadyExist = errors.New("DownTrack already exist") ErrBufferNotFound = errors.New("buffer not found") + ErrDuplicateLayer = errors.New("duplicate layer") ) type AudioLevelHandle func(level uint8, duration uint32) @@ -114,11 +115,9 @@ type WebRTCReceiver struct { bufferMu sync.RWMutex buffers [buffer.DefaultMaxLayerSpatial + 1]*buffer.Buffer + upTracks [buffer.DefaultMaxLayerSpatial + 1]*webrtc.TrackRemote rtt uint32 - upTrackMu sync.RWMutex - upTracks [buffer.DefaultMaxLayerSpatial + 1]*webrtc.TrackRemote - lbThreshold int streamTrackerManager *StreamTrackerManager @@ -265,9 +264,16 @@ func (w *WebRTCReceiver) OnStatsUpdate(fn func(w *WebRTCReceiver, stat *livekit. } func (w *WebRTCReceiver) OnMaxLayerChange(fn func(maxLayer int32)) { - w.upTrackMu.Lock() + w.bufferMu.Lock() w.onMaxLayerChange = fn - w.upTrackMu.Unlock() + w.bufferMu.Unlock() +} + +func (w *WebRTCReceiver) getOnMaxLayerChange() func(maxLayer int32) { + w.bufferMu.RLock() + defer w.bufferMu.RUnlock() + + return w.onMaxLayerChange } func (w *WebRTCReceiver) GetConnectionScoreAndQuality() (float32, livekit.ConnectionQuality) { @@ -306,10 +312,7 @@ func (w *WebRTCReceiver) TrackID() livekit.TrackID { return w.trackID } -func (w *WebRTCReceiver) SSRC(layer int) uint32 { - w.upTrackMu.RLock() - defer w.upTrackMu.RUnlock() - +func (w *WebRTCReceiver) ssrc(layer int) uint32 { if track := w.upTracks[layer]; track != nil { return uint32(track.SSRC()) } @@ -328,9 +331,9 @@ func (w *WebRTCReceiver) Kind() webrtc.RTPCodecType { return w.kind } -func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buffer) { +func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buffer) error { if w.closed.Load() { - return + return ErrReceiverClosed } layer := int32(0) @@ -369,14 +372,16 @@ func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buff buff.SetPLIThrottle(duration.Nanoseconds()) } - w.upTrackMu.Lock() - w.upTracks[layer] = track - w.upTrackMu.Unlock() - w.bufferMu.Lock() + if w.upTracks[layer] != nil { + w.bufferMu.Unlock() + return ErrDuplicateLayer + } + w.upTracks[layer] = track w.buffers[layer] = buff rtt := w.rtt w.bufferMu.Unlock() + buff.SetRTT(rtt) buff.SetPaused(w.streamTrackerManager.IsPaused()) @@ -385,6 +390,7 @@ func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buff } go w.forwardRTP(layer) + return nil } // SetUpTrackPaused indicates upstream will not be sending any data. @@ -471,11 +477,7 @@ func (w *WebRTCReceiver) OnMaxTemporalLayerSeenChanged(maxTemporalLayerSeen int3 // StreamTrackerManagerListener.OnMaxAvailableLayerChanged func (w *WebRTCReceiver) OnMaxAvailableLayerChanged(maxAvailableLayer int32) { - w.upTrackMu.RLock() - onMaxLayerChange := w.onMaxLayerChange - w.upTrackMu.RUnlock() - - if onMaxLayerChange != nil { + if onMaxLayerChange := w.getOnMaxLayerChange(); onMaxLayerChange != nil { onMaxLayerChange(maxAvailableLayer) } } @@ -619,7 +621,7 @@ func (w *WebRTCReceiver) GetDeltaStats() map[uint32]*buffer.StreamStatsWithLayer patched[int32(layer)] = sswl.Layers[0] sswl.Layers = patched - deltaStats[w.SSRC(layer)] = sswl + deltaStats[w.ssrc(layer)] = sswl } return deltaStats @@ -711,7 +713,7 @@ func (w *WebRTCReceiver) DebugInfo() map[string]interface{} { "Simulcast": isSimulcast, } - w.upTrackMu.RLock() + w.bufferMu.RLock() upTrackInfo := make([]map[string]interface{}, 0, len(w.upTracks)) for layer, ut := range w.upTracks { if ut != nil { @@ -723,7 +725,7 @@ func (w *WebRTCReceiver) DebugInfo() map[string]interface{} { }) } } - w.upTrackMu.RUnlock() + w.bufferMu.RUnlock() info["UpTracks"] = upTrackInfo return info