StreamAllocator - tracking more things (#1652)

* WIP commit * Add a probe cluster mode * better variable naming * fix units * WIP commit * WIP commit * WIP commit * new file * WIP commit * Maintain history of a few things * correct signal * fix typo * WIP commmit * gofmt * rate not sum * adjust edges of rate monitor * fmt * remove debug
2026-08-28 00:44:12 +00:00 · 2023-04-25 07:31:22 +05:30
parent b4ea4de5c0
commit 8ed193b231
9 changed files with 625 additions and 57 deletions
@@ -18,7 +18,7 @@ require (
 	github.com/jxskiss/base62 v1.1.0
 	github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1
 	github.com/livekit/mediatransportutil v0.0.0-20230326055817-ed569ca13d26
-	github.com/livekit/protocol v1.5.5
+	github.com/livekit/protocol v1.5.6-0.20230424073901-c54f5f7f4182
 	github.com/livekit/psrpc v0.3.1-0.20230424064451-65c6a2dd048b
 	github.com/mackerelio/go-osstat v0.2.4
 	github.com/magefile/mage v1.14.0
@@ -35,7 +35,7 @@ require (
 	github.com/pion/stun v0.4.0
 	github.com/pion/transport/v2 v2.2.0
 	github.com/pion/turn/v2 v2.1.0
-	github.com/pion/webrtc/v3 v3.1.60
+	github.com/pion/webrtc/v3 v3.1.61
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.15.0
 	github.com/redis/go-redis/v9 v9.0.3
@@ -121,8 +121,8 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD
 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
 github.com/livekit/mediatransportutil v0.0.0-20230326055817-ed569ca13d26 h1:QlQFyMwCDgjyySsrgmrMcVbEBA6KZcyTzvK+z346tUA=
 github.com/livekit/mediatransportutil v0.0.0-20230326055817-ed569ca13d26/go.mod h1:eDA41kiySZoG+wy4Etsjb3w0jjLx69i/vAmSjG4bteA=
-github.com/livekit/protocol v1.5.5 h1:vuSU3TI/w58WnAWnyC59nMzY/JE+ZznU6W/iRgWw4JQ=
-github.com/livekit/protocol v1.5.5/go.mod h1:iZ289+6H5xn/9kP2iqpRvVWxuc8GXBMqN0qI7LdN9HI=
+github.com/livekit/protocol v1.5.6-0.20230424073901-c54f5f7f4182 h1:rpaYN8Jy5F1ZhxH4Q61+6Cc42I/evI63SlSXUMZ+VdU=
+github.com/livekit/protocol v1.5.6-0.20230424073901-c54f5f7f4182/go.mod h1:B7Ns8diIKB3y39oRHm7ZluU9ZGCxCWQT+uKcbY3MCG4=
 github.com/livekit/psrpc v0.3.1-0.20230424064451-65c6a2dd048b h1:WqRJoXeycdrhEuA9C0jJ7aN6i5wx8B3c+M0hIKnnFZE=
 github.com/livekit/psrpc v0.3.1-0.20230424064451-65c6a2dd048b/go.mod h1:n6JntEg+zT6Ji8InoyTpV7wusPNwGqqtxmHlkNhDN0U=
 github.com/mackerelio/go-osstat v0.2.4 h1:qxGbdPkFo65PXOb/F/nhDKpF2nGmGaCFDLXoZjJTtUs=
@@ -214,8 +214,8 @@ github.com/pion/turn/v2 v2.1.0 h1:5wGHSgGhJhP/RpabkUb/T9PdsAjkGLS6toYz5HNzoSI=
 github.com/pion/turn/v2 v2.1.0/go.mod h1:yrT5XbXSGX1VFSF31A3c1kCNB5bBZgk/uu5LET162qs=
 github.com/pion/udp/v2 v2.0.1 h1:xP0z6WNux1zWEjhC7onRA3EwwSliXqu1ElUZAQhUP54=
 github.com/pion/udp/v2 v2.0.1/go.mod h1:B7uvTMP00lzWdyMr/1PVZXtV3wpPIxBRd4Wl6AksXn8=
-github.com/pion/webrtc/v3 v3.1.60 h1:FLF6HT3x3CMHtPz5JbdAARfIUpMZu2YeOSzkVxaeF+k=
-github.com/pion/webrtc/v3 v3.1.60/go.mod h1:65gfOgxrmszb6ec7kEiZp32QwnmDNIrJK8hgo/0niWY=
+github.com/pion/webrtc/v3 v3.1.61 h1:WG6p786t7jxXO/3miw6HmAQmO3p/n+QLRa2xLaovcr8=
+github.com/pion/webrtc/v3 v3.1.61/go.mod h1:uk/4AJmgEUpSExaP7aexCyODwfbHap8hAnQzRV7zKcE=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -119,6 +119,12 @@ func (d DownTrackState) String() string {

 // -------------------------------------------------------------------

+type NackInfo struct {
+	Timestamp      uint32
+	SequenceNumber uint16
+	Attempts       uint8
+}
+
 type DownTrackStreamAllocatorListener interface {
 	// RTCP received
 	OnREMB(dt *DownTrack, remb *rtcp.ReceiverEstimatedMaximumBitrate)
@@ -147,6 +153,12 @@ type DownTrackStreamAllocatorListener interface {

 	// packet(s) sent
 	OnPacketsSent(dt *DownTrack, size int)
+
+	// NACKs received
+	OnNACK(dt *DownTrack, nackInfos []NackInfo)
+
+	// RTCP Receiver Report received
+	OnRTCPReceiverReport(dt *DownTrack, rr rtcp.ReceptionReport)
 }

 type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport)
@@ -198,8 +210,7 @@ type DownTrack struct {

 	rtpStats *buffer.RTPStats

-	statsLock          sync.RWMutex
-	totalRepeatedNACKs uint32
+	totalRepeatedNACKs atomic.Uint32

 	keyFrameRequestGeneration atomic.Uint32

@@ -219,6 +230,8 @@ type DownTrack struct {
 	streamAllocatorListener         DownTrackStreamAllocatorListener
 	streamAllocatorReportGeneration int
 	streamAllocatorBytesCounter     atomic.Uint32
+	bytesSent                       atomic.Uint32
+	bytesRetransmitted              atomic.Uint32

 	// update stats
 	onStatsUpdate func(dt *DownTrack, stat *livekit.AnalyticsStat)
@@ -593,7 +606,9 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
 		return err
 	}

+	// STREAM-ALLOCATOR-TODO: remove this stream allocator bytes counter once stream allocator changes fully to pull bytes counter
 	d.streamAllocatorBytesCounter.Add(uint32(hdr.MarshalSize() + len(payload)))
+	d.bytesSent.Add(uint32(hdr.MarshalSize() + len(payload)))

 	if tp.isSwitchingToMaxSpatial && d.onMaxSubscribedLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo {
 		d.onMaxSubscribedLayerChanged(d, layer)
@@ -1166,6 +1181,7 @@ func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan
 				}

 				d.streamAllocatorBytesCounter.Add(uint32(pktSize))
+				d.bytesSent.Add(uint32(pktSize))

 				// only the first frame will need frameEndNeeded to close out the
 				// previous picture, rest are small key frames (for the video case)
@@ -1314,6 +1330,10 @@ func (d *DownTrack) handleRTCP(bytes []byte) {
 				if isRttChanged {
 					rttToReport = rtt
 				}
+
+				if sal := d.getStreamAllocatorListener(); sal != nil {
+					sal.OnRTCPReceiverReport(d, r)
+				}
 			}
 			if len(rr.Reports) > 0 {
 				d.listenerLock.RLock()
@@ -1399,12 +1419,18 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
 	nackAcks := uint32(0)
 	nackMisses := uint32(0)
 	numRepeatedNACKs := uint32(0)
+	nackInfos := make([]NackInfo, 0, len(filtered))
 	for _, meta := range d.sequencer.getPacketsMeta(filtered) {
 		if disallowedLayers[meta.layer] {
 			continue
 		}

 		nackAcks++
+		nackInfos = append(nackInfos, NackInfo{
+			SequenceNumber: meta.targetSeqNo,
+			Timestamp:      meta.timestamp,
+			Attempts:       meta.nacked,
+		})

 		if pool != nil {
 			PacketFactory.Put(pool)
@@ -1465,16 +1491,30 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
 			d.logger.Errorw("writing rtx packet err", err)
 		} else {
 			d.streamAllocatorBytesCounter.Add(uint32(pkt.Header.MarshalSize() + len(payload)))
+			d.bytesRetransmitted.Add(uint32(pkt.Header.MarshalSize() + len(payload)))

 			d.rtpStats.Update(&pkt.Header, len(payload), 0, time.Now().UnixNano())
 		}
 	}

-	d.statsLock.Lock()
-	d.totalRepeatedNACKs += numRepeatedNACKs
-	d.statsLock.Unlock()
+	d.totalRepeatedNACKs.Add(numRepeatedNACKs)

 	d.rtpStats.UpdateNackProcessed(nackAcks, nackMisses, numRepeatedNACKs)
+	// STREAM-ALLOCATOR-EXPERIMENTAL-TODO-START
+	// Need to check on the following
+	//   - get all NACKs from sequencer even if SFU is not acknowledging,
+	//     i. e. SFU does not acknowledge even same sequence number is NACKed too closely,
+	//     but if sequencer return those also (even if not actually retransmitting),
+	//     will that provide a signal?
+	//   - get padding NACKs also? Maybe only look at them when their NACK count is 2?
+	//     because padding runs in a separate path, it could get out of order with
+	//     primary packets. So, it could be NACKed once. But, a repeat NACK means they
+	//     were probably lost. But, as we do not retransmit padding packets, more than
+	//     the second try does not provide any useful signal.
+	// STREAM-ALLOCATOR-EXPERIMENTAL-TODO-END
+	if sal := d.getStreamAllocatorListener(); sal != nil && len(nackInfos) != 0 {
+		sal.OnNACK(d, nackInfos)
+	}
 }

 type extensionData struct {
@@ -1606,14 +1646,14 @@ func (d *DownTrack) getDeltaStats() map[uint32]*buffer.StreamStatsWithLayers {

 func (d *DownTrack) GetNackStats() (totalPackets uint32, totalRepeatedNACKs uint32) {
 	totalPackets = d.rtpStats.GetTotalPacketsPrimary()
-
-	d.statsLock.RLock()
-	totalRepeatedNACKs = d.totalRepeatedNACKs
-	d.statsLock.RUnlock()
-
+	totalRepeatedNACKs = d.totalRepeatedNACKs.Load()
 	return
 }

+func (d *DownTrack) GetAndResetBytesSent() (uint32, uint32) {
+	return d.bytesSent.Swap(0), d.bytesRetransmitted.Swap(0)
+}
+
 func (d *DownTrack) onBindAndConnected() {
 	if d.connected.Load() && d.bound.Load() && !d.bindAndConnectedOnce.Swap(true) {
 		if d.kind == webrtc.RTPCodecTypeVideo {
@@ -89,7 +89,7 @@ func NewChannelObserver(params ChannelObserverParams, logger logger.Logger) *Cha
 			CollapseThreshold:      params.EstimateCollapseThreshold,
 		}),
 		nackTracker: NewNackTracker(NackTrackerParams{
-			Name:              params.Name + "-estimate",
+			Name:              params.Name + "-nack",
 			Logger:            logger,
 			WindowMinDuration: params.NackWindowMinDuration,
 			WindowMaxDuration: params.NackWindowMaxDuration,
@@ -122,6 +122,10 @@ func (c *ChannelObserver) GetNackRatio() float64 {
 	return c.nackTracker.GetRatio()
 }

+func (c *ChannelObserver) GetNackHistory() []string {
+	return c.nackTracker.GetHistory()
+}
+
 func (c *ChannelObserver) GetTrend() (ChannelTrend, ChannelCongestionReason) {
 	estimateDirection := c.estimateTrend.GetDirection()

@@ -23,16 +23,22 @@ type NackTracker struct {
 	windowStartTime time.Time
 	packets         uint32
 	repeatedNacks   uint32
+
+	// STREAM-ALLOCATOR-EXPERIMENTAL-TODO: remove when cleaning up experimental stuff
+	history []string
 }

 func NewNackTracker(params NackTrackerParams) *NackTracker {
 	return &NackTracker{
-		params: params,
+		params:  params,
+		history: make([]string, 0, 10),
 	}
 }

 func (n *NackTracker) Add(packets uint32, repeatedNacks uint32) {
 	if n.params.WindowMaxDuration != 0 && !n.windowStartTime.IsZero() && time.Since(n.windowStartTime) > n.params.WindowMaxDuration {
+		n.updateHistory()
+
 		n.windowStartTime = time.Time{}
 		n.packets = 0
 		n.repeatedNacks = 0
@@ -81,7 +87,19 @@ func (n *NackTracker) ToString() string {
 		elapsed := now.Sub(n.windowStartTime).Seconds()
 		window = fmt.Sprintf("t: %+v|%+v|%.2fs", n.windowStartTime.Format(time.UnixDate), now.Format(time.UnixDate), elapsed)
 	}
-	return fmt.Sprintf("n: %s, t: %s, p: %d,  rn: %d, rn/p: %.2f", n.params.Name, window, n.packets, n.repeatedNacks, n.GetRatio())
+	return fmt.Sprintf("n: %s, %s, p: %d, rn: %d, rn/p: %.2f", n.params.Name, window, n.packets, n.repeatedNacks, n.GetRatio())
+}
+
+func (n *NackTracker) GetHistory() []string {
+	return n.history
+}
+
+func (n *NackTracker) updateHistory() {
+	if len(n.history) >= 10 {
+		n.history = n.history[1:]
+	}
+
+	n.history = append(n.history, n.ToString())
 }

 // ------------------------------------------------
@@ -195,13 +195,13 @@ func (p *Prober) Reset() {
 	p.processActiveStateQueue()
 }

-func (p *Prober) AddCluster(desiredRateBps int, expectedRateBps int, minDuration time.Duration, maxDuration time.Duration) ProbeClusterId {
+func (p *Prober) AddCluster(mode ProbeClusterMode, desiredRateBps int, expectedRateBps int, minDuration time.Duration, maxDuration time.Duration) ProbeClusterId {
 	if desiredRateBps <= 0 {
 		return ProbeClusterIdInvalid
 	}

 	clusterId := ProbeClusterId(p.clusterId.Inc())
-	cluster := NewCluster(clusterId, desiredRateBps, expectedRateBps, minDuration, maxDuration)
+	cluster := NewCluster(clusterId, mode, desiredRateBps, expectedRateBps, minDuration, maxDuration)
 	p.logger.Debugw("cluster added", "cluster", cluster.String())

 	p.pushBackClusterAndMaybeStart(cluster)
@@ -353,47 +353,117 @@ type ProbeClusterId uint32

 const (
 	ProbeClusterIdInvalid ProbeClusterId = 0
+
+	bucketDuration  = time.Second
+	bytesPerProbe   = 1000
+	minProbeRateBps = 10000
 )

+// -----------------------------------
+
+type ProbeClusterMode int
+
+const (
+	ProbeClusterModeUniform ProbeClusterMode = iota
+	ProbeClusterModeLinearChirp
+)
+
+func (p ProbeClusterMode) String() string {
+	switch p {
+	case ProbeClusterModeUniform:
+		return "UNIFORM"
+	case ProbeClusterModeLinearChirp:
+		return "LINEAR_CHIRP"
+	default:
+		return fmt.Sprintf("%d", int(p))
+	}
+}
+
+// ---------------------------------------------------------------------------
+
 type ProbeClusterInfo struct {
 	Id        ProbeClusterId
 	BytesSent int
 	Duration  time.Duration
 }

+type clusterBucket struct {
+	desiredBytes       int
+	desiredElapsedTime time.Duration
+	sleepDuration      time.Duration
+}
+
 type Cluster struct {
 	lock sync.RWMutex

 	id           ProbeClusterId
+	mode         ProbeClusterMode
 	desiredBytes int
 	minDuration  time.Duration
 	maxDuration  time.Duration

-	sleepDuration time.Duration
+	buckets   []clusterBucket
+	bucketIdx int

 	bytesSentProbe    int
 	bytesSentNonProbe int
 	startTime         time.Time
 }

-func NewCluster(id ProbeClusterId, desiredRateBps int, expectedRateBps int, minDuration time.Duration, maxDuration time.Duration) *Cluster {
-	minDurationMs := minDuration.Milliseconds()
-	desiredBytes := int((int64(desiredRateBps)*minDurationMs/time.Second.Milliseconds() + 7) / 8)
-	expectedBytes := int((int64(expectedRateBps)*minDurationMs/time.Second.Milliseconds() + 7) / 8)
-
-	// pace based on sending approximately 1000 bytes per probe
-	numProbes := (desiredBytes - expectedBytes + 999) / 1000
-	sleepDurationMicroSeconds := int(float64(minDurationMs*1000)/float64(numProbes) + 0.5)
+func NewCluster(id ProbeClusterId, mode ProbeClusterMode, desiredRateBps int, expectedRateBps int, minDuration time.Duration, maxDuration time.Duration) *Cluster {
 	c := &Cluster{
-		id:            id,
-		desiredBytes:  desiredBytes,
-		minDuration:   minDuration,
-		maxDuration:   maxDuration,
-		sleepDuration: time.Duration(sleepDurationMicroSeconds) * time.Microsecond,
+		id:          id,
+		mode:        mode,
+		minDuration: minDuration,
+		maxDuration: maxDuration,
 	}
+	c.initBuckets(desiredRateBps, expectedRateBps, minDuration)
+	c.desiredBytes = c.buckets[len(c.buckets)-1].desiredBytes
 	return c
 }

+func (c *Cluster) initBuckets(desiredRateBps int, expectedRateBps int, minDuration time.Duration) {
+	// split into 1-second bucket
+	// NOTE: splitting even if mode is unitform
+	numBuckets := int((minDuration.Milliseconds() + bucketDuration.Milliseconds() - 1) / bucketDuration.Milliseconds())
+	if numBuckets < 1 {
+		numBuckets = 1
+	}
+
+	expectedRateBytes := (expectedRateBps + 7) / 8
+	baseProbeRateBps := (desiredRateBps - expectedRateBps + numBuckets - 1) / numBuckets
+
+	runningDesiredBytes := 0
+	runningDesiredElapsedTime := time.Duration(0)
+
+	c.buckets = make([]clusterBucket, 0, numBuckets)
+	for bucketIdx := 0; bucketIdx < numBuckets; bucketIdx++ {
+		multiplier := numBuckets
+		if c.mode == ProbeClusterModeLinearChirp {
+			multiplier = bucketIdx + 1
+		}
+
+		bucketProbeRateBps := baseProbeRateBps * multiplier
+		if bucketProbeRateBps < minProbeRateBps {
+			bucketProbeRateBps = minProbeRateBps
+		}
+		bucketProbeRateBytes := (bucketProbeRateBps + 7) / 8
+
+		// pace based on bytes per probe
+		numProbes := (bucketProbeRateBytes + bytesPerProbe - 1) / bytesPerProbe
+		sleepDurationMicroSeconds := int(float64(1_000_000)/float64(numProbes) + 0.5)
+
+		runningDesiredBytes += bucketProbeRateBytes + expectedRateBytes
+		runningDesiredElapsedTime += bucketDuration
+
+		c.buckets = append(c.buckets, clusterBucket{
+			desiredBytes:       runningDesiredBytes,
+			desiredElapsedTime: runningDesiredElapsedTime,
+			sleepDuration:      time.Duration(sleepDurationMicroSeconds) * time.Microsecond,
+		})
+	}
+}
+
 func (c *Cluster) Start() {
 	c.lock.Lock()
 	defer c.lock.Unlock()
@@ -407,7 +477,7 @@ func (c *Cluster) GetSleepDuration() time.Duration {
 	c.lock.RLock()
 	defer c.lock.RUnlock()

-	return c.sleepDuration
+	return c.buckets[c.bucketIdx].sleepDuration
 }

 func (c *Cluster) PacketsSent(size int) {
@@ -456,7 +526,6 @@ func (c *Cluster) GetInfo() ProbeClusterInfo {

 func (c *Cluster) Process(pl ProberListener) {
 	c.lock.RLock()
-
 	timeElapsed := time.Since(c.startTime)

 	// Calculate number of probe bytes that should have been sent since start.
@@ -464,14 +533,7 @@ func (c *Cluster) Process(pl ProberListener) {
 	// However, it is possible that timeElapsed is more than minDuration due
 	// to scheduling variance. When overshooting time budget, use a capped
 	// short fall if there is a grace period given.
-	windowDone := float64(timeElapsed) / float64(c.minDuration)
-	if windowDone > 1.0 {
-		// cluster has been running for longer than  minDuration
-		windowDone = 1.0
-	}
-
-	bytesShouldHaveBeenSent := int(windowDone * float64(c.desiredBytes))
-	bytesShortFall := bytesShouldHaveBeenSent - c.bytesSentProbe - c.bytesSentNonProbe
+	bytesShortFall := c.buckets[c.bucketIdx].desiredBytes - c.bytesSentProbe - c.bytesSentNonProbe
 	if bytesShortFall < 0 {
 		bytesShortFall = 0
 	}
@@ -482,6 +544,14 @@ func (c *Cluster) Process(pl ProberListener) {
 	}
 	// round up to packet size
 	bytesShortFall = ((bytesShortFall + 274) / 275) * 275
+
+	// move to next bucket if necessary
+	if timeElapsed > c.buckets[c.bucketIdx].desiredElapsedTime {
+		c.bucketIdx++
+		if c.bucketIdx >= len(c.buckets) {
+			c.bucketIdx = len(c.buckets) - 1
+		}
+	}
 	c.lock.RUnlock()

 	if bytesShortFall > 0 && pl != nil {
@@ -497,8 +567,9 @@ func (c *Cluster) String() string {
 		activeTimeMs = time.Since(c.startTime).Milliseconds()
 	}

-	return fmt.Sprintf("id: %d, bytes: desired %d / probe %d / non-probe %d / remaining: %d, time(ms): active %d / min %d / max %d",
+	return fmt.Sprintf("id: %d, mode: %s, bytes: desired %d / probe %d / non-probe %d / remaining: %d, time(ms): active %d / min %d / max %d",
 		c.id,
+		c.mode,
 		c.desiredBytes,
 		c.bytesSentProbe,
 		c.bytesSentNonProbe,
@@ -507,3 +578,5 @@ func (c *Cluster) String() string {
 		c.minDuration.Milliseconds(),
 		c.maxDuration.Milliseconds())
 }
+
+// ----------------------------------------------------------------------
@@ -0,0 +1,158 @@
+package streamallocator
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/livekit/protocol/utils/timeseries"
+)
+
+// ------------------------------------------------
+
+const (
+	rateMonitorWindow  = 10 * time.Second
+	queueMonitorWindow = 2 * time.Second
+)
+
+// ------------------------------------------------
+
+type RateMonitor struct {
+	bitrateEstimate             *timeseries.TimeSeries[int64]
+	managedBytesSent            *timeseries.TimeSeries[uint32]
+	managedBytesRetransmitted   *timeseries.TimeSeries[uint32]
+	unmanagedBytesSent          *timeseries.TimeSeries[uint32]
+	unmanagedBytesRetransmitted *timeseries.TimeSeries[uint32]
+
+	// STREAM-ALLOCATOR-EXPERIMENTAL-TODO: remove after experimental
+	history []string
+}
+
+func NewRateMonitor() *RateMonitor {
+	return &RateMonitor{
+		bitrateEstimate: timeseries.NewTimeSeries[int64](timeseries.TimeSeriesParams{
+			UpdateOp: timeseries.TimeSeriesUpdateOpLatest,
+			Window:   rateMonitorWindow,
+		}),
+		managedBytesSent: timeseries.NewTimeSeries[uint32](timeseries.TimeSeriesParams{
+			UpdateOp: timeseries.TimeSeriesUpdateOpAdd,
+			Window:   rateMonitorWindow,
+		}),
+		managedBytesRetransmitted: timeseries.NewTimeSeries[uint32](timeseries.TimeSeriesParams{
+			UpdateOp: timeseries.TimeSeriesUpdateOpAdd,
+			Window:   rateMonitorWindow,
+		}),
+		unmanagedBytesSent: timeseries.NewTimeSeries[uint32](timeseries.TimeSeriesParams{
+			UpdateOp: timeseries.TimeSeriesUpdateOpAdd,
+			Window:   rateMonitorWindow,
+		}),
+		unmanagedBytesRetransmitted: timeseries.NewTimeSeries[uint32](timeseries.TimeSeriesParams{
+			UpdateOp: timeseries.TimeSeriesUpdateOpAdd,
+			Window:   rateMonitorWindow,
+		}),
+	}
+}
+
+func (r *RateMonitor) Update(estimate int64, managedBytesSent uint32, managedBytesRetransmitted uint32, unmanagedBytesSent uint32, unmanagedBytesRetransmitted uint32) {
+	now := time.Now()
+	r.bitrateEstimate.AddSampleAt(estimate, now)
+	r.managedBytesSent.AddSampleAt(managedBytesSent, now)
+	r.managedBytesRetransmitted.AddSampleAt(managedBytesRetransmitted, now)
+	r.unmanagedBytesSent.AddSampleAt(unmanagedBytesSent, now)
+	r.unmanagedBytesRetransmitted.AddSampleAt(unmanagedBytesRetransmitted, now)
+
+	r.updateHistory()
+}
+
+// STREAM-ALLOCATOR-TODO:
+// This should be updated periodically to flush any pending.
+// Reason is that the estimate could be higher than the actual rate by a significant amount.
+// So, updating periodically to flush out samples that will not contribute to queueing would be good.
+func (r *RateMonitor) GetQueuingGuess() float64 {
+	_, _, _, _, _, qd := r.getRates(queueMonitorWindow)
+	return qd
+}
+
+func (r *RateMonitor) getRates(monitorDuration time.Duration) (float64, float64, float64, float64, float64, float64) {
+	threshold := time.Now().Add(-monitorDuration)
+	bitrateEstimateSamples := r.bitrateEstimate.GetSamplesAfter(threshold)
+	managedBytesSentSamples := r.managedBytesSent.GetSamplesAfter(threshold)
+	managedBytesRetransmittedSamples := r.managedBytesRetransmitted.GetSamplesAfter(threshold)
+	unmanagedBytesSentSamples := r.unmanagedBytesSent.GetSamplesAfter(threshold)
+	unmanagedBytesRetransmittedSamples := r.unmanagedBytesRetransmitted.GetSamplesAfter(threshold)
+
+	if len(bitrateEstimateSamples) == 0 || (len(managedBytesSentSamples)+len(managedBytesRetransmittedSamples)+len(unmanagedBytesSentSamples)+len(unmanagedBytesRetransmittedSamples)) == 0 {
+		return 0.0, 0.0, 0.0, 0.0, 0.0, 0.0
+	}
+
+	totalBitrateEstimate := getTimeWeightedSum(bitrateEstimateSamples)
+	totalManagedSent := getRate(managedBytesSentSamples) * 8
+	totalManagedRetransmitted := getRate(managedBytesRetransmittedSamples) * 8
+	totalUnmanagedSent := getRate(unmanagedBytesSentSamples) * 8
+	totalUnmanagedRetransmitted := getRate(unmanagedBytesRetransmittedSamples) * 8
+	totalBits := totalManagedSent + totalManagedRetransmitted + totalUnmanagedSent + totalUnmanagedRetransmitted
+
+	queuingDelay := float64(0.0)
+	if totalBits > totalBitrateEstimate {
+		latestBitrateEstimate := bitrateEstimateSamples[len(bitrateEstimateSamples)-1].Value
+		excessBits := totalBits - totalBitrateEstimate
+		queuingDelay = excessBits / float64(latestBitrateEstimate)
+	}
+	return totalBitrateEstimate, totalManagedSent, totalManagedRetransmitted, totalUnmanagedSent, totalUnmanagedRetransmitted, queuingDelay
+}
+
+func (r *RateMonitor) updateHistory() {
+	if len(r.history) >= 10 {
+		r.history = r.history[1:]
+	}
+
+	e, m, mr, um, umr, qd := r.getRates(time.Second)
+	if e == 0.0 {
+		return
+	}
+
+	r.history = append(
+		r.history,
+		fmt.Sprintf("t: %+v, e: %.2f, m: %.2f/%.2f, um: %.2f/%.2f, qd: %.2f", time.Now().UnixMilli(), e, m, mr, um, umr, qd),
+	)
+}
+
+func (r *RateMonitor) GetHistory() []string {
+	return r.history
+}
+
+// ------------------------------------------------
+
+func getTimeWeightedSum[T int64 | uint32](samples []timeseries.TimeSeriesSample[T]) float64 {
+	if len(samples) < 2 {
+		return 0.0
+	}
+
+	sum := 0.0
+	for i := 1; i < len(samples); i++ {
+		diff := samples[i].At.Sub(samples[i-1].At).Seconds()
+		sum += diff * float64(samples[i-1].Value)
+	}
+
+	diff := time.Now().Sub(samples[len(samples)-1].At).Seconds()
+	sum += diff * float64(samples[len(samples)-1].Value)
+	return sum
+}
+
+func getRate[T int64 | uint32](samples []timeseries.TimeSeriesSample[T]) float64 {
+	if len(samples) < 2 {
+		return 0.0
+	}
+
+	sum := 0.0
+	// start at 1 as the first sample duration is not available
+	for i := 1; i < len(samples); i++ {
+		sum += float64(samples[i].Value)
+	}
+
+	duration := samples[len(samples)-1].At.Sub(samples[0].At)
+	if duration == 0 {
+		return 0.0
+	}
+
+	return sum / duration.Seconds()
+}
@@ -66,7 +66,7 @@ var (
 		Name:                           "non-probe",
 		EstimateRequiredSamples:        8,
 		EstimateDownwardTrendThreshold: -0.5,
-		EstimateCollapseThreshold:      500 * time.Millisecond,
+		EstimateCollapseThreshold:      250 * time.Millisecond,
 		NackWindowMinDuration:          1 * time.Second,
 		NackWindowMaxDuration:          2 * time.Second,
 		NackRatioThreshold:             0.08,
@@ -108,6 +108,8 @@ const (
 	streamAllocatorSignalResume
 	streamAllocatorSignalSetAllowPause
 	streamAllocatorSignalSetChannelCapacity
+	streamAllocatorSignalNACK
+	streamAllocatorSignalRTCPReceiverReport
 )

 func (s streamAllocatorSignal) String() string {
@@ -132,6 +134,10 @@ func (s streamAllocatorSignal) String() string {
 		return "SET_ALLOW_PAUSE"
 	case streamAllocatorSignalSetChannelCapacity:
 		return "SET_CHANNEL_CAPACITY"
+	case streamAllocatorSignalNACK:
+		return "NACK"
+	case streamAllocatorSignalRTCPReceiverReport:
+		return "RTCP_RECEIVER_REPORT"
 	default:
 		return fmt.Sprintf("%d", int(s))
 	}
@@ -180,6 +186,7 @@ type StreamAllocator struct {
 	prober *Prober

 	channelObserver *ChannelObserver
+	rateMonitor     *RateMonitor

 	videoTracksMu        sync.RWMutex
 	videoTracks          map[livekit.TrackID]*Track
@@ -201,8 +208,9 @@ func NewStreamAllocator(params StreamAllocatorParams) *StreamAllocator {
 		prober: NewProber(ProberParams{
 			Logger: params.Logger,
 		}),
+		rateMonitor: NewRateMonitor(),
 		videoTracks: make(map[livekit.TrackID]*Track),
-		eventCh:     make(chan Event, 200),
+		eventCh:     make(chan Event, 1000),
 	}

 	s.resetState()
@@ -463,11 +471,30 @@ func (s *StreamAllocator) OnResume(downTrack *sfu.DownTrack) {
 	})
 }

-// called when a video DownTrack sends a packet
+// called by a video DownTrack to report packet send
 func (s *StreamAllocator) OnPacketsSent(downTrack *sfu.DownTrack, size int) {
 	s.prober.PacketsSent(size)
 }

+// called by a video DownTrack when it processes NACKs
+func (s *StreamAllocator) OnNACK(downTrack *sfu.DownTrack, nackInfos []sfu.NackInfo) {
+	s.postEvent(Event{
+		Signal:  streamAllocatorSignalNACK,
+		TrackID: livekit.TrackID(downTrack.ID()),
+		Data:    nackInfos,
+	})
+}
+
+// called by a video DownTrack when it receives an RTCP Receiver Report
+// STREAM-ALLOCATOR-TODO: this should probably be done for audio tracks also
+func (s *StreamAllocator) OnRTCPReceiverReport(downTrack *sfu.DownTrack, rr rtcp.ReceptionReport) {
+	s.postEvent(Event{
+		Signal:  streamAllocatorSignalRTCPReceiverReport,
+		TrackID: livekit.TrackID(downTrack.ID()),
+		Data:    rr,
+	})
+}
+
 // called when prober wants to send packet(s)
 func (s *StreamAllocator) OnSendProbe(bytesToSend int) {
 	s.postEvent(Event{
@@ -538,7 +565,7 @@ func (s *StreamAllocator) processEvents() {
 }

 func (s *StreamAllocator) ping() {
-	ticker := time.NewTicker(time.Second)
+	ticker := time.NewTicker(500 * time.Millisecond)
 	defer ticker.Stop()

 	for {
@@ -575,6 +602,10 @@ func (s *StreamAllocator) handleEvent(event *Event) {
 		s.handleSignalSetAllowPause(event)
 	case streamAllocatorSignalSetChannelCapacity:
 		s.handleSignalSetChannelCapacity(event)
+	case streamAllocatorSignalNACK:
+		s.handleSignalNACK(event)
+	case streamAllocatorSignalRTCPReceiverReport:
+		s.handleSignalRTCPReceiverReport(event)
 	}
 }

@@ -608,6 +639,7 @@ func (s *StreamAllocator) handleSignalAdjustState(event *Event) {
 func (s *StreamAllocator) handleSignalEstimate(event *Event) {
 	receivedEstimate, _ := event.Data.(int64)
 	s.lastReceivedEstimate = receivedEstimate
+	s.monitorRate(receivedEstimate)

 	// while probing, maintain estimate separately to enable keeping current committed estimate if probe fails
 	if s.isInProbe() {
@@ -627,6 +659,8 @@ func (s *StreamAllocator) handleSignalPeriodicPing(event *Event) {
 	if s.state == streamAllocatorStateDeficient {
 		s.maybeProbe()
 	}
+
+	s.updateTracksHistory()
 }

 func (s *StreamAllocator) handleSignalSendProbe(event *Event) {
@@ -702,6 +736,30 @@ func (s *StreamAllocator) handleSignalSetChannelCapacity(event *Event) {
 	}
 }

+func (s *StreamAllocator) handleSignalNACK(event *Event) {
+	nackInfos := event.Data.([]sfu.NackInfo)
+
+	s.videoTracksMu.Lock()
+	track := s.videoTracks[event.TrackID]
+	s.videoTracksMu.Unlock()
+
+	if track != nil {
+		track.UpdateNack(nackInfos)
+	}
+}
+
+func (s *StreamAllocator) handleSignalRTCPReceiverReport(event *Event) {
+	rr := event.Data.(rtcp.ReceptionReport)
+
+	s.videoTracksMu.Lock()
+	track := s.videoTracks[event.TrackID]
+	s.videoTracksMu.Unlock()
+
+	if track != nil {
+		track.ProcessRTCPReceiverReport(rr)
+	}
+}
+
 func (s *StreamAllocator) setState(state streamAllocatorState) {
 	if s.state == state {
 		return
@@ -801,6 +859,13 @@ func (s *StreamAllocator) handleNewEstimateInNonProbe() {
 		"expectedUsage(bps)", expectedBandwidthUsage,
 		"channel", s.channelObserver.ToString(),
 	)
+	s.params.Logger.Infow(
+		"stream allocator: channel congestion detected, updating channel capacity: experimental",
+		"rateHistory", s.rateMonitor.GetHistory(),
+		"expectedQueuing", s.rateMonitor.GetQueuingGuess(),
+		"nackHistory", s.channelObserver.GetNackHistory(),
+		"trackHistory", s.getTracksHistory(),
+	)
 	s.committedChannelCapacity = estimateToCommit

 	// reset to get new set of samples for next trend
@@ -919,7 +984,7 @@ func (s *StreamAllocator) finalizeProbe() {

 	//
 	// Reset estimator at the end of a probe irrespective of probe result to get fresh readings.
-	// With a failed probe, the latest estimate would be lower than committed estimate.
+	// With a failed probe, the latest estimate could be lower than committed estimate.
 	// As bandwidth estimator (remote in REMB case, local in TWCC case) holds state,
 	// subsequent estimates could start from the lower point. That should not trigger a
 	// downward trend and get latched to committed estimate as that would trigger a re-allocation.
@@ -1179,6 +1244,7 @@ func (s *StreamAllocator) initProbe(probeGoalDeltaBps int64) {
 	s.channelObserver.SeedEstimate(s.lastReceivedEstimate)

 	s.probeClusterId = s.prober.AddCluster(
+		ProbeClusterModeUniform,
 		int(s.probeGoalBps),
 		int(expectedBandwidthUsage),
 		ProbeMinDuration,
@@ -1342,4 +1408,47 @@ func (s *StreamAllocator) getMaxDistanceSortedDeficient() MaxDistanceSorter {
 	return maxDistanceSorter
 }

+// STREAM-ALLOCATOR-EXPERIMENTAL-TODO
+// Monitor sent rate vs estimate to figure out queuing on congestion.
+// Idea here is to pause all managed tracks on congestion detection immediately till queue drains.
+// That will allow channel to clear up without more traffic added and a re-allocation can start afresh.
+// Some bits to work out
+//   - how good is queuing estimate?
+//   - should we pause unmanaged tracks also? But, they will restart at highest layer and request a key frame.
+//   - what should be the channel capacity to use when resume re-allocation happens?
+func (s *StreamAllocator) monitorRate(estimate int64) {
+	managedBytesSent := uint32(0)
+	managedBytesRetransmitted := uint32(0)
+	unmanagedBytesSent := uint32(0)
+	unmanagedBytesRetransmitted := uint32(0)
+	for _, track := range s.getTracks() {
+		b, r := track.GetAndResetBytesSent()
+		if track.IsManaged() {
+			managedBytesSent += b
+			managedBytesRetransmitted += r
+		} else {
+			unmanagedBytesSent += b
+			unmanagedBytesRetransmitted += r
+		}
+	}
+
+	s.rateMonitor.Update(estimate, managedBytesSent, managedBytesRetransmitted, unmanagedBytesSent, unmanagedBytesRetransmitted)
+}
+
+func (s *StreamAllocator) updateTracksHistory() {
+	for _, track := range s.getTracks() {
+		track.UpdateHistory()
+	}
+}
+
+func (s *StreamAllocator) getTracksHistory() map[livekit.TrackID]string {
+	tracks := s.getTracks()
+	history := make(map[livekit.TrackID]string, len(tracks))
+	for _, track := range tracks {
+		history[track.ID()] = track.GetHistory()
+	}
+
+	return history
+}
+
 // ------------------------------------------------
@@ -1,8 +1,14 @@
 package streamallocator

 import (
+	"fmt"
+	"sort"
+	"time"
+
+	"github.com/livekit/mediatransportutil"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
+	"github.com/pion/rtcp"

 	"github.com/livekit/livekit-server/pkg/sfu"
 	"github.com/livekit/livekit-server/pkg/sfu/buffer"
@@ -21,6 +27,19 @@ type Track struct {
 	totalPackets       uint32
 	totalRepeatedNacks uint32

+	nackInfos map[uint16]sfu.NackInfo
+	// STREAM-ALLOCATOR-EXPERIMENTAL-TODO: remove after experimental
+	nackHistory []string
+
+	receiverReportInitialized       bool
+	totalLostAtLastRead             uint32
+	totalLost                       uint32
+	highestSequenceNumberAtLastRead uint32
+	highestSequenceNumber           uint32
+	maxRTT                          uint32
+	// STREAM-ALLOCATOR-EXPERIMENTAL-TODO: remove after experimental
+	receiverReportHistory []string
+
 	isDirty bool

 	isPaused bool
@@ -34,12 +53,15 @@ func NewTrack(
 	logger logger.Logger,
 ) *Track {
 	t := &Track{
-		downTrack:   downTrack,
-		source:      source,
-		isSimulcast: isSimulcast,
-		publisherID: publisherID,
-		logger:      logger,
-		isPaused:    true,
+		downTrack:             downTrack,
+		source:                source,
+		isSimulcast:           isSimulcast,
+		publisherID:           publisherID,
+		logger:                logger,
+		nackInfos:             make(map[uint16]sfu.NackInfo),
+		nackHistory:           make([]string, 0, 10),
+		receiverReportHistory: make([]string, 0, 10),
+		isPaused:              true,
 	}
 	t.SetPriority(0)
 	t.SetMaxLayer(downTrack.MaxLayer())
@@ -176,6 +198,150 @@ func (t *Track) GetNackDelta() (uint32, uint32) {
 	return packetDelta, nackDelta
 }

+func (t *Track) UpdateNack(nackInfos []sfu.NackInfo) {
+	for _, ni := range nackInfos {
+		t.nackInfos[ni.SequenceNumber] = ni
+	}
+}
+
+func (t *Track) GetAndResetNackStats() (lowest uint16, highest uint16, numNacked int, numNacks int, numRuns int) {
+	if len(t.nackInfos) == 0 {
+		return
+	}
+
+	sns := make([]uint16, 0, len(t.nackInfos))
+	for _, ni := range t.nackInfos {
+		if lowest == 0 || ni.SequenceNumber-lowest > (1<<15) {
+			lowest = ni.SequenceNumber
+		}
+		if highest == 0 || highest-ni.SequenceNumber > (1<<15) {
+			highest = ni.SequenceNumber
+		}
+		numNacks += int(ni.Attempts)
+		sns = append(sns, ni.SequenceNumber)
+	}
+	numNacked = len(t.nackInfos)
+
+	// find number of runs, i. e. bursts of contiguous sequence numbers NACKed, does not include isolated NACKs
+	sort.Slice(sns, func(i, j int) bool {
+		return (sns[i] - sns[j]) > (1 << 15)
+	})
+
+	rsn := sns[0]
+	rsi := 0
+	for i := 1; i < len(sns); i++ {
+		if sns[i] == rsn+1 {
+			continue
+		}
+
+		if (i - rsi - 1) > 0 {
+			numRuns++
+		}
+
+		rsn = sns[i]
+		rsi = i
+	}
+
+	t.nackInfos = make(map[uint16]sfu.NackInfo)
+	return
+}
+
+func (t *Track) ProcessRTCPReceiverReport(rr rtcp.ReceptionReport) {
+	if !t.receiverReportInitialized {
+		t.receiverReportInitialized = true
+		t.totalLostAtLastRead = rr.TotalLost
+		t.highestSequenceNumberAtLastRead = rr.LastSequenceNumber
+	}
+
+	t.totalLost = rr.TotalLost
+	t.highestSequenceNumber = rr.LastSequenceNumber
+
+	if rtt, err := mediatransportutil.GetRttMsFromReceiverReportOnly(&rr); err != nil {
+		if rtt > t.maxRTT {
+			t.maxRTT = rtt
+		}
+	}
+
+	t.updateReceiverReportHistory()
+}
+
+func (t *Track) GetRTCPReceiverReportDelta() (uint32, uint32, uint32) {
+	deltaPackets := t.highestSequenceNumber - t.highestSequenceNumberAtLastRead
+	t.highestSequenceNumberAtLastRead = t.highestSequenceNumber
+
+	deltaLost := t.totalLost - t.totalLostAtLastRead
+	t.totalLostAtLastRead = t.totalLost
+
+	maxRTT := t.maxRTT
+	t.maxRTT = 0
+
+	return deltaLost, deltaPackets, maxRTT
+}
+
+func (t *Track) GetAndResetBytesSent() (uint32, uint32) {
+	return t.downTrack.GetAndResetBytesSent()
+}
+
+func (t *Track) UpdateHistory() {
+	t.updateNackHistory()
+}
+
+func (t *Track) GetHistory() string {
+	return fmt.Sprintf("t: %+v, n: %+v, rr: %+v", time.Now(), t.nackHistory, t.receiverReportHistory)
+}
+
+// STREAM-ALLOCATOR-EXPERIMENTAL-TODO:
+// Idea is to check if this provides a good signal to detect congestion.
+// This measures a few things
+//   1. Spread: sequence number difference between highest and lowest NACK
+//      - shows how widespread the losses are
+//   2. Number of runs of length more than 1: Counts number of burst losses.
+//      - could be a sign of congestion when losses are bursty
+//   3. NACK density: how many sequence numbers in the spread were NACKed.
+//      - a high density could be a sign of congestion
+//   4. NACK intensity: how many times those sequence numbers were NACKed.
+//      - high intensity could be a sign of congestion
+//
+// While these all could be good signals, some challenges in making use of these
+// - aggregating across tracks
+// - proper thresholing, i. e. something based on averages should not trip
+//   because of small numbers, e. g. a single NACK run of 2 sequence numbers
+//   is technically a burst, but is it a signal of congestion?
+func (t *Track) updateNackHistory() {
+	if len(t.nackHistory) >= 10 {
+		t.nackHistory = t.nackHistory[1:]
+	}
+
+	l, h, nnd, nns, nr := t.GetAndResetNackStats()
+	spread := h - l + 1
+	density := float64(0.0)
+	if nnd != 0 {
+		density = float64(nnd) / float64(spread)
+	} else {
+		spread = 0
+	}
+	intensity := float64(0.0)
+	if nnd != 0 {
+		intensity = float64(nns) / float64(nnd)
+	}
+	t.nackHistory = append(
+		t.nackHistory,
+		fmt.Sprintf("t: %+v, l: %d, h: %d, sp: %d, nnd: %d, dens: %.2f, nns: %d, int: %.2f, nr: %d", time.Now().UnixMilli(), l, h, spread, nnd, density, nns, intensity, nr),
+	)
+}
+
+func (t *Track) updateReceiverReportHistory() {
+	if len(t.receiverReportHistory) >= 10 {
+		t.receiverReportHistory = t.receiverReportHistory[1:]
+	}
+
+	dl, dp, maxRTT := t.GetRTCPReceiverReportDelta()
+	t.receiverReportHistory = append(
+		t.receiverReportHistory,
+		fmt.Sprintf("t: %+v, l: %d, p: %d, rtt: %d", time.Now().Format(time.UnixDate), dl, dp, maxRTT),
+	)
+}
+
 // ------------------------------------------------

 type TrackSorter []*Track