From ea1a46719175f594718e36ad248d3dd416236295 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sun, 5 Mar 2023 14:10:19 +0530 Subject: [PATCH] Bitrate based quality tracking for DownTrack (#1491) * Make connection quality not too optimistic. With score normalization, the quality indicator showed good under conditions which should have normally showed some badness. So, a few things in this PR - Do not normalize scores - Pick the weakest link as the representative score (moving away from averaging) - For down track direction, when reporting delta stats, take the number of packets sent actually. If there are holes in the feed (upstream packet loss), down tracks should not be penalised for that loss. State of things in connection quality feature - Audio uses rtcscore-go (with a change to accommodate RED codec). This follows the E-model. - Camera uses rtcscore-go. No change here. NOTE: THe rtscore here is purely based on bits per pixel per frame (bpf). This has the following existing issues (no change, these were already there) o Does not take packet loss, jitter, rtt into account o Expected frame rate is not available. So, measured frame rate is used as expected frame rate also. If expected frame rate were available, the score could be reduced for lower frame rates. - Screen share tracks: No change. This uses the very old simple loss based thresholding for scoring. As the bit rate varies a lot based on content and rtcscore video algorithm used for camera relies on bits per pixel per frame, this could produce a very low value (large width/height encoded in a small number of bits because of static content) and hence a low score. So, the old loss based thresholding is used. * clean up * update rtcscore pointer * fix tests * log lines reformat * WIP commit * WIP commit * update mute of receiver * WIP commit * WIP commit * start adding tests * take min score if quality matches * start adding bytes based scoring * clean up * more clean up * Use Fuse * log quality drop * Periodically report bitrate to down track. For connection quality based on bitrate for down tracks, the measured rate should be used. That is to ensure that down track quality measurement does not get affected by publisher side changes negatively (or positively). Report the optimal bit rate to connection quality scorer every second so that scorer has a continuously updating picture of the stream and can compare the actual bit rate against expected optimal bitrate more reliably. Doing it at time like allocation, the bitrate may not be accurate (or may not even be available). So, a periodic update is necessary. * add transition at allocation times * clean up debug log * - Use number of windows for wait to make things simpler - track no layer expected case - always update transition - always call updateScore --- pkg/sfu/downtrack.go | 17 ++++++++ pkg/sfu/forwarder.go | 46 +++++++++++++++----- pkg/sfu/forwarder_test.go | 76 ++++++++++++++++++++------------- pkg/sfu/receiver.go | 8 ++++ pkg/sfu/streamtrackermanager.go | 33 ++++++++++++++ 5 files changed, 139 insertions(+), 41 deletions(-) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index a2901f90d..8c8a7a39c 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -29,6 +29,7 @@ type TrackSender interface { UpTrackLayersChange() UpTrackBitrateAvailabilityChange() UpTrackMaxPublishedLayerChange(maxPublishedLayer int32) + UpTrackBitrateReport(availableLayers []int32, bitrates Bitrates) WriteRTP(p *buffer.ExtPacket, layer int32) error Close() IsClosed() bool @@ -876,6 +877,18 @@ func (d *DownTrack) UpTrackMaxPublishedLayerChange(maxPublishedLayer int32) { } } +func (d *DownTrack) maybeAddTransition(bitrate int64) { + if d.kind == webrtc.RTPCodecTypeAudio { + return + } + + d.connectionStats.AddTransition(bitrate, time.Now()) +} + +func (d *DownTrack) UpTrackBitrateReport(_availableLayers []int32, bitrates Bitrates) { + d.maybeAddTransition(d.forwarder.GetOptimalBandwidthNeeded(bitrates)) +} + // OnCloseHandler method to be called on remote tracked removed func (d *DownTrack) OnCloseHandler(fn func(willBeResumed bool)) { d.onCloseHandler = fn @@ -964,6 +977,7 @@ func (d *DownTrack) AllocateOptimal(allowOvershoot bool) VideoAllocation { al, brs := d.receiver.GetLayeredBitrate() allocation := d.forwarder.AllocateOptimal(al, brs, allowOvershoot) d.maybeStartKeyFrameRequester() + d.maybeAddTransition(allocation.bandwidthNeeded) return allocation } @@ -991,6 +1005,7 @@ func (d *DownTrack) ProvisionalAllocateGetBestWeightedTransition() VideoTransiti func (d *DownTrack) ProvisionalAllocateCommit() VideoAllocation { allocation := d.forwarder.ProvisionalAllocateCommit() d.maybeStartKeyFrameRequester() + d.maybeAddTransition(allocation.bandwidthNeeded) return allocation } @@ -998,6 +1013,7 @@ func (d *DownTrack) AllocateNextHigher(availableChannelCapacity int64, allowOver _, brs := d.receiver.GetLayeredBitrate() allocation, available := d.forwarder.AllocateNextHigher(availableChannelCapacity, brs, allowOvershoot) d.maybeStartKeyFrameRequester() + d.maybeAddTransition(allocation.bandwidthNeeded) return allocation, available } @@ -1012,6 +1028,7 @@ func (d *DownTrack) Pause() VideoAllocation { _, brs := d.receiver.GetLayeredBitrate() allocation := d.forwarder.Pause(brs) d.maybeStartKeyFrameRequester() + d.maybeAddTransition(allocation.bandwidthNeeded) return allocation } diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 115b9f04d..768cc9353 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -59,6 +59,7 @@ type VideoAllocation struct { isDeficient bool bandwidthRequested int64 bandwidthDelta int64 + bandwidthNeeded int64 bitrates Bitrates targetLayers VideoLayers requestLayerSpatial int32 @@ -67,8 +68,18 @@ type VideoAllocation struct { } func (v VideoAllocation) String() string { - return fmt.Sprintf("VideoAllocation{pause: %s, def: %+v, bw: %d, del: %d, rates: %+v, target: %s, req: %d, max: %s, dist: %d}", - v.pauseReason, v.isDeficient, v.bandwidthRequested, v.bandwidthDelta, v.bitrates, v.targetLayers, v.requestLayerSpatial, v.maxLayers, v.distanceToDesired) + return fmt.Sprintf("VideoAllocation{pause: %s, def: %+v, bwr: %d, del: %d, bwn: %d, rates: %+v, target: %s, req: %d, max: %s, dist: %d}", + v.pauseReason, + v.isDeficient, + v.bandwidthRequested, + v.bandwidthDelta, + v.bandwidthNeeded, + v.bitrates, + v.targetLayers, + v.requestLayerSpatial, + v.maxLayers, + v.distanceToDesired, + ) } var ( @@ -478,6 +489,13 @@ func (f *Forwarder) DistanceToDesired() int32 { return f.lastAllocation.distanceToDesired } +func (f *Forwarder) GetOptimalBandwidthNeeded(brs Bitrates) int64 { + f.lock.RLock() + defer f.lock.RUnlock() + + return getOptimalBandwidthNeeded(f.muted, f.pubMuted, f.maxPublishedLayer, brs, f.maxLayers) +} + func (f *Forwarder) AllocateOptimal(availableLayers []int32, brs Bitrates, allowOvershoot bool) VideoAllocation { f.lock.Lock() defer f.lock.Unlock() @@ -497,6 +515,7 @@ func (f *Forwarder) AllocateOptimal(availableLayers []int32, brs Bitrates, allow if optimalBandwidthNeeded == 0 { alloc.pauseReason = VideoPauseReasonFeedDry } + alloc.bandwidthNeeded = optimalBandwidthNeeded opportunisticAlloc := func() { // opportunistically latch on to anything @@ -890,10 +909,18 @@ func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation { f.lock.Lock() defer f.lock.Unlock() + optimalBandwidthNeeded := getOptimalBandwidthNeeded( + f.provisional.muted, + f.provisional.pubMuted, + f.provisional.maxPublishedLayer, + f.provisional.bitrates, + f.provisional.maxLayers, + ) alloc := VideoAllocation{ bandwidthRequested: 0, bandwidthDelta: -f.lastAllocation.bandwidthRequested, bitrates: f.provisional.bitrates, + bandwidthNeeded: optimalBandwidthNeeded, targetLayers: f.provisional.allocatedLayers, requestLayerSpatial: f.provisional.allocatedLayers.Spatial, maxLayers: f.provisional.maxLayers, @@ -914,13 +941,7 @@ func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation { case f.provisional.pubMuted: alloc.pauseReason = VideoPauseReasonPubMuted - case getOptimalBandwidthNeeded( - f.provisional.muted, - f.provisional.pubMuted, - f.provisional.maxPublishedLayer, - f.provisional.bitrates, - f.provisional.maxLayers, - ) == 0: + case optimalBandwidthNeeded == 0: if f.provisional.allocatedLayers.IsValid() { // overshoot alloc.bandwidthRequested = f.provisional.bitrates[f.provisional.allocatedLayers.Spatial][f.provisional.allocatedLayers.Temporal] @@ -1010,6 +1031,7 @@ func (f *Forwarder) AllocateNextHigher(availableChannelCapacity int64, brs Bitra isDeficient: true, bandwidthRequested: bandwidthRequested, bandwidthDelta: bandwidthRequested - alreadyAllocated, + bandwidthNeeded: optimalBandwidthNeeded, bitrates: brs, targetLayers: targetLayers, requestLayerSpatial: targetLayers.Spatial, @@ -1152,10 +1174,12 @@ func (f *Forwarder) Pause(brs Bitrates) VideoAllocation { f.lock.Lock() defer f.lock.Unlock() + optimalBandwidthNeeded := getOptimalBandwidthNeeded(f.muted, f.pubMuted, f.maxPublishedLayer, brs, f.maxLayers) alloc := VideoAllocation{ bandwidthRequested: 0, bandwidthDelta: 0 - f.lastAllocation.bandwidthRequested, bitrates: brs, + bandwidthNeeded: optimalBandwidthNeeded, targetLayers: InvalidLayers, requestLayerSpatial: InvalidLayerSpatial, maxLayers: f.maxLayers, @@ -1169,7 +1193,7 @@ func (f *Forwarder) Pause(brs Bitrates) VideoAllocation { case f.pubMuted: alloc.pauseReason = VideoPauseReasonPubMuted - case getOptimalBandwidthNeeded(f.muted, f.pubMuted, f.maxPublishedLayer, brs, f.maxLayers) == 0: + case optimalBandwidthNeeded == 0: alloc.pauseReason = VideoPauseReasonFeedDry default: @@ -1660,7 +1684,7 @@ func getDistanceToDesired( targetLayers VideoLayers, maxLayers VideoLayers, ) int32 { - if muted || pubMuted || maxPublishedLayer == InvalidLayerSpatial { + if muted || pubMuted || maxPublishedLayer == InvalidLayerSpatial || !maxLayers.IsValid() { return 0 } diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 82c4a4b08..0ba5241dc 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -218,41 +218,43 @@ func TestForwarderAllocateOptimal(t *testing.T) { // when max layers changes, target is opportunistic, but requested spatial layer should be at max f.maxLayers = VideoLayers{Spatial: 1, Temporal: 3} expectedResult = VideoAllocation{ - pauseReason: VideoPauseReasonFeedDry, - bandwidthRequested: 0, - bandwidthDelta: 0, - bitrates: emptyBitrates, + pauseReason: VideoPauseReasonNone, + bandwidthRequested: bitrates[1][3], + bandwidthDelta: bitrates[1][3], + bandwidthNeeded: bitrates[1][3], + bitrates: bitrates, targetLayers: DefaultMaxLayers, requestLayerSpatial: f.maxLayers.Spatial, maxLayers: f.maxLayers, - distanceToDesired: 0, + distanceToDesired: -1, } - result = f.AllocateOptimal(nil, emptyBitrates, true) + result = f.AllocateOptimal(nil, bitrates, true) require.Equal(t, expectedResult, result) require.Equal(t, expectedResult, f.lastAllocation) require.Equal(t, DefaultMaxLayers, f.TargetLayers()) // reset max layers for rest of the tests below f.maxLayers = DefaultMaxLayers - f.lastAllocation.maxLayers = DefaultMaxLayers // when feed is dry and current is not valid, should set up for opportunistic forwarding + // NOTE: feed is dry due to availableLayers = nil, some valid bitrates may be passed in here for testing purposes only disable(f) expectedTargetLayers := VideoLayers{ Spatial: 2, Temporal: DefaultMaxLayerTemporal, } expectedResult = VideoAllocation{ - pauseReason: VideoPauseReasonFeedDry, - bandwidthRequested: 0, - bandwidthDelta: 0, - bitrates: emptyBitrates, + pauseReason: VideoPauseReasonNone, + bandwidthRequested: bitrates[2][1], + bandwidthDelta: bitrates[2][1] - bitrates[1][3], + bandwidthNeeded: bitrates[2][1], + bitrates: bitrates, targetLayers: expectedTargetLayers, requestLayerSpatial: expectedTargetLayers.Spatial, maxLayers: DefaultMaxLayers, distanceToDesired: 0, } - result = f.AllocateOptimal(nil, emptyBitrates, true) + result = f.AllocateOptimal(nil, bitrates, true) require.Equal(t, expectedResult, result) require.Equal(t, expectedResult, f.lastAllocation) require.Equal(t, expectedTargetLayers, f.TargetLayers()) @@ -268,7 +270,7 @@ func TestForwarderAllocateOptimal(t *testing.T) { expectedResult = VideoAllocation{ pauseReason: VideoPauseReasonFeedDry, bandwidthRequested: 0, - bandwidthDelta: 0, + bandwidthDelta: 0 - bitrates[2][1], bitrates: emptyBitrates, targetLayers: expectedTargetLayers, requestLayerSpatial: expectedTargetLayers.Spatial, @@ -284,16 +286,17 @@ func TestForwarderAllocateOptimal(t *testing.T) { // opportunistic target if feed is not dry and current is not valid, i. e. not forwarding expectedResult = VideoAllocation{ - pauseReason: VideoPauseReasonFeedDry, - bandwidthRequested: 0, - bandwidthDelta: 0, - bitrates: emptyBitrates, + pauseReason: VideoPauseReasonNone, + bandwidthRequested: bitrates[2][1], + bandwidthDelta: bitrates[2][1], + bandwidthNeeded: bitrates[2][1], + bitrates: bitrates, targetLayers: DefaultMaxLayers, requestLayerSpatial: 2, maxLayers: DefaultMaxLayers, distanceToDesired: 0, } - result = f.AllocateOptimal([]int32{0, 1}, emptyBitrates, true) + result = f.AllocateOptimal([]int32{0, 1}, bitrates, true) require.Equal(t, expectedResult, result) require.Equal(t, expectedResult, f.lastAllocation) require.Equal(t, DefaultMaxLayers, f.TargetLayers()) @@ -303,7 +306,7 @@ func TestForwarderAllocateOptimal(t *testing.T) { expectedResult = VideoAllocation{ pauseReason: VideoPauseReasonFeedDry, bandwidthRequested: 0, - bandwidthDelta: 0, + bandwidthDelta: 0 - bitrates[2][1], bitrates: emptyBitrates, targetLayers: DefaultMaxLayers, requestLayerSpatial: 2, @@ -320,16 +323,17 @@ func TestForwarderAllocateOptimal(t *testing.T) { Temporal: DefaultMaxLayerTemporal, } expectedResult = VideoAllocation{ - pauseReason: VideoPauseReasonFeedDry, - bandwidthRequested: 0, - bandwidthDelta: 0, - bitrates: emptyBitrates, + pauseReason: VideoPauseReasonNone, + bandwidthRequested: bitrates[2][1], + bandwidthDelta: bitrates[2][1], + bandwidthNeeded: bitrates[2][1], + bitrates: bitrates, targetLayers: expectedTargetLayers, requestLayerSpatial: 2, maxLayers: DefaultMaxLayers, distanceToDesired: 0, } - result = f.AllocateOptimal([]int32{0, 1}, emptyBitrates, true) + result = f.AllocateOptimal([]int32{0, 1}, bitrates, true) require.Equal(t, expectedResult, result) require.Equal(t, expectedResult, f.lastAllocation) @@ -340,16 +344,17 @@ func TestForwarderAllocateOptimal(t *testing.T) { Temporal: DefaultMaxLayerTemporal, } expectedResult = VideoAllocation{ - pauseReason: VideoPauseReasonFeedDry, - bandwidthRequested: 0, + pauseReason: VideoPauseReasonNone, + bandwidthRequested: bitrates[2][1], bandwidthDelta: 0, - bitrates: emptyBitrates, + bandwidthNeeded: bitrates[2][1], + bitrates: bitrates, targetLayers: expectedTargetLayers, requestLayerSpatial: 1, maxLayers: DefaultMaxLayers, - distanceToDesired: 0, + distanceToDesired: 1, } - result = f.AllocateOptimal([]int32{1}, emptyBitrates, true) + result = f.AllocateOptimal([]int32{1}, bitrates, true) require.Equal(t, expectedResult, result) require.Equal(t, expectedResult, f.lastAllocation) @@ -364,7 +369,7 @@ func TestForwarderAllocateOptimal(t *testing.T) { expectedResult = VideoAllocation{ pauseReason: VideoPauseReasonFeedDry, bandwidthRequested: 0, - bandwidthDelta: 0, + bandwidthDelta: 0 - bitrates[2][1], bitrates: emptyBitrates, targetLayers: expectedTargetLayers, requestLayerSpatial: 0, @@ -437,6 +442,7 @@ func TestForwarderProvisionalAllocate(t *testing.T) { isDeficient: true, bandwidthRequested: bitrates[1][2], bandwidthDelta: bitrates[1][2], + bandwidthNeeded: bitrates[2][3], bitrates: bitrates, targetLayers: expectedTargetLayers, requestLayerSpatial: expectedTargetLayers.Spatial, @@ -463,6 +469,7 @@ func TestForwarderProvisionalAllocate(t *testing.T) { isDeficient: true, bandwidthRequested: bitrates[0][0], bandwidthDelta: bitrates[0][0] - bitrates[1][2], + bandwidthNeeded: bitrates[2][3], bitrates: bitrates, targetLayers: expectedTargetLayers, requestLayerSpatial: expectedTargetLayers.Spatial, @@ -666,6 +673,7 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { isDeficient: true, bandwidthRequested: 1, bandwidthDelta: 1, + bandwidthNeeded: bitrates[2][1], bitrates: bitrates, targetLayers: expectedLayers, requestLayerSpatial: expectedLayers.Spatial, @@ -695,6 +703,7 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { bandwidthRequested: 10, bandwidthDelta: 0, bitrates: bitrates, + bandwidthNeeded: bitrates[2][1], targetLayers: expectedLayers, requestLayerSpatial: expectedLayers.Spatial, maxLayers: DefaultMaxLayers, @@ -904,6 +913,7 @@ func TestForwarderAllocateNextHigher(t *testing.T) { isDeficient: true, bandwidthRequested: 3, bandwidthDelta: 1, + bandwidthNeeded: bitrates[2][1], bitrates: bitrates, targetLayers: expectedTargetLayers, requestLayerSpatial: expectedTargetLayers.Spatial, @@ -931,6 +941,7 @@ func TestForwarderAllocateNextHigher(t *testing.T) { isDeficient: true, bandwidthRequested: 4, bandwidthDelta: 1, + bandwidthNeeded: bitrates[2][1], bitrates: bitrates, targetLayers: expectedTargetLayers, requestLayerSpatial: expectedTargetLayers.Spatial, @@ -954,6 +965,7 @@ func TestForwarderAllocateNextHigher(t *testing.T) { isDeficient: true, bandwidthRequested: 5, bandwidthDelta: 1, + bandwidthNeeded: bitrates[2][1], bitrates: bitrates, targetLayers: expectedTargetLayers, requestLayerSpatial: expectedTargetLayers.Spatial, @@ -976,6 +988,7 @@ func TestForwarderAllocateNextHigher(t *testing.T) { bandwidthRequested: 7, bandwidthDelta: 2, bitrates: bitrates, + bandwidthNeeded: bitrates[2][1], targetLayers: expectedTargetLayers, requestLayerSpatial: expectedTargetLayers.Spatial, maxLayers: DefaultMaxLayers, @@ -1009,6 +1022,7 @@ func TestForwarderAllocateNextHigher(t *testing.T) { isDeficient: true, bandwidthRequested: 2, bandwidthDelta: 2, + bandwidthNeeded: bitrates[2][1], bitrates: bitrates, targetLayers: expectedTargetLayers, requestLayerSpatial: expectedTargetLayers.Spatial, @@ -1026,6 +1040,7 @@ func TestForwarderAllocateNextHigher(t *testing.T) { isDeficient: true, bandwidthRequested: 2, bandwidthDelta: 2, + bandwidthNeeded: bitrates[2][1], bitrates: bitrates, targetLayers: expectedTargetLayers, requestLayerSpatial: expectedTargetLayers.Spatial, @@ -1096,6 +1111,7 @@ func TestForwarderPause(t *testing.T) { isDeficient: true, bandwidthRequested: 0, bandwidthDelta: 0 - bitrates[0][0], + bandwidthNeeded: bitrates[2][3], bitrates: bitrates, targetLayers: InvalidLayers, requestLayerSpatial: InvalidLayerSpatial, diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 05bd9ebf4..e09ebee45 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -195,6 +195,7 @@ func NewWebRTCReceiver( w.streamTrackerManager.OnAvailableLayersChanged(w.downTrackLayerChange) w.streamTrackerManager.OnBitrateAvailabilityChanged(w.downTrackBitrateAvailabilityChange) w.streamTrackerManager.OnMaxPublishedLayerChanged(w.downTrackMaxPublishedLayerChange) + w.streamTrackerManager.OnBitrateReport(w.downTrackBitrateReport) for _, opt := range opts { w = opt(w) @@ -427,6 +428,12 @@ func (w *WebRTCReceiver) downTrackMaxPublishedLayerChange(maxPublishedLayer int3 w.notifyMaxExpectedLayer(maxPublishedLayer) } +func (w *WebRTCReceiver) downTrackBitrateReport(availableLayers []int32, bitrates Bitrates) { + for _, dt := range w.downTrackSpreader.GetDownTracks() { + dt.UpTrackBitrateReport(availableLayers, bitrates) + } +} + func (w *WebRTCReceiver) GetLayeredBitrate() ([]int32, Bitrates) { return w.streamTrackerManager.GetLayeredBitrate() } @@ -634,6 +641,7 @@ func (w *WebRTCReceiver) forwardRTP(layer int32) { // closeTracks close all tracks from Receiver func (w *WebRTCReceiver) closeTracks() { w.connectionStats.Close() + w.streamTrackerManager.Close() for _, dt := range w.downTrackSpreader.ResetAndGetDownTracks() { dt.Close() diff --git a/pkg/sfu/streamtrackermanager.go b/pkg/sfu/streamtrackermanager.go index dd15e88e3..bda1c0705 100644 --- a/pkg/sfu/streamtrackermanager.go +++ b/pkg/sfu/streamtrackermanager.go @@ -4,7 +4,9 @@ import ( "fmt" "sort" "sync" + "time" + "github.com/frostbyte73/core" "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/livekit-server/pkg/sfu/streamtracker" @@ -32,10 +34,13 @@ type StreamTrackerManager struct { senderReportMu sync.RWMutex senderReports [DefaultMaxLayerSpatial + 1]*buffer.RTCPSenderReportDataExt + closed core.Fuse + onAvailableLayersChanged func() onBitrateAvailabilityChanged func() onMaxPublishedLayerChanged func(maxPublishedLayer int32) onMaxAvailableLayerChanged func(maxAvailableLayer int32) + onBitrateReport func(availableLayers []int32, bitrates Bitrates) } func NewStreamTrackerManager( @@ -51,6 +56,7 @@ func NewStreamTrackerManager( isSVC: isSVC, maxPublishedLayer: InvalidLayerSpatial, clockRate: clockRate, + closed: core.NewFuse(), } switch s.trackInfo.Source { @@ -63,9 +69,15 @@ func NewStreamTrackerManager( } s.maxExpectedLayerFromTrackInfo() + + go s.bitrateReporter() return s } +func (s *StreamTrackerManager) Close() { + s.closed.Break() +} + func (s *StreamTrackerManager) OnAvailableLayersChanged(f func()) { s.onAvailableLayersChanged = f } @@ -82,6 +94,10 @@ func (s *StreamTrackerManager) OnMaxLayerChanged(f func(maxAvailableLayer int32) s.onMaxAvailableLayerChanged = f } +func (s *StreamTrackerManager) OnBitrateReport(f func(availableLayers []int32, bitrates Bitrates)) { + s.onBitrateReport = f +} + func (s *StreamTrackerManager) createStreamTrackerPacket(layer int32) streamtracker.StreamTrackerImpl { packetTrackerConfig, ok := s.trackerConfig.PacketTracker[layer] if !ok { @@ -467,3 +483,20 @@ func (s *StreamTrackerManager) GetReferenceLayerRTPTimestamp(ts uint32, layer in // the reference layer. return ts + (srRef.SenderReportData.RTPTimestamp - normalizedTS), nil } + +func (s *StreamTrackerManager) bitrateReporter() { + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + + for { + select { + case <-s.closed.Watch(): + return + + case <-ticker.C: + if s.onBitrateReport != nil { + s.onBitrateReport(s.GetLayeredBitrate()) + } + } + } +}