diff --git a/pkg/rtc/dynacastmanager.go b/pkg/rtc/dynacastmanager.go index 8943646fe..1163cb09a 100644 --- a/pkg/rtc/dynacastmanager.go +++ b/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 diff --git a/pkg/sfu/streamallocator/streamallocator.go b/pkg/sfu/streamallocator/streamallocator.go index 364dd0895..6370692b0 100644 --- a/pkg/sfu/streamallocator/streamallocator.go +++ b/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() { diff --git a/pkg/telemetry/telemetryservice.go b/pkg/telemetry/telemetryservice.go index 572bba118..fe5e329c2 100644 --- a/pkg/telemetry/telemetryservice.go +++ b/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) { diff --git a/pkg/utils/opsqueue.go b/pkg/utils/opsqueue.go index 3e992461c..01f9a12ff 100644 --- a/pkg/utils/opsqueue.go +++ b/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() + } } }