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()) + } + } + } +}