From 8ed193b231aca1b145c1cba0d0bfb148c18e326a Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 25 Apr 2023 07:31:22 +0530 Subject: [PATCH] `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 --- go.mod | 4 +- go.sum | 8 +- pkg/sfu/downtrack.go | 60 +++++-- pkg/sfu/streamallocator/channelobserver.go | 6 +- pkg/sfu/streamallocator/nacktracker.go | 22 ++- pkg/sfu/streamallocator/prober.go | 127 +++++++++++---- pkg/sfu/streamallocator/ratemonitor.go | 158 ++++++++++++++++++ pkg/sfu/streamallocator/streamallocator.go | 119 +++++++++++++- pkg/sfu/streamallocator/track.go | 178 ++++++++++++++++++++- 9 files changed, 625 insertions(+), 57 deletions(-) create mode 100644 pkg/sfu/streamallocator/ratemonitor.go diff --git a/go.mod b/go.mod index 83e9abbc2..d8b16b22a 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index dbb146402..d36f3ccad 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 87498d987..538b524e1 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -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 { diff --git a/pkg/sfu/streamallocator/channelobserver.go b/pkg/sfu/streamallocator/channelobserver.go index 25dfe7cac..1960b6c2d 100644 --- a/pkg/sfu/streamallocator/channelobserver.go +++ b/pkg/sfu/streamallocator/channelobserver.go @@ -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() diff --git a/pkg/sfu/streamallocator/nacktracker.go b/pkg/sfu/streamallocator/nacktracker.go index cc91bfe98..74104d625 100644 --- a/pkg/sfu/streamallocator/nacktracker.go +++ b/pkg/sfu/streamallocator/nacktracker.go @@ -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()) } // ------------------------------------------------ diff --git a/pkg/sfu/streamallocator/prober.go b/pkg/sfu/streamallocator/prober.go index dd18401bd..a1b1a3205 100644 --- a/pkg/sfu/streamallocator/prober.go +++ b/pkg/sfu/streamallocator/prober.go @@ -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()) } + +// ---------------------------------------------------------------------- diff --git a/pkg/sfu/streamallocator/ratemonitor.go b/pkg/sfu/streamallocator/ratemonitor.go new file mode 100644 index 000000000..06445ea44 --- /dev/null +++ b/pkg/sfu/streamallocator/ratemonitor.go @@ -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() +} diff --git a/pkg/sfu/streamallocator/streamallocator.go b/pkg/sfu/streamallocator/streamallocator.go index cb73976d3..b680ae817 100644 --- a/pkg/sfu/streamallocator/streamallocator.go +++ b/pkg/sfu/streamallocator/streamallocator.go @@ -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 +} + // ------------------------------------------------ diff --git a/pkg/sfu/streamallocator/track.go b/pkg/sfu/streamallocator/track.go index e309baec6..d3aedcef4 100644 --- a/pkg/sfu/streamallocator/track.go +++ b/pkg/sfu/streamallocator/track.go @@ -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