From 11bdda07792ebd84bd1e5f1e93eacf28b4b552be Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Mon, 31 Jul 2023 12:45:21 +0530 Subject: [PATCH] Temper stream allocator more (#1920) * Temper stream allocator more * gofmt * AllowPause default --- pkg/config/config.go | 14 ++++++++++---- pkg/sfu/streamallocator/channelobserver.go | 2 ++ pkg/sfu/streamallocator/trenddetector.go | 8 +++++--- 3 files changed, 17 insertions(+), 7 deletions(-) diff --git a/pkg/config/config.go b/pkg/config/config.go index 6b285da12..c280423e4 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -139,7 +139,9 @@ type CongestionControlProbeConfig struct { type CongestionControlChannelObserverConfig struct { EstimateRequiredSamples int `yaml:"estimate_required_samples,omitempty"` + EstimateRequiredSamplesMin int `yaml:"estimate_required_samples_min,omitempty"` EstimateDownwardTrendThreshold float64 `yaml:"estimate_downward_trend_threshold,omitempty"` + EstimateDownwardTrendMaxWait time.Duration `yaml:"estimate_downward_trend_max_wait,omitempty"` EstimateCollapseThreshold time.Duration `yaml:"estimate_collapse_threshold,omitempty"` EstimateValidityWindow time.Duration `yaml:"estimate_validity_window,omitempty"` NackWindowMinDuration time.Duration `yaml:"nack_window_min_duration,omitempty"` @@ -331,7 +333,9 @@ var DefaultConfig = Config{ }, ChannelObserverProbeConfig: CongestionControlChannelObserverConfig{ EstimateRequiredSamples: 3, + EstimateRequiredSamplesMin: 3, EstimateDownwardTrendThreshold: 0.0, + EstimateDownwardTrendMaxWait: 5 * time.Second, EstimateCollapseThreshold: 0, EstimateValidityWindow: 10 * time.Second, NackWindowMinDuration: 500 * time.Millisecond, @@ -339,12 +343,14 @@ var DefaultConfig = Config{ NackRatioThreshold: 0.04, }, ChannelObserverNonProbeConfig: CongestionControlChannelObserverConfig{ - EstimateRequiredSamples: 8, - EstimateDownwardTrendThreshold: -0.5, + EstimateRequiredSamples: 12, + EstimateRequiredSamplesMin: 8, + EstimateDownwardTrendThreshold: -0.6, + EstimateDownwardTrendMaxWait: 5 * time.Second, EstimateCollapseThreshold: 500 * time.Millisecond, EstimateValidityWindow: 10 * time.Second, - NackWindowMinDuration: 1 * time.Second, - NackWindowMaxDuration: 2 * time.Second, + NackWindowMinDuration: 2 * time.Second, + NackWindowMaxDuration: 3 * time.Second, NackRatioThreshold: 0.08, }, }, diff --git a/pkg/sfu/streamallocator/channelobserver.go b/pkg/sfu/streamallocator/channelobserver.go index e8c432dc8..9afab9c49 100644 --- a/pkg/sfu/streamallocator/channelobserver.go +++ b/pkg/sfu/streamallocator/channelobserver.go @@ -90,7 +90,9 @@ func NewChannelObserver(params ChannelObserverParams, logger logger.Logger) *Cha Name: params.Name + "-estimate", Logger: logger, RequiredSamples: params.Config.EstimateRequiredSamples, + RequiredSamplesMin: params.Config.EstimateRequiredSamplesMin, DownwardTrendThreshold: params.Config.EstimateDownwardTrendThreshold, + DownwardTrendMaxWait: params.Config.EstimateDownwardTrendMaxWait, CollapseThreshold: params.Config.EstimateCollapseThreshold, ValidityWindow: params.Config.EstimateValidityWindow, }), diff --git a/pkg/sfu/streamallocator/trenddetector.go b/pkg/sfu/streamallocator/trenddetector.go index c54d29efa..019716f27 100644 --- a/pkg/sfu/streamallocator/trenddetector.go +++ b/pkg/sfu/streamallocator/trenddetector.go @@ -57,7 +57,9 @@ type TrendDetectorParams struct { Name string Logger logger.Logger RequiredSamples int + RequiredSamplesMin int DownwardTrendThreshold float64 + DownwardTrendMaxWait time.Duration CollapseThreshold time.Duration ValidityWindow time.Duration } @@ -203,7 +205,7 @@ func (t *TrendDetector) prune() { } func (t *TrendDetector) updateDirection() { - if len(t.samples) < t.params.RequiredSamples { + if len(t.samples) < t.params.RequiredSamplesMin { t.direction = TrendDirectionNeutral return } @@ -213,9 +215,9 @@ func (t *TrendDetector) updateDirection() { t.direction = TrendDirectionNeutral switch { - case kt > 0: + case kt > 0 && len(t.samples) >= t.params.RequiredSamples: t.direction = TrendDirectionUpward - case kt < t.params.DownwardTrendThreshold: + case kt < t.params.DownwardTrendThreshold && (len(t.samples) >= t.params.RequiredSamples || t.samples[len(t.samples)-1].at.Sub(t.samples[0].at) > t.params.DownwardTrendMaxWait): t.direction = TrendDirectionDownward } }