diff --git a/pkg/sfu/bwe/bwe.go b/pkg/sfu/bwe/bwe.go index a47f05bf4..f6857e649 100644 --- a/pkg/sfu/bwe/bwe.go +++ b/pkg/sfu/bwe/bwe.go @@ -89,6 +89,7 @@ type BWE interface { ProbeDuration() time.Duration ProbeClusterStarting(pci ccutils.ProbeClusterInfo) ProbeClusterDone(pci ccutils.ProbeClusterInfo) + ProbeClusterIsGoalReached() bool ProbeClusterFinalize() (ccutils.ProbeSignal, int64, bool) } diff --git a/pkg/sfu/bwe/null_bwe.go b/pkg/sfu/bwe/null_bwe.go index e0242edcf..140635cda 100644 --- a/pkg/sfu/bwe/null_bwe.go +++ b/pkg/sfu/bwe/null_bwe.go @@ -62,6 +62,10 @@ func (n *NullBWE) ProbeClusterStarting(_pci ccutils.ProbeClusterInfo) {} func (n *NullBWE) ProbeClusterDone(_pci ccutils.ProbeClusterInfo) {} +func (n *NullBWE) ProbeClusterIsGoalReached() bool { + return false +} + func (n *NullBWE) ProbeClusterFinalize() (ccutils.ProbeSignal, int64, bool) { return ccutils.ProbeSignalInconclusive, 0, false } diff --git a/pkg/sfu/bwe/remotebwe/probe_controller.go b/pkg/sfu/bwe/remotebwe/probe_controller.go index 00f8c6f1c..50a59979c 100644 --- a/pkg/sfu/bwe/remotebwe/probe_controller.go +++ b/pkg/sfu/bwe/remotebwe/probe_controller.go @@ -62,8 +62,8 @@ var ( ProbeRegulator: ccutils.DefaultProbeRegulatorConfig, SettleWaitNumRTT: 5, - SettleWaitMin: 500 * time.Millisecond, - SettleWaitMax: 10 * time.Second, + SettleWaitMin: 250 * time.Millisecond, + SettleWaitMax: 5 * time.Second, } ) @@ -140,6 +140,10 @@ func (p *probeController) ProbeClusterDone(pci ccutils.ProbeClusterInfo) { p.setState(probeControllerStateHangover) } +func (p *probeController) ProbeClusterInfo() ccutils.ProbeClusterInfo { + return p.pci +} + func (p *probeController) MaybeFinalizeProbe() (ccutils.ProbeClusterInfo, bool) { if p.state != probeControllerStateHangover { return ccutils.ProbeClusterInfoInvalid, false diff --git a/pkg/sfu/bwe/remotebwe/remote_bwe.go b/pkg/sfu/bwe/remotebwe/remote_bwe.go index 2b78d3edb..bce129710 100644 --- a/pkg/sfu/bwe/remotebwe/remote_bwe.go +++ b/pkg/sfu/bwe/remotebwe/remote_bwe.go @@ -158,11 +158,6 @@ func (r *RemoteBWE) congestionDetectionStateMachine() (bool, bwe.CongestionState newState := r.congestionState update := false trend, reason := r.channelObserver.GetTrend() - if trend == channelTrendCongesting && r.congestionState == bwe.CongestionStateNone { - r.params.Logger.Debugw("remote bwe, channel congesting", "channel", r.channelObserver) - } else if trend == channelTrendClearing && r.congestionState != bwe.CongestionStateNone { - r.params.Logger.Debugw("remote bwe, channel congestion relieving", "channel", r.channelObserver) - } switch r.congestionState { case bwe.CongestionStateNone: @@ -285,6 +280,19 @@ func (r *RemoteBWE) ProbeClusterDone(pci ccutils.ProbeClusterInfo) { r.probeController.ProbeClusterDone(pci) } +func (r *RemoteBWE) ProbeClusterIsGoalReached() bool { + r.lock.Lock() + defer r.lock.Unlock() + + if !r.probeController.IsInProbe() || + r.congestionState != bwe.CongestionStateNone || + !r.channelObserver.HasEnoughEstimateSamples() { + return false + } + + return r.channelObserver.GetHighestEstimate() > int64(r.probeController.ProbeClusterInfo().Goal.DesiredBps) +} + func (r *RemoteBWE) ProbeClusterFinalize() (ccutils.ProbeSignal, int64, bool) { r.lock.Lock() defer r.lock.Unlock() diff --git a/pkg/sfu/bwe/sendsidebwe/probe_packet_group.go b/pkg/sfu/bwe/sendsidebwe/probe_packet_group.go index f47b6fad6..2d4201bcf 100644 --- a/pkg/sfu/bwe/sendsidebwe/probe_packet_group.go +++ b/pkg/sfu/bwe/sendsidebwe/probe_packet_group.go @@ -42,8 +42,8 @@ var ( }, SettleWaitNumRTT: 5, - SettleWaitMin: 500 * time.Millisecond, - SettleWaitMax: 10 * time.Second, + SettleWaitMin: 250 * time.Millisecond, + SettleWaitMax: 5 * time.Second, } ) diff --git a/pkg/sfu/ccutils/prober.go b/pkg/sfu/ccutils/prober.go index 210a4b0cc..b6c92662e 100644 --- a/pkg/sfu/ccutils/prober.go +++ b/pkg/sfu/ccutils/prober.go @@ -578,7 +578,6 @@ func (c *Cluster) MarshalLogObject(e zapcore.ObjectEncoder) error { e.AddObject("info", c.info) e.AddDuration("baseSleepDuration", c.baseSleepDuration) e.AddInt("numBuckets", len(c.buckets)) - e.AddArray("buckets", logger.ObjectSlice(c.buckets)) e.AddInt("bucketIdx", c.bucketIdx) e.AddInt("probeBytesSent", c.probeBytesSent) e.AddTime("startTime", c.startTime) diff --git a/pkg/sfu/streamallocator/streamallocator.go b/pkg/sfu/streamallocator/streamallocator.go index 72b750adf..470c98cb4 100644 --- a/pkg/sfu/streamallocator/streamallocator.go +++ b/pkg/sfu/streamallocator/streamallocator.go @@ -202,9 +202,11 @@ type StreamAllocator struct { isAllocateAllPending bool rembTrackingSSRC uint32 - state streamAllocatorState - isHolding bool - activeProbeClusterId ccutils.ProbeClusterId + state streamAllocatorState + isHolding bool + activeProbeClusterId ccutils.ProbeClusterId + activeProbeGoalReached bool + activeProbeCongesting bool eventsQueue *utils.TypedOpsQueue[Event] @@ -679,12 +681,21 @@ func (s *StreamAllocator) handleSignalPeriodicPing(Event) { s.params.BWE.Reset() } - // finalize any probe that may have finished/aborted if s.activeProbeClusterId != ccutils.ProbeClusterIdInvalid { + if !s.activeProbeCongesting && !s.activeProbeGoalReached && s.params.BWE.ProbeClusterIsGoalReached() { + s.params.Logger.Debugw( + "stream allocator: probe goal reached", + "activeProbeClusterId", s.activeProbeClusterId, + ) + s.activeProbeGoalReached = true + s.maybeStopProbe() + } + + // finalize any probe that may have finished/aborted if probeSignal, channelCapacity, isFinalized := s.params.BWE.ProbeClusterFinalize(); isFinalized { s.params.Logger.Debugw( "stream allocator: probe result", - "probeClusterId", s.activeProbeClusterId, + "activeProbeClusterId", s.activeProbeClusterId, "probeSignal", probeSignal, "channelCapacity", channelCapacity, ) @@ -720,6 +731,8 @@ func (s *StreamAllocator) handleSignalPeriodicPing(Event) { func (s *StreamAllocator) handleSignalProbeClusterSwitch(event Event) { pci := event.Data.(ccutils.ProbeClusterInfo) s.activeProbeClusterId = pci.Id + s.activeProbeGoalReached = false + s.activeProbeCongesting = false s.params.BWE.ProbeClusterStarting(pci) @@ -817,12 +830,15 @@ func (s *StreamAllocator) handleSignalCongestionStateChange(event Event) { if cscd.congestionState == bwe.CongestionStateCongested { if s.activeProbeClusterId != ccutils.ProbeClusterIdInvalid { - s.params.Logger.Infow( - "stream allocator: channel congestion detected, not updating channel capacity in active probe", - "old(bps)", s.committedChannelCapacity, - "new(bps)", cscd.estimatedAvailableChannelCapacity, - "expectedUsage(bps)", s.getExpectedBandwidthUsage(), - ) + if !s.activeProbeCongesting { + s.activeProbeCongesting = true + s.params.Logger.Infow( + "stream allocator: channel congestion detected, not updating channel capacity in active probe", + "old(bps)", s.committedChannelCapacity, + "new(bps)", cscd.estimatedAvailableChannelCapacity, + "expectedUsage(bps)", s.getExpectedBandwidthUsage(), + ) + } } else { s.params.Logger.Infow( "stream allocator: channel congestion detected, updating channel capacity", diff --git a/pkg/sfu/streamallocator/track.go b/pkg/sfu/streamallocator/track.go index 1c28e9b8d..237faf528 100644 --- a/pkg/sfu/streamallocator/track.go +++ b/pkg/sfu/streamallocator/track.go @@ -135,7 +135,7 @@ func (t *Track) WritePaddingRTP(bytesToSend int) int { } func (t *Track) WriteProbePackets(bytesToSend int) int { - return t.downTrack.WriteProbePackets(bytesToSend, true) + return t.downTrack.WriteProbePackets(bytesToSend, false) } func (t *Track) AllocateOptimal(allowOvershoot bool, hold bool) sfu.VideoAllocation {