Cleaning up unused stream allocator experiments. (#3237)

Not sure if we will ever use it. Can bring it back if needed.

pkg/sfu/bwe/remotebwe/channel_observer.go

@@ -167,12 +167,6 @@ func (c *channelObserver) GetNackRatio() float64 {
 	return c.nackTracker.GetRatio()
 }
 
-/* REMOTE-BWE-DATA
-func (c *channelObserver) GetNackHistory() []string {
-	return c.nackTracker.GetHistory()
-}
-*/
-
 func (c *channelObserver) GetTrend() (channelTrend, channelCongestionReason) {
 	estimateDirection := c.estimateTrend.GetDirection()
 

pkg/sfu/bwe/remotebwe/nack_tracker.go

@@ -58,24 +58,16 @@ type nackTracker struct {
 	windowStartTime time.Time
 	packets         uint32
 	repeatedNacks   uint32
-
-	/* REMOTE-BWE-DATA
-	// REMOTE-BWE-EXPERIMENTAL-TODO: remove when cleaning up experimental stuff
-	history []string
-	*/
 }
 
 func newNackTracker(params nackTrackerParams) *nackTracker {
 	return &nackTracker{
 		params: params,
-		// REMOTE-BWE-DATA history: make([]string, 0, 10),
 	}
 }
 
 func (n *nackTracker) Add(packets uint32, repeatedNacks uint32) {
 	if n.params.Config.WindowMaxDuration != 0 && !n.windowStartTime.IsZero() && time.Since(n.windowStartTime) > n.params.Config.WindowMaxDuration {
-		// REMOTE-BWE-DATA n.updateHistory()
-
 		n.windowStartTime = time.Time{}
 		n.packets = 0
 		n.repeatedNacks = 0
@@ -135,18 +127,4 @@ func (n *nackTracker) MarshalLogObject(e zapcore.ObjectEncoder) error {
 	return nil
 }
 
-/* REMOTE-BWE-DATA
-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.String())
-}
-*/
-
 // ------------------------------------------------

pkg/sfu/bwe/remotebwe/remote_bwe.go

@@ -229,13 +229,6 @@ func (r *RemoteBWE) estimateAvailableChannelCapacity(reason channelCongestionRea
 	}
 
 	ulgr.Infow("remote bwe: channel congestion detected, applying channel capacity update")
-	/* REMOTE-BWE-DATA
-	r.params.Logger.Debugw(
-		fmt.Sprintf("remote bwe: channel congestion detected, %s channel capacity: experimental", action),
-		"nackHistory", r.channelObserver.GetNackHistory(),
-	)
-	*/
-
 	r.committedChannelCapacity = estimateToCommit
 
 	// reset to get new set of samples for next trend

pkg/sfu/bwe/sendsidebwe/congestion_detector.go

@@ -817,7 +817,6 @@ func (c *congestionDetector) processFeedbackReport(fbr feedbackReport) {
 			// previous group ended, start a new group
 			c.updateCTRTrend(pg)
 
-			// SSBWE-REMOVE c.params.Logger.Infow("packet group done", "group", pg, "numGroups", len(c.packetGroups)) // SSBWE-REMOVE
 			pg = newPacketGroup(
 				packetGroupParams{
 					Config:       c.params.Config.PacketGroup,

pkg/sfu/bwe/sendsidebwe/packet_tracker.go

@@ -76,7 +76,6 @@ func (p *packetTracker) RecordPacketSendAndGetSequenceNumber(
 		probeClusterId: probeClusterId,
 		isProbe:        isProbe,
 	}
-	//p.params.Logger.Infow("send side bwe: packet sent", "packetInfo", pi) // SSBWE-REMOVE
 
 	p.sequenceNumber++
 

pkg/sfu/bwe/sendsidebwe/twcc_feedback.go

@@ -66,7 +66,6 @@ func newTWCCFeedback(params twccFeedbackParams) *twccFeedback {
 }
 
 func (t *twccFeedback) ProcessReport(report *rtcp.TransportLayerCC, at time.Time) (int64, bool) {
-	// t.params.Logger.Infow("send side bwe: TWCC feedback", "report", report.String()) // SSBWE-REMOVE
 	t.numReports++
 	if t.lastFeedbackTime.IsZero() {
 		t.lastFeedbackTime = at
@@ -99,7 +98,6 @@ func (t *twccFeedback) ProcessReport(report *rtcp.TransportLayerCC, at time.Time
 
 	if !isOutOfOrder {
 		sinceLast := at.Sub(t.lastFeedbackTime)
-		// t.params.Logger.Infow("send side bwe: report received", "at", at, "sinceLast", sinceLast, "pktCount", report.FbPktCount) // SSBWE-REMOVE
 		if t.estimatedFeedbackInterval == 0 {
 			t.estimatedFeedbackInterval = sinceLast
 		} else {

pkg/sfu/downtrack.go

@@ -157,14 +157,6 @@ func (d DownTrackState) MarshalLogObject(e zapcore.ObjectEncoder) error {
 
 // -------------------------------------------------------------------
 
-/* STREAM-ALLOCATOR-DATA
-type NackInfo struct {
-	Timestamp      uint32
-	SequenceNumber uint16
-	Attempts       uint8
-}
-*/
-
 type DownTrackStreamAllocatorListener interface {
 	// RTCP received
 	OnREMB(dt *DownTrack, remb *rtcp.ReceiverEstimatedMaximumBitrate)
@@ -191,14 +183,6 @@ type DownTrackStreamAllocatorListener interface {
 	// stream resumed
 	OnResume(dt *DownTrack)
 
-	/* STREAM-ALLOCATOR-DATA
-	// NACKs received
-	OnNACK(dt *DownTrack, nackInfos []NackInfo)
-
-	// RTCP Receiver Report received
-	OnRTCPReceiverReport(dt *DownTrack, rr rtcp.ReceptionReport)
-	*/
-
 	// check if track should participate in BWE
 	IsBWEEnabled(dt *DownTrack) bool
 
@@ -291,11 +275,7 @@ type DownTrack struct {
 
 	streamAllocatorLock     sync.RWMutex
 	streamAllocatorListener DownTrackStreamAllocatorListener
-	/* STREAM-ALLOCATOR-DATA
-	bytesSent                       atomic.Uint32
-	bytesRetransmitted              atomic.Uint32
-	*/
-	probeClusterId atomic.Uint32
+	probeClusterId          atomic.Uint32
 
 	playoutDelay *PlayoutDelayController
 
@@ -904,15 +884,16 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
 	}
 
 	headerSize := hdr.MarshalSize()
-	d.updateStats(updateStatsParams{
-		packetTime:        extPkt.Arrival,
-		extSequenceNumber: tp.rtp.extSequenceNumber,
-		extTimestamp:      tp.rtp.extTimestamp,
-		isOutOfOrder:      extPkt.IsOutOfOrder,
-		headerSize:        headerSize,
-		payloadSize:       len(payload),
-		marker:            hdr.Marker,
-	})
+	d.rtpStats.Update(
+		extPkt.Arrival,
+		tp.rtp.extSequenceNumber,
+		tp.rtp.extTimestamp,
+		hdr.Marker,
+		headerSize,
+		len(payload),
+		0,
+		extPkt.IsOutOfOrder,
+	)
 	d.pacer.Enqueue(&pacer.Packet{
 		Header:             hdr,
 		HeaderSize:         headerSize,
@@ -1027,15 +1008,16 @@ func (d *DownTrack) WritePaddingRTP(
 
 		hdrSize := hdr.MarshalSize()
 		payloadSize := len(payload)
-		d.updateStats(updateStatsParams{
-			packetTime:        mono.UnixNano(),
-			extSequenceNumber: snts[i].extSequenceNumber,
-			extTimestamp:      snts[i].extTimestamp,
-			headerSize:        hdrSize,
-			payloadSize:       payloadSize,
-			isPadding:         true,
-			disableCounter:    true,
-		})
+		d.rtpStats.Update(
+			mono.UnixNano(),
+			snts[i].extSequenceNumber,
+			snts[i].extTimestamp,
+			hdr.Marker,
+			hdrSize,
+			0,
+			payloadSize,
+			false,
+		)
 		d.pacer.Enqueue(&pacer.Packet{
 			Header:             hdr,
 			HeaderSize:         hdrSize,
@@ -1584,13 +1566,16 @@ func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan
 				}
 
 				headerSize := hdr.MarshalSize()
-				d.updateStats(updateStatsParams{
-					packetTime:        mono.UnixNano(),
-					extSequenceNumber: snts[i].extSequenceNumber,
-					extTimestamp:      snts[i].extTimestamp,
-					headerSize:        headerSize,
-					payloadSize:       len(payload),
-				})
+				d.rtpStats.Update(
+					mono.UnixNano(),
+					snts[i].extSequenceNumber,
+					snts[i].extTimestamp,
+					hdr.Marker,
+					headerSize,
+					len(payload),
+					0,
+					false,
+				)
 				d.pacer.Enqueue(&pacer.Packet{
 					Header:             hdr,
 					HeaderSize:         headerSize,
@@ -1747,12 +1732,6 @@ func (d *DownTrack) handleRTCP(bytes []byte) {
 					rttToReport = rtt
 				}
 
-				/* STREAM-ALLOCATOR-DATA
-				if sal := d.getStreamAllocatorListener(); sal != nil {
-					sal.OnRTCPReceiverReport(d, r)
-				}
-				*/
-
 				if d.playoutDelay != nil {
 					d.playoutDelay.OnSeqAcked(uint16(r.LastSequenceNumber))
 					// screen share track has inaccuracy jitter due to its low frame rate and bursty traffic
@@ -1870,20 +1849,12 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
 	nackAcks := uint32(0)
 	nackMisses := uint32(0)
 	numRepeatedNACKs := uint32(0)
-	// STREAM-ALLOCATOR-DATA nackInfos := make([]NackInfo, 0, len(filtered))
 	for _, epm := range d.sequencer.getExtPacketMetas(filtered) {
 		if disallowedLayers[epm.layer] {
 			continue
 		}
 
 		nackAcks++
-		/* STREAM-ALLOCATOR-DATA
-		nackInfos = append(nackInfos, NackInfo{
-			SequenceNumber: epm.targetSeqNo,
-			Timestamp:      epm.timestamp,
-			Attempts:       epm.nacked,
-		})
-		*/
 
 		pktBuff := *src
 		n, err := d.params.Receiver.ReadRTP(pktBuff, uint8(epm.layer), epm.sourceSeqNo)
@@ -1937,15 +1908,16 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
 		d.addDummyExtensions(&pkt.Header)
 
 		headerSize := pkt.Header.MarshalSize()
-		d.updateStats(updateStatsParams{
-			packetTime:        mono.UnixNano(),
-			extSequenceNumber: epm.extSequenceNumber,
-			extTimestamp:      epm.extTimestamp,
-			isOutOfOrder:      true,
-			headerSize:        headerSize,
-			payloadSize:       len(payload),
-			isRTX:             true,
-		})
+		d.rtpStats.Update(
+			mono.UnixNano(),
+			epm.extSequenceNumber,
+			epm.extTimestamp,
+			pkt.Header.Marker,
+			headerSize,
+			len(payload),
+			0,
+			true,
+		)
 		d.pacer.Enqueue(&pacer.Packet{
 			Header:             &pkt.Header,
 			HeaderSize:         headerSize,
@@ -1963,23 +1935,6 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
 	d.totalRepeatedNACKs.Add(numRepeatedNACKs)
 
 	d.rtpStats.UpdateNackProcessed(nackAcks, nackMisses, numRepeatedNACKs)
-	/* STREAM-ALLOCATOR-DATA
-	// 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)
-	}
-	*/
 }
 
 func (d *DownTrack) addDummyExtensions(hdr *rtp.Header) {
@@ -2070,12 +2025,6 @@ func (d *DownTrack) GetNackStats() (totalPackets uint32, totalRepeatedNACKs uint
 	return
 }
 
-/* STREAM-ALLOCATOR-DATA
-func (d *DownTrack) GetAndResetBytesSent() (uint32, uint32) {
-	return d.bytesSent.Swap(0), d.bytesRetransmitted.Swap(0)
-}
-*/
-
 func (d *DownTrack) onBindAndConnectedChange() {
 	if d.writeStopped.Load() {
 		return
@@ -2154,15 +2103,16 @@ func (d *DownTrack) sendSilentFrameOnMuteForOpus() {
 			}
 
 			headerSize := hdr.MarshalSize()
-			d.updateStats(updateStatsParams{
-				packetTime:        mono.UnixNano(),
-				extSequenceNumber: snts[i].extSequenceNumber,
-				extTimestamp:      snts[i].extTimestamp,
-				headerSize:        headerSize,
-				payloadSize:       len(payload),
-				// although this is using empty frames, mark as padding as these are used to trigger Pion OnTrack only
-				isPadding: true,
-			})
+			d.rtpStats.Update(
+				mono.UnixNano(),
+				snts[i].extSequenceNumber,
+				snts[i].extTimestamp,
+				hdr.Marker,
+				headerSize,
+				0,
+				len(payload), // although this is using empty frames, mark as padding as these are used to trigger Pion OnTrack only
+				false,
+			)
 			d.pacer.Enqueue(&pacer.Packet{
 				Header:             hdr,
 				HeaderSize:         headerSize,
@@ -2198,48 +2148,4 @@ func (d *DownTrack) handleRTCPSenderReportData(publisherSRData *livekit.RTCPSend
 	d.rtpStats.MaybeAdjustFirstPacketTime(publisherSRData, tsOffset)
 }
 
-type updateStatsParams struct {
-	packetTime        int64
-	extSequenceNumber uint64
-	extTimestamp      uint64
-	isOutOfOrder      bool
-	headerSize        int
-	payloadSize       int
-	marker            bool
-	isRTX             bool
-	isPadding         bool
-	disableCounter    bool
-}
-
-func (d *DownTrack) updateStats(params updateStatsParams) {
-	if !params.disableCounter {
-		/* STREAM-ALLOCATOR-DATA
-		if params.isRTX {
-			d.bytesRetransmitted.Add(size)
-		} else {
-			d.bytesSent.Add(size)
-		}
-		*/
-	}
-
-	// update RTPStats
-	payloadSize := params.payloadSize
-	paddingSize := params.payloadSize
-	if params.isPadding {
-		payloadSize = 0
-	} else {
-		paddingSize = 0
-	}
-	d.rtpStats.Update(
-		params.packetTime,
-		params.extSequenceNumber,
-		params.extTimestamp,
-		params.marker,
-		params.headerSize,
-		payloadSize,
-		paddingSize,
-		params.isOutOfOrder,
-	)
-}
-
 // -------------------------------------------------------------------------------

pkg/sfu/streamallocator/ratemonitor.go

@@ -1,172 +0,0 @@
-// Copyright 2023 LiveKit, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package streamallocator
-
-import (
-	"fmt"
-	"sync"
-	"time"
-
-	"github.com/livekit/protocol/utils/timeseries"
-)
-
-// ------------------------------------------------
-
-const (
-	rateMonitorWindow  = 10 * time.Second
-	queueMonitorWindow = 2 * time.Second
-)
-
-// ------------------------------------------------
-
-type RateMonitor struct {
-	mu                          sync.Mutex
-	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) {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-
-	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 {
-	_, _, _, _, _, queuingDelay := r.getRates(queueMonitorWindow)
-	return queuingDelay
-}
-
-func (r *RateMonitor) getRates(monitorDuration time.Duration) (totalBitrateEstimate, totalManagedSent, totalManagedRetransmitted, totalUnmanagedSent, totalUnmanagedRetransmitted, queuingDelay float64) {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-
-	threshold := time.Now().Add(-monitorDuration)
-	if !r.bitrateEstimate.HasSamplesAfter(threshold) ||
-		!(r.managedBytesSent.HasSamplesAfter(threshold) ||
-			r.managedBytesRetransmitted.HasSamplesAfter(threshold) ||
-			r.unmanagedBytesSent.HasSamplesAfter(threshold) ||
-			r.unmanagedBytesRetransmitted.HasSamplesAfter(threshold)) {
-		return
-	}
-
-	totalBitrateEstimate = getTimeWeightedSum(r.bitrateEstimate.ReverseIterateSamplesAfter(threshold))
-	totalManagedSent = getRate(r.managedBytesSent.ReverseIterateSamplesAfter(threshold)) * 8
-	totalManagedRetransmitted = getRate(r.managedBytesRetransmitted.ReverseIterateSamplesAfter(threshold)) * 8
-	totalUnmanagedSent = getRate(r.unmanagedBytesSent.ReverseIterateSamplesAfter(threshold)) * 8
-	totalUnmanagedRetransmitted = getRate(r.unmanagedBytesRetransmitted.ReverseIterateSamplesAfter(threshold)) * 8
-	totalBits := totalManagedSent + totalManagedRetransmitted + totalUnmanagedSent + totalUnmanagedRetransmitted
-
-	if totalBits > totalBitrateEstimate {
-		latestBitrateEstimate := r.bitrateEstimate.Back().Value
-		excessBits := totalBits - totalBitrateEstimate
-		queuingDelay = excessBits / float64(latestBitrateEstimate)
-	}
-	return
-}
-
-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.mu.Lock()
-	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),
-	)
-	r.mu.Unlock()
-}
-
-func (r *RateMonitor) GetHistory() []string {
-	return r.history
-}
-
-// ------------------------------------------------
-
-func getTimeWeightedSum[T int64 | uint32](it timeseries.ReverseIterator[T]) float64 {
-	sum := 0.0
-	next := time.Now()
-	for it.Next() {
-		diff := next.Sub(it.Value().At).Seconds()
-		sum += diff * float64(it.Value().Value)
-		next = it.Value().At
-	}
-	return sum
-}
-
-func getRate[T int64 | uint32](it timeseries.ReverseIterator[T]) float64 {
-	var sum float64
-	var first, last time.Time
-	for it.Next() {
-		if last.IsZero() {
-			last = it.Value().At
-		}
-		first = it.Value().At
-		sum += float64(it.Value().Value)
-	}
-
-	if duration := last.Sub(first); duration > 0 {
-		return sum / duration.Seconds()
-	}
-	return 0
-}

pkg/sfu/streamallocator/streamallocator.go

@@ -90,8 +90,6 @@ const (
 	streamAllocatorSignalResume
 	streamAllocatorSignalSetAllowPause
 	streamAllocatorSignalSetChannelCapacity
-	// STREAM-ALLOCATOR-DATA streamAllocatorSignalNACK
-	// STREAM-ALLOCATOR-DATA streamAllocatorSignalRTCPReceiverReport
 	streamAllocatorSignalCongestionStateChange
 )
 
@@ -119,12 +117,6 @@ func (s streamAllocatorSignal) String() string {
 		return "SET_ALLOW_PAUSE"
 	case streamAllocatorSignalSetChannelCapacity:
 		return "SET_CHANNEL_CAPACITY"
-		/* STREAM-ALLOCATOR-DATA
-		case streamAllocatorSignalNACK:
-			return "NACK"
-		case streamAllocatorSignalRTCPReceiverReport:
-			return "RTCP_RECEIVER_REPORT"
-		*/
 	case streamAllocatorSignalCongestionStateChange:
 		return "CONGESTION_STATE_CHANGE"
 	default:
@@ -199,8 +191,6 @@ type StreamAllocator struct {
 
 	prober *ccutils.Prober
 
-	// STREAM-ALLOCATOR-DATA rateMonitor     *RateMonitor
-
 	videoTracksMu        sync.RWMutex
 	videoTracks          map[livekit.TrackID]*Track
 	isAllocateAllPending bool
@@ -219,10 +209,9 @@ type StreamAllocator struct {
 
 func NewStreamAllocator(params StreamAllocatorParams, enabled bool, allowPause bool) *StreamAllocator {
 	s := &StreamAllocator{
-		params:     params,
-		enabled:    enabled,
-		allowPause: allowPause,
-		// STREAM-ALLOCATOR-DATA rateMonitor: NewRateMonitor(),
+		params:               params,
+		enabled:              enabled,
+		allowPause:           allowPause,
 		videoTracks:          make(map[livekit.TrackID]*Track),
 		state:                streamAllocatorStateStable,
 		activeProbeClusterId: ccutils.ProbeClusterIdInvalid,
@@ -508,27 +497,6 @@ func (s *StreamAllocator) OnResume(downTrack *sfu.DownTrack) {
 	})
 }
 
-/* STREAM-ALLOCATOR-DATA
-// 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 probe cluster changes
 func (s *StreamAllocator) OnProbeClusterSwitch(pci ccutils.ProbeClusterInfo) {
 	s.postEvent(Event{
@@ -639,12 +607,6 @@ func (s *StreamAllocator) postEvent(event Event) {
 			event.handleSignalSetAllowPause(event)
 		case streamAllocatorSignalSetChannelCapacity:
 			event.handleSignalSetChannelCapacity(event)
-			/* STREAM-ALLOCATOR-DATA
-			case streamAllocatorSignalNACK:
-				event.s.handleSignalNACK(event)
-			case streamAllocatorSignalRTCPReceiverReport:
-				event.s.handleSignalRTCPReceiverReport(event)
-			*/
 		case streamAllocatorSignalCongestionStateChange:
 			s.handleSignalCongestionStateChange(event)
 		}
@@ -729,11 +691,6 @@ func (s *StreamAllocator) handleSignalPeriodicPing(Event) {
 			}
 		}
 	}
-
-	/* STREAM-ALLOCATOR-DATA
-	s.monitorRate(s.committedChannelCapacity)
-	s.updateTracksHistory()
-	*/
 }
 
 func (s *StreamAllocator) handleSignalProbeClusterSwitch(event Event) {
@@ -805,32 +762,6 @@ func (s *StreamAllocator) handleSignalSetChannelCapacity(event Event) {
 	}
 }
 
-/* STREAM-ALLOCATOR-DATA
-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) handleSignalCongestionStateChange(event Event) {
 	cscd := event.Data.(congestionStateChangeData)
 	if cscd.congestionState != bwe.CongestionStateNone {
@@ -873,14 +804,6 @@ func (s *StreamAllocator) handleSignalCongestionStateChange(event Event) {
 				"new(bps)", cscd.estimatedAvailableChannelCapacity,
 				"expectedUsage(bps)", s.getExpectedBandwidthUsage(),
 			)
-			/* STREAM-ALLOCATOR-DATA
-			s.params.Logger.Debugw(
-				fmt.Sprintf("stream allocator: channel congestion detected, %s channel capacity: experimental", action),
-				"rateHistory", s.rateMonitor.GetHistory(),
-				"expectedQueuing", s.rateMonitor.GetQueuingGuess(),
-				"trackHistory", s.getTracksHistory(),
-			)
-			*/
 			s.committedChannelCapacity = cscd.estimatedAvailableChannelCapacity
 
 			s.allocateAllTracks()
@@ -1427,51 +1350,6 @@ func (s *StreamAllocator) getMaxDistanceSortedDeficient() MaxDistanceSorter {
 	return maxDistanceSorter
 }
 
-/* STREAM-ALLOCATOR-DATA
-// 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
-}
-*/
-
 // ------------------------------------------------
 
 func updateStreamStateChange(track *Track, allocation sfu.VideoAllocation, update *StreamStateUpdate) {

pkg/sfu/streamallocator/track.go

@@ -35,21 +35,6 @@ type Track struct {
 	totalPackets       uint32
 	totalRepeatedNacks uint32
 
-	/* STREAM-ALLOCATOR-DATA
-	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
 
 	streamState StreamState
@@ -68,11 +53,6 @@ func NewTrack(
 		isSimulcast: isSimulcast,
 		publisherID: publisherID,
 		logger:      logger,
-		/* STREAM-ALLOCATOR-DATA
-		nackInfos:             make(map[uint16]sfu.NackInfo),
-		nackHistory:           make([]string, 0, 10),
-		receiverReportHistory: make([]string, 0, 10),
-		*/
 		streamState: StreamStateInactive,
 	}
 	t.SetPriority(0)
@@ -218,152 +198,6 @@ func (t *Track) GetNackDelta() (uint32, uint32) {
 	return packetDelta, nackDelta
 }
 
-/* STREAM-ALLOCATOR-DATA
-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