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