From 710e06b7da4c2907b6fb09d65349b370fb93b55b Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Fri, 13 May 2022 10:18:18 +0530 Subject: [PATCH] Tweak screen share stream tracker. (#683) * Tweak screen share stream tracker. Screen share uses extremely low frame rate when content is static. This leads to stream tracker changes and layer switches. Use a longer measurement interval for screen share tracks. Setting it to 4 seconds. The screen share frame rate is approx. 1 fps and with two temporal layers, it should get a frame every 2 seconds or so in every layer. But, that does not work. Even three second measurement window was reporting 0 rate. So, it is getting clumped. 4 seconds works fine. Also modifying the bit rate availability change to trigger on a temporal layer becoming available or going away. With these changes, on a local test, not seeing any unnecessary PLIs from client and hence no unnecessary key frames from publisher. * Fix test --- pkg/sfu/streamtracker.go | 26 ++++++++----------- pkg/sfu/streamtracker_test.go | 9 ++++--- pkg/sfu/streamtrackermanager.go | 45 +++++++++++++++++++-------------- 3 files changed, 41 insertions(+), 39 deletions(-) diff --git a/pkg/sfu/streamtracker.go b/pkg/sfu/streamtracker.go index 41db4bcbc..356743d4c 100644 --- a/pkg/sfu/streamtracker.go +++ b/pkg/sfu/streamtracker.go @@ -27,10 +27,6 @@ const ( StreamStatusActive StreamStatus = 1 ) -const ( - bitrateReportInterval = 1 * time.Second -) - type StreamTrackerParams struct { // number of samples needed per cycle SamplesRequired uint32 @@ -40,6 +36,8 @@ type StreamTrackerParams struct { CycleDuration time.Duration + BitrateReportInterval time.Duration + Logger logger.Logger } @@ -235,7 +233,7 @@ func (s *StreamTracker) detectWorker(generation uint32) { ticker := time.NewTicker(s.params.CycleDuration) defer ticker.Stop() - tickerBitrate := time.NewTicker(bitrateReportInterval / 2) + tickerBitrate := time.NewTicker(s.params.BitrateReportInterval) defer tickerBitrate.Stop() for { @@ -282,26 +280,22 @@ func (s *StreamTracker) detectChanges() { func (s *StreamTracker) bitrateReport() { // run this even if paused to drain out bitrate if there are no packets coming in s.lock.Lock() - if time.Since(s.lastBitrateReport) < bitrateReportInterval { - s.lock.Unlock() - return - } - now := time.Now() diff := now.Sub(s.lastBitrateReport) s.lastBitrateReport = now - aggLast := int64(0) - aggNow := int64(0) + bitrateAvailabilityChanged := false for i := 0; i < len(s.bytesForBitrate); i++ { - aggLast += s.bitrate[i] - s.bitrate[i] = int64(float64(s.bytesForBitrate[i]*8) / diff.Seconds()) - aggNow += s.bitrate[i] + bitrate := int64(float64(s.bytesForBitrate[i]*8) / diff.Seconds()) + if (s.bitrate[i] == 0 && bitrate > 0) || (s.bitrate[i] > 0 && bitrate == 0) { + bitrateAvailabilityChanged = true + } + s.bitrate[i] = bitrate s.bytesForBitrate[i] = 0 } s.lock.Unlock() - if aggLast == 0 && aggNow > 0 && s.onBitrateAvailable != nil { + if bitrateAvailabilityChanged && s.onBitrateAvailable != nil { s.onBitrateAvailable() } } diff --git a/pkg/sfu/streamtracker_test.go b/pkg/sfu/streamtracker_test.go index 6d9b2c8b0..4b6f546a8 100644 --- a/pkg/sfu/streamtracker_test.go +++ b/pkg/sfu/streamtracker_test.go @@ -14,10 +14,11 @@ import ( func newStreamTracker(samplesRequired uint32, cyclesRequired uint32, cycleDuration time.Duration) *StreamTracker { return NewStreamTracker(StreamTrackerParams{ - SamplesRequired: samplesRequired, - CyclesRequired: cyclesRequired, - CycleDuration: cycleDuration, - Logger: logger.GetDefaultLogger(), + SamplesRequired: samplesRequired, + CyclesRequired: cyclesRequired, + CycleDuration: cycleDuration, + BitrateReportInterval: 1 * time.Second, + Logger: logger.GetDefaultLogger(), }) } diff --git a/pkg/sfu/streamtrackermanager.go b/pkg/sfu/streamtrackermanager.go index 995133895..0a54fb339 100644 --- a/pkg/sfu/streamtrackermanager.go +++ b/pkg/sfu/streamtrackermanager.go @@ -5,6 +5,7 @@ import ( "sync" "time" + "github.com/go-logr/logr" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" ) @@ -16,38 +17,44 @@ var ( var ( ConfigVideo = []StreamTrackerParams{ { - SamplesRequired: 1, - CyclesRequired: 4, - CycleDuration: 500 * time.Millisecond, + SamplesRequired: 1, + CyclesRequired: 4, + CycleDuration: 500 * time.Millisecond, + BitrateReportInterval: 1 * time.Second, }, { - SamplesRequired: 5, - CyclesRequired: 60, - CycleDuration: 500 * time.Millisecond, + SamplesRequired: 5, + CyclesRequired: 60, + CycleDuration: 500 * time.Millisecond, + BitrateReportInterval: 1 * time.Second, }, { - SamplesRequired: 5, - CyclesRequired: 60, - CycleDuration: 500 * time.Millisecond, + SamplesRequired: 5, + CyclesRequired: 60, + CycleDuration: 500 * time.Millisecond, + BitrateReportInterval: 1 * time.Second, }, } // be very forgiving for screen share to account for cases like static screen where there could be only one packet per second ConfigScreenshare = []StreamTrackerParams{ { - SamplesRequired: 1, - CyclesRequired: 1, - CycleDuration: 2 * time.Second, + SamplesRequired: 1, + CyclesRequired: 1, + CycleDuration: 2 * time.Second, + BitrateReportInterval: 4 * time.Second, }, { - SamplesRequired: 1, - CyclesRequired: 1, - CycleDuration: 2 * time.Second, + SamplesRequired: 1, + CyclesRequired: 1, + CycleDuration: 2 * time.Second, + BitrateReportInterval: 4 * time.Second, }, { - SamplesRequired: 1, - CyclesRequired: 1, - CycleDuration: 2 * time.Second, + SamplesRequired: 1, + CyclesRequired: 1, + CycleDuration: 2 * time.Second, + BitrateReportInterval: 4 * time.Second, }, } ) @@ -98,7 +105,7 @@ func (s *StreamTrackerManager) AddTracker(layer int32) *StreamTracker { params = ConfigVideo[layer] } - params.Logger = s.logger + params.Logger = logger.Logger(logr.Logger(s.logger).WithValues("layer", layer)) tracker := NewStreamTracker(params) tracker.OnStatusChanged(func(status StreamStatus) { if status == StreamStatusStopped {