Use Deque in ops queue. (#2418)

* Use Seque in ops queue.

Standardizing some uses
- Change OpsQueue to use Deque so that it can grow/shrink as necessary and
  need not worry about channel getting full and dropping events.
- Change StreamAllocator and TelemetryService to use OpsQueue so that
  they also need not worry about channel size and overflows.

* Address feedback

* delete obvious comment

* clean up

pkg/rtc/dynacastmanager.go

@@ -59,7 +59,7 @@ func NewDynacastManager(params DynacastManagerParams) *DynacastManager {
 		maxSubscribedQuality:          make(map[string]livekit.VideoQuality),
 		committedMaxSubscribedQuality: make(map[string]livekit.VideoQuality),
 		maxSubscribedQualityDebounce:  debounce.New(params.DynacastPauseDelay),
-		qualityNotifyOpQueue:          utils.NewOpsQueue(params.Logger, "quality-notify", 100),
+		qualityNotifyOpQueue:          utils.NewOpsQueue("quality-notify", 0, true),
 	}
 	d.qualityNotifyOpQueue.Start()
 	return d

pkg/sfu/streamallocator/streamallocator.go

@@ -31,6 +31,7 @@ import (
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/sfu"
 	"github.com/livekit/livekit-server/pkg/sfu/buffer"
+	"github.com/livekit/livekit-server/pkg/utils"
 )
 
 const (
@@ -165,8 +166,7 @@ type StreamAllocator struct {
 
 	state streamAllocatorState
 
-	eventChMu sync.RWMutex
-	eventCh   chan Event
+	eventsQueue *utils.OpsQueue
 
 	isStopped atomic.Bool
 }
@@ -180,7 +180,7 @@ func NewStreamAllocator(params StreamAllocatorParams) *StreamAllocator {
 		}),
 		rateMonitor: NewRateMonitor(),
 		videoTracks: make(map[livekit.TrackID]*Track),
-		eventCh:     make(chan Event, 1000),
+		eventsQueue: utils.NewOpsQueue("stream-allocator", 64, true),
 	}
 
 	s.probeController = NewProbeController(ProbeControllerParams{
@@ -197,19 +197,18 @@ func NewStreamAllocator(params StreamAllocatorParams) *StreamAllocator {
 }
 
 func (s *StreamAllocator) Start() {
-	go s.processEvents()
+	s.eventsQueue.Start()
 	go s.ping()
 }
 
 func (s *StreamAllocator) Stop() {
-	s.eventChMu.Lock()
 	if s.isStopped.Swap(true) {
-		s.eventChMu.Unlock()
 		return
 	}
 
-	close(s.eventCh)
-	s.eventChMu.Unlock()
+	// wait for eventsQueue to be done
+	<-s.eventsQueue.Stop()
+	s.probeController.StopProbe()
 }
 
 func (s *StreamAllocator) OnStreamStateChange(f func(update *StreamStateUpdate) error) {
@@ -546,30 +545,9 @@ func (s *StreamAllocator) maybePostEventAllocateTrack(downTrack *sfu.DownTrack)
 }
 
 func (s *StreamAllocator) postEvent(event Event) {
-	s.eventChMu.RLock()
-	if s.isStopped.Load() {
-		s.eventChMu.RUnlock()
-		return
-	}
-
-	select {
-	case s.eventCh <- event:
-	default:
-		s.params.Logger.Warnw("stream allocator: event queue full", nil, "event", event.String())
-	}
-	s.eventChMu.RUnlock()
-}
-
-func (s *StreamAllocator) processEvents() {
-	for event := range s.eventCh {
-		if s.isStopped.Load() {
-			break
-		}
-
+	s.eventsQueue.Enqueue(func() {
 		s.handleEvent(&event)
-	}
-
-	s.probeController.StopProbe()
+	})
 }
 
 func (s *StreamAllocator) ping() {

pkg/telemetry/telemetryservice.go

@@ -20,6 +20,7 @@ import (
 	"time"
 
 	"github.com/livekit/livekit-server/pkg/config"
+	"github.com/livekit/livekit-server/pkg/utils"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 	"github.com/livekit/protocol/webhook"
@@ -82,15 +83,15 @@ type TelemetryService interface {
 }
 
 const (
-	workerCleanupWait  = 3 * time.Minute
-	jobQueueBufferSize = 10000
+	workerCleanupWait = 3 * time.Minute
+	jobsQueueMinSize  = 2048
 )
 
 type telemetryService struct {
 	AnalyticsService
 
-	notifier webhook.QueuedNotifier
-	jobsChan chan func()
+	notifier  webhook.QueuedNotifier
+	jobsQueue *utils.OpsQueue
 
 	lock    sync.RWMutex
 	workers map[livekit.ParticipantID]*StatsWorker
@@ -100,11 +101,12 @@ func NewTelemetryService(notifier webhook.QueuedNotifier, analytics AnalyticsSer
 	t := &telemetryService{
 		AnalyticsService: analytics,
 
-		notifier: notifier,
-		jobsChan: make(chan func(), jobQueueBufferSize),
-		workers:  make(map[livekit.ParticipantID]*StatsWorker),
+		notifier:  notifier,
+		jobsQueue: utils.NewOpsQueue("telemetry", jobsQueueMinSize, true),
+		workers:   make(map[livekit.ParticipantID]*StatsWorker),
 	}
 
+	t.jobsQueue.Start()
 	go t.run()
 
 	return t
@@ -132,19 +134,12 @@ func (t *telemetryService) run() {
 			t.FlushStats()
 		case <-cleanupTicker.C:
 			t.cleanupWorkers()
-		case op := <-t.jobsChan:
-			op()
 		}
 	}
 }
 
 func (t *telemetryService) enqueue(op func()) {
-	select {
-	case t.jobsChan <- op:
-	// success
-	default:
-		logger.Warnw("telemetry queue full", nil)
-	}
+	t.jobsQueue.Enqueue(op)
 }
 
 func (t *telemetryService) getWorker(participantID livekit.ParticipantID) (worker *StatsWorker, ok bool) {

pkg/utils/opsqueue.go

@@ -15,33 +15,34 @@
 package utils
 
 import (
+	"math/bits"
 	"sync"
 
-	"github.com/livekit/protocol/logger"
+	"github.com/gammazero/deque"
+	"github.com/livekit/protocol/utils"
 )
 
 type OpsQueue struct {
-	logger logger.Logger
-	name   string
-	size   int
+	name        string
+	flushOnStop bool
 
-	lock      sync.RWMutex
-	ops       chan func()
+	lock      sync.Mutex
+	ops       deque.Deque[func()]
+	wake      chan struct{}
 	isStarted bool
+	doneChan  chan struct{}
 	isStopped bool
 }
 
-func NewOpsQueue(logger logger.Logger, name string, size int) *OpsQueue {
-	return &OpsQueue{
-		logger: logger,
-		name:   name,
-		size:   size,
-		ops:    make(chan func(), size),
+func NewOpsQueue(name string, minSize uint, flushOnStop bool) *OpsQueue {
+	oq := &OpsQueue{
+		name:        name,
+		flushOnStop: flushOnStop,
+		wake:        make(chan struct{}, 1),
+		doneChan:    make(chan struct{}),
 	}
-}
-
-func (oq *OpsQueue) SetLogger(logger logger.Logger) {
-	oq.logger = logger
+	oq.ops.SetMinCapacity(uint(utils.Min(bits.Len64(uint64(minSize-1)), 16)))
+	return oq
 }
 
 func (oq *OpsQueue) Start() {
@@ -57,42 +58,52 @@ func (oq *OpsQueue) Start() {
 	go oq.process()
 }
 
-func (oq *OpsQueue) Stop() {
+func (oq *OpsQueue) Stop() <-chan struct{} {
 	oq.lock.Lock()
 	if oq.isStopped {
 		oq.lock.Unlock()
-		return
+		return oq.doneChan
 	}
 
 	oq.isStopped = true
-	close(oq.ops)
+	close(oq.wake)
 	oq.lock.Unlock()
-}
-
-func (oq *OpsQueue) IsStarted() bool {
-	oq.lock.RLock()
-	defer oq.lock.RUnlock()
-
-	return oq.isStarted
+	return oq.doneChan
 }
 
 func (oq *OpsQueue) Enqueue(op func()) {
-	oq.lock.RLock()
-	if oq.isStopped {
-		oq.lock.RUnlock()
-		return
-	}
+	oq.lock.Lock()
+	defer oq.lock.Unlock()
 
-	select {
-	case oq.ops <- op:
-	default:
-		oq.logger.Errorw("ops queue full", nil, "name", oq.name, "size", oq.size)
+	oq.ops.PushBack(op)
+	if oq.ops.Len() == 1 && !oq.isStopped {
+		select {
+		case oq.wake <- struct{}{}:
+		default:
+		}
 	}
-	oq.lock.RUnlock()
 }
 
 func (oq *OpsQueue) process() {
-	for op := range oq.ops {
-		op()
+	defer close(oq.doneChan)
+
+	for {
+		<-oq.wake
+		for {
+			oq.lock.Lock()
+			if oq.isStopped && (!oq.flushOnStop || oq.ops.Len() == 0) {
+				oq.lock.Unlock()
+				return
+			}
+
+			if oq.ops.Len() == 0 {
+				oq.lock.Unlock()
+				break
+			}
+			op := oq.ops.PopFront()
+			oq.lock.Unlock()
+
+			op()
+		}
 	}
 }