From c743d2ab3a72a52ea9b148ad07908c94560f428b Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 2 Mar 2022 19:08:07 +0530 Subject: [PATCH] Warn on ops queue full (#481) Pass logger so that we can log with proper context. --- pkg/sfu/buffer/buffer.go | 6 ++++-- pkg/sfu/downtrack.go | 2 +- pkg/sfu/receiver.go | 2 +- pkg/sfu/streamtracker.go | 5 +++-- pkg/sfu/streamtracker_test.go | 11 ++++++----- pkg/sfu/streamtrackermanager.go | 9 +++++++-- pkg/utils/opsqueue.go | 19 ++++++++++++++++--- 7 files changed, 38 insertions(+), 16 deletions(-) diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 0d40cf488..9af90a916 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/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) { diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 9fd9f2c76..2a516e5cb 100644 --- a/pkg/sfu/downtrack.go +++ b/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{}), } diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 5ac5d647f..0dfdf7eaa 100644 --- a/pkg/sfu/receiver.go +++ b/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) diff --git a/pkg/sfu/streamtracker.go b/pkg/sfu/streamtracker.go index 552b4efb3..13e385dd7 100644 --- a/pkg/sfu/streamtracker.go +++ b/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 } diff --git a/pkg/sfu/streamtracker_test.go b/pkg/sfu/streamtracker_test.go index 90582d6fe..04beb0525 100644 --- a/pkg/sfu/streamtracker_test.go +++ b/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() diff --git a/pkg/sfu/streamtrackermanager.go b/pkg/sfu/streamtrackermanager.go index f8b6a7682..55df1ed2b 100644 --- a/pkg/sfu/streamtrackermanager.go +++ b/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) diff --git a/pkg/utils/opsqueue.go b/pkg/utils/opsqueue.go index a09c91528..ce5fc661e 100644 --- a/pkg/utils/opsqueue.go +++ b/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() }