From 017ed1333842316935abe945a06e40288d8cce7a Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 12 Jan 2022 23:19:56 +0530 Subject: [PATCH] Prevent multiple resume notifications on track (#334) * Prevent multiple resume notifications on track When returning previous allocation as is, reset change to none. Ideally, would like to have change as a separate return and not in last allocation. But, also prefer to have last allocation state. For now, resetting change. * log bandwidth estimate only on decrease --- pkg/sfu/forwarder.go | 10 ++++++++++ pkg/sfu/forwarder_test.go | 2 +- pkg/sfu/streamallocator.go | 5 ++++- 3 files changed, 15 insertions(+), 2 deletions(-) diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 39538acb1..e2d283384 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -406,6 +406,10 @@ func (f *Forwarder) Allocate(availableChannelCapacity int64, allowPause bool, br // when pause is disallowed, pick the lowest available targetLayers.spatial = int32(math.Min(float64(f.maxLayers.spatial), float64(f.availableLayers[0]))) targetLayers.temporal = int32(math.Min(0, float64(f.maxLayers.temporal))) + + if f.targetLayers == InvalidLayers { + change = VideoStreamingChangeResuming + } } else { // disable forwarding as it is not known how big this stream is // and if it will fit in the available channel capacity @@ -817,6 +821,7 @@ func (f *Forwarder) FinalizeAllocate(brs Bitrates) VideoAllocation { defer f.lock.Unlock() if f.lastAllocation.state != VideoAllocationStateAwaitingMeasurement { + f.lastAllocation.change = VideoStreamingChangeNone return f.lastAllocation } @@ -886,22 +891,26 @@ func (f *Forwarder) AllocateNextHigher(brs Bitrates) (VideoAllocation, bool) { defer f.lock.Unlock() if f.kind == webrtc.RTPCodecTypeAudio { + f.lastAllocation.change = VideoStreamingChangeNone return f.lastAllocation, false } // if not deficient, nothing to do if f.lastAllocation.state != VideoAllocationStateDeficient { + f.lastAllocation.change = VideoStreamingChangeNone return f.lastAllocation, false } // if targets are still pending, don't increase if f.targetLayers != InvalidLayers && f.targetLayers != f.currentLayers { + f.lastAllocation.change = VideoStreamingChangeNone return f.lastAllocation, false } optimalBandwidthNeeded := f.getOptimalBandwidthNeeded(brs) if optimalBandwidthNeeded == 0 { // either feed is dry or awaiting measurement, don't hunt for higher + f.lastAllocation.change = VideoStreamingChangeNone return f.lastAllocation, false } @@ -968,6 +977,7 @@ func (f *Forwarder) AllocateNextHigher(brs Bitrates) (VideoAllocation, bool) { } } + f.lastAllocation.change = VideoStreamingChangeNone return f.lastAllocation, false } diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 5d497bcd7..1b5751376 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -227,7 +227,7 @@ func TestForwarderAllocate(t *testing.T) { } expectedResult = VideoAllocation{ state: VideoAllocationStateAwaitingMeasurement, - change: VideoStreamingChangeNone, + change: VideoStreamingChangeResuming, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: []uint16{0}, diff --git a/pkg/sfu/streamallocator.go b/pkg/sfu/streamallocator.go index ff4c03c53..58a79a265 100644 --- a/pkg/sfu/streamallocator.go +++ b/pkg/sfu/streamallocator.go @@ -687,10 +687,13 @@ func (s *StreamAllocator) maybeCommitEstimate() (isDecreasing bool) { s.lastEstimateDecreaseTime = time.Now() isDecreasing = true } + + if s.committedChannelCapacity > s.receivedEstimate { + s.params.Logger.Debugw("committing channel capacity(bps)", "from", s.committedChannelCapacity, "to", s.receivedEstimate) + } s.committedChannelCapacity = s.receivedEstimate s.lastCommitTime = time.Now() - s.params.Logger.Debugw("committing channel capacity", "capacity(bps)", s.committedChannelCapacity) return }