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

@@ -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()

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,
 	}

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()

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
 	}
 

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,
 	}

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))
 	}

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() {