From 641858832a892979ec34966dff9230dcc7b82802 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Mon, 21 Mar 2022 14:53:31 +0530 Subject: [PATCH] Address edge case stream allocation (#544) * Handle an edge in layer lock. A very edge case - Available layer: [0, 1, 2], but bitrate is not yet available. We set it to layer 2 awaiting measurement. - Measurement for layers 0 and 1 come through. - Still no key frame for layer 2. - Finalize layers runs and sees that bitrate is available for 0 and 1. It finalizes layer 1. - Layer 1 key frame comes (because we asked key frame of layer 2, publisher sends key frame for all layers). Locks to layer 1. - No more events happen to switch to layer 2. Changes ------- - Move bit rate measurement to StreamTrackerManager. Allows re-use in relay. - Make bit rate availability (from zero -> non-zero) an event and let it flow through the stream allocation flow so that we always have an event when bit rate measurement becomes available. This gets rid of finalization which I was unhappy with it anyway as it was polling every second. - Removing REMB stuff from buffer. We do not use it. It is incorrect anyway. REMB should be ay peer connection level. * Fix test * fix test * Simplify allocate * Simplify/clean up --- pkg/sfu/buffer/buffer.go | 140 +++------------ pkg/sfu/buffer/rtpstats.go | 15 +- pkg/sfu/downtrack.go | 26 ++- pkg/sfu/forwarder.go | 290 ++++++++------------------------ pkg/sfu/forwarder_test.go | 241 +++----------------------- pkg/sfu/receiver.go | 31 ++-- pkg/sfu/streamallocator.go | 48 +++--- pkg/sfu/streamtracker.go | 200 ++++++++++++++++------ pkg/sfu/streamtracker_test.go | 30 ++-- pkg/sfu/streamtrackermanager.go | 43 ++++- 10 files changed, 390 insertions(+), 674 deletions(-) diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 34b5df90b..0eaa1009e 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -28,12 +28,14 @@ type pendingPacket struct { } type ExtPacket struct { - Head bool - Arrival int64 - Packet *rtp.Packet - Payload interface{} - KeyFrame bool - RawPacket []byte + Head bool + Arrival int64 + Packet *rtp.Packet + Payload interface{} + KeyFrame bool + RawPacket []byte + SpatialLayer int32 + TemporalLayer int32 } // Buffer contains all packets @@ -49,7 +51,6 @@ type Buffer struct { closeOnce sync.Once mediaSSRC uint32 clockRate uint32 - maxBitrate int64 lastReport int64 twccExt uint8 audioExt uint8 @@ -66,14 +67,11 @@ type Buffer struct { latestTSForAudioLevel uint32 lastPacketRead int - bitrate atomic.Value - bitrateHelper [4]int64 pliThrottle int64 rtpStats *RTPStats rrSnapshotId uint32 - rembSnapshotId uint32 connectionQualitySnapshotId uint32 lastFractionLostToReport uint8 // Last fraction lost from subscribers, should report to publisher; Audio only @@ -106,7 +104,6 @@ func NewBuffer(ssrc uint32, vp, ap *sync.Pool) *Buffer { logger: logger, callbacksQueue: utils.NewOpsQueue(logger), } - b.bitrate.Store(make([]int64, len(b.bitrateHelper))) b.extPackets.SetMinCapacity(7) return b } @@ -127,13 +124,12 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili ClockRate: codec.ClockRate, }) b.rrSnapshotId = b.rtpStats.NewSnapshotId() - b.rembSnapshotId = b.rtpStats.NewSnapshotId() b.connectionQualitySnapshotId = b.rtpStats.NewSnapshotId() b.callbacksQueue.Start() b.clockRate = codec.ClockRate - b.maxBitrate = int64(o.MaxBitRate) + b.lastReport = time.Now().UnixNano() b.mime = strings.ToLower(codec.MimeType) switch { @@ -159,6 +155,7 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili switch fb.Type { case webrtc.TypeRTCPFBGoogREMB: b.logger.Debugw("Setting feedback", "type", webrtc.TypeRTCPFBGoogREMB) + b.logger.Warnw("REMB not supported, RTCP feedback will not be generated", nil) b.remb = true case webrtc.TypeRTCPFBTransportCC: b.logger.Debugw("Setting feedback", "type", webrtc.TypeRTCPFBTransportCC) @@ -357,16 +354,12 @@ func (b *Buffer) calc(pkt []byte, arrivalTime int64) { flowState := b.updateStreamState(&p, arrivalTime) b.processHeaderExtensions(&p, arrivalTime) - ep, temporalLayer := b.getExtPacket(pb, &p, arrivalTime, flowState.IsHighestSN) + ep := b.getExtPacket(pb, &p, arrivalTime, flowState.IsHighestSN) if ep == nil { return } b.extPackets.PushBack(ep) - if temporalLayer >= 0 { - b.bitrateHelper[temporalLayer] += int64(len(pkt)) - } - b.doNACKs() b.doReports(arrivalTime) @@ -424,30 +417,32 @@ func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime int64) { } } -func (b *Buffer) getExtPacket(rawPacket []byte, rtpPacket *rtp.Packet, arrivalTime int64, isHighestSN bool) (*ExtPacket, int32) { +func (b *Buffer) getExtPacket(rawPacket []byte, rtpPacket *rtp.Packet, arrivalTime int64, isHighestSN bool) *ExtPacket { ep := &ExtPacket{ - Head: isHighestSN, - Packet: rtpPacket, - Arrival: arrivalTime, - RawPacket: rawPacket, + Head: isHighestSN, + Packet: rtpPacket, + Arrival: arrivalTime, + RawPacket: rawPacket, + SpatialLayer: -1, + TemporalLayer: -1, } if len(rtpPacket.Payload) == 0 { // padding only packet, nothing else to do - return ep, -1 + return ep } - temporalLayer := int32(0) + ep.TemporalLayer = 0 switch b.mime { case "video/vp8": vp8Packet := VP8{} if err := vp8Packet.Unmarshal(rtpPacket.Payload); err != nil { b.logger.Warnw("could not unmarshal VP8 packet", err) - return nil, -1 + return nil } ep.Payload = vp8Packet ep.KeyFrame = vp8Packet.IsKeyFrame - temporalLayer = int32(vp8Packet.TID) + ep.TemporalLayer = int32(vp8Packet.TID) case "video/h264": ep.KeyFrame = IsH264Keyframe(rtpPacket.Payload) } @@ -458,7 +453,7 @@ func (b *Buffer) getExtPacket(rawPacket []byte, rtpPacket *rtp.Packet, arrivalTi } } - return ep, temporalLayer + return ep } func (b *Buffer) doNACKs() { @@ -484,23 +479,6 @@ func (b *Buffer) doReports(arrivalTime int64) { b.lastReport = arrivalTime - // - // As this happens in the data path, if there are no packets received - // in an interval, the bitrate will be stuck with the old value. - // GetBitrate() method in sfu.Receiver uses the availableLayers - // set by stream tracker to report 0 bitrate if a layer is not available. - // - bitrates, ok := b.bitrate.Load().([]int64) - if !ok { - bitrates = make([]int64, len(b.bitrateHelper)) - } - for i := 0; i < len(b.bitrateHelper); i++ { - br := (8 * b.bitrateHelper[i] * int64(ReportDelta)) / timeDiff - bitrates[i] = br - b.bitrateHelper[i] = 0 - } - b.bitrate.Store(bitrates) - // RTCP reports pkts := b.getRTCP() if pkts != nil { @@ -525,40 +503,6 @@ func (b *Buffer) buildNACKPacket() ([]rtcp.Packet, int) { return nil, 0 } -func (b *Buffer) buildREMBPacket() *rtcp.ReceiverEstimatedMaximumBitrate { - if b.rtpStats == nil { - return nil - } - - br := b.Bitrate() - s := b.rtpStats.SnapshotInfo(b.rembSnapshotId) - if s == nil { - return nil - } - - lostRate := float32(0.0) - if s.PacketsExpected != 0 { - lostRate = float32(s.PacketsLost) / float32(s.PacketsExpected) - } - if lostRate < 0.02 { - br = int64(float64(br)*1.09) + 2000 - } - if lostRate > 0.1 { - br = int64(float64(br) * float64(1-0.5*lostRate)) - } - if br > b.maxBitrate { - br = b.maxBitrate - } - if br < 100000 { - br = 100000 - } - - return &rtcp.ReceiverEstimatedMaximumBitrate{ - Bitrate: float32(br), - SSRCs: []uint32{b.mediaSSRC}, - } -} - func (b *Buffer) buildReceptionReport() *rtcp.ReceptionReport { if b.rtpStats == nil { return nil @@ -592,10 +536,6 @@ func (b *Buffer) getRTCP() []rtcp.Packet { }) } - if b.remb && !b.twcc { - pkts = append(pkts, b.buildREMBPacket()) - } - return pkts } @@ -608,40 +548,6 @@ func (b *Buffer) GetPacket(buff []byte, sn uint16) (int, error) { return b.bucket.GetPacket(buff, sn) } -// Bitrate returns the current publisher stream bitrate. -func (b *Buffer) Bitrate() int64 { - bitrates, ok := b.bitrate.Load().([]int64) - bitrate := int64(0) - if ok { - for _, b := range bitrates { - bitrate += b - } - } - return bitrate -} - -// BitrateTemporalCumulative returns the current publisher stream bitrate temporal layer accumulated with lower temporal layers. -func (b *Buffer) BitrateTemporalCumulative() []int64 { - bitrates, ok := b.bitrate.Load().([]int64) - if !ok { - return make([]int64, len(b.bitrateHelper)) - } - - // copy and process - brs := make([]int64, len(bitrates)) - copy(brs, bitrates) - - for i := len(brs) - 1; i >= 1; i-- { - if brs[i] != 0 { - for j := i - 1; j >= 0; j-- { - brs[i] += brs[j] - } - } - } - - return brs -} - func (b *Buffer) OnTransportWideCC(fn func(sn uint16, timeNS int64, marker bool)) { b.feedbackTWCC = fn } diff --git a/pkg/sfu/buffer/rtpstats.go b/pkg/sfu/buffer/rtpstats.go index b99734f75..67df98436 100644 --- a/pkg/sfu/buffer/rtpstats.go +++ b/pkg/sfu/buffer/rtpstats.go @@ -157,6 +157,7 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa return } + first := false if !r.initialized { r.initialized = true @@ -168,6 +169,8 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa r.extStartSN = uint32(rtph.SequenceNumber) r.cycles = 0 + + first = true } pktSize := uint64(rtph.MarshalSize() + payloadSize + paddingSize) @@ -232,7 +235,7 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa } r.packetsLost += uint32(diff - 1) - if rtph.SequenceNumber < r.highestSN { + if rtph.SequenceNumber < r.highestSN && !first { r.cycles++ } r.highestSN = rtph.SequenceNumber @@ -890,9 +893,11 @@ func (r *RTPStats) getAndResetSnapshot(snapshotId uint32) *Snapshot { snapshot := r.snapshots[snapshotId] if snapshot == nil { snapshot = &Snapshot{ - extStartSN: r.extStartSN, - maxJitter: 0.0, - maxRtt: 0, + extStartSN: r.extStartSN, + maxJitter: 0.0, + isJitterOverridden: false, + maxJitterOverridden: 0.0, + maxRtt: 0, } r.snapshots[snapshotId] = snapshot } @@ -901,6 +906,8 @@ func (r *RTPStats) getAndResetSnapshot(snapshotId uint32) *Snapshot { snapshot.extStartSN = r.getExtHighestSN() + 1 snapshot.maxJitter = 0.0 + snapshot.isJitterOverridden = false + snapshot.maxJitterOverridden = 0.0 snapshot.maxRtt = 0 return &toReturn diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index c5b5c5274..eb1cb2fc1 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -26,6 +26,7 @@ import ( // TrackSender defines an interface send media to remote peer type TrackSender interface { UpTrackLayersChange(availableLayers []int32) + UpTrackBitrateAvailabilityChange() WriteRTP(p *buffer.ExtPacket, layer int32) error Close() // ID is the globally unique identifier for this Track. @@ -123,6 +124,9 @@ type DownTrack struct { // simulcast layer availability change callback onAvailableLayersChanged func(dt *DownTrack) + // layer bitrate availability change callback + onBitrateAvailabilityChanged func(dt *DownTrack) + // subscription change callback onSubscriptionChanged func(dt *DownTrack) @@ -643,6 +647,14 @@ func (d *DownTrack) UpTrackLayersChange(availableLayers []int32) { } } +func (d *DownTrack) UpTrackBitrateAvailabilityChange() { + if d.onBitrateAvailabilityChanged != nil { + d.callbacksQueue.Enqueue(func() { + d.onBitrateAvailabilityChanged(d) + }) + } +} + // OnCloseHandler method to be called on remote tracked removed func (d *DownTrack) OnCloseHandler(fn func()) { d.onCloseHandler = fn @@ -671,6 +683,10 @@ func (d *DownTrack) OnAvailableLayersChanged(fn func(dt *DownTrack)) { d.onAvailableLayersChanged = fn } +func (d *DownTrack) OnBitrateAvailabilityChanged(fn func(dt *DownTrack)) { + d.onBitrateAvailabilityChanged = fn +} + func (d *DownTrack) OnSubscriptionChanged(fn func(dt *DownTrack)) { d.onSubscriptionChanged = fn } @@ -711,9 +727,9 @@ func (d *DownTrack) DistanceToDesired() int32 { return d.forwarder.DistanceToDesired() } -func (d *DownTrack) Allocate(availableChannelCapacity int64, allowPause bool) VideoAllocation { - allocation := d.forwarder.Allocate(availableChannelCapacity, allowPause, d.receiver.GetBitrateTemporalCumulative()) - d.logger.Debugw("stream: allocation", "channel", availableChannelCapacity, "allocation", allocation) +func (d *DownTrack) AllocateOptimal() VideoAllocation { + allocation := d.forwarder.AllocateOptimal(d.receiver.GetBitrateTemporalCumulative()) + d.logger.Debugw("stream: allocation optimal available", "allocation", allocation) d.maybeStartKeyFrameRequester() return allocation } @@ -745,10 +761,6 @@ func (d *DownTrack) ProvisionalAllocateCommit() VideoAllocation { return allocation } -func (d *DownTrack) FinalizeAllocate() VideoAllocation { - return d.forwarder.FinalizeAllocate(d.receiver.GetBitrateTemporalCumulative()) -} - func (d *DownTrack) AllocateNextHigher(availableChannelCapacity int64) (VideoAllocation, bool) { allocation, available := d.forwarder.AllocateNextHigher(availableChannelCapacity, d.receiver.GetBitrateTemporalCumulative()) d.logger.Debugw("stream: allocation next higher layer", "allocation", allocation, "available", available) diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 6d5a5f8fe..dc41946e0 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -102,9 +102,10 @@ var ( ) type VideoAllocationProvisional struct { - layers VideoLayers - muted bool - bitrates Bitrates + layers VideoLayers + muted bool + bitrates Bitrates + availableLayers []int32 } type VideoTransition struct { @@ -325,6 +326,26 @@ func (f *Forwarder) getOptimalBandwidthNeeded(brs Bitrates) int64 { return 0 } +func (f *Forwarder) bitrateAvailable(brs Bitrates, availableLayers []int32) bool { + neededLayers := 0 + var bitrateAvailableLayers []int32 + for _, layer := range f.availableLayers { + if layer > f.maxLayers.spatial { + continue + } + + neededLayers++ + for t := f.maxLayers.temporal; t >= 0; t-- { + if brs[layer][t] != 0 { + bitrateAvailableLayers = append(bitrateAvailableLayers, layer) + break + } + } + } + + return len(bitrateAvailableLayers) == neededLayers +} + func (f *Forwarder) getDistanceToDesired(brs Bitrates, targetLayers VideoLayers) int32 { if f.muted { return 0 @@ -378,7 +399,7 @@ func (f *Forwarder) DistanceToDesired() int32 { return f.lastAllocation.distanceToDesired } -func (f *Forwarder) Allocate(availableChannelCapacity int64, allowPause bool, brs Bitrates) VideoAllocation { +func (f *Forwarder) AllocateOptimal(brs Bitrates) VideoAllocation { f.lock.Lock() defer f.lock.Unlock() @@ -386,8 +407,6 @@ func (f *Forwarder) Allocate(availableChannelCapacity int64, allowPause bool, br return f.lastAllocation } - optimalBandwidthNeeded := f.getOptimalBandwidthNeeded(brs) - state := VideoAllocationStateNone change := VideoStreamingChangeNone bandwidthRequested := int64(0) @@ -396,110 +415,47 @@ func (f *Forwarder) Allocate(availableChannelCapacity int64, allowPause bool, br switch { case f.muted: state = VideoAllocationStateMuted - case optimalBandwidthNeeded == 0: - if len(f.availableLayers) == 0 { - // feed is dry - state = VideoAllocationStateFeedDry - } else { - // feed bitrate is not yet calculated - state = VideoAllocationStateAwaitingMeasurement + case len(f.availableLayers) == 0: + // feed is dry + state = VideoAllocationStateFeedDry + case !f.bitrateAvailable(brs, f.availableLayers): + // feed bitrate not yet calculated for all available layers + state = VideoAllocationStateAwaitingMeasurement - if availableChannelCapacity == ChannelCapacityInfinity { - // - // Channel capacity allows a free pass. - // So, resume with the highest layer available <= max subscribed layer - // If already resumed, move allocation to the highest available layer <= max subscribed layer - // - targetLayers.spatial = int32(math.Min(float64(f.maxLayers.spatial), float64(f.availableLayers[len(f.availableLayers)-1]))) - targetLayers.temporal = int32(math.Max(0, float64(f.maxLayers.temporal))) + // + // Resume with the highest layer available <= max subscribed layer + // If already resumed, move allocation to the highest available layer <= max subscribed layer + // + targetLayers.spatial = int32(math.Min(float64(f.maxLayers.spatial), float64(f.availableLayers[len(f.availableLayers)-1]))) + targetLayers.temporal = int32(math.Max(0, float64(f.maxLayers.temporal))) - if f.targetLayers == InvalidLayers { - change = VideoStreamingChangeResuming - } - } else { - if !allowPause { - // when pause is disallowed, pick the lowest available - targetLayers.spatial = int32(math.Min(float64(f.maxLayers.spatial), float64(f.availableLayers[0]))) - targetLayers.temporal = int32(math.Min(0, float64(f.maxLayers.temporal))) - - if f.targetLayers == InvalidLayers { - change = VideoStreamingChangeResuming - } - } else { - // disable forwarding as it is not known how big this stream is - // and if it will fit in the available channel capacity - state = VideoAllocationStateDeficient - - if f.targetLayers != InvalidLayers { - change = VideoStreamingChangePausing - } - } - } + if f.targetLayers == InvalidLayers { + change = VideoStreamingChangeResuming } default: - // allocate best layer that fits + // allocate best layer available for s := f.maxLayers.spatial; s >= 0; s-- { for t := f.maxLayers.temporal; t >= 0; t-- { if brs[s][t] == 0 { continue } - if brs[s][t] <= availableChannelCapacity { - targetLayers = VideoLayers{ - spatial: s, - temporal: t, - } - - bandwidthRequested = brs[s][t] - if bandwidthRequested == optimalBandwidthNeeded { - state = VideoAllocationStateOptimal - } else { - state = VideoAllocationStateDeficient - } - - if f.targetLayers == InvalidLayers { - change = VideoStreamingChangeResuming - } - break + targetLayers = VideoLayers{ + spatial: s, + temporal: t, + } + + bandwidthRequested = brs[s][t] + state = VideoAllocationStateOptimal + + if f.targetLayers == InvalidLayers { + change = VideoStreamingChangeResuming } - } - if bandwidthRequested != 0 { break } - } - if bandwidthRequested == 0 { - state = VideoAllocationStateDeficient - - if !allowPause { - // find the lowest layer to prevent pausing - for s := int32(0); s <= f.maxLayers.spatial; s++ { - for t := int32(0); t <= f.maxLayers.temporal; t++ { - if brs[s][t] == 0 { - continue - } - - targetLayers = VideoLayers{ - spatial: s, - temporal: t, - } - - bandwidthRequested = brs[s][t] - - if f.targetLayers == InvalidLayers { - change = VideoStreamingChangeResuming - } - break - } - - if bandwidthRequested != 0 { - break - } - } - } else { - if f.targetLayers != InvalidLayers { - change = VideoStreamingChangePausing - } + if bandwidthRequested != 0 { + break } } } @@ -527,9 +483,10 @@ func (f *Forwarder) ProvisionalAllocatePrepare(bitrates Bitrates) { defer f.lock.Unlock() f.provisional = &VideoAllocationProvisional{ - layers: InvalidLayers, - muted: f.muted, - bitrates: bitrates, + layers: InvalidLayers, + muted: f.muted, + bitrates: bitrates, + availableLayers: f.availableLayers, } } @@ -586,7 +543,7 @@ func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition() VideoTransitio // 4. If not currently streaming, find the minimum layers that can unpause the stream. // // To summarize, co-operative streaming means - // - Try to keep tracks streaming, i.e. no pauses even if not at optimal layers + // - Try to keep tracks streaming, i.e. no pauses at the expense of some streams not being at optimal layers // - Do not make an upgrade as it could affect other tracks // f.lock.Lock() @@ -645,7 +602,8 @@ func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition() VideoTransitio } // currently not streaming, find minimal - // NOTE: a layer in feed could have paused and there could be other options than going back to minimal, but the cooperative scheme knocks things back to minimal + // NOTE: a layer in feed could have paused and there could be other options than going back to minimal, + // but the cooperative scheme knocks things back to minimal minimalLayers := InvalidLayers bandwidthRequired := int64(0) for s := int32(0); s <= f.maxLayers.spatial; s++ { @@ -703,7 +661,7 @@ func (f *Forwarder) ProvisionalAllocateGetBestWeightedTransition() VideoTransiti } } - maxReachableLayerTemporal := int32(-1) + maxReachableLayerTemporal := InvalidLayerTemporal for t := f.maxLayers.temporal; t >= 0; t-- { for s := f.maxLayers.spatial; s >= 0; s-- { if f.provisional.bitrates[s][t] != 0 { @@ -711,18 +669,19 @@ func (f *Forwarder) ProvisionalAllocateGetBestWeightedTransition() VideoTransiti break } } - if maxReachableLayerTemporal != -1 { + if maxReachableLayerTemporal != InvalidLayerTemporal { break } } - if maxReachableLayerTemporal == -1 { + if maxReachableLayerTemporal == InvalidLayerTemporal { // feed has gone dry, f.provisional.layers = InvalidLayers return VideoTransition{ from: f.targetLayers, to: InvalidLayers, bandwidthDelta: 0 - f.lastAllocation.bandwidthRequested, + // LK-TODO should this take current bitrate of current target layers? } } @@ -770,8 +729,6 @@ func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation { f.lock.Lock() defer f.lock.Unlock() - optimalBandwidthNeeded := f.getOptimalBandwidthNeeded(f.provisional.bitrates) - state := VideoAllocationStateNone change := VideoStreamingChangeNone bandwidthRequested := int64(0) @@ -779,21 +736,9 @@ func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation { switch { case f.provisional.muted: state = VideoAllocationStateMuted - case optimalBandwidthNeeded == 0: - if len(f.availableLayers) == 0 { - // feed is dry - state = VideoAllocationStateFeedDry - } else { - // feed bitrate is not yet calculated - state = VideoAllocationStateDeficient - - // disable forwarding as it is not known how big this stream is - // and if it will fit in the available channel capacity - - if f.targetLayers != InvalidLayers { - change = VideoStreamingChangePausing - } - } + case len(f.provisional.availableLayers) == 0: + // feed is dry + state = VideoAllocationStateFeedDry case f.provisional.layers == InvalidLayers: state = VideoAllocationStateDeficient @@ -802,7 +747,7 @@ func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation { } default: bandwidthRequested = f.provisional.bitrates[f.provisional.layers.spatial][f.provisional.layers.temporal] - if bandwidthRequested == optimalBandwidthNeeded { + if bandwidthRequested == f.getOptimalBandwidthNeeded(f.provisional.bitrates) { state = VideoAllocationStateOptimal } else { state = VideoAllocationStateDeficient @@ -818,7 +763,7 @@ func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation { change: change, bandwidthRequested: bandwidthRequested, bandwidthDelta: bandwidthRequested - f.lastAllocation.bandwidthRequested, - availableLayers: f.availableLayers, + availableLayers: f.provisional.availableLayers, bitrates: f.provisional.bitrates, targetLayers: f.provisional.layers, distanceToDesired: f.getDistanceToDesired(f.provisional.bitrates, f.provisional.layers), @@ -831,76 +776,6 @@ func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation { return f.lastAllocation } -func (f *Forwarder) FinalizeAllocate(brs Bitrates) VideoAllocation { - f.lock.Lock() - defer f.lock.Unlock() - - if f.lastAllocation.state != VideoAllocationStateAwaitingMeasurement { - f.lastAllocation.change = VideoStreamingChangeNone - return f.lastAllocation - } - - optimalBandwidthNeeded := f.getOptimalBandwidthNeeded(brs) - if optimalBandwidthNeeded == 0 { - if len(f.availableLayers) == 0 { - // feed dry - f.lastAllocation = VideoAllocation{ - state: VideoAllocationStateFeedDry, - change: VideoStreamingChangeNone, - bandwidthRequested: 0, - bandwidthDelta: 0 - f.lastAllocation.bandwidthRequested, - availableLayers: f.availableLayers, - bitrates: brs, - targetLayers: InvalidLayers, - distanceToDesired: f.getDistanceToDesired(brs, InvalidLayers), - } - f.targetLayers = f.lastAllocation.targetLayers - if f.targetLayers == InvalidLayers { - f.resyncLocked() - } - } - - // still awaiting measurement - return f.lastAllocation - } - - // finalize using optimal layer - for s := f.maxLayers.spatial; s >= 0; s-- { - for t := f.maxLayers.temporal; t >= 0; t-- { - bandwidthRequested := brs[s][t] - if bandwidthRequested == 0 { - continue - } - - state := VideoAllocationStateOptimal - if bandwidthRequested != optimalBandwidthNeeded { - state = VideoAllocationStateDeficient - } - - change := VideoStreamingChangeNone - if f.targetLayers == InvalidLayers { - change = VideoStreamingChangeResuming - } - - targetLayers := VideoLayers{spatial: s, temporal: t} - f.lastAllocation = VideoAllocation{ - state: state, - change: change, - bandwidthRequested: bandwidthRequested, - bandwidthDelta: bandwidthRequested - f.lastAllocation.bandwidthRequested, - availableLayers: f.availableLayers, - bitrates: brs, - targetLayers: targetLayers, - distanceToDesired: f.getDistanceToDesired(brs, targetLayers), - } - f.targetLayers = f.lastAllocation.targetLayers - return f.lastAllocation - } - } - - return f.lastAllocation -} - func (f *Forwarder) AllocateNextHigher(availableChannelCapacity int64, brs Bitrates) (VideoAllocation, bool) { f.lock.Lock() defer f.lock.Unlock() @@ -923,11 +798,6 @@ func (f *Forwarder) AllocateNextHigher(availableChannelCapacity int64, brs Bitra } optimalBandwidthNeeded := f.getOptimalBandwidthNeeded(brs) - if optimalBandwidthNeeded == 0 { - // either feed is dry or awaiting measurement, don't hunt for higher - f.lastAllocation.change = VideoStreamingChangeNone - return f.lastAllocation, false - } alreadyAllocated := int64(0) if f.targetLayers != InvalidLayers { @@ -1031,12 +901,6 @@ func (f *Forwarder) GetNextHigherTransition(brs Bitrates) (VideoTransition, bool return VideoTransition{}, false } - optimalBandwidthNeeded := f.getOptimalBandwidthNeeded(brs) - if optimalBandwidthNeeded == 0 { - // either feed is dry or awaiting measurement, don't hunt for higher - return VideoTransition{}, false - } - alreadyAllocated := int64(0) if f.targetLayers != InvalidLayers { alreadyAllocated = brs[f.targetLayers.spatial][f.targetLayers.temporal] @@ -1085,27 +949,17 @@ func (f *Forwarder) Pause(brs Bitrates) VideoAllocation { f.lock.Lock() defer f.lock.Unlock() - optimalBandwidthNeeded := f.getOptimalBandwidthNeeded(brs) - state := VideoAllocationStateNone change := VideoStreamingChangeNone switch { case f.muted: state = VideoAllocationStateMuted - case optimalBandwidthNeeded == 0: - if len(f.availableLayers) == 0 { - // feed is dry - state = VideoAllocationStateFeedDry - } else { - // feed bitrate is not yet calculated - state = VideoAllocationStateDeficient - - if f.targetLayers != InvalidLayers { - change = VideoStreamingChangePausing - } - } + case len(f.availableLayers) == 0: + // feed is dry + state = VideoAllocationStateFeedDry default: + // feed bitrate is not yet calculated or pausing due to lack of bandwidth state = VideoAllocationStateDeficient if f.targetLayers != InvalidLayers { diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 599ed0473..15c94ee63 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -183,7 +183,7 @@ func TestForwarderAllocate(t *testing.T) { targetLayers: InvalidLayers, distanceToDesired: 0, } - result := f.Allocate(ChannelCapacityInfinity, true, bitrates) + result := f.AllocateOptimal(bitrates) require.Equal(t, expectedResult, result) require.Equal(t, expectedResult, f.lastAllocation) @@ -201,7 +201,7 @@ func TestForwarderAllocate(t *testing.T) { targetLayers: InvalidLayers, distanceToDesired: 0, } - result = f.Allocate(ChannelCapacityInfinity, true, emptyBitrates) + result = f.AllocateOptimal(emptyBitrates) require.Equal(t, expectedResult, result) require.Equal(t, expectedResult, f.lastAllocation) @@ -223,51 +223,13 @@ func TestForwarderAllocate(t *testing.T) { targetLayers: expectedTargetLayers, distanceToDesired: 0, } - result = f.Allocate(ChannelCapacityInfinity, true, emptyBitrates) + result = f.AllocateOptimal(emptyBitrates) require.Equal(t, expectedResult, result) require.Equal(t, expectedResult, f.lastAllocation) require.Equal(t, expectedTargetLayers, f.TargetLayers()) require.Equal(t, InvalidLayers, f.CurrentLayers()) - // while awaiting measurement, less than infinite channel capacity should pause the stream when pause is allowed - expectedResult = VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangePausing, - bandwidthRequested: 0, - bandwidthDelta: 0, - availableLayers: []int32{0}, - bitrates: emptyBitrates, - targetLayers: InvalidLayers, - distanceToDesired: 0, - } - result = f.Allocate(ChannelCapacityInfinity-1, true, emptyBitrates) - require.Equal(t, expectedResult, result) - require.Equal(t, expectedResult, f.lastAllocation) - require.Equal(t, InvalidLayers, f.CurrentLayers()) - require.Equal(t, InvalidLayers, f.TargetLayers()) - - // while awaiting measurement, less than infinite channel capacity should not pause the stream when pause is not allowed - expectedTargetLayers = VideoLayers{ - spatial: 0, - temporal: 0, - } - expectedResult = VideoAllocation{ - state: VideoAllocationStateAwaitingMeasurement, - change: VideoStreamingChangeResuming, - bandwidthRequested: 0, - bandwidthDelta: 0, - availableLayers: []int32{0}, - bitrates: emptyBitrates, - targetLayers: expectedTargetLayers, - distanceToDesired: 0, - } - result = f.Allocate(ChannelCapacityInfinity-1, false, emptyBitrates) - require.Equal(t, expectedResult, result) - require.Equal(t, expectedResult, f.lastAllocation) - require.Equal(t, InvalidLayers, f.CurrentLayers()) - require.Equal(t, expectedTargetLayers, f.TargetLayers()) - - // allocate using bitrates and less than infinite channel capacity, but enough for optimal + // allocate using bitrates, allocation should choose optimal expectedTargetLayers = VideoLayers{ spatial: 2, temporal: 1, @@ -282,66 +244,7 @@ func TestForwarderAllocate(t *testing.T) { targetLayers: expectedTargetLayers, distanceToDesired: 0, } - result = f.Allocate(ChannelCapacityInfinity-1, true, bitrates) - require.Equal(t, expectedResult, result) - require.Equal(t, expectedResult, f.lastAllocation) - require.Equal(t, InvalidLayers, f.CurrentLayers()) - require.Equal(t, expectedTargetLayers, f.TargetLayers()) - - // give it a bitrate that is less than optimal - expectedTargetLayers = VideoLayers{ - spatial: 1, - temporal: 3, - } - expectedResult = VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangeNone, - bandwidthRequested: bitrates[1][3], - bandwidthDelta: bitrates[1][3] - bitrates[2][1], - availableLayers: []int32{0}, - bitrates: bitrates, - targetLayers: expectedTargetLayers, - distanceToDesired: 1, - } - result = f.Allocate(bitrates[2][1]-1, true, bitrates) - require.Equal(t, expectedResult, result) - require.Equal(t, expectedResult, f.lastAllocation) - require.Equal(t, InvalidLayers, f.CurrentLayers()) - require.Equal(t, expectedTargetLayers, f.TargetLayers()) - - // give it a bitrate that cannot fit any layer - expectedResult = VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangePausing, - bandwidthRequested: 0, - bandwidthDelta: 0 - bitrates[1][3], - availableLayers: []int32{0}, - bitrates: bitrates, - targetLayers: InvalidLayers, - distanceToDesired: 5, - } - result = f.Allocate(bitrates[0][0]-1, true, bitrates) - require.Equal(t, expectedResult, result) - require.Equal(t, expectedResult, f.lastAllocation) - require.Equal(t, InvalidLayers, f.CurrentLayers()) - require.Equal(t, InvalidLayers, f.TargetLayers()) - - // give it a bitrate that cannot fit any layer, but disallow pause, should allocate the lowest available layer - expectedTargetLayers = VideoLayers{ - spatial: 0, - temporal: 0, - } - expectedResult = VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangeResuming, - bandwidthRequested: bitrates[0][0], - bandwidthDelta: bitrates[0][0], - availableLayers: []int32{0}, - bitrates: bitrates, - targetLayers: expectedTargetLayers, - distanceToDesired: 4, - } - result = f.Allocate(bitrates[0][0]-1, false, bitrates) + result = f.AllocateOptimal(bitrates) require.Equal(t, expectedResult, result) require.Equal(t, expectedResult, f.lastAllocation) require.Equal(t, InvalidLayers, f.CurrentLayers()) @@ -351,12 +254,14 @@ func TestForwarderAllocate(t *testing.T) { func TestForwarderProvisionalAllocate(t *testing.T) { f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) + availableLayers := []int32{0, 1, 2} bitrates := Bitrates{ {1, 2, 3, 4}, {5, 6, 7, 8}, {9, 10, 11, 12}, } + f.availableLayers = availableLayers f.ProvisionalAllocatePrepare(bitrates) usedBitrate := f.ProvisionalAllocate(bitrates[2][3], VideoLayers{spatial: 0, temporal: 0}, true) @@ -385,7 +290,7 @@ func TestForwarderProvisionalAllocate(t *testing.T) { change: VideoStreamingChangeResuming, bandwidthRequested: bitrates[1][2], bandwidthDelta: bitrates[1][2], - availableLayers: nil, + availableLayers: availableLayers, bitrates: bitrates, targetLayers: expectedTargetLayers, distanceToDesired: 5, @@ -411,7 +316,7 @@ func TestForwarderProvisionalAllocate(t *testing.T) { change: VideoStreamingChangeResuming, bandwidthRequested: bitrates[0][0], bandwidthDelta: bitrates[0][0] - bitrates[1][2], - availableLayers: nil, + availableLayers: availableLayers, bitrates: bitrates, targetLayers: expectedTargetLayers, distanceToDesired: 11, @@ -460,12 +365,14 @@ func TestForwarderProvisionalAllocateMute(t *testing.T) { func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) + availableLayers := []int32{0, 1, 2} bitrates := Bitrates{ {1, 2, 3, 4}, {5, 6, 7, 8}, {9, 10, 0, 0}, } + f.availableLayers = availableLayers f.ProvisionalAllocatePrepare(bitrates) // from scratch (InvalidLayers) should give back layer (0, 0) @@ -484,7 +391,7 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { change: VideoStreamingChangeResuming, bandwidthRequested: 1, bandwidthDelta: 1, - availableLayers: nil, + availableLayers: availableLayers, bitrates: bitrates, targetLayers: expectedLayers, distanceToDesired: 9, @@ -513,7 +420,7 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { change: VideoStreamingChangeNone, bandwidthRequested: 10, bandwidthDelta: 0, - availableLayers: nil, + availableLayers: availableLayers, bitrates: bitrates, targetLayers: expectedLayers, distanceToDesired: 0, @@ -572,120 +479,6 @@ func TestForwarderProvisionalAllocateGetBestWeightedTransition(t *testing.T) { require.Equal(t, expectedTransition, transition) } -func TestForwarderFinalizeAllocate(t *testing.T) { - f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) - - bitrates := Bitrates{ - {1, 2, 3, 4}, - {5, 6, 7, 8}, - {9, 10, 11, 12}, - } - // FinalizeAllocate should do nothing unless Forwarder allocation state is VideoAllocationStateAwaitingMeasurement - result := f.FinalizeAllocate(bitrates) - require.Equal(t, VideoAllocationDefault, result) - require.Equal(t, VideoAllocationDefault, f.lastAllocation) - - f.lastAllocation.state = VideoAllocationStateMuted - disable(f) - expectedResult := VideoAllocation{ - state: VideoAllocationStateMuted, - change: VideoStreamingChangeNone, - bandwidthRequested: 0, - bandwidthDelta: 0, - availableLayers: nil, - bitrates: Bitrates{}, - targetLayers: InvalidLayers, - distanceToDesired: 0, - } - result = f.FinalizeAllocate(bitrates) - require.Equal(t, expectedResult, result) - require.Equal(t, expectedResult, f.lastAllocation) - - f.lastAllocation.state = VideoAllocationStateAwaitingMeasurement - disable(f) - expectedTargetLayers := VideoLayers{ - spatial: 2, - temporal: 3, - } - expectedResult = VideoAllocation{ - state: VideoAllocationStateOptimal, - change: VideoStreamingChangeResuming, - bandwidthRequested: bitrates[2][3], - bandwidthDelta: bitrates[2][3], - availableLayers: nil, - bitrates: bitrates, - targetLayers: expectedTargetLayers, - distanceToDesired: 0, - } - result = f.FinalizeAllocate(bitrates) - require.Equal(t, expectedResult, result) - require.Equal(t, expectedResult, f.lastAllocation) - require.Equal(t, InvalidLayers, f.CurrentLayers()) - require.Equal(t, expectedTargetLayers, f.TargetLayers()) - - // no layers available => feed dry - f.lastAllocation.state = VideoAllocationStateAwaitingMeasurement - disable(f) - expectedResult = VideoAllocation{ - state: VideoAllocationStateFeedDry, - change: VideoStreamingChangeNone, - bandwidthRequested: 0, - bandwidthDelta: 0 - bitrates[2][3], - availableLayers: nil, - bitrates: Bitrates{}, - targetLayers: InvalidLayers, - distanceToDesired: 0, - } - result = f.FinalizeAllocate(Bitrates{}) - require.Equal(t, expectedResult, result) - require.Equal(t, expectedResult, f.lastAllocation) - - // layers available, but still awaiting measurement - f.lastAllocation.state = VideoAllocationStateAwaitingMeasurement - disable(f) - f.UpTrackLayersChange([]int32{0, 1}) - expectedResult = VideoAllocation{ - state: VideoAllocationStateAwaitingMeasurement, - change: VideoStreamingChangeNone, - bandwidthRequested: 0, - bandwidthDelta: -12, - availableLayers: nil, - bitrates: Bitrates{}, - targetLayers: InvalidLayers, - distanceToDesired: 0, - } - result = f.FinalizeAllocate(Bitrates{}) - require.Equal(t, expectedResult, result) - require.Equal(t, expectedResult, f.lastAllocation) - - // sparse layers - bitrates = Bitrates{ - {1, 2, 0, 0}, - {5, 0, 0, 6}, - {0, 0, 0, 0}, - } - disable(f) - expectedTargetLayers = VideoLayers{ - spatial: 1, - temporal: 3, - } - expectedResult = VideoAllocation{ - state: VideoAllocationStateOptimal, - change: VideoStreamingChangeResuming, - bandwidthRequested: bitrates[1][3], - bandwidthDelta: bitrates[1][3], - availableLayers: []int32{0, 1}, - bitrates: bitrates, - targetLayers: expectedTargetLayers, - distanceToDesired: 0, - } - result = f.FinalizeAllocate(bitrates) - require.Equal(t, expectedResult, result) - require.Equal(t, expectedResult, f.lastAllocation) - require.Equal(t, InvalidLayers, f.CurrentLayers()) - require.Equal(t, expectedTargetLayers, f.TargetLayers()) -} - func TestForwarderAllocateNextHigher(t *testing.T) { f := newForwarder(testutils.TestOpusCodec, webrtc.RTPCodecTypeAudio) @@ -895,12 +688,14 @@ func TestForwarderAllocateNextHigher(t *testing.T) { func TestForwarderPause(t *testing.T) { f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) + availableLayers := []int32{0, 1, 2} bitrates := Bitrates{ {1, 2, 3, 4}, {5, 6, 7, 8}, {9, 10, 11, 12}, } + f.availableLayers = availableLayers f.ProvisionalAllocatePrepare(bitrates) f.ProvisionalAllocate(bitrates[2][3], VideoLayers{spatial: 0, temporal: 0}, true) // should have set target at (0, 0) @@ -911,7 +706,7 @@ func TestForwarderPause(t *testing.T) { change: VideoStreamingChangePausing, bandwidthRequested: 0, bandwidthDelta: 0 - bitrates[0][0], - availableLayers: nil, + availableLayers: availableLayers, bitrates: bitrates, targetLayers: InvalidLayers, distanceToDesired: 12, @@ -925,12 +720,14 @@ func TestForwarderPause(t *testing.T) { func TestForwarderPauseMute(t *testing.T) { f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) + availableLayers := []int32{0, 1, 2} bitrates := Bitrates{ {1, 2, 3, 4}, {5, 6, 7, 8}, {9, 10, 11, 12}, } + f.availableLayers = availableLayers f.ProvisionalAllocatePrepare(bitrates) f.ProvisionalAllocate(bitrates[2][3], VideoLayers{spatial: 0, temporal: 0}, true) // should have set target at (0, 0) @@ -942,7 +739,7 @@ func TestForwarderPauseMute(t *testing.T) { change: VideoStreamingChangeNone, bandwidthRequested: 0, bandwidthDelta: 0 - bitrates[0][0], - availableLayers: nil, + availableLayers: availableLayers, bitrates: bitrates, targetLayers: InvalidLayers, distanceToDesired: 0, diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 403edebb4..572707858 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -157,6 +157,7 @@ func NewWebRTCReceiver( streamTrackerManager: NewStreamTrackerManager(logger, source), } w.streamTrackerManager.OnAvailableLayersChanged(w.downTrackLayerChange) + w.streamTrackerManager.OnBitrateAvailabilityChanged(w.downTrackBitrateAvailabilityChange) if runtime.GOMAXPROCS(0) < w.numProcs { w.numProcs = runtime.GOMAXPROCS(0) @@ -331,24 +332,20 @@ func (w *WebRTCReceiver) downTrackLayerChange(layers []int32) { } } -func (w *WebRTCReceiver) GetBitrateTemporalCumulative() Bitrates { - // LK-TODO: For SVC tracks, need to accumulate across spatial layers also - var br Bitrates - w.bufferMu.RLock() - defer w.bufferMu.RUnlock() - for i, buff := range w.buffers { - if buff != nil { - tls := make([]int64, DefaultMaxLayerTemporal+1) - if w.streamTrackerManager.HasSpatialLayer(int32(i)) { - tls = buff.BitrateTemporalCumulative() - } +func (w *WebRTCReceiver) downTrackBitrateAvailabilityChange() { + w.downTrackMu.RLock() + downTracks := w.downTracks + w.downTrackMu.RUnlock() - for j := 0; j < len(br[i]); j++ { - br[i][j] = tls[j] - } + for _, dt := range downTracks { + if dt != nil { + dt.UpTrackBitrateAvailabilityChange() } } - return br +} + +func (w *WebRTCReceiver) GetBitrateTemporalCumulative() Bitrates { + return w.streamTrackerManager.GetBitrateTemporalCumulative() } // OnCloseHandler method to be called on remote tracked removed @@ -498,8 +495,8 @@ func (w *WebRTCReceiver) forwardRTP(layer int32) { return } - if tracker != nil { - tracker.Observe(pkt.Packet.SequenceNumber) + if tracker != nil && len(pkt.Packet.Payload) > 0 { + tracker.Observe(pkt.Packet.SequenceNumber, pkt.TemporalLayer, len(pkt.RawPacket)) } w.downTrackMu.RLock() diff --git a/pkg/sfu/streamallocator.go b/pkg/sfu/streamallocator.go index 1a1df527e..90486d743 100644 --- a/pkg/sfu/streamallocator.go +++ b/pkg/sfu/streamallocator.go @@ -72,6 +72,7 @@ const ( SignalEstimate SignalTargetBitrate SignalAvailableLayersChange + SignalBitrateAvailabilityChange SignalSubscriptionChange SignalSubscribedLayersChange SignalPeriodicPing @@ -91,6 +92,10 @@ func (s Signal) String() string { return "ESTIMATE" case SignalTargetBitrate: return "TARGET_BITRATE" + case SignalAvailableLayersChange: + return "AVAILABLE_LAYERS_CHANGE" + case SignalBitrateAvailabilityChange: + return "BITRATE_AVAILABILITY_CHANGE" case SignalSubscriptionChange: return "SUBSCRIPTION_CHANGE" case SignalSubscribedLayersChange: @@ -220,6 +225,7 @@ func (s *StreamAllocator) AddTrack(downTrack *DownTrack, params AddTrackParams) downTrack.OnREMB(s.onREMB) downTrack.OnTransportCCFeedback(s.onTransportCCFeedback) downTrack.OnAvailableLayersChanged(s.onAvailableLayersChanged) + downTrack.OnBitrateAvailabilityChanged(s.onBitrateAvailabilityChanged) downTrack.OnSubscriptionChanged(s.onSubscriptionChanged) downTrack.OnSubscribedLayersChanged(s.onSubscribedLayersChanged) downTrack.OnPacketSentUnsafe(s.onPacketSent) @@ -280,6 +286,14 @@ func (s *StreamAllocator) onAvailableLayersChanged(downTrack *DownTrack) { }) } +// called when feeding track's bitrate measurement of any layer is available +func (s *StreamAllocator) onBitrateAvailabilityChanged(downTrack *DownTrack) { + s.postEvent(Event{ + Signal: SignalBitrateAvailabilityChange, + DownTrack: downTrack, + }) +} + // called when subscription settings changes (muting/unmuting of track) func (s *StreamAllocator) onSubscriptionChanged(downTrack *DownTrack) { s.postEvent(Event{ @@ -364,6 +378,8 @@ func (s *StreamAllocator) handleEvent(event *Event) { s.handleSignalTargetBitrate(event) case SignalAvailableLayersChange: s.handleSignalAvailableLayersChange(event) + case SignalBitrateAvailabilityChange: + s.handleSignalBitrateAvailabilityChange(event) case SignalSubscriptionChange: s.handleSignalSubscriptionChange(event) case SignalSubscribedLayersChange: @@ -510,6 +526,15 @@ func (s *StreamAllocator) handleSignalAvailableLayersChange(event *Event) { s.allocateTrack(track) } +func (s *StreamAllocator) handleSignalBitrateAvailabilityChange(event *Event) { + track, ok := s.videoTracks[livekit.TrackID(event.DownTrack.ID())] + if !ok { + return + } + + s.allocateTrack(track) +} + func (s *StreamAllocator) handleSignalSubscriptionChange(event *Event) { track, ok := s.videoTracks[livekit.TrackID(event.DownTrack.ID())] if !ok { @@ -537,9 +562,6 @@ func (s *StreamAllocator) handleSignalPeriodicPing(event *Event) { s.finalizeProbe() } - // catch up on all optimistically streamed tracks - s.finalizeTracks() - // probe if necessary and timing is right if s.state == StateDeficient { s.maybeProbe() @@ -704,7 +726,7 @@ func (s *StreamAllocator) allocateTrack(track *Track) { // if not deficient, free pass allocate track if !s.params.Config.Enabled || s.state == StateStable || !track.IsManaged() { update := NewStreamStateUpdate() - allocation := track.Allocate(ChannelCapacityInfinity, s.params.Config.AllowPause) + allocation := track.AllocateOptimal() update.HandleStreamingChange(allocation.change, track) s.maybeSendUpdate(update) return @@ -885,7 +907,7 @@ func (s *StreamAllocator) allocateAllTracks() { continue } - allocation := track.Allocate(ChannelCapacityInfinity, s.params.Config.AllowPause) + allocation := track.AllocateOptimal() update.HandleStreamingChange(allocation.change, track) // LK-TODO: optimistic allocation before bitrate is available will return 0. How to account for that? @@ -953,14 +975,6 @@ func (s *StreamAllocator) maybeSendUpdate(update *StreamStateUpdate) { } } -func (s *StreamAllocator) finalizeTracks() { - for _, t := range s.videoTracks { - t.FinalizeAllocate() - } - - s.adjustState() -} - func (s *StreamAllocator) getExpectedBandwidthUsage() int64 { expected := int64(0) for _, track := range s.videoTracks { @@ -1277,8 +1291,8 @@ func (t *Track) WritePaddingRTP(bytesToSend int) int { return t.downTrack.WritePaddingRTP(bytesToSend) } -func (t *Track) Allocate(availableChannelCapacity int64, allowPause bool) VideoAllocation { - return t.downTrack.Allocate(availableChannelCapacity, allowPause) +func (t *Track) AllocateOptimal() VideoAllocation { + return t.downTrack.AllocateOptimal() } func (t *Track) ProvisionalAllocatePrepare() { @@ -1309,10 +1323,6 @@ func (t *Track) GetNextHigherTransition() (VideoTransition, bool) { return t.downTrack.GetNextHigherTransition() } -func (t *Track) FinalizeAllocate() { - t.downTrack.FinalizeAllocate() -} - func (t *Track) Pause() VideoAllocation { return t.downTrack.Pause() } diff --git a/pkg/sfu/streamtracker.go b/pkg/sfu/streamtracker.go index 4b5d48300..9990ae63b 100644 --- a/pkg/sfu/streamtracker.go +++ b/pkg/sfu/streamtracker.go @@ -27,6 +27,10 @@ const ( StreamStatusActive StreamStatus = 1 ) +const ( + bitrateReportInterval = 1 * time.Second +) + type StreamTrackerParams struct { // number of samples needed per cycle SamplesRequired uint32 @@ -44,17 +48,18 @@ type StreamTrackerParams struct { type StreamTracker struct { params StreamTrackerParams - onStatusChanged func(status StreamStatus) + onStatusChanged func(status StreamStatus) + onBitrateAvailable func() - paused atomic.Bool - countSinceLast atomic.Uint32 // number of packets received since last check + lock sync.RWMutex + + paused bool + countSinceLast uint32 // number of packets received since last check generation atomic.Uint32 - initMu sync.Mutex initialized bool - statusMu sync.RWMutex - status StreamStatus + status StreamStatus // only access within detectWorker cycleCount uint32 @@ -62,9 +67,13 @@ type StreamTracker struct { // only access by the same goroutine as Observe lastSN uint16 + lastBitrateReport time.Time + bytesForBitrate [4]int64 + bitrate [4]int64 + callbacksQueue *utils.OpsQueue - isStopped atomic.Bool + isStopped bool } func NewStreamTracker(params StreamTrackerParams) *StreamTracker { @@ -80,22 +89,28 @@ func (s *StreamTracker) OnStatusChanged(f func(status StreamStatus)) { s.onStatusChanged = f } +func (s *StreamTracker) OnBitrateAvailable(f func()) { + s.onBitrateAvailable = f +} + func (s *StreamTracker) Status() StreamStatus { - s.statusMu.RLock() - defer s.statusMu.RUnlock() + s.lock.RLock() + defer s.lock.RUnlock() return s.status } -func (s *StreamTracker) maybeSetStatus(status StreamStatus) { +func (s *StreamTracker) maybeSetStatus(status StreamStatus) (StreamStatus, bool) { changed := false - s.statusMu.Lock() if s.status != status { s.status = status changed = true } - s.statusMu.Unlock() + return status, changed +} + +func (s *StreamTracker) maybeNotifyStatus(status StreamStatus, changed bool) { if changed && s.onStatusChanged != nil { s.callbacksQueue.Enqueue(func() { s.onStatusChanged(status) @@ -103,16 +118,12 @@ func (s *StreamTracker) maybeSetStatus(status StreamStatus) { } } -func (s *StreamTracker) maybeSetActive() { - s.maybeSetStatus(StreamStatusActive) -} - -func (s *StreamTracker) maybeSetStopped() { - s.maybeSetStatus(StreamStatusStopped) -} - func (s *StreamTracker) init() { - s.maybeSetActive() + s.lock.Lock() + status, changed := s.maybeSetStatus(StreamStatusActive) + s.lock.Unlock() + + s.maybeNotifyStatus(status, changed) go s.detectWorker(s.generation.Load()) } @@ -122,9 +133,13 @@ func (s *StreamTracker) Start() { } func (s *StreamTracker) Stop() { - if s.isStopped.Swap(true) { + s.lock.Lock() + defer s.lock.Unlock() + + if s.isStopped { return } + s.isStopped = true s.callbacksQueue.Stop() @@ -133,90 +148,175 @@ func (s *StreamTracker) Stop() { } func (s *StreamTracker) Reset() { - if s.isStopped.Load() { + s.lock.Lock() + defer s.lock.Unlock() + + if s.isStopped { return } + s.resetLocked() +} + +func (s *StreamTracker) resetLocked() { // bump generation to trigger exit of current worker s.generation.Inc() - s.countSinceLast.Store(0) + s.countSinceLast = 0 s.cycleCount = 0 - s.initMu.Lock() s.initialized = false - s.initMu.Unlock() - s.statusMu.Lock() s.status = StreamStatusStopped - s.statusMu.Unlock() + + for i := 0; i < len(s.bytesForBitrate); i++ { + s.bytesForBitrate[i] = 0 + } + for i := 0; i < len(s.bitrate); i++ { + s.bitrate[i] = 0 + } } func (s *StreamTracker) SetPaused(paused bool) { - s.paused.Store(paused) + s.lock.Lock() + defer s.lock.Unlock() + + s.paused = paused + + if !paused { + s.resetLocked() + } } // Observe a packet that's received -func (s *StreamTracker) Observe(sn uint16) { - if s.paused.Load() { +func (s *StreamTracker) Observe(sn uint16, temporalLayer int32, pktSize int) { + s.lock.Lock() + defer s.lock.Unlock() + + if s.isStopped || s.paused { return } - s.initMu.Lock() if !s.initialized { // first packet s.initialized = true - s.initMu.Unlock() s.lastSN = sn - s.countSinceLast.Inc() + s.countSinceLast = 1 + + s.lastBitrateReport = time.Now() + if temporalLayer >= 0 { + s.bytesForBitrate[temporalLayer] += int64(pktSize) + } // declare stream active and start the detection worker go s.init() return } - s.initMu.Unlock() // ignore out-of-order SNs if (sn - s.lastSN) > uint16(1<<15) { return } s.lastSN = sn - s.countSinceLast.Inc() + s.countSinceLast++ + + if temporalLayer >= 0 { + s.bytesForBitrate[temporalLayer] += int64(pktSize) + } +} + +// BitrateTemporalCumulative returns the current stream bitrate temporal layer accumulated with lower temporal layers. +func (s *StreamTracker) BitrateTemporalCumulative() [4]int64 { + s.lock.RLock() + defer s.lock.RUnlock() + + // copy and process + var brs [4]int64 + for i := 0; i < len(s.bitrate); i++ { + brs[i] = s.bitrate[i] + } + + for i := len(brs) - 1; i >= 1; i-- { + if brs[i] != 0 { + for j := i - 1; j >= 0; j-- { + brs[i] += brs[j] + } + } + } + + return brs } func (s *StreamTracker) detectWorker(generation uint32) { ticker := time.NewTicker(s.params.CycleDuration) + tickerBitrate := time.NewTicker(bitrateReportInterval / 2) for { - <-ticker.C - if generation != s.generation.Load() { - return - } + select { + case <-ticker.C: + if generation != s.generation.Load() { + return + } + s.detectChanges() - s.detectChanges() + case <-tickerBitrate.C: + if generation != s.generation.Load() { + return + } + s.bitrateReport() + } } } func (s *StreamTracker) detectChanges() { - if s.paused.Load() { - return - } - - if s.countSinceLast.Load() >= s.params.SamplesRequired { - s.cycleCount += 1 + s.lock.Lock() + if s.countSinceLast >= s.params.SamplesRequired { + s.cycleCount++ } else { s.cycleCount = 0 } + status := s.status + changed := false if s.cycleCount == 0 { // flip to stopped - s.maybeSetStopped() + status, changed = s.maybeSetStatus(StreamStatusStopped) } else if s.cycleCount >= s.params.CyclesRequired { // flip to active - s.maybeSetActive() + status, changed = s.maybeSetStatus(StreamStatusActive) } - s.countSinceLast.Store(0) + s.countSinceLast = 0 + s.lock.Unlock() + + s.maybeNotifyStatus(status, changed) +} + +func (s *StreamTracker) bitrateReport() { + // run this even if paused to drain out bitrate if there are no packets coming in + s.lock.Lock() + if time.Since(s.lastBitrateReport) < bitrateReportInterval { + s.lock.Unlock() + return + } + + now := time.Now() + diff := now.Sub(s.lastBitrateReport) + s.lastBitrateReport = now + + aggLast := int64(0) + aggNow := int64(0) + for i := 0; i < len(s.bytesForBitrate); i++ { + aggLast += s.bitrate[i] + s.bitrate[i] = int64(float64(s.bytesForBitrate[i]*8) / diff.Seconds()) + aggNow += s.bitrate[i] + s.bytesForBitrate[i] = 0 + } + s.lock.Unlock() + + if aggLast == 0 && aggNow > 0 && s.onBitrateAvailable != nil { + s.onBitrateAvailable() + } } diff --git a/pkg/sfu/streamtracker_test.go b/pkg/sfu/streamtracker_test.go index 6ba5c5d83..6b707ac64 100644 --- a/pkg/sfu/streamtracker_test.go +++ b/pkg/sfu/streamtracker_test.go @@ -32,7 +32,7 @@ func TestStreamTracker(t *testing.T) { require.Equal(t, StreamStatusStopped, tracker.Status()) // observe first packet - tracker.Observe(1) + tracker.Observe(1, 0, 0) testutils.WithTimeout(t, func() string { if callbackCalled.Load() { @@ -53,7 +53,7 @@ func TestStreamTracker(t *testing.T) { tracker.Start() require.Equal(t, StreamStatusStopped, tracker.Status()) - tracker.Observe(1) + tracker.Observe(1, 0, 0) testutils.WithTimeout(t, func() string { if tracker.Status() == StreamStatusActive { return "" @@ -89,7 +89,7 @@ func TestStreamTracker(t *testing.T) { tracker.Start() require.Equal(t, StreamStatusStopped, tracker.Status()) - tracker.Observe(1) + tracker.Observe(1, 0, 0) testutils.WithTimeout(t, func() string { if tracker.Status() == StreamStatusActive { return "" @@ -98,23 +98,23 @@ func TestStreamTracker(t *testing.T) { } }) - tracker.maybeSetStopped() + tracker.maybeSetStatus(StreamStatusStopped) - tracker.Observe(2) + tracker.Observe(2, 0, 0) tracker.detectChanges() require.Equal(t, StreamStatusStopped, tracker.Status()) - tracker.Observe(3) + tracker.Observe(3, 0, 0) tracker.detectChanges() require.Equal(t, StreamStatusActive, tracker.Status()) tracker.Stop() }) - t.Run("does not change to inactive when paused", func(t *testing.T) { + t.Run("changes to inactive when paused", func(t *testing.T) { tracker := newStreamTracker(5, 60, 500*time.Millisecond) tracker.Start() - tracker.Observe(1) + tracker.Observe(1, 0, 0) testutils.WithTimeout(t, func() string { if tracker.Status() == StreamStatusActive { return "" @@ -125,7 +125,7 @@ func TestStreamTracker(t *testing.T) { tracker.SetPaused(true) tracker.detectChanges() - require.Equal(t, StreamStatusActive, tracker.Status()) + require.Equal(t, StreamStatusStopped, tracker.Status()) tracker.Stop() }) @@ -140,7 +140,7 @@ func TestStreamTracker(t *testing.T) { require.Equal(t, StreamStatusStopped, tracker.Status()) // observe first packet - tracker.Observe(1) + tracker.Observe(1, 0, 0) testutils.WithTimeout(t, func() string { if callbackCalled.Load() == 1 { @@ -154,10 +154,10 @@ func TestStreamTracker(t *testing.T) { require.Equal(t, uint32(1), callbackCalled.Load()) // observe a few more - tracker.Observe(2) - tracker.Observe(3) - tracker.Observe(4) - tracker.Observe(5) + tracker.Observe(2, 0, 0) + tracker.Observe(3, 0, 0) + tracker.Observe(4, 0, 0) + tracker.Observe(5, 0, 0) tracker.detectChanges() // should still be active @@ -168,7 +168,7 @@ func TestStreamTracker(t *testing.T) { require.Equal(t, StreamStatusStopped, tracker.Status()) // first packet after reset - tracker.Observe(1) + tracker.Observe(1, 0, 0) testutils.WithTimeout(t, func() string { if callbackCalled.Load() == 2 { diff --git a/pkg/sfu/streamtrackermanager.go b/pkg/sfu/streamtrackermanager.go index 58f91d761..695959389 100644 --- a/pkg/sfu/streamtrackermanager.go +++ b/pkg/sfu/streamtrackermanager.go @@ -63,7 +63,8 @@ type StreamTrackerManager struct { availableLayers []int32 maxExpectedLayer int32 - onAvailableLayersChanged func(availableLayers []int32) + onAvailableLayersChanged func(availableLayers []int32) + onBitrateAvailabilityChanged func() } func NewStreamTrackerManager(logger logger.Logger, source livekit.TrackSource) *StreamTrackerManager { @@ -78,6 +79,10 @@ func (s *StreamTrackerManager) OnAvailableLayersChanged(f func(availableLayers [ s.onAvailableLayersChanged = f } +func (s *StreamTrackerManager) OnBitrateAvailabilityChanged(f func()) { + s.onBitrateAvailabilityChanged = f +} + func (s *StreamTrackerManager) AddTracker(layer int32) { var params StreamTrackerParams if s.source == livekit.TrackSource_SCREEN_SHARE { @@ -113,6 +118,11 @@ func (s *StreamTrackerManager) AddTracker(layer int32) { s.addAvailableLayer(layer) } }) + tracker.OnBitrateAvailable(func() { + if s.onBitrateAvailabilityChanged != nil { + s.onBitrateAvailabilityChanged() + } + }) s.lock.Lock() s.trackers[layer] = tracker @@ -155,9 +165,9 @@ func (s *StreamTrackerManager) GetTracker(layer int32) *StreamTracker { } func (s *StreamTrackerManager) SetPaused(paused bool) { - s.lock.Lock() + s.lock.RLock() trackers := s.trackers - s.lock.Unlock() + s.lock.RUnlock() for _, tracker := range trackers { if tracker != nil { @@ -211,6 +221,29 @@ func (s *StreamTrackerManager) IsReducedQuality() bool { return int32(len(s.availableLayers)) < (s.maxExpectedLayer + 1) } +func (s *StreamTrackerManager) GetBitrateTemporalCumulative() Bitrates { + s.lock.RLock() + defer s.lock.RUnlock() + + // LK-TODO: For SVC tracks, need to accumulate across spatial layers also + var br Bitrates + + for i, tracker := range s.trackers { + if tracker != nil { + var tls [DefaultMaxLayerTemporal + 1]int64 + if s.hasSpatialLayerLocked(int32(i)) { + tls = tracker.BitrateTemporalCumulative() + } + + for j := 0; j < len(br[i]); j++ { + br[i][j] = tls[j] + } + } + } + + return br +} + func (s *StreamTrackerManager) GetAvailableLayers() []int32 { s.lock.RLock() defer s.lock.RUnlock() @@ -254,7 +287,7 @@ func (s *StreamTrackerManager) addAvailableLayer(layer int32) { layers := s.availableLayers s.lock.Unlock() - s.logger.Debugw("available layers changed", "layers", layers) + s.logger.Debugw("available layers changed - layer seen", "layers", layers) if s.onAvailableLayersChanged != nil { s.onAvailableLayersChanged(layers) @@ -273,7 +306,7 @@ func (s *StreamTrackerManager) removeAvailableLayer(layer int32) { s.availableLayers = newLayers s.lock.Unlock() - s.logger.Debugw("available layers changed", "layers", newLayers) + s.logger.Debugw("available layers changed - layer gone", "layers", newLayers) // need to immediately switch off unavailable layers if s.onAvailableLayersChanged != nil {