Callback queue for a couple of modules (#468)

* callback queue for stream tracker

* Ops queue for down track

* fix test

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

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

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

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