diff --git a/pkg/sfu/bwe/remotebwe/probe_controller.go b/pkg/sfu/bwe/remotebwe/probe_controller.go index 50a59979c..52a9ff940 100644 --- a/pkg/sfu/bwe/remotebwe/probe_controller.go +++ b/pkg/sfu/bwe/remotebwe/probe_controller.go @@ -140,8 +140,12 @@ func (p *probeController) ProbeClusterDone(pci ccutils.ProbeClusterInfo) { p.setState(probeControllerStateHangover) } -func (p *probeController) ProbeClusterInfo() ccutils.ProbeClusterInfo { - return p.pci +func (p *probeController) ProbeClusterIsGoalReached(estimate int64) bool { + if p.pci.Id == ccutils.ProbeClusterIdInvalid { + return false + } + + return estimate > int64(p.pci.Goal.DesiredBps) } func (p *probeController) MaybeFinalizeProbe() (ccutils.ProbeClusterInfo, bool) { diff --git a/pkg/sfu/bwe/remotebwe/remote_bwe.go b/pkg/sfu/bwe/remotebwe/remote_bwe.go index bce129710..9d19866fa 100644 --- a/pkg/sfu/bwe/remotebwe/remote_bwe.go +++ b/pkg/sfu/bwe/remotebwe/remote_bwe.go @@ -290,7 +290,7 @@ func (r *RemoteBWE) ProbeClusterIsGoalReached() bool { return false } - return r.channelObserver.GetHighestEstimate() > int64(r.probeController.ProbeClusterInfo().Goal.DesiredBps) + return r.probeController.ProbeClusterIsGoalReached(r.channelObserver.GetHighestEstimate()) } func (r *RemoteBWE) ProbeClusterFinalize() (ccutils.ProbeSignal, int64, bool) { diff --git a/pkg/sfu/streamallocator/streamallocator.go b/pkg/sfu/streamallocator/streamallocator.go index 470c98cb4..062f13b9a 100644 --- a/pkg/sfu/streamallocator/streamallocator.go +++ b/pkg/sfu/streamallocator/streamallocator.go @@ -678,6 +678,7 @@ func (s *StreamAllocator) handleSignalPeriodicPing(Event) { // reset BWE if that persists for a while if s.state == streamAllocatorStateDeficient && s.params.Pacer.TimeSinceLastSentPacket() > s.params.Config.PausedMinWait { s.params.Logger.Infow("stream allocator: resetting bwe to enable probing") + s.maybeStopProbe() s.params.BWE.Reset() } @@ -1041,6 +1042,11 @@ func (s *StreamAllocator) maybeStopProbe() { } pci := s.params.Pacer.EndProbeCluster(s.activeProbeClusterId) + + for _, t := range s.getTracks() { + t.DownTrack().SwapProbeClusterId(pci.Id, ccutils.ProbeClusterIdInvalid) + } + s.params.BWE.ProbeClusterDone(pci) s.prober.Reset(pci) }