Warn on ops queue full (#481)

Pass logger so that we can log with proper context.

pkg/sfu/buffer/buffer.go

@@ -110,13 +110,14 @@ type Options struct {
 
 // NewBuffer constructs a new Buffer
 func NewBuffer(ssrc uint32, vp, ap *sync.Pool) *Buffer {
+	logger := logger.Logger(logger.GetLogger()) // will be reset with correct context via SetLogger
 	b := &Buffer{
 		mediaSSRC:      ssrc,
 		videoPool:      vp,
 		audioPool:      ap,
 		pliThrottle:    int64(500 * time.Millisecond),
-		logger:         logger.Logger(logger.GetLogger()), // will be reset with correct context via SetLogger
-		callbacksQueue: utils.NewOpsQueue(),
+		logger:         logger,
+		callbacksQueue: utils.NewOpsQueue(logger),
 	}
 	b.bitrate.Store(make([]int64, len(b.bitrateHelper)))
 	b.extPackets.SetMinCapacity(7)
@@ -125,6 +126,7 @@ func NewBuffer(ssrc uint32, vp, ap *sync.Pool) *Buffer {
 
 func (b *Buffer) SetLogger(logger logger.Logger) {
 	b.logger = logger
+	b.callbacksQueue.SetLogger(logger)
 }
 
 func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapability, o Options) {

pkg/sfu/downtrack.go

@@ -174,7 +174,7 @@ func NewDownTrack(
 		codec:          c,
 		kind:           kind,
 		forwarder:      NewForwarder(c, kind, logger),
-		callbacksQueue: utils.NewOpsQueue(),
+		callbacksQueue: utils.NewOpsQueue(logger),
 		closed:         make(chan struct{}),
 	}
 

pkg/sfu/receiver.go

@@ -154,7 +154,7 @@ func NewWebRTCReceiver(
 		index:                make(map[livekit.ParticipantID]int),
 		free:                 make(map[int]struct{}),
 		numProcs:             runtime.NumCPU(),
-		streamTrackerManager: NewStreamTrackerManager(),
+		streamTrackerManager: NewStreamTrackerManager(logger),
 	}
 	w.streamTrackerManager.OnAvailableLayersChanged(w.downTrackLayerChange)
 

pkg/sfu/streamtracker.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/livekit/livekit-server/pkg/utils"
+	"github.com/livekit/protocol/logger"
 	"go.uber.org/atomic"
 )
 
@@ -58,13 +59,13 @@ type StreamTracker struct {
 	isStopped atomic.Bool
 }
 
-func NewStreamTracker(samplesRequired uint32, cyclesRequired uint64, cycleDuration time.Duration) *StreamTracker {
+func NewStreamTracker(logger logger.Logger, samplesRequired uint32, cyclesRequired uint64, cycleDuration time.Duration) *StreamTracker {
 	s := &StreamTracker{
 		samplesRequired: samplesRequired,
 		cyclesRequired:  cyclesRequired,
 		cycleDuration:   cycleDuration,
 		status:          StreamStatusStopped,
-		callbacksQueue:  utils.NewOpsQueue(),
+		callbacksQueue:  utils.NewOpsQueue(logger),
 	}
 	return s
 }

pkg/sfu/streamtracker_test.go

@@ -9,12 +9,13 @@ import (
 	"go.uber.org/atomic"
 
 	"github.com/livekit/livekit-server/pkg/testutils"
+	"github.com/livekit/protocol/logger"
 )
 
 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 := NewStreamTracker(logger.Logger(logger.GetLogger()), 5, 60, 500*time.Millisecond)
 		tracker.Start()
 		tracker.OnStatusChanged(func(status StreamStatus) {
 			callbackCalled.Store(true)
@@ -39,7 +40,7 @@ func TestStreamTracker(t *testing.T) {
 	})
 
 	t.Run("flips to inactive immediately", func(t *testing.T) {
-		tracker := NewStreamTracker(5, 60, 500*time.Millisecond)
+		tracker := NewStreamTracker(logger.Logger(logger.GetLogger()), 5, 60, 500*time.Millisecond)
 		tracker.Start()
 		require.Equal(t, StreamStatusStopped, tracker.Status())
 
@@ -75,7 +76,7 @@ func TestStreamTracker(t *testing.T) {
 	})
 
 	t.Run("flips back to active after iterations", func(t *testing.T) {
-		tracker := NewStreamTracker(1, 2, 500*time.Millisecond)
+		tracker := NewStreamTracker(logger.Logger(logger.GetLogger()), 1, 2, 500*time.Millisecond)
 		tracker.Start()
 		require.Equal(t, StreamStatusStopped, tracker.Status())
 
@@ -102,7 +103,7 @@ func TestStreamTracker(t *testing.T) {
 	})
 
 	t.Run("does not change to inactive when paused", func(t *testing.T) {
-		tracker := NewStreamTracker(5, 60, 500*time.Millisecond)
+		tracker := NewStreamTracker(logger.Logger(logger.GetLogger()), 5, 60, 500*time.Millisecond)
 		tracker.Start()
 		tracker.Observe(1)
 		testutils.WithTimeout(t, func() string {
@@ -122,7 +123,7 @@ func TestStreamTracker(t *testing.T) {
 
 	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 := NewStreamTracker(logger.Logger(logger.GetLogger()), 5, 60, 500*time.Millisecond)
 		tracker.Start()
 		tracker.OnStatusChanged(func(status StreamStatus) {
 			callbackCalled.Inc()

pkg/sfu/streamtrackermanager.go

@@ -4,9 +4,13 @@ import (
 	"sort"
 	"sync"
 	"time"
+
+	"github.com/livekit/protocol/logger"
 )
 
 type StreamTrackerManager struct {
+	logger logger.Logger
+
 	lock sync.RWMutex
 
 	trackers [DefaultMaxLayerSpatial + 1]*StreamTracker
@@ -17,8 +21,9 @@ type StreamTrackerManager struct {
 	onAvailableLayersChanged func(availableLayers []int32)
 }
 
-func NewStreamTrackerManager() *StreamTrackerManager {
+func NewStreamTrackerManager(logger logger.Logger) *StreamTrackerManager {
 	return &StreamTrackerManager{
+		logger:           logger,
 		maxExpectedLayer: DefaultMaxLayerSpatial,
 	}
 }
@@ -37,7 +42,7 @@ func (s *StreamTrackerManager) AddTracker(layer int32) {
 		cyclesRequired = 1 // 1 packet in 2 seconds
 		cycleDuration = 2 * time.Second
 	}
-	tracker := NewStreamTracker(samplesRequired, cyclesRequired, cycleDuration)
+	tracker := NewStreamTracker(s.logger, samplesRequired, cyclesRequired, cycleDuration)
 	tracker.OnStatusChanged(func(status StreamStatus) {
 		if status == StreamStatusStopped {
 			s.removeAvailableLayer(layer)

pkg/utils/opsqueue.go

@@ -2,6 +2,8 @@ package utils
 
 import (
 	"sync"
+
+	"github.com/livekit/protocol/logger"
 )
 
 const (
@@ -9,17 +11,24 @@ const (
 )
 
 type OpsQueue struct {
+	logger logger.Logger
+
 	lock      sync.RWMutex
 	ops       chan func()
 	isStopped bool
 }
 
-func NewOpsQueue() *OpsQueue {
+func NewOpsQueue(logger logger.Logger) *OpsQueue {
 	return &OpsQueue{
-		ops: make(chan func(), MaxOps),
+		logger: logger,
+		ops:    make(chan func(), MaxOps),
 	}
 }
 
+func (oq *OpsQueue) SetLogger(logger logger.Logger) {
+	oq.logger = logger
+}
+
 func (oq *OpsQueue) Start() {
 	go oq.process()
 }
@@ -43,7 +52,11 @@ func (oq *OpsQueue) Enqueue(op func()) {
 		return
 	}
 
-	oq.ops <- op
+	select {
+	case oq.ops <- op:
+	default:
+		oq.logger.Warnw("ops queue full", nil)
+	}
 	oq.lock.RUnlock()
 }