mirror of
https://github.com/livekit/livekit.git
synced 2026-10-22 15:22:30 +00:00
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
This commit is contained in:
1 parent
9e327b1f3c
commit
ea1a467191
5 files changed
+139
-41
No files matched your search
@@ -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
|
||||
}
|
||||
|
||||
|
||||
+35
-11
@@ -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
|
||||
}
|
||||
|
||||
|
||||
+46
-30
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user