From 7c5a558a4815d0700db857064973634ee9c59b49 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Fri, 6 Dec 2024 10:52:24 +0530 Subject: [PATCH] Try up-allocation on neutral trend. (#3235) * Try up-allocation on neutral trend. Some probes end up with neutral trend due to getting much estimates of same value. It is okay to try up-allocating in those cases. Otherwise, the stream allocator some times gets stuck and does not up-allocate at all as all probes end up neutral. Changing the name of the signal to `NotCongesting` to signify it is either neutral or clearing. * wait 5 RTT for probe to finalize * trend detector object encoder --- pkg/sfu/bwe/remotebwe/channel_observer.go | 3 +- pkg/sfu/bwe/remotebwe/probe_controller.go | 2 +- pkg/sfu/bwe/remotebwe/remote_bwe.go | 4 +- .../bwe/sendsidebwe/congestion_detector.go | 4 +- pkg/sfu/bwe/sendsidebwe/probe_packet_group.go | 3 +- pkg/sfu/ccutils/probe_signal.go | 6 +-- pkg/sfu/ccutils/trenddetector.go | 53 ++++++++++++------- 7 files changed, 45 insertions(+), 30 deletions(-) diff --git a/pkg/sfu/bwe/remotebwe/channel_observer.go b/pkg/sfu/bwe/remotebwe/channel_observer.go index 584abae20..52b8cc1a8 100644 --- a/pkg/sfu/bwe/remotebwe/channel_observer.go +++ b/pkg/sfu/bwe/remotebwe/channel_observer.go @@ -48,6 +48,7 @@ func (c channelTrend) String() string { } // ------------------------------------------------ + type channelCongestionReason int const ( @@ -195,7 +196,7 @@ func (c *channelObserver) MarshalLogObject(e zapcore.ObjectEncoder) error { } e.AddString("name", c.params.Name) - e.AddString("estimate", c.estimateTrend.String()) + e.AddObject("estimate", c.estimateTrend) e.AddObject("nack", c.nackTracker) channelTrend, channelCongestionReason := c.GetTrend() diff --git a/pkg/sfu/bwe/remotebwe/probe_controller.go b/pkg/sfu/bwe/remotebwe/probe_controller.go index 0bf94fa8e..00f8c6f1c 100644 --- a/pkg/sfu/bwe/remotebwe/probe_controller.go +++ b/pkg/sfu/bwe/remotebwe/probe_controller.go @@ -61,7 +61,7 @@ var ( DefaultProbeControllerConfig = ProbeControllerConfig{ ProbeRegulator: ccutils.DefaultProbeRegulatorConfig, - SettleWaitNumRTT: 10, + SettleWaitNumRTT: 5, SettleWaitMin: 500 * time.Millisecond, SettleWaitMax: 10 * time.Second, } diff --git a/pkg/sfu/bwe/remotebwe/remote_bwe.go b/pkg/sfu/bwe/remotebwe/remote_bwe.go index 156338714..05ab3d781 100644 --- a/pkg/sfu/bwe/remotebwe/remote_bwe.go +++ b/pkg/sfu/bwe/remotebwe/remote_bwe.go @@ -320,10 +320,10 @@ func (r *RemoteBWE) ProbeClusterFinalize() (ccutils.ProbeSignal, int64, bool) { "probeClusterInfo", pci, ) - probeSignal := ccutils.ProbeSignalClearing + probeSignal := ccutils.ProbeSignalNotCongesting if probeCongestionState != bwe.CongestionStateNone { probeSignal = ccutils.ProbeSignalCongesting - } else if trend, _ := pco.GetTrend(); !pco.HasEnoughEstimateSamples() || trend == channelTrendNeutral { + } else if !pco.HasEnoughEstimateSamples() { probeSignal = ccutils.ProbeSignalInconclusive } else { highestEstimate := pco.GetHighestEstimate() diff --git a/pkg/sfu/bwe/sendsidebwe/congestion_detector.go b/pkg/sfu/bwe/sendsidebwe/congestion_detector.go index da01845ac..4d2e41a05 100644 --- a/pkg/sfu/bwe/sendsidebwe/congestion_detector.go +++ b/pkg/sfu/bwe/sendsidebwe/congestion_detector.go @@ -94,7 +94,7 @@ func (p ProbeSignalConfig) ProbeSignal(ppg *probePacketGroup) (ccutils.ProbeSign } if pqd < p.DQRMaxDelay.Microseconds() { - return ccutils.ProbeSignalClearing, ts.AcknowledgedBitrate() + return ccutils.ProbeSignalNotCongesting, ts.AcknowledgedBitrate() } return ccutils.ProbeSignalInconclusive, ts.AcknowledgedBitrate() @@ -521,7 +521,7 @@ func (c *congestionDetector) ProbeClusterFinalize() (ccutils.ProbeSignal, int64, ) probeSignal, estimatedAvailableChannelCapacity := c.params.Config.ProbeSignal.ProbeSignal(c.probePacketGroup) - if probeSignal == ccutils.ProbeSignalClearing && estimatedAvailableChannelCapacity > c.estimatedAvailableChannelCapacity { + if probeSignal == ccutils.ProbeSignalNotCongesting && estimatedAvailableChannelCapacity > c.estimatedAvailableChannelCapacity { c.estimatedAvailableChannelCapacity = estimatedAvailableChannelCapacity } diff --git a/pkg/sfu/bwe/sendsidebwe/probe_packet_group.go b/pkg/sfu/bwe/sendsidebwe/probe_packet_group.go index 22dd2d096..17efc88b6 100644 --- a/pkg/sfu/bwe/sendsidebwe/probe_packet_group.go +++ b/pkg/sfu/bwe/sendsidebwe/probe_packet_group.go @@ -40,7 +40,8 @@ var ( MinPackets: 16384, MaxWindowDuration: time.Minute, }, - SettleWaitNumRTT: 10, + + SettleWaitNumRTT: 5, SettleWaitMin: 500 * time.Millisecond, SettleWaitMax: 10 * time.Second, } diff --git a/pkg/sfu/ccutils/probe_signal.go b/pkg/sfu/ccutils/probe_signal.go index c71b7616e..22c3bf82e 100644 --- a/pkg/sfu/ccutils/probe_signal.go +++ b/pkg/sfu/ccutils/probe_signal.go @@ -23,7 +23,7 @@ type ProbeSignal int const ( ProbeSignalInconclusive ProbeSignal = iota ProbeSignalCongesting - ProbeSignalClearing + ProbeSignalNotCongesting ) func (p ProbeSignal) String() string { @@ -32,8 +32,8 @@ func (p ProbeSignal) String() string { return "INCONCLUSIVE" case ProbeSignalCongesting: return "CONGESTING" - case ProbeSignalClearing: - return "CLEARING" + case ProbeSignalNotCongesting: + return "NOT_CONGESTING" default: return fmt.Sprintf("%d", int(p)) } diff --git a/pkg/sfu/ccutils/trenddetector.go b/pkg/sfu/ccutils/trenddetector.go index 3f76fcb97..f6855a233 100644 --- a/pkg/sfu/ccutils/trenddetector.go +++ b/pkg/sfu/ccutils/trenddetector.go @@ -19,6 +19,7 @@ import ( "time" "github.com/livekit/protocol/logger" + "go.uber.org/zap/zapcore" ) // ------------------------------------------------ @@ -57,6 +58,17 @@ type trendDetectorSample[T trendDetectorNumber] struct { at time.Time } +type trendDetectorSampleElapsed[T trendDetectorNumber] struct { + value T + sinceFirst time.Duration +} + +func (t trendDetectorSampleElapsed[T]) MarshalLogObject(e zapcore.ObjectEncoder) error { + e.AddFloat64("value", float64(t.value)) + e.AddDuration("sinceFirst", t.sinceFirst) + return nil +} + // ------------------------------------------------ type TrendDetectorConfig struct { @@ -154,28 +166,29 @@ func (t *TrendDetector[T]) HasEnoughSamples() bool { return t.numSamples >= t.params.Config.RequiredSamples } -func (t *TrendDetector[T]) String() string { - samplesStr := "" - if len(t.samples) > 0 { - firstTime := t.samples[0].at - samplesStr += "[" - for i, sample := range t.samples { - suffix := ", " - if i == len(t.samples)-1 { - suffix = "" - } - samplesStr += fmt.Sprintf("%v(%d)%s", sample.value, sample.at.Sub(firstTime).Milliseconds(), suffix) - } - samplesStr += "]" +func (t *TrendDetector[T]) MarshalLogObject(e zapcore.ObjectEncoder) error { + if t == nil { + return nil } - now := time.Now() - elapsed := now.Sub(t.startTime).Seconds() - return fmt.Sprintf("n: %s, t: %+v|%+v|%.2fs, v: %d|%v|%v|%s|%.2f", - t.params.Name, - t.startTime.Format(time.UnixDate), now.Format(time.UnixDate), elapsed, - t.numSamples, t.lowestValue, t.highestValue, samplesStr, t.kendallsTau(), - ) + var samples []trendDetectorSampleElapsed[T] + if len(t.samples) > 0 { + firstTime := t.samples[0].at + for _, sample := range t.samples { + samples = append(samples, trendDetectorSampleElapsed[T]{sample.value, sample.at.Sub(firstTime)}) + } + } + + e.AddString("name", t.params.Name) + e.AddTime("startTime", t.startTime) + e.AddDuration("elapsed", time.Since(t.startTime)) + e.AddInt("numSamples", t.numSamples) + e.AddArray("samples", logger.ObjectSlice(samples)) + e.AddFloat64("lowestValue", float64(t.lowestValue)) + e.AddFloat64("highestValue", float64(t.highestValue)) + e.AddFloat64("kendallsTau", t.kendallsTau()) + e.AddString("direction", t.direction.String()) + return nil } func (t *TrendDetector[T]) prune() {