From c172ba13e6879735e36e4fcdc36afd7d68d0d65d Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sun, 8 Dec 2024 13:00:58 +0530 Subject: [PATCH] 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 | 6 - pkg/sfu/bwe/remotebwe/nack_tracker.go | 22 -- pkg/sfu/bwe/remotebwe/remote_bwe.go | 7 - .../bwe/sendsidebwe/congestion_detector.go | 1 - pkg/sfu/bwe/sendsidebwe/packet_tracker.go | 1 - pkg/sfu/bwe/sendsidebwe/twcc_feedback.go | 2 - pkg/sfu/downtrack.go | 196 +++++------------- pkg/sfu/streamallocator/ratemonitor.go | 172 --------------- pkg/sfu/streamallocator/streamallocator.go | 128 +----------- pkg/sfu/streamallocator/track.go | 166 --------------- 10 files changed, 54 insertions(+), 647 deletions(-) delete mode 100644 pkg/sfu/streamallocator/ratemonitor.go diff --git a/pkg/sfu/bwe/remotebwe/channel_observer.go b/pkg/sfu/bwe/remotebwe/channel_observer.go index 52b8cc1a8..24155cf11 100644 --- a/pkg/sfu/bwe/remotebwe/channel_observer.go +++ b/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() diff --git a/pkg/sfu/bwe/remotebwe/nack_tracker.go b/pkg/sfu/bwe/remotebwe/nack_tracker.go index d2d5b8d6d..106115f80 100644 --- a/pkg/sfu/bwe/remotebwe/nack_tracker.go +++ b/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()) -} -*/ - // ------------------------------------------------ diff --git a/pkg/sfu/bwe/remotebwe/remote_bwe.go b/pkg/sfu/bwe/remotebwe/remote_bwe.go index 05ab3d781..d6316929a 100644 --- a/pkg/sfu/bwe/remotebwe/remote_bwe.go +++ b/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 diff --git a/pkg/sfu/bwe/sendsidebwe/congestion_detector.go b/pkg/sfu/bwe/sendsidebwe/congestion_detector.go index 4d2e41a05..a894313d8 100644 --- a/pkg/sfu/bwe/sendsidebwe/congestion_detector.go +++ b/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, diff --git a/pkg/sfu/bwe/sendsidebwe/packet_tracker.go b/pkg/sfu/bwe/sendsidebwe/packet_tracker.go index 76c0e5503..17acbd482 100644 --- a/pkg/sfu/bwe/sendsidebwe/packet_tracker.go +++ b/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++ diff --git a/pkg/sfu/bwe/sendsidebwe/twcc_feedback.go b/pkg/sfu/bwe/sendsidebwe/twcc_feedback.go index 4f8761f07..f6e10e1b8 100644 --- a/pkg/sfu/bwe/sendsidebwe/twcc_feedback.go +++ b/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 { diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index fb8c7212c..1d03122eb 100644 --- a/pkg/sfu/downtrack.go +++ b/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, - ) -} - // ------------------------------------------------------------------------------- diff --git a/pkg/sfu/streamallocator/ratemonitor.go b/pkg/sfu/streamallocator/ratemonitor.go deleted file mode 100644 index f2275822f..000000000 --- a/pkg/sfu/streamallocator/ratemonitor.go +++ /dev/null @@ -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 -} diff --git a/pkg/sfu/streamallocator/streamallocator.go b/pkg/sfu/streamallocator/streamallocator.go index 0c8fc0aa3..4afa9c6e6 100644 --- a/pkg/sfu/streamallocator/streamallocator.go +++ b/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) { diff --git a/pkg/sfu/streamallocator/track.go b/pkg/sfu/streamallocator/track.go index 77b560109..6f2af870b 100644 --- a/pkg/sfu/streamallocator/track.go +++ b/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