diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 0c3ee7b64..39f596f01 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -20,6 +20,7 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/livekit-server/pkg/sfu/connectionquality" + "github.com/livekit/livekit-server/pkg/utils" ) // TrackSender defines an interface send media to remote peer @@ -110,6 +111,8 @@ type DownTrack struct { isNACKThrottled atomic.Bool + callbacksQueue *utils.OpsQueue + // RTCP callbacks onREMB func(dt *DownTrack, remb *rtcp.ReceiverEstimatedMaximumBitrate) onTransportCCFeedback func(dt *DownTrack, cc *rtcp.TransportLayerCC) @@ -124,10 +127,10 @@ type DownTrack struct { onSubscribedLayersChanged func(dt *DownTrack, layers VideoLayers) // packet sent callback - onPacketSent []func(dt *DownTrack, size int) + onPacketSentUnsafe []func(dt *DownTrack, size int) // padding packet sent callback - onPaddingSent []func(dt *DownTrack, size int) + onPaddingSentUnsafe []func(dt *DownTrack, size int) // update stats onStatsUpdate func(dt *DownTrack, stat *livekit.AnalyticsStat) @@ -161,17 +164,18 @@ func NewDownTrack( } d := &DownTrack{ - logger: logger, - id: r.TrackID(), - peerID: peerID, - maxTrack: mt, - streamID: r.StreamID(), - bufferFactory: bf, - receiver: r, - codec: c, - kind: kind, - forwarder: NewForwarder(c, kind, logger), - closed: make(chan struct{}), + logger: logger, + id: r.TrackID(), + peerID: peerID, + maxTrack: mt, + streamID: r.StreamID(), + bufferFactory: bf, + receiver: r, + codec: c, + kind: kind, + forwarder: NewForwarder(c, kind, logger), + callbacksQueue: utils.NewOpsQueue(), + closed: make(chan struct{}), } d.connectionStats = connectionquality.NewConnectionStats(connectionquality.ConnectionStatsParams{ @@ -185,7 +189,9 @@ func NewDownTrack( }) d.connectionStats.OnStatsUpdate(func(_cs *connectionquality.ConnectionStats, stat *livekit.AnalyticsStat) { if d.onStatsUpdate != nil { - d.onStatsUpdate(d, stat) + d.callbacksQueue.Enqueue(func() { + d.onStatsUpdate(d, stat) + }) } }) d.connectionStats.Start() @@ -201,27 +207,33 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, return webrtc.RTPCodecParameters{}, ErrTrackAlreadyBind } parameters := webrtc.RTPCodecParameters{RTPCodecCapability: d.codec} - if codec, err := codecParametersFuzzySearch(parameters, t.CodecParameters()); err == nil { - d.ssrc = uint32(t.SSRC()) - d.payloadType = uint8(codec.PayloadType) - d.writeStream = t.WriteStream() - d.mime = strings.ToLower(codec.MimeType) - if rr := d.bufferFactory.GetOrNew(packetio.RTCPBufferPacket, uint32(t.SSRC())).(*buffer.RTCPReader); rr != nil { - rr.OnPacket(func(pkt []byte) { - d.handleRTCP(pkt) - }) - } - if strings.HasPrefix(d.codec.MimeType, "video/") { - d.sequencer = newSequencer(d.maxTrack, d.logger) - } - if d.onBind != nil { - d.onBind() - } - d.bound.Store(true) - go d.requestFirstKeyframe() - return codec, nil + codec, err := codecParametersFuzzySearch(parameters, t.CodecParameters()) + if err != nil { + return webrtc.RTPCodecParameters{}, webrtc.ErrUnsupportedCodec } - return webrtc.RTPCodecParameters{}, webrtc.ErrUnsupportedCodec + + d.ssrc = uint32(t.SSRC()) + d.payloadType = uint8(codec.PayloadType) + d.writeStream = t.WriteStream() + d.mime = strings.ToLower(codec.MimeType) + if rr := d.bufferFactory.GetOrNew(packetio.RTCPBufferPacket, uint32(t.SSRC())).(*buffer.RTCPReader); rr != nil { + rr.OnPacket(func(pkt []byte) { + d.handleRTCP(pkt) + }) + } + if strings.HasPrefix(d.codec.MimeType, "video/") { + d.sequencer = newSequencer(d.maxTrack, d.logger) + } + if d.onBind != nil { + d.callbacksQueue.Enqueue(d.onBind) + } + d.bound.Store(true) + + go d.requestFirstKeyframe() + + d.callbacksQueue.Start() + + return codec, nil } // Unbind implements the teardown logic when the track is no longer needed. This happens @@ -350,12 +362,14 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { _, err = d.writeStream.WriteRTP(hdr, payload) if err == nil { pktSize := hdr.MarshalSize() + len(payload) - for _, f := range d.onPacketSent { + for _, f := range d.onPacketSentUnsafe { f(d, pktSize) } if tp.isSwitchingToMaxLayer && d.onMaxLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo { - d.onMaxLayerChanged(d, layer) + d.callbacksQueue.Enqueue(func() { + d.onMaxLayerChanged(d, layer) + }) } d.updatePrimaryStats(pktSize, hdr.Marker) @@ -449,7 +463,7 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int) int { size := hdr.MarshalSize() + len(payload) d.updatePaddingStats(size) - for _, f := range d.onPaddingSent { + for _, f := range d.onPaddingSentUnsafe { f(d, size) } @@ -477,19 +491,25 @@ func (d *DownTrack) Mute(muted bool) { if d.onMaxLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo { if muted { - d.onMaxLayerChanged(d, InvalidLayerSpatial) + d.callbacksQueue.Enqueue(func() { + d.onMaxLayerChanged(d, InvalidLayerSpatial) + }) } else { // // When unmuting, don't wait for layer lock as // client might need to be notified to start layers // before locking can happen in the forwarder. // - d.onMaxLayerChanged(d, maxLayers.spatial) + d.callbacksQueue.Enqueue(func() { + d.onMaxLayerChanged(d, maxLayers.spatial) + }) } } if d.onSubscriptionChanged != nil { - d.onSubscriptionChanged(d) + d.callbacksQueue.Enqueue(func() { + d.onSubscriptionChanged(d) + }) } } @@ -521,12 +541,16 @@ func (d *DownTrack) CloseWithFlush(flush bool) { d.connectionStats.Close() if d.onMaxLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo { - d.onMaxLayerChanged(d, InvalidLayerSpatial) + d.callbacksQueue.Enqueue(func() { + d.onMaxLayerChanged(d, InvalidLayerSpatial) + }) } if d.onCloseHandler != nil { - d.onCloseHandler() + d.callbacksQueue.Enqueue(d.onCloseHandler) } + + d.callbacksQueue.Stop() close(d.closed) }) } @@ -545,11 +569,15 @@ func (d *DownTrack) SetMaxSpatialLayer(spatialLayer int32) { // a. is higher than previous max -> client may need to start higher layer before forwarder can lock // b. is lower than previous max -> client can stop higher layer(s) // - d.onMaxLayerChanged(d, maxLayers.spatial) + d.callbacksQueue.Enqueue(func() { + d.onMaxLayerChanged(d, maxLayers.spatial) + }) } if d.onSubscribedLayersChanged != nil { - d.onSubscribedLayersChanged(d, maxLayers) + d.callbacksQueue.Enqueue(func() { + d.onSubscribedLayersChanged(d, maxLayers) + }) } } @@ -560,7 +588,9 @@ func (d *DownTrack) SetMaxTemporalLayer(temporalLayer int32) { } if d.onSubscribedLayersChanged != nil { - d.onSubscribedLayersChanged(d, maxLayers) + d.callbacksQueue.Enqueue(func() { + d.onSubscribedLayersChanged(d, maxLayers) + }) } } @@ -576,7 +606,9 @@ func (d *DownTrack) UpTrackLayersChange(availableLayers []int32) { d.forwarder.UpTrackLayersChange(availableLayers) if d.onAvailableLayersChanged != nil { - d.onAvailableLayersChanged(d) + d.callbacksQueue.Enqueue(func() { + d.onAvailableLayersChanged(d) + }) } } @@ -616,12 +648,12 @@ func (d *DownTrack) OnSubscribedLayersChanged(fn func(dt *DownTrack, layers Vide d.onSubscribedLayersChanged = fn } -func (d *DownTrack) OnPacketSent(fn func(dt *DownTrack, size int)) { - d.onPacketSent = append(d.onPacketSent, fn) +func (d *DownTrack) OnPacketSentUnsafe(fn func(dt *DownTrack, size int)) { + d.onPacketSentUnsafe = append(d.onPacketSentUnsafe, fn) } -func (d *DownTrack) OnPaddingSent(fn func(dt *DownTrack, size int)) { - d.onPaddingSent = append(d.onPaddingSent, fn) +func (d *DownTrack) OnPaddingSentUnsafe(fn func(dt *DownTrack, size int)) { + d.onPaddingSentUnsafe = append(d.onPaddingSentUnsafe, fn) } func (d *DownTrack) OnStatsUpdate(fn func(dt *DownTrack, stat *livekit.AnalyticsStat)) { @@ -836,7 +868,7 @@ func (d *DownTrack) writeBlankFrameRTP() error { return err } - for _, f := range d.onPacketSent { + for _, f := range d.onPacketSentUnsafe { f(d, pktSize) } @@ -925,7 +957,9 @@ func (d *DownTrack) handleRTCP(bytes []byte) { case *rtcp.ReceiverEstimatedMaximumBitrate: if d.onREMB != nil { - d.onREMB(d, p) + d.callbacksQueue.Enqueue(func() { + d.onREMB(d, p) + }) } case *rtcp.ReceiverReport: @@ -973,7 +1007,9 @@ func (d *DownTrack) handleRTCP(bytes []byte) { case *rtcp.TransportLayerCC: if p.MediaSSRC == d.ssrc && d.onTransportCCFeedback != nil { - d.onTransportCCFeedback(d, p) + d.callbacksQueue.Enqueue(func() { + d.onTransportCCFeedback(d, p) + }) } } } @@ -990,7 +1026,9 @@ func (d *DownTrack) handleRTCP(bytes []byte) { } if d.onRttUpdate != nil { - d.onRttUpdate(d, rttToReport) + d.callbacksQueue.Enqueue(func() { + d.onRttUpdate(d, rttToReport) + }) } } } @@ -1092,7 +1130,7 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) { d.logger.Errorw("writing rtx packet err", err) } else { pktSize := pkt.Header.MarshalSize() + len(payload) - for _, f := range d.onPacketSent { + for _, f := range d.onPacketSentUnsafe { f(d, pktSize) } diff --git a/pkg/sfu/streamallocator.go b/pkg/sfu/streamallocator.go index 2b9ec7331..3c0acff35 100644 --- a/pkg/sfu/streamallocator.go +++ b/pkg/sfu/streamallocator.go @@ -222,7 +222,7 @@ func (s *StreamAllocator) AddTrack(downTrack *DownTrack, params AddTrackParams) downTrack.OnAvailableLayersChanged(s.onAvailableLayersChanged) downTrack.OnSubscriptionChanged(s.onSubscriptionChanged) downTrack.OnSubscribedLayersChanged(s.onSubscribedLayersChanged) - downTrack.OnPacketSent(s.onPacketSent) + downTrack.OnPacketSentUnsafe(s.onPacketSent) } } diff --git a/pkg/sfu/streamtracker.go b/pkg/sfu/streamtracker.go index 7d3aeeffa..552b4efb3 100644 --- a/pkg/sfu/streamtracker.go +++ b/pkg/sfu/streamtracker.go @@ -4,6 +4,7 @@ import ( "sync" "time" + "github.com/livekit/livekit-server/pkg/utils" "go.uber.org/atomic" ) @@ -52,6 +53,8 @@ type StreamTracker struct { // only access by the same goroutine as Observe lastSN uint16 + callbacksQueue *utils.OpsQueue + isStopped atomic.Bool } @@ -61,6 +64,7 @@ func NewStreamTracker(samplesRequired uint32, cyclesRequired uint64, cycleDurati cyclesRequired: cyclesRequired, cycleDuration: cycleDuration, status: StreamStatusStopped, + callbacksQueue: utils.NewOpsQueue(), } return s } @@ -86,7 +90,9 @@ func (s *StreamTracker) maybeSetStatus(status StreamStatus) { s.statusMu.Unlock() if changed && s.onStatusChanged != nil { - s.onStatusChanged(status) + s.callbacksQueue.Enqueue(func() { + s.onStatusChanged(status) + }) } } @@ -105,6 +111,7 @@ func (s *StreamTracker) init() { } func (s *StreamTracker) Start() { + s.callbacksQueue.Start() } func (s *StreamTracker) Stop() { @@ -112,6 +119,8 @@ func (s *StreamTracker) Stop() { return } + s.callbacksQueue.Stop() + // bump generation to trigger exit of worker s.generation.Inc() } diff --git a/pkg/sfu/streamtracker_test.go b/pkg/sfu/streamtracker_test.go index e45836481..90582d6fe 100644 --- a/pkg/sfu/streamtracker_test.go +++ b/pkg/sfu/streamtracker_test.go @@ -15,6 +15,7 @@ func TestStreamTracker(t *testing.T) { t.Run("flips to active on first observe", func(t *testing.T) { callbackCalled := atomic.NewBool(false) tracker := NewStreamTracker(5, 60, 500*time.Millisecond) + tracker.Start() tracker.OnStatusChanged(func(status StreamStatus) { callbackCalled.Store(true) }) @@ -33,10 +34,13 @@ func TestStreamTracker(t *testing.T) { require.Equal(t, StreamStatusActive, tracker.Status()) require.True(t, callbackCalled.Load()) + + tracker.Stop() }) t.Run("flips to inactive immediately", func(t *testing.T) { tracker := NewStreamTracker(5, 60, 500*time.Millisecond) + tracker.Start() require.Equal(t, StreamStatusStopped, tracker.Status()) tracker.Observe(1) @@ -56,12 +60,23 @@ func TestStreamTracker(t *testing.T) { // run a single iteration tracker.detectChanges() + + testutils.WithTimeout(t, func() string { + if callbackCalled.Load() { + return "" + } else { + return "inactive cycle did not declare stream stopped" + } + }) require.Equal(t, StreamStatusStopped, tracker.Status()) require.True(t, callbackCalled.Load()) + + tracker.Stop() }) t.Run("flips back to active after iterations", func(t *testing.T) { tracker := NewStreamTracker(1, 2, 500*time.Millisecond) + tracker.Start() require.Equal(t, StreamStatusStopped, tracker.Status()) tracker.Observe(1) @@ -82,10 +97,13 @@ func TestStreamTracker(t *testing.T) { tracker.Observe(3) tracker.detectChanges() require.Equal(t, StreamStatusActive, tracker.Status()) + + tracker.Stop() }) t.Run("does not change to inactive when paused", func(t *testing.T) { tracker := NewStreamTracker(5, 60, 500*time.Millisecond) + tracker.Start() tracker.Observe(1) testutils.WithTimeout(t, func() string { if tracker.Status() == StreamStatusActive { @@ -98,11 +116,14 @@ func TestStreamTracker(t *testing.T) { tracker.SetPaused(true) tracker.detectChanges() require.Equal(t, StreamStatusActive, tracker.Status()) + + tracker.Stop() }) t.Run("flips back to active on first observe after reset", func(t *testing.T) { callbackCalled := atomic.NewUint32(0) tracker := NewStreamTracker(5, 60, 500*time.Millisecond) + tracker.Start() tracker.OnStatusChanged(func(status StreamStatus) { callbackCalled.Inc() }) @@ -149,5 +170,7 @@ func TestStreamTracker(t *testing.T) { require.Equal(t, StreamStatusActive, tracker.Status()) require.Equal(t, uint32(2), callbackCalled.Load()) + + tracker.Stop() }) }